bae4b80730816f3994b5294f53d4ee4f9492e160
6 from juju
.client
.jujudata
import FileJujuData
7 from juju
.controller
import Controller
12 def is_bootstrapped():
13 result
= subprocess
.run(['juju', 'switch'], stdout
=subprocess
.PIPE
)
15 result
.returncode
== 0 and
16 len(result
.stdout
.decode().strip()) > 0)
19 bootstrapped
= pytest
.mark
.skipif(
20 not is_bootstrapped(),
21 reason
='bootstrapped Juju environment required')
23 test_run_nonce
= uuid
.uuid4().hex[-4:]
26 class CleanController():
28 self
._controller
= None
30 async def __aenter__(self
):
31 self
._controller
= Controller()
32 await self
._controller
.connect()
33 return self
._controller
35 async def __aexit__(self
, exc_type
, exc
, tb
):
36 await self
._controller
.disconnect()
40 def __init__(self
, bakery_client
=None):
41 self
._controller
= None
43 self
._model
_uuid
= None
44 self
._bakery
_client
= bakery_client
46 async def __aenter__(self
):
47 model_nonce
= uuid
.uuid4().hex[-4:]
48 frame
= inspect
.stack()[1]
49 test_name
= frame
.function
.replace('_', '-')
50 jujudata
= TestJujuData()
51 self
._controller
= Controller(
53 bakery_client
=self
._bakery
_client
,
55 controller_name
= jujudata
.current_controller()
56 user_name
= jujudata
.accounts()[controller_name
]['user']
57 await self
._controller
.connect(controller_name
)
59 model_name
= 'test-{}-{}-{}'.format(
64 self
._model
= await self
._controller
.add_model(model_name
)
66 # Change the JujuData instance so that it will return the new
67 # model as the current model name, so that we'll connect
71 user_name
+ "/" + model_name
,
72 self
._model
.info
.uuid
,
75 # save the model UUID in case test closes model
76 self
._model
_uuid
= self
._model
.info
.uuid
81 result
= self
._orig
_models
()
82 models
= result
[self
.controller_name
]['models']
83 full_model_name
= '{}/{}'.format(self
.user_name
, self
.model_name
)
84 if full_model_name
not in models
:
85 models
[full_model_name
] = {'uuid': self
.model_uuid
}
88 async def __aexit__(self
, exc_type
, exc
, tb
):
89 await self
._model
.disconnect()
90 await self
._controller
.destroy_model(self
._model
_uuid
)
91 await self
._controller
.disconnect()
94 class TestJujuData(FileJujuData
):
96 self
.__controller
_name
= None
97 self
.__model
_name
= None
98 self
.__model
_uuid
= None
101 def set_model(self
, controller_name
, model_name
, model_uuid
):
102 self
.__controller
_name
= controller_name
103 self
.__model
_name
= model_name
104 self
.__model
_uuid
= model_uuid
106 def current_model(self
, *args
, **kwargs
):
107 return self
.__model
_name
or super().current_model(*args
, **kwargs
)
110 all_models
= super().models()
111 if self
.__model
_name
is None:
113 all_models
.setdefault(self
.__controller
_name
, {})
114 all_models
[self
.__controller
_name
].setdefault('models', {})
115 cmodels
= all_models
[self
.__controller
_name
]['models']
116 cmodels
[self
.__model
_name
] = {'uuid': self
.__model
_uuid
}
118 >>>>>>> New N2VC interface
+ updated libjuju
121 class AsyncMock(mock
.MagicMock
):
122 async def __call__(self
, *args
, **kwargs
):
123 return super().__call
__(*args
, **kwargs
)