await self.ws.close()
async def recv(self, request_id):
+ if not self.is_open:
+ raise websockets.exceptions.ConnectionClosed(0, 'websocket closed')
return await self.messages.get(request_id)
async def receiver(self):
while self.is_open:
- result = await self.ws.recv()
- if result is not None:
- result = json.loads(result)
- await self.messages.put(result['request-id'], result)
+ try:
+ result = await self.ws.recv()
+ if result is not None:
+ result = json.loads(result)
+ await self.messages.put(result['request-id'], result)
+ except Exception as e:
+ await self.messages.put_all(e)
+ raise
+ await self.messages.put_all(websockets.exceptions.ConnectionClosed(
+ 0, 'websocket closed'))
async def rpc(self, msg, encoder=None):
self.__request_id__ += 1
# API Error Response
raise JujuAPIError(result)
- if not 'response' in result:
+ if 'response' not in result:
# This may never happen
return result
async def get(self, id):
value = await self._queues[id].get()
del self._queues[id]
+ if isinstance(value, Exception):
+ raise value
return value
async def put(self, id, value):
await self._queues[id].put(value)
+
+ async def put_all(self, value):
+ for queue in self._queues.values():
+ await queue.put(value)
import pytest
from collections import deque
+from websockets.exceptions import ConnectionClosed
+
from .. import base
from juju.client.connection import Connection
pass
async def recv(self):
+ if not self.responses:
+ raise ConnectionClosed(0, 'no reason')
return json.dumps(self.responses.popleft())
async def close(self):