modules/libjuju/tests/base.py

   1 import inspect
   2 import subprocess
   3 import uuid
   4
   5 import mock
   6 from juju.client.jujudata import FileJujuData
   7 from juju.controller import Controller
   8
   9 import pytest
  10
  11
  12 def is_bootstrapped():
  13     result = subprocess.run(['juju', 'switch'], stdout=subprocess.PIPE)
  14     return (
  15         result.returncode == 0 and
  16         len(result.stdout.decode().strip()) > 0)
  17
  18
  19 bootstrapped = pytest.mark.skipif(
  20     not is_bootstrapped(),
  21     reason='bootstrapped Juju environment required')
  22
  23 test_run_nonce = uuid.uuid4().hex[-4:]
  24
  25
  26 class CleanController():
  27     def __init__(self):
  28         self._controller = None
  29
  30     async def __aenter__(self):
  31         self._controller = Controller()
  32         await self._controller.connect()
  33         return self._controller
  34
  35     async def __aexit__(self, exc_type, exc, tb):
  36         await self._controller.disconnect()
  37
  38
  39 class CleanModel():
  40     def __init__(self, bakery_client=None):
  41         self._controller = None
  42         self._model = None
  43         self._model_uuid = None
  44         self._bakery_client = bakery_client
  45
  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(
  52             jujudata=jujudata,
  53             bakery_client=self._bakery_client,
  54         )
  55         controller_name = jujudata.current_controller()
  56         user_name = jujudata.accounts()[controller_name]['user']
  57         await self._controller.connect(controller_name)
  58
  59         model_name = 'test-{}-{}-{}'.format(
  60             test_run_nonce,
  61             test_name,
  62             model_nonce,
  63         )
  64         self._model = await self._controller.add_model(model_name)
  65
  66         # Change the JujuData instance so that it will return the new
  67         # model as the current model name, so that we'll connect
  68         # to it by default.
  69         jujudata.set_model(
  70             controller_name,
  71             user_name + "/" + model_name,
  72             self._model.info.uuid,
  73         )
  74
  75         # save the model UUID in case test closes model
  76         self._model_uuid = self._model.info.uuid
  77
  78         return self._model
  79
  80     def _models(self):
  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}
  86         return result
  87
  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()
  92
  93
  94 class TestJujuData(FileJujuData):
  95     def __init__(self):
  96         self.__controller_name = None
  97         self.__model_name = None
  98         self.__model_uuid = None
  99         super().__init__()
 100
 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
 105
 106     def current_model(self, *args, **kwargs):
 107         return self.__model_name or super().current_model(*args, **kwargs)
 108
 109     def models(self):
 110         all_models = super().models()
 111         if self.__model_name is None:
 112             return all_models
 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}
 117         return all_models
 118 >>>>>>> New N2VC interface + updated libjuju
 119
 120
 121 class AsyncMock(mock.MagicMock):
 122     async def __call__(self, *args, **kwargs):
 123         return super().__call__(*args, **kwargs)