X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=modules%2Flibjuju%2Ftests%2Fintegration%2Ftest_controller.py;h=d5593134b35cae20dd2209f892bc6c402ac33607;hp=f3840cc4f6fd10c671753630cd8be185207fad26;hb=1a15d1c84fc826fa7996c1c9d221a324edd33432;hpb=fd577a36aaf408c845f3bb8b3f28ecbbf3332107 diff --git a/modules/libjuju/tests/integration/test_controller.py b/modules/libjuju/tests/integration/test_controller.py index f3840cc..d559313 100644 --- a/modules/libjuju/tests/integration/test_controller.py +++ b/modules/libjuju/tests/integration/test_controller.py @@ -1,3 +1,4 @@ +import asyncio import pytest import uuid @@ -8,13 +9,21 @@ from juju.errors import JujuAPIError @base.bootstrapped @pytest.mark.asyncio -async def test_add_user(event_loop): +async def test_add_remove_user(event_loop): async with base.CleanController() as controller: username = 'test{}'.format(uuid.uuid4()) - await controller.add_user(username) - result = await controller.get_user(username) - res_ser = result.serialize()['results'][0].serialize() - assert res_ser['result'] is not None + user = await controller.get_user(username) + assert user is None + user = await controller.add_user(username) + assert user is not None + assert user.username == username + users = await controller.get_users() + assert any(u.username == username for u in users) + await controller.remove_user(username) + user = await controller.get_user(username) + assert user is None + users = await controller.get_users() + assert not any(u.username == username for u in users) @base.bootstrapped @@ -22,15 +31,23 @@ async def test_add_user(event_loop): async def test_disable_enable_user(event_loop): async with base.CleanController() as controller: username = 'test-disable{}'.format(uuid.uuid4()) - await controller.add_user(username) - await controller.disable_user(username) - result = await controller.get_user(username) - res_ser = result.serialize()['results'][0].serialize() - assert res_ser['result'].serialize()['disabled'] is True - await controller.enable_user(username) - result = await controller.get_user(username) - res_ser = result.serialize()['results'][0].serialize() - assert res_ser['result'].serialize()['disabled'] is False + user = await controller.add_user(username) + + await user.disable() + assert not user.enabled + assert user.disabled + + fresh = await controller.get_user(username) # fetch fresh copy + assert not fresh.enabled + assert fresh.disabled + + await user.enable() + assert user.enabled + assert not user.disabled + + fresh = await controller.get_user(username) # fetch fresh copy + assert fresh.enabled + assert not fresh.disabled @base.bootstrapped @@ -38,40 +55,98 @@ async def test_disable_enable_user(event_loop): async def test_change_user_password(event_loop): async with base.CleanController() as controller: username = 'test-password{}'.format(uuid.uuid4()) - await controller.add_user(username) - await controller.change_user_password(username, 'password') + user = await controller.add_user(username) + await user.set_password('password') try: new_controller = Controller() await new_controller.connect( controller.connection.endpoint, username, 'password') - result = True - await new_controller.disconnect() except JujuAPIError: - result = False - assert result is True + raise AssertionError('Unable to connect with new password') + finally: + await new_controller.disconnect() @base.bootstrapped @pytest.mark.asyncio -async def test_grant(event_loop): +async def test_grant_revoke(event_loop): async with base.CleanController() as controller: username = 'test-grant{}'.format(uuid.uuid4()) - await controller.add_user(username) - await controller.grant(username, 'superuser') - result = await controller.get_user(username) - result = result.serialize()['results'][0].serialize()['result']\ - .serialize() - assert result['access'] == 'superuser' - await controller.grant(username, 'login') - result = await controller.get_user(username) - result = result.serialize()['results'][0].serialize()['result']\ - .serialize() - assert result['access'] == 'login' + user = await controller.add_user(username) + await user.grant('superuser') + assert user.access == 'superuser' + fresh = await controller.get_user(username) # fetch fresh copy + assert fresh.access == 'superuser' + await user.grant('login') + assert user.access == 'login' + fresh = await controller.get_user(username) # fetch fresh copy + assert fresh.access == 'login' + await user.revoke() + assert user.access is '' + fresh = await controller.get_user(username) # fetch fresh copy + assert fresh.access is '' + + +@base.bootstrapped +@pytest.mark.asyncio +async def test_list_models(event_loop): + async with base.CleanController() as controller: + async with base.CleanModel() as model: + result = await controller.list_models() + assert model.info.name in result + + +@base.bootstrapped +@pytest.mark.asyncio +async def test_get_model(event_loop): + async with base.CleanController() as controller: + by_name, by_uuid = None, None + model_name = 'test-{}'.format(uuid.uuid4()) + model = await controller.add_model(model_name) + model_uuid = model.info.uuid + await model.disconnect() + try: + by_name = await controller.get_model(model_name) + by_uuid = await controller.get_model(model_uuid) + assert by_name.info.name == model_name + assert by_name.info.uuid == model_uuid + assert by_uuid.info.name == model_name + assert by_uuid.info.uuid == model_uuid + finally: + if by_name: + await by_name.disconnect() + if by_uuid: + await by_uuid.disconnect() + await controller.destroy_model(model_name) + + +async def _wait_for_model_gone(controller, model_name): + while model_name in await controller.list_models(): + await asyncio.sleep(0.5, loop=controller.loop) + + +@base.bootstrapped +@pytest.mark.asyncio +async def test_destroy_model_by_name(event_loop): + async with base.CleanController() as controller: + model_name = 'test-{}'.format(uuid.uuid4()) + model = await controller.add_model(model_name) + await model.disconnect() + await controller.destroy_model(model_name) + await asyncio.wait_for(_wait_for_model_gone(controller, + model_name), + timeout=60) @base.bootstrapped @pytest.mark.asyncio -async def test_get_models(event_loop): +async def test_add_destroy_model_by_uuid(event_loop): async with base.CleanController() as controller: - result = await controller.get_models() - assert isinstance(result.serialize()['user-models'], list) + model_name = 'test-{}'.format(uuid.uuid4()) + model = await controller.add_model(model_name) + model_uuid = model.info.uuid + await model.disconnect() + await controller.destroy_model(model_uuid) + await asyncio.wait_for(_wait_for_model_gone(controller, + model_name), + timeout=60)