X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=tests%2Fintegration%2Ftest_model.py;h=92d205529ed5f668ee0ca393348002d2edd790a2;hb=refs%2Ftags%2F0.4.0;hp=4d45a1c00cf7e2e887677a14ec069fb001af5efc;hpb=77e2f1321fa65dab4fa63bdc7262273810fa0f5e;p=osm%2FN2VC.git diff --git a/tests/integration/test_model.py b/tests/integration/test_model.py index 4d45a1c..92d2055 100644 --- a/tests/integration/test_model.py +++ b/tests/integration/test_model.py @@ -1,5 +1,6 @@ import asyncio from concurrent.futures import ThreadPoolExecutor +from pathlib import Path import pytest from .. import base @@ -7,7 +8,7 @@ from juju.model import Model MB = 1 GB = 1024 - +SSH_KEY = 'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCsYMJGNGG74HAJha3n2CFmWYsOOaORnJK6VqNy86pj0MIpvRXBzFzVy09uPQ66GOQhTEoJHEqE77VMui7+62AcMXT+GG7cFHcnU8XVQsGM6UirCcNyWNysfiEMoAdZScJf/GvoY87tMEszhZIUV37z8PUBx6twIqMdr31W1J0IaPa+sV6FEDadeLaNTvancDcHK1zuKsL39jzAg7+LYjKJfEfrsQP+lj/EQcjtKqlhVS5kzsJVfx8ZEd0xhW5G7N6bCdKNalS8mKCMaBXJpijNQ82AiyqCIDCRrre2To0/i7pTjRiL0U9f9mV3S4NJaQaokR050w/ZLySFf6F7joJT mathijs@Qrama-Mathijs' @base.bootstrapped @pytest.mark.asyncio @@ -135,3 +136,74 @@ async def test_explicit_loop_threaded(event_loop): 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_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'