modules/libjuju/tests/integration/test_controller.py

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