modules/libjuju/tests/integration/test_connection.py

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