+++ /dev/null
-import asyncio
-import subprocess
-import uuid
-
-from juju.client.connection import Connection
-from juju.client.jujudata import FileJujuData
-from juju.controller import Controller
-from juju.errors import JujuAPIError
-
-import pytest
-
-from .. import base
-
-
-@base.bootstrapped
-@pytest.mark.asyncio
-async def test_add_remove_user(event_loop):
- async with base.CleanController() as controller:
- username = 'test{}'.format(uuid.uuid4())
- user = await controller.get_user(username)
- assert user is None
- user = await controller.add_user(username)
- assert user is not None
- assert user.secret_key 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
-@pytest.mark.asyncio
-async def test_disable_enable_user(event_loop):
- async with base.CleanController() as controller:
- username = 'test-disable{}'.format(uuid.uuid4())
- 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
-@pytest.mark.asyncio
-async def test_change_user_password(event_loop):
- async with base.CleanController() as controller:
- username = 'test-password{}'.format(uuid.uuid4())
- user = await controller.add_user(username)
- await user.set_password('password')
- # Check that we can connect with the new password.
- new_connection = None
- try:
- kwargs = controller.connection().connect_params()
- kwargs['username'] = username
- kwargs['password'] = 'password'
- new_connection = await Connection.connect(**kwargs)
- except JujuAPIError:
- raise AssertionError('Unable to connect with new password')
- finally:
- if new_connection:
- await new_connection.close()
-
-
-@base.bootstrapped
-@pytest.mark.asyncio
-async def test_reset_user_password(event_loop):
- async with base.CleanController() as controller:
- username = 'test{}'.format(uuid.uuid4())
- user = await controller.add_user(username)
- origin_secret_key = user.secret_key
- await user.set_password('password')
- await controller.reset_user_password(username)
- user = await controller.get_user(username)
- new_secret_key = user.secret_key
- # Check secret key is different after the reset.
- assert origin_secret_key != new_secret_key
- # Check that we can't connect with the old password.
- new_connection = None
- try:
- kwargs = controller.connection().connect_params()
- kwargs['username'] = username
- kwargs['password'] = 'password'
- new_connection = await Connection.connect(**kwargs)
- except JujuAPIError:
- pass
- finally:
- # No connection with old password
- assert new_connection is None
-
-
-@base.bootstrapped
-@pytest.mark.asyncio
-async def test_grant_revoke(event_loop):
- async with base.CleanController() as controller:
- username = 'test-grant{}'.format(uuid.uuid4())
- 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') # already has 'superuser', so no-op
- assert user.access == 'superuser'
- fresh = await controller.get_user(username) # fetch fresh copy
- assert fresh.access == 'superuser'
- await user.revoke()
- assert user.access == ''
- fresh = await controller.get_user(username) # fetch fresh copy
- assert fresh.access == ''
-
-
-@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(controller, model_name):
- while model_name not in await controller.list_models():
- await asyncio.sleep(0.5, loop=controller.loop)
-
-
-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 asyncio.wait_for(_wait_for_model(controller,
- model_name),
- timeout=60)
- 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_add_destroy_model_by_uuid(event_loop):
- async with base.CleanController() as controller:
- model_name = 'test-{}'.format(uuid.uuid4())
- model = await controller.add_model(model_name)
- model_uuid = model.info.uuid
- await model.disconnect()
- await asyncio.wait_for(_wait_for_model(controller,
- model_name),
- timeout=60)
- await controller.destroy_model(model_uuid)
- await asyncio.wait_for(_wait_for_model_gone(controller,
- model_name),
- timeout=60)
-
-
-# this test must be run serially because it modifies the login password
-@pytest.mark.serial
-@base.bootstrapped
-@pytest.mark.asyncio
-async def test_macaroon_auth(event_loop):
- jujudata = FileJujuData()
- account = jujudata.accounts()[jujudata.current_controller()]
- with base.patch_file('~/.local/share/juju/accounts.yaml'):
- if 'password' in account:
- # force macaroon auth by "changing" password to current password
- result = subprocess.run(
- ['juju', 'change-user-password'],
- input='{0}\n{0}\n'.format(account['password']),
- universal_newlines=True,
- stderr=subprocess.PIPE)
- assert result.returncode == 0, ('Failed to change password: '
- '{}'.format(result.stderr))
- controller = Controller()
- try:
- await controller.connect()
- assert controller.is_connected()
- finally:
- if controller.is_connected():
- await controller.disconnect()
- async with base.CleanModel():
- pass # create and login to model works
projects / osm / N2VC.git / blobdiff