modules/libjuju/tests/integration/test_controller.py

   1 import asyncio
   2 import pytest
   3 import uuid
   4
   5 from .. import base
   6 from juju.controller import Controller
   7 from juju.errors import JujuAPIError
   8
   9
  10 @base.bootstrapped
  11 @pytest.mark.asyncio
  12 async def test_add_remove_user(event_loop):
  13     async with base.CleanController() as controller:
  14         username = 'test{}'.format(uuid.uuid4())
  15         user = await controller.get_user(username)
  16         assert user is None
  17         user = await controller.add_user(username)
  18         assert user is not None
  19         assert user.username == username
  20         users = await controller.get_users()
  21         assert any(u.username == username for u in users)
  22         await controller.remove_user(username)
  23         user = await controller.get_user(username)
  24         assert user is None
  25         users = await controller.get_users()
  26         assert not any(u.username == username for u in users)
  27
  28
  29 @base.bootstrapped
  30 @pytest.mark.asyncio
  31 async def test_disable_enable_user(event_loop):
  32     async with base.CleanController() as controller:
  33         username = 'test-disable{}'.format(uuid.uuid4())
  34         user = await controller.add_user(username)
  35
  36         await user.disable()
  37         assert not user.enabled
  38         assert user.disabled
  39
  40         fresh = await controller.get_user(username)  # fetch fresh copy
  41         assert not fresh.enabled
  42         assert fresh.disabled
  43
  44         await user.enable()
  45         assert user.enabled
  46         assert not user.disabled
  47
  48         fresh = await controller.get_user(username)  # fetch fresh copy
  49         assert fresh.enabled
  50         assert not fresh.disabled
  51
  52
  53 @base.bootstrapped
  54 @pytest.mark.asyncio
  55 async def test_change_user_password(event_loop):
  56     async with base.CleanController() as controller:
  57         username = 'test-password{}'.format(uuid.uuid4())
  58         user = await controller.add_user(username)
  59         await user.set_password('password')
  60         try:
  61             new_controller = Controller()
  62             await new_controller.connect(
  63                 controller.connection.endpoint, username, 'password')
  64         except JujuAPIError:
  65             raise AssertionError('Unable to connect with new password')
  66         finally:
  67             await new_controller.disconnect()
  68
  69
  70 @base.bootstrapped
  71 @pytest.mark.asyncio
  72 async def test_grant_revoke(event_loop):
  73     async with base.CleanController() as controller:
  74         username = 'test-grant{}'.format(uuid.uuid4())
  75         user = await controller.add_user(username)
  76         await user.grant('superuser')
  77         assert user.access == 'superuser'
  78         fresh = await controller.get_user(username)  # fetch fresh copy
  79         assert fresh.access == 'superuser'
  80         await user.grant('login')
  81         assert user.access == 'login'
  82         fresh = await controller.get_user(username)  # fetch fresh copy
  83         assert fresh.access == 'login'
  84         await user.revoke()
  85         assert user.access is ''
  86         fresh = await controller.get_user(username)  # fetch fresh copy
  87         assert fresh.access is ''
  88
  89
  90 @base.bootstrapped
  91 @pytest.mark.asyncio
  92 async def test_list_models(event_loop):
  93     async with base.CleanController() as controller:
  94         async with base.CleanModel() as model:
  95             result = await controller.list_models()
  96             assert model.info.name in result
  97
  98
  99 @base.bootstrapped
 100 @pytest.mark.asyncio
 101 async def test_get_model(event_loop):
 102     async with base.CleanController() as controller:
 103         by_name, by_uuid = None, None
 104         model_name = 'test-{}'.format(uuid.uuid4())
 105         model = await controller.add_model(model_name)
 106         model_uuid = model.info.uuid
 107         await model.disconnect()
 108         try:
 109             by_name = await controller.get_model(model_name)
 110             by_uuid = await controller.get_model(model_uuid)
 111             assert by_name.info.name == model_name
 112             assert by_name.info.uuid == model_uuid
 113             assert by_uuid.info.name == model_name
 114             assert by_uuid.info.uuid == model_uuid
 115         finally:
 116             if by_name:
 117                 await by_name.disconnect()
 118             if by_uuid:
 119                 await by_uuid.disconnect()
 120             await controller.destroy_model(model_name)
 121
 122
 123 async def _wait_for_model_gone(controller, model_name):
 124     while model_name in await controller.list_models():
 125         await asyncio.sleep(0.5, loop=controller.loop)
 126
 127
 128 @base.bootstrapped
 129 @pytest.mark.asyncio
 130 async def test_destroy_model_by_name(event_loop):
 131     async with base.CleanController() as controller:
 132         model_name = 'test-{}'.format(uuid.uuid4())
 133         model = await controller.add_model(model_name)
 134         await model.disconnect()
 135         await controller.destroy_model(model_name)
 136         await asyncio.wait_for(_wait_for_model_gone(controller,
 137                                                     model_name),
 138                                timeout=60)
 139
 140
 141 @base.bootstrapped
 142 @pytest.mark.asyncio
 143 async def test_add_destroy_model_by_uuid(event_loop):
 144     async with base.CleanController() as controller:
 145         model_name = 'test-{}'.format(uuid.uuid4())
 146         model = await controller.add_model(model_name)
 147         model_uuid = model.info.uuid
 148         await model.disconnect()
 149         await controller.destroy_model(model_uuid)
 150         await asyncio.wait_for(_wait_for_model_gone(controller,
 151                                                     model_name),
 152                                timeout=60)