tests/integration/test_connection.py

   1 import asyncio
   2 import pytest
   3
   4 from juju.client.connection import Connection
   5 from juju.client import client
   6 from .. import base
   7
   8
   9 @base.bootstrapped
  10 @pytest.mark.asyncio
  11 async def test_connect_current(event_loop):
  12     async with base.CleanModel():
  13         conn = await Connection.connect_current()
  14
  15         assert isinstance(conn, Connection)
  16         await conn.close()
  17
  18
  19 @base.bootstrapped
  20 @pytest.mark.asyncio
  21 async def test_monitor(event_loop):
  22
  23     async with base.CleanModel():
  24         conn = await Connection.connect_current()
  25
  26         assert conn.monitor.status == 'connected'
  27         await conn.close()
  28
  29         assert conn.monitor.status == 'disconnected'
  30
  31
  32 @base.bootstrapped
  33 @pytest.mark.asyncio
  34 async def test_monitor_catches_error(event_loop):
  35
  36     async with base.CleanModel():
  37         conn = await Connection.connect_current()
  38
  39         assert conn.monitor.status == 'connected'
  40         await conn.ws.close()
  41
  42         assert conn.monitor.status == 'error'
  43
  44         await conn.close()
  45
  46
  47 @base.bootstrapped
  48 @pytest.mark.asyncio
  49 async def test_full_status(event_loop):
  50     async with base.CleanModel() as model:
  51         await model.deploy(
  52             'ubuntu-0',
  53             application_name='ubuntu',
  54             series='trusty',
  55             channel='stable',
  56         )
  57
  58         c = client.ClientFacade.from_connection(model.connection)
  59
  60         await c.FullStatus(None)
  61
  62
  63 @base.bootstrapped
  64 @pytest.mark.asyncio
  65 async def test_reconnect(event_loop):
  66     async with base.CleanModel() as model:
  67         conn = await Connection.connect(
  68             model.connection.endpoint,
  69             model.connection.uuid,
  70             model.connection.username,
  71             model.connection.password,
  72             model.connection.cacert,
  73             model.connection.macaroons,
  74             model.connection.loop,
  75             model.connection.max_frame_size)
  76         try:
  77             await asyncio.sleep(0.1)
  78             assert conn.is_open
  79             await conn.ws.close()
  80             assert not conn.is_open
  81             await model.block_until(lambda: conn.is_open, timeout=3)
  82         finally:
  83             await conn.close()