modules/libjuju/tests/integration/test_connection.py

import asyncio
import pytest

from juju.client.connection import Connection
from juju.client import client
from .. import base


@base.bootstrapped
@pytest.mark.asyncio
async def test_connect_current(event_loop):
    async with base.CleanModel():
        conn = await Connection.connect_current()

        assert isinstance(conn, Connection)
        await conn.close()


@base.bootstrapped
@pytest.mark.asyncio
async def test_monitor(event_loop):

    async with base.CleanModel():
        conn = await Connection.connect_current()

        assert conn.monitor.status == 'connected'
        await conn.close()

        assert conn.monitor.status == 'disconnected'


@base.bootstrapped
@pytest.mark.asyncio
async def test_monitor_catches_error(event_loop):

    async with base.CleanModel():
        conn = await Connection.connect_current()

        assert conn.monitor.status == 'connected'
        await conn.ws.close()

        assert conn.monitor.status == 'error'

        await conn.close()


@base.bootstrapped
@pytest.mark.asyncio
async def test_full_status(event_loop):
    async with base.CleanModel() as model:
        await model.deploy(
            'ubuntu-0',
            application_name='ubuntu',
            series='trusty',
            channel='stable',
        )

        c = client.ClientFacade.from_connection(model.connection)

        await c.FullStatus(None)


@base.bootstrapped
@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)
        try:
            await asyncio.sleep(0.1)
            assert conn.is_open
            await conn.ws.close()
            assert not conn.is_open
            await model.block_until(lambda: conn.is_open, timeout=3)
        finally:
            await conn.close()