3 from collections
import deque
6 from juju
.client
.connection
import Connection
7 from websockets
.exceptions
import ConnectionClosed
15 def __init__(self
, responses
):
17 self
.responses
= deque(responses
)
20 async def send(self
, message
):
24 if not self
.responses
:
25 await asyncio
.sleep(1) # delay to give test time to finish
26 raise ConnectionClosed(0, 'ran out of responses')
27 return json
.dumps(self
.responses
.popleft())
29 async def close(self
):
34 async def test_out_of_order(event_loop
):
40 expected_responses
= [
45 minimal_facades
= [{'name': 'Pinger', 'versions': [1]}]
49 mock
.patch('websockets.connect', base
.AsyncMock(return_value
=ws
)), \
51 'juju.client.connection.Connection.login',
52 base
.AsyncMock(return_value
={'response': {
53 'facades': minimal_facades
,
56 mock
.patch('juju.client.connection.Connection._get_ssl'), \
57 mock
.patch('juju.client.connection.Connection._pinger', base
.AsyncMock()):
58 con
= await Connection
.connect('0.1.2.3:999')
61 actual_responses
.append(await con
.rpc({'version': 1}))
62 assert actual_responses
== expected_responses