tests/integration/test_controller.py

   1 import asyncio
   2 from concurrent.futures import ThreadPoolExecutor
   3 import pytest
   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_user(event_loop):
  13     async with base.CleanController() as controller:
  14         await controller.add_user('test')
  15         result = await controller.get_user('test')
  16         res_ser = result.serialize()['results'][0].serialize()
  17         assert res_ser['result'] is not None
  18
  19
  20 @base.bootstrapped
  21 @pytest.mark.asyncio
  22 async def test_disable_enable_user(event_loop):
  23     async with base.CleanController() as controller:
  24         await controller.add_user('test-disable')
  25         await controller.disable_user('test-disable')
  26         result = await controller.get_user('test-disable')
  27         res_ser = result.serialize()['results'][0].serialize()
  28         assert res_ser['result'].serialize()['disabled'] is True
  29         await controller.enable_user('test-disable')
  30         result = await controller.get_user('test-disable')
  31         res_ser = result.serialize()['results'][0].serialize()
  32         assert res_ser['result'].serialize()['disabled'] is False
  33
  34
  35 @base.bootstrapped
  36 @pytest.mark.asyncio
  37 async def test_change_user_password(event_loop):
  38     async with base.CleanController() as controller:
  39         await controller.add_user('test-password')
  40         await controller.change_user_password('test-password', 'password')
  41         try:
  42             con = await controller.connect(controller.connection.endpoint, 'test-password', 'password')
  43             result = True
  44         except JujuAPIError:
  45             result = False
  46         assert result is True
  47
  48
  49 @base.bootstrapped
  50 @pytest.mark.asyncio
  51 async def test_grant(event_loop):
  52     async with base.CleanController() as controller:
  53         await controller.add_user('test-grant')
  54         await controller.grant('test-grant', 'superuser')
  55         result = await controller.get_user('test-grant')
  56         result = result.serialize()['results'][0].serialize()['result'].serialize()
  57         assert result['access'] == 'superuser'
  58         await controller.grant('test-grant', 'login')
  59         result = await controller.get_user('test-grant')
  60         result = result.serialize()['results'][0].serialize()['result'].serialize()
  61         assert result['access'] == 'login'
  62
  63
  64 @base.bootstrapped
  65 @pytest.mark.asyncio
  66 async def test_get_models(event_loop):
  67     async with base.CleanController() as controller:
  68         result = await controller.get_models()
  69         assert isinstance(result.serialize()['user-models'], list)