import asyncio
-import pytest
-from juju.client.connection import Connection
from juju.client import client
-from .. import base
-
+from juju.client.connection import Connection
-@base.bootstrapped
-@pytest.mark.asyncio
-async def test_connect_current(event_loop):
- async with base.CleanModel():
- conn = await Connection.connect_current()
+import pytest
- assert isinstance(conn, Connection)
- await conn.close()
+from .. import base
@base.bootstrapped
@pytest.mark.asyncio
async def test_monitor(event_loop):
- async with base.CleanModel():
- conn = await Connection.connect_current()
-
+ async with base.CleanModel() as model:
+ conn = model.connection()
assert conn.monitor.status == 'connected'
await conn.close()
@pytest.mark.asyncio
async def test_monitor_catches_error(event_loop):
- async with base.CleanModel():
- conn = await Connection.connect_current()
+ async with base.CleanModel() as model:
+ conn = model.connection()
assert conn.monitor.status == 'connected'
- await conn.ws.close()
-
- assert conn.monitor.status == 'error'
-
- await conn.close()
+ try:
+ async with conn.monitor.reconnecting:
+ await conn.ws.close()
+ await asyncio.sleep(1)
+ assert conn.monitor.status == 'error'
+ finally:
+ await conn.close()
@base.bootstrapped
channel='stable',
)
- c = client.ClientFacade.from_connection(model.connection)
+ c = client.ClientFacade.from_connection(model.connection())
await c.FullStatus(None)
@pytest.mark.asyncio
async def test_reconnect(event_loop):
async with base.CleanModel() as model:
- conn = await Connection.connect(
- model.connection.endpoint,
- model.connection.uuid,
- model.connection.username,
- model.connection.password,
- model.connection.cacert,
- model.connection.macaroons,
- model.connection.loop,
- model.connection.max_frame_size)
+ kwargs = model.connection().connect_params()
+ conn = await Connection.connect(**kwargs)
try:
await asyncio.sleep(0.1)
assert conn.is_open