tests/unit/test_connection.py

   1 import json
   2 import mock
   3 import pytest
   4 from collections import deque
   5
   6 from .. import base
   7 from juju.client.connection import Connection
   8
   9
  10 class WebsocketMock:
  11     def __init__(self, responses):
  12         super().__init__()
  13         self.responses = deque(responses)
  14         self.open = True
  15
  16     async def send(self, message):
  17         pass
  18
  19     async def recv(self):
  20         return json.dumps(self.responses.popleft())
  21
  22     async def close(self):
  23         self.open = False
  24
  25
  26 @pytest.mark.asyncio
  27 async def test_out_of_order(event_loop):
  28     con = Connection(*[None]*4)
  29     ws = WebsocketMock([
  30         {'request-id': 1},
  31         {'request-id': 3},
  32         {'request-id': 2},
  33     ])
  34     expected_responses = [
  35         {'request-id': 1},
  36         {'request-id': 2},
  37         {'request-id': 3},
  38     ]
  39     con._get_sll = mock.MagicMock()
  40     with mock.patch('websockets.connect', base.AsyncMock(return_value=ws)):
  41         await con.open()
  42     actual_responses = []
  43     for i in range(3):
  44         actual_responses.append(await con.rpc({'version': 1}))
  45     assert actual_responses == expected_responses