tests/integration/test_connection.py

   1 import pytest
   2
   3 from juju.client.connection import Connection
   4 from .. import base
   5
   6
   7 @base.bootstrapped
   8 @pytest.mark.asyncio
   9 async def test_connect_current(event_loop):
  10     async with base.CleanModel():
  11         conn = await Connection.connect_current()
  12
  13         assert isinstance(conn, Connection)
  14         await conn.close()
  15
  16
  17 @base.bootstrapped
  18 @pytest.mark.asyncio
  19 async def test_monitor(event_loop):
  20
  21     async with base.CleanModel():
  22         conn = await Connection.connect_current()
  23
  24         assert conn.monitor.status == 'connected'
  25         await conn.close()
  26
  27         assert conn.monitor.status == 'disconnected'
  28
  29
  30 @base.bootstrapped
  31 @pytest.mark.asyncio
  32 async def test_monitor_catches_error(event_loop):
  33
  34     async with base.CleanModel():
  35         conn = await Connection.connect_current()
  36
  37         assert conn.monitor.status == 'connected'
  38         await conn.ws.close()
  39
  40         assert conn.monitor.status == 'error'
  41
  42         await conn.close()