7 from juju
.controller
import Controller
8 from juju
.client
.connection
import JujuData
11 def is_bootstrapped():
12 result
= subprocess
.run(['juju', 'switch'], stdout
=subprocess
.PIPE
)
14 result
.returncode
== 0 and
15 len(result
.stdout
.decode().strip()) > 0)
17 bootstrapped
= pytest
.mark
.skipif(
18 not is_bootstrapped(),
19 reason
='bootstrapped Juju environment required')
22 class CleanController():
24 self
.controller
= None
26 async def __aenter__(self
):
27 self
.controller
= Controller()
28 await self
.controller
.connect_current()
29 return self
.controller
31 async def __aexit__(self
, exc_type
, exc
, tb
):
32 await self
.controller
.disconnect()
37 self
.controller
= None
40 async def __aenter__(self
):
41 self
.controller
= Controller()
42 await self
.controller
.connect_current()
44 model_name
= 'model-{}'.format(uuid
.uuid4())
45 self
.model
= await self
.controller
.add_model(model_name
)
47 # save the model UUID in case test closes model
48 self
.model_uuid
= self
.model
.info
.uuid
50 # Ensure that we connect to the new model by default. This also
51 # prevents failures if test was started with no current model.
52 self
._patch
_cm
= mock
.patch
.object(JujuData
, 'current_model',
53 return_value
=model_name
)
54 self
._patch
_cm
.start()
58 async def __aexit__(self
, exc_type
, exc
, tb
):
60 await self
.model
.disconnect()
61 await self
.controller
.destroy_model(self
.model_uuid
)
62 await self
.controller
.disconnect()
65 class AsyncMock(mock
.MagicMock
):
66 async def __call__(self
, *args
, **kwargs
):
67 return super().__call
__(*args
, **kwargs
)