X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=tests%2Fintegration%2Ftest_model.py;h=85067864dc3b76d3786e6f423d59f569b37226fa;hb=17b26ef759a99c9010ea30e47205bfb332400e74;hp=2fe97d0bcd0852e8c1dedd04e3590218f62d315e;hpb=f5341c9db31e56527fcb0bb320abea86fe17a550;p=osm%2FN2VC.git diff --git a/tests/integration/test_model.py b/tests/integration/test_model.py index 2fe97d0..8506786 100644 --- a/tests/integration/test_model.py +++ b/tests/integration/test_model.py @@ -1,9 +1,15 @@ +import asyncio +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path import pytest from .. import base +from juju.model import Model +from juju.client.client import ConfigValue MB = 1 GB = 1024 +SSH_KEY = 'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCsYMJGNGG74HAJha3n2CFmWYsOOaORnJK6VqNy86pj0MIpvRXBzFzVy09uPQ66GOQhTEoJHEqE77VMui7+62AcMXT+GG7cFHcnU8XVQsGM6UirCcNyWNysfiEMoAdZScJf/GvoY87tMEszhZIUV37z8PUBx6twIqMdr31W1J0IaPa+sV6FEDadeLaNTvancDcHK1zuKsL39jzAg7+LYjKJfEfrsQP+lj/EQcjtKqlhVS5kzsJVfx8ZEd0xhW5G7N6bCdKNalS8mKCMaBXJpijNQ82AiyqCIDCRrre2To0/i7pTjRiL0U9f9mV3S4NJaQaokR050w/ZLySFf6F7joJT mathijs@Qrama-Mathijs' # noqa @base.bootstrapped @@ -30,6 +36,22 @@ async def test_deploy_bundle(event_loop): assert app in model.applications +@base.bootstrapped +@pytest.mark.asyncio +async def test_deploy_channels_revs(event_loop): + async with base.CleanModel() as model: + charm = 'cs:~johnsca/libjuju-test' + stable = await model.deploy(charm, 'a1') + edge = await model.deploy(charm, 'a2', channel='edge') + rev = await model.deploy(charm+'-2', 'a3') + + assert [a.charm_url for a in (stable, edge, rev)] == [ + 'cs:~johnsca/libjuju-test-1', + 'cs:~johnsca/libjuju-test-2', + 'cs:~johnsca/libjuju-test-2', + ] + + @base.bootstrapped @pytest.mark.asyncio async def test_add_machine(event_loop): @@ -95,3 +117,133 @@ async def test_relate(event_loop): ) assert isinstance(my_relation, Relation) + + +async def _deploy_in_loop(new_loop, model_name): + new_model = Model(new_loop) + await new_model.connect_model(model_name) + try: + await new_model.deploy('cs:xenial/ubuntu') + assert 'ubuntu' in new_model.applications + finally: + await new_model.disconnect() + + +@base.bootstrapped +@pytest.mark.asyncio +async def test_explicit_loop(event_loop): + async with base.CleanModel() as model: + model_name = model.info.name + new_loop = asyncio.new_event_loop() + new_loop.run_until_complete( + _deploy_in_loop(new_loop, model_name)) + await model._wait_for_new('application', 'ubuntu') + assert 'ubuntu' in model.applications + + +@base.bootstrapped +@pytest.mark.asyncio +async def test_explicit_loop_threaded(event_loop): + async with base.CleanModel() as model: + model_name = model.info.name + new_loop = asyncio.new_event_loop() + with ThreadPoolExecutor(1) as executor: + f = executor.submit( + new_loop.run_until_complete, + _deploy_in_loop(new_loop, model_name)) + f.result() + await model._wait_for_new('application', 'ubuntu') + assert 'ubuntu' in model.applications + + +@base.bootstrapped +@pytest.mark.asyncio +async def test_store_resources_charm(event_loop): + async with base.CleanModel() as model: + ghost = await model.deploy('cs:ghost-19') + assert 'ghost' in model.applications + terminal_statuses = ('active', 'error', 'blocked') + await model.block_until( + lambda: ( + len(ghost.units) > 0 and + ghost.units[0].workload_status in terminal_statuses) + ) + # ghost will go in to blocked (or error, for older + # charm revs) if the resource is missing + assert ghost.units[0].workload_status == 'active' + + +@base.bootstrapped +@pytest.mark.asyncio +async def test_store_resources_bundle(event_loop): + async with base.CleanModel() as model: + bundle = str(Path(__file__).parent / 'bundle') + await model.deploy(bundle) + assert 'ghost' in model.applications + ghost = model.applications['ghost'] + terminal_statuses = ('active', 'error', 'blocked') + await model.block_until( + lambda: ( + len(ghost.units) > 0 and + ghost.units[0].workload_status in terminal_statuses) + ) + # ghost will go in to blocked (or error, for older + # charm revs) if the resource is missing + assert ghost.units[0].workload_status == 'active' + + +@base.bootstrapped +@pytest.mark.asyncio +async def test_ssh_key(event_loop): + async with base.CleanModel() as model: + await model.add_ssh_key('admin', SSH_KEY) + result = await model.get_ssh_key(True) + result = result.serialize()['results'][0].serialize()['result'] + assert SSH_KEY in result + await model.remove_ssh_key('admin', SSH_KEY) + result = await model.get_ssh_key(True) + result = result.serialize()['results'][0].serialize()['result'] + assert result is None + + +@base.bootstrapped +@pytest.mark.asyncio +async def test_get_machines(event_loop): + async with base.CleanModel() as model: + result = await model.get_machines() + assert isinstance(result, list) + + +@base.bootstrapped +@pytest.mark.asyncio +async def test_watcher_reconnect(event_loop): + async with base.CleanModel() as model: + await model.connection.ws.close() + await asyncio.sleep(0.1) + assert model.connection.is_open + + +@base.bootstrapped +@pytest.mark.asyncio +async def test_config(event_loop): + async with base.CleanModel() as model: + await model.set_config({ + 'extra-info': 'booyah', + 'test-mode': ConfigValue(value=True), + }) + result = await model.get_config() + assert 'extra-info' in result + assert result['extra-info'].source == 'model' + assert result['extra-info'].value == 'booyah' + +# @base.bootstrapped +# @pytest.mark.asyncio +# async def test_grant(event_loop) +# async with base.CleanController() as controller: +# await controller.add_user('test-model-grant') +# await controller.grant('test-model-grant', 'superuser') +# async with base.CleanModel() as model: +# await model.grant('test-model-grant', 'admin') +# assert model.get_user('test-model-grant')['access'] == 'admin' +# await model.grant('test-model-grant', 'login') +# assert model.get_user('test-model-grant')['access'] == 'login'