X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=modules%2Flibjuju%2Ftests%2Fbase.py;fp=modules%2Flibjuju%2Ftests%2Fbase.py;h=0000000000000000000000000000000000000000;hp=600372cfba017a8e0cef83cdc3465281e7fb4472;hb=9d18c22a0dc9e295adda50601fc5e2f45d2c9b8a;hpb=19c5cfca317615597be6bf1051e9d2fa903adb97 diff --git a/modules/libjuju/tests/base.py b/modules/libjuju/tests/base.py deleted file mode 100644 index 600372c..0000000 --- a/modules/libjuju/tests/base.py +++ /dev/null @@ -1,148 +0,0 @@ -import inspect -import subprocess -import uuid -from contextlib import contextmanager -from pathlib import Path - -import mock -from juju.client.jujudata import FileJujuData -from juju.controller import Controller - -import pytest - - -def is_bootstrapped(): - try: - result = subprocess.run(['juju', 'switch'], stdout=subprocess.PIPE) - return ( - result.returncode == 0 and - len(result.stdout.decode().strip()) > 0) - except FileNotFoundError: - return False - - -bootstrapped = pytest.mark.skipif( - not is_bootstrapped(), - reason='bootstrapped Juju environment required') - -test_run_nonce = uuid.uuid4().hex[-4:] - - -class CleanController(): - """ - Context manager that automatically connects and disconnects from - the currently active controller. - - Note: Unlike CleanModel, this will not create a new controller for you, - and an active controller must already be available. - """ - def __init__(self): - self._controller = None - - async def __aenter__(self): - self._controller = Controller() - await self._controller.connect() - return self._controller - - async def __aexit__(self, exc_type, exc, tb): - await self._controller.disconnect() - - -class CleanModel(): - """ - Context manager that automatically connects to the currently active - controller, adds a fresh model, returns the connection to that model, - and automatically disconnects and cleans up the model. - - The new model is also set as the current default for the controller - connection. - """ - def __init__(self, bakery_client=None): - self._controller = None - self._model = None - self._model_uuid = None - self._bakery_client = bakery_client - - async def __aenter__(self): - model_nonce = uuid.uuid4().hex[-4:] - frame = inspect.stack()[1] - test_name = frame.function.replace('_', '-') - jujudata = TestJujuData() - self._controller = Controller( - jujudata=jujudata, - bakery_client=self._bakery_client, - ) - controller_name = jujudata.current_controller() - user_name = jujudata.accounts()[controller_name]['user'] - await self._controller.connect(controller_name) - - model_name = 'test-{}-{}-{}'.format( - test_run_nonce, - test_name, - model_nonce, - ) - self._model = await self._controller.add_model(model_name) - - # Change the JujuData instance so that it will return the new - # model as the current model name, so that we'll connect - # to it by default. - jujudata.set_model( - controller_name, - user_name + "/" + model_name, - self._model.info.uuid, - ) - - # save the model UUID in case test closes model - self._model_uuid = self._model.info.uuid - - return self._model - - async def __aexit__(self, exc_type, exc, tb): - await self._model.disconnect() - await self._controller.destroy_model(self._model_uuid) - await self._controller.disconnect() - - -class TestJujuData(FileJujuData): - def __init__(self): - self.__controller_name = None - self.__model_name = None - self.__model_uuid = None - super().__init__() - - def set_model(self, controller_name, model_name, model_uuid): - self.__controller_name = controller_name - self.__model_name = model_name - self.__model_uuid = model_uuid - - def current_model(self, *args, **kwargs): - return self.__model_name or super().current_model(*args, **kwargs) - - def models(self): - all_models = super().models() - if self.__model_name is None: - return all_models - all_models.setdefault(self.__controller_name, {}) - all_models[self.__controller_name].setdefault('models', {}) - cmodels = all_models[self.__controller_name]['models'] - cmodels[self.__model_name] = {'uuid': self.__model_uuid} - return all_models - - -class AsyncMock(mock.MagicMock): - async def __call__(self, *args, **kwargs): - return super().__call__(*args, **kwargs) - - -@contextmanager -def patch_file(filename): - """ - "Patch" a file so that its current contents are automatically restored - when the context is exited. - """ - filepath = Path(filename).expanduser() - data = filepath.read_bytes() - try: - yield - finally: - filepath.write_bytes(data)