modules/libjuju/tests/integration/test_controller.py

   1 import asyncio
   2 import subprocess
   3 import uuid
   4
   5 from juju.client.connection import Connection
   6 from juju.client.jujudata import FileJujuData
   7 from juju.controller import Controller
   8 from juju.errors import JujuAPIError
   9
  10 import pytest
  11
  12 from .. import base
  13
  14
  15 @base.bootstrapped
  16 @pytest.mark.asyncio
  17 async def test_add_remove_user(event_loop):
  18     async with base.CleanController() as controller:
  19         username = 'test{}'.format(uuid.uuid4())
  20         user = await controller.get_user(username)
  21         assert user is None
  22         user = await controller.add_user(username)
  23         assert user is not None
  24         assert user.secret_key is not None
  25         assert user.username == username
  26         users = await controller.get_users()
  27         assert any(u.username == username for u in users)
  28         await controller.remove_user(username)
  29         user = await controller.get_user(username)
  30         assert user is None
  31         users = await controller.get_users()
  32         assert not any(u.username == username for u in users)
  33
  34
  35 @base.bootstrapped
  36 @pytest.mark.asyncio
  37 async def test_disable_enable_user(event_loop):
  38     async with base.CleanController() as controller:
  39         username = 'test-disable{}'.format(uuid.uuid4())
  40         user = await controller.add_user(username)
  41
  42         await user.disable()
  43         assert not user.enabled
  44         assert user.disabled
  45
  46         fresh = await controller.get_user(username)  # fetch fresh copy
  47         assert not fresh.enabled
  48         assert fresh.disabled
  49
  50         await user.enable()
  51         assert user.enabled
  52         assert not user.disabled
  53
  54         fresh = await controller.get_user(username)  # fetch fresh copy
  55         assert fresh.enabled
  56         assert not fresh.disabled
  57
  58
  59 @base.bootstrapped
  60 @pytest.mark.asyncio
  61 async def test_change_user_password(event_loop):
  62     async with base.CleanController() as controller:
  63         username = 'test-password{}'.format(uuid.uuid4())
  64         user = await controller.add_user(username)
  65         await user.set_password('password')
  66         # Check that we can connect with the new password.
  67         new_connection = None
  68         try:
  69             kwargs = controller.connection().connect_params()
  70             kwargs['username'] = username
  71             kwargs['password'] = 'password'
  72             new_connection = await Connection.connect(**kwargs)
  73         except JujuAPIError:
  74             raise AssertionError('Unable to connect with new password')
  75         finally:
  76             if new_connection:
  77                 await new_connection.close()
  78
  79
  80 @base.bootstrapped
  81 @pytest.mark.asyncio
  82 async def test_reset_user_password(event_loop):
  83     async with base.CleanController() as controller:
  84         username = 'test{}'.format(uuid.uuid4())
  85         user = await controller.add_user(username)
  86         origin_secret_key = user.secret_key
  87         await user.set_password('password')
  88         await controller.reset_user_password(username)
  89         user = await controller.get_user(username)
  90         new_secret_key = user.secret_key
  91         # Check secret key is different after the reset.
  92         assert origin_secret_key != new_secret_key
  93         # Check that we can't connect with the old password.
  94         new_connection = None
  95         try:
  96             kwargs = controller.connection().connect_params()
  97             kwargs['username'] = username
  98             kwargs['password'] = 'password'
  99             new_connection = await Connection.connect(**kwargs)
 100         except JujuAPIError:
 101             pass
 102         finally:
 103             # No connection with old password
 104             assert new_connection is None
 105
 106
 107 @base.bootstrapped
 108 @pytest.mark.asyncio
 109 async def test_grant_revoke(event_loop):
 110     async with base.CleanController() as controller:
 111         username = 'test-grant{}'.format(uuid.uuid4())
 112         user = await controller.add_user(username)
 113         await user.grant('superuser')
 114         assert user.access == 'superuser'
 115         fresh = await controller.get_user(username)  # fetch fresh copy
 116         assert fresh.access == 'superuser'
 117         await user.grant('login')  # already has 'superuser', so no-op
 118         assert user.access == 'superuser'
 119         fresh = await controller.get_user(username)  # fetch fresh copy
 120         assert fresh.access == 'superuser'
 121         await user.revoke()
 122         assert user.access == ''
 123         fresh = await controller.get_user(username)  # fetch fresh copy
 124         assert fresh.access == ''
 125
 126
 127 @base.bootstrapped
 128 @pytest.mark.asyncio
 129 async def test_list_models(event_loop):
 130     async with base.CleanController() as controller:
 131         async with base.CleanModel() as model:
 132             result = await controller.list_models()
 133             assert model.info.name in result
 134
 135
 136 @base.bootstrapped
 137 @pytest.mark.asyncio
 138 async def test_get_model(event_loop):
 139     async with base.CleanController() as controller:
 140         by_name, by_uuid = None, None
 141         model_name = 'test-{}'.format(uuid.uuid4())
 142         model = await controller.add_model(model_name)
 143         model_uuid = model.info.uuid
 144         await model.disconnect()
 145         try:
 146             by_name = await controller.get_model(model_name)
 147             by_uuid = await controller.get_model(model_uuid)
 148             assert by_name.info.name == model_name
 149             assert by_name.info.uuid == model_uuid
 150             assert by_uuid.info.name == model_name
 151             assert by_uuid.info.uuid == model_uuid
 152         finally:
 153             if by_name:
 154                 await by_name.disconnect()
 155             if by_uuid:
 156                 await by_uuid.disconnect()
 157             await controller.destroy_model(model_name)
 158
 159
 160 async def _wait_for_model(controller, model_name):
 161     while model_name not in await controller.list_models():
 162         await asyncio.sleep(0.5, loop=controller.loop)
 163
 164
 165 async def _wait_for_model_gone(controller, model_name):
 166     while model_name in await controller.list_models():
 167         await asyncio.sleep(0.5, loop=controller.loop)
 168
 169
 170 @base.bootstrapped
 171 @pytest.mark.asyncio
 172 async def test_destroy_model_by_name(event_loop):
 173     async with base.CleanController() as controller:
 174         model_name = 'test-{}'.format(uuid.uuid4())
 175         model = await controller.add_model(model_name)
 176         await model.disconnect()
 177         await asyncio.wait_for(_wait_for_model(controller,
 178                                                model_name),
 179                                timeout=60)
 180         await controller.destroy_model(model_name)
 181         await asyncio.wait_for(_wait_for_model_gone(controller,
 182                                                     model_name),
 183                                timeout=60)
 184
 185
 186 @base.bootstrapped
 187 @pytest.mark.asyncio
 188 async def test_add_destroy_model_by_uuid(event_loop):
 189     async with base.CleanController() as controller:
 190         model_name = 'test-{}'.format(uuid.uuid4())
 191         model = await controller.add_model(model_name)
 192         model_uuid = model.info.uuid
 193         await model.disconnect()
 194         await asyncio.wait_for(_wait_for_model(controller,
 195                                                model_name),
 196                                timeout=60)
 197         await controller.destroy_model(model_uuid)
 198         await asyncio.wait_for(_wait_for_model_gone(controller,
 199                                                     model_name),
 200                                timeout=60)
 201
 202
 203 # this test must be run serially because it modifies the login password
 204 @pytest.mark.serial
 205 @base.bootstrapped
 206 @pytest.mark.asyncio
 207 async def test_macaroon_auth(event_loop):
 208     jujudata = FileJujuData()
 209     account = jujudata.accounts()[jujudata.current_controller()]
 210     with base.patch_file('~/.local/share/juju/accounts.yaml'):
 211         if 'password' in account:
 212             # force macaroon auth by "changing" password to current password
 213             result = subprocess.run(
 214                 ['juju', 'change-user-password'],
 215                 input='{0}\n{0}\n'.format(account['password']),
 216                 universal_newlines=True,
 217                 stderr=subprocess.PIPE)
 218             assert result.returncode == 0, ('Failed to change password: '
 219                                             '{}'.format(result.stderr))
 220         controller = Controller()
 221         try:
 222             await controller.connect()
 223             assert controller.is_connected()
 224         finally:
 225             if controller.is_connected():
 226                 await controller.disconnect()
 227         async with base.CleanModel():
 228             pass  # create and login to model works