diff --git a/awxkit/awxkit/ws.py b/awxkit/awxkit/ws.py index f6fde6211..3a0f357af 100644 --- a/awxkit/awxkit/ws.py +++ b/awxkit/awxkit/ws.py @@ -105,7 +105,7 @@ def __init__( self._add_received_time = add_received_time def connect(self): - wst = threading.Thread(target=self._ws_run_forever, args=(self.ws, {"cert_reqs": ssl.CERT_NONE})) + wst = threading.Thread(target=self._ws_run_forever, args=({"cert_reqs": ssl.CERT_NONE},)) wst.daemon = True wst.start() atexit.register(self.close) @@ -205,7 +205,7 @@ def unsubscribe(self, wait=True, timeout=10): else: self._send(json.dumps(dict(groups={}, xrftoken=self.csrftoken))) - def _on_message(self, message): + def _on_message(self, ws, message): message = json.loads(message) log.debug('received message: {}'.format(message)) if self._add_received_time: @@ -230,17 +230,17 @@ def _update_subscription(self, job_id): self.subscribe(**subscription) self._should_subscribe_to_pending_job = False - def _on_open(self): + def _on_open(self, ws): self._ws_connected_flag.set() - def _on_error(self, error): + def _on_error(self, ws, error): log.info('Error received: {}'.format(error)) - def _on_close(self): + def _on_close(self, ws, close_status_code, close_msg): log.info('Successfully closed ws.') self._ws_closed = True - def _ws_run_forever(self, sockopt=None, sslopt=None): + def _ws_run_forever(self, sslopt=None): self.ws.run_forever(sslopt=sslopt) log.debug('ws.run_forever finished') diff --git a/awxkit/setup.py b/awxkit/setup.py index 70ceec7e9..e8552b75a 100644 --- a/awxkit/setup.py +++ b/awxkit/setup.py @@ -93,7 +93,7 @@ def run(self): 'setuptools', ], python_requires=">=3.11", - extras_require={'formatting': ['jq'], 'websockets': ['websocket-client==0.57.0'], 'crypto': ['cryptography']}, + extras_require={'formatting': ['jq'], 'websockets': ['websocket-client>=1.0.0'], 'crypto': ['cryptography']}, license='Apache 2.0', classifiers=[ 'Development Status :: 5 - Production/Stable', diff --git a/awxkit/test/test_ws.py b/awxkit/test/test_ws.py index c2e69fbc5..9ad280c0a 100644 --- a/awxkit/test/test_ws.py +++ b/awxkit/test/test_ws.py @@ -39,3 +39,33 @@ def test_urlparsing(url, result): assert client.port == result.port assert client.hostname == result.hostname assert client._use_ssl == result.secure + + +def test_callbacks_follow_websocket_client_1x_convention(caplog): + """websocket-client 1.x always invokes WebSocketApp callbacks with the app + as the first argument; these calls mirror exactly how run_forever fires + them and fail if the signatures regress to the 0.x style.""" + client = WSClient("token", "hostname", 566, False) + app = client.ws + + client._on_open(app) + assert client._ws_connected_flag.is_set() + + client._on_message(app, '{"group_name": "jobs", "status": "successful"}') + assert client._recv(wait=True, timeout=1) == {"group_name": "jobs", "status": "successful"} + + import logging + + with caplog.at_level(logging.INFO, logger="awxkit.ws"): + client._on_error(app, Exception("boom")) + assert "boom" in caplog.text + + client._on_close(app, 1000, "normal closure") + assert client._ws_closed + + +def test_unsubscribe_ack_sets_event(): + client = WSClient("token", "hostname", 566, False) + client._on_message(client.ws, '{"groups_current": {}}') + assert client._pending_unsubscribe.is_set() +