forked from saltstack/salt
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Better testing of ssl opts and ws transport
- Loading branch information
Showing
5 changed files
with
83 additions
and
48 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2,6 +2,7 @@ | |
| :codeauthor: Thomas Jackson <[email protected]> | ||
| """ | ||
|
|
||
| import asyncio | ||
| import hashlib | ||
| import logging | ||
|
|
||
|
|
@@ -10,6 +11,7 @@ | |
|
|
||
| import salt.crypt | ||
| import salt.transport.tcp | ||
| import salt.transport.ws | ||
| import salt.transport.zeromq | ||
| import salt.utils.stringutils | ||
| from tests.support.mock import MagicMock, patch | ||
|
|
@@ -26,7 +28,7 @@ def transport_ids(value): | |
| return "Transport({})".format(value) | ||
|
|
||
|
|
||
| @pytest.fixture(params=("zeromq", "tcp"), ids=transport_ids) | ||
| @pytest.fixture(params=("zeromq", "tcp", "ws"), ids=transport_ids) | ||
| def transport(request): | ||
| return request.param | ||
|
|
||
|
|
@@ -168,31 +170,28 @@ async def test_publish_client_connect_server_down(transport, io_loop): | |
| await client.connect() | ||
| assert client._socket | ||
| elif transport == "tcp": | ||
| client = salt.transport.tcp.TCPPubClient(opts, io_loop, host=host, port=port) | ||
| try: | ||
| # XXX: This is an implimentation detail of the tcp transport. | ||
| # await client.connect(port) | ||
| io_loop.spawn_callback(client.connect) | ||
| except TimeoutError: | ||
| pass | ||
| except Exception: # pylint: disable=broad-except | ||
| log.error("Got exception", exc_info=True) | ||
| client = salt.transport.tcp.PublishClient(opts, io_loop, host=host, port=port) | ||
| io_loop.spawn_callback(client.connect) | ||
| assert client._stream is None | ||
| elif transport == "ws": | ||
| client = salt.transport.ws.PublishClient(opts, io_loop, host=host, port=port) | ||
| io_loop.spawn_callback(client.connect) | ||
| assert client._ws is None | ||
| assert client._session is None | ||
| client.close() | ||
| await asyncio.sleep(0.03) | ||
|
|
||
|
|
||
| async def test_publish_client_connect_server_comes_up(transport, io_loop): | ||
| opts = {"master_ip": "127.0.0.1"} | ||
| host = "127.0.0.1" | ||
| port = 11122 | ||
| msg = salt.payload.dumps({"meh": 123}) | ||
| if transport == "zeromq": | ||
| import asyncio | ||
|
|
||
| import zmq | ||
|
|
||
| ctx = zmq.asyncio.Context() | ||
| uri = f"tcp://{opts['master_ip']}:{port}" | ||
| msg = salt.payload.dumps({"meh": 123}) | ||
| log.debug("TEST - Senging %r", msg) | ||
| client = salt.transport.zeromq.PublishClient( | ||
| opts, io_loop, host=host, port=port | ||
|
|
@@ -213,6 +212,7 @@ async def recv(): | |
| task = asyncio.create_task(recv()) | ||
| # Sleep to allow zmq to do it's thing. | ||
| await socket.send(msg) | ||
| await asyncio.sleep(0.03) | ||
| await task | ||
| response = task.result() | ||
| assert response | ||
|
|
@@ -221,10 +221,9 @@ async def recv(): | |
| await asyncio.sleep(0.03) | ||
| ctx.term() | ||
| elif transport == "tcp": | ||
| import asyncio | ||
| import socket | ||
|
|
||
| client = salt.transport.tcp.TCPPubClient(opts, io_loop, host=host, port=port) | ||
| client = salt.transport.tcp.PublishClient(opts, io_loop, host=host, port=port) | ||
| # XXX: This is an implimentation detail of the tcp transport. | ||
| # await client.connect(port) | ||
| io_loop.spawn_callback(client.connect) | ||
|
|
@@ -238,11 +237,45 @@ async def recv(): | |
| sock.listen(128) | ||
| await asyncio.sleep(0.03) | ||
|
|
||
| msg = salt.payload.dumps({"meh": 123}) | ||
| msg = salt.transport.frame.frame_msg(msg, header=None) | ||
| data = salt.transport.frame.frame_msg(msg, header=None) | ||
| conn, addr = sock.accept() | ||
| conn.send(msg) | ||
| conn.send(data) | ||
| response = await client.recv() | ||
| assert response | ||
| assert msg == response | ||
| elif transport == "ws": | ||
| import socket | ||
|
|
||
| import aiohttp | ||
|
|
||
| client = salt.transport.ws.PublishClient(opts, io_loop, host=host, port=port) | ||
| io_loop.spawn_callback(client.connect) | ||
| assert client._ws is None | ||
| assert client._session is None | ||
| await asyncio.sleep(2) | ||
|
|
||
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | ||
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) | ||
| sock.setblocking(0) | ||
| sock.bind((opts["master_ip"], port)) | ||
| sock.listen(128) | ||
|
|
||
| async def handler(request): | ||
| ws = aiohttp.web.WebSocketResponse() | ||
| await ws.prepare(request) | ||
| data = salt.transport.frame.frame_msg(msg, header=None) | ||
| await ws.send_bytes(data) | ||
|
|
||
| server = aiohttp.web.Server(handler) | ||
| runner = aiohttp.web.ServerRunner(server) | ||
| await runner.setup() | ||
| site = aiohttp.web.SockSite(runner, sock) | ||
| await site.start() | ||
|
|
||
| await asyncio.sleep(0.03) | ||
|
|
||
| response = await client.recv() | ||
| assert msg == response | ||
| else: | ||
| raise Exception(f"Unknown transport {transport}") | ||
| client.close() | ||
| await asyncio.sleep(0.03) | ||