5 from juju
.client
.connection
import Connection
6 from juju
.errors
import JujuAPIError
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
)
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
)
28 users
= await controller
.get_users()
29 assert not any(u
.username
== username
for u
in users
)
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
)
40 assert not user
.enabled
43 fresh
= await controller
.get_user(username
) # fetch fresh copy
44 assert not fresh
.enabled
49 assert not user
.disabled
51 fresh
= await controller
.get_user(username
) # fetch fresh copy
53 assert not fresh
.disabled
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.
66 kwargs
= controller
.connection().connect_params()
67 kwargs
['username'] = username
68 kwargs
['password'] = 'password'
69 new_connection
= await Connection
.connect(**kwargs
)
71 raise AssertionError('Unable to connect with new password')
74 await new_connection
.close()
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'
92 assert user
.access
is ''
93 fresh
= await controller
.get_user(username
) # fetch fresh copy
94 assert fresh
.access
is ''
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
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()
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
124 await by_name
.disconnect()
126 await by_uuid
.disconnect()
127 await controller
.destroy_model(model_name
)
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
)
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
)
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
,
150 await controller
.destroy_model(model_name
)
151 await asyncio
.wait_for(_wait_for_model_gone(controller
,
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
,
167 await controller
.destroy_model(model_uuid
)
168 await asyncio
.wait_for(_wait_for_model_gone(controller
,