import asyncio
import base64
import logging
+from dataclasses import dataclass
from juju.application import Application
from juju.controller import Controller
from n2vc.config import EnvironConfig
-from n2vc.temporal_libjuju import ConnectionInfo, Libjuju
from osm_common.dataclasses.temporal_dataclasses import (
CharmInfo,
CheckCharmStatusInput,
)
from osm_common.temporal_constants import (
ACTIVITY_TEST_VIM_CONNECTIVITY,
+ ACTIVITY_CHECK_CHARM_STATUS,
ACTIVITY_CREATE_MODEL,
ACTIVITY_DEPLOY_CHARM,
- ACTIVITY_CHECK_CHARM_STATUS,
)
from osm_lcm.data_utils.database.database import Database
from temporalio import activity
+@dataclass
+class ConnectionInfo:
+ """Information to connect to juju controller"""
+
+ endpoint: str
+ user: str
+ password: str
+ cacert: str
+
+ def __repr__(self):
+ return f"{self.__class__.__name__}(endpoint: {self.endpoint}, user: {self.user}, password: ******, cacert: ******)"
+
+ def __str__(self):
+ return f"{self.__class__.__name__}(endpoint: {self.endpoint}, user: {self.user}, password: ******, cacert: ******)"
+
+
class JujuPaasConnector:
"""Handles Juju Controller operations.
username = vim_content["vim_user"]
vim_config = vim_content["config"]
cacert = vim_config["ca_cert_content"]
- cloud_name = vim_config["cloud"]
- cloud_credentials = vim_config["cloud_credentials"]
password = self._decrypt_password(vim_content)
- return ConnectionInfo(
- endpoint, username, password, cacert, cloud_name, cloud_credentials
- )
-
- def _get_libjuju(self, vim_uuid):
- connection_info = self._get_connection_info(vim_uuid)
- return Libjuju(connection_info)
+ return ConnectionInfo(endpoint, username, password, cacert)
async def _get_controller(self, vim_uuid) -> Controller:
connection_info = self._get_connection_info(vim_uuid)
and wait on each connection attempt.
"""
vim_id = test_connectivity_input.vim_uuid
- controller = None
- try:
- libjuju = self._get_libjuju(vim_id)
- controller = await libjuju.get_controller()
- message = f"Connection to juju controller succeeded for {vim_id}"
- self.logger.info(message)
- finally:
- await libjuju.disconnect_controller(controller)
+ self._get_controller(vim_id)
+ message = f"Connection to juju controller succeeded for {vim_id}"
+ self.logger.info(message)
@activity.defn(name=ACTIVITY_CREATE_MODEL)
async def create_model(self, create_model_input: ModelInfo) -> None:
from juju.model import Model
from juju.errors import JujuError
from juju.unit import Unit
-from n2vc.temporal_libjuju import ConnectionInfo
from osm_common.dataclasses.temporal_dataclasses import (
CharmInfo,
CheckCharmStatusInput,
ModelInfo,
+ TestVimConnectivityInput,
+ VduComputeConstraints,
VduInstantiateInput,
)
+from osm_common.dbbase import DbException
from osm_lcm.temporal.juju_paas_activities import JujuPaasConnector
from parameterized import parameterized
from temporalio.testing import ActivityEnvironment
from unittest.mock import ANY, AsyncMock, Mock
-vim_id = "some-vim-uuid"
namespace = "some-namespace"
-model_info = ModelInfo(vim_id, namespace)
-connection_info = ConnectionInfo(
- "1.2.3.4:17070", "user", "password", "cacert", "cloud_name", "cloud_credentials"
-)
vim_content = {
"_id": "82258772-0145-47cf-9a56-98a83aab38cc",
"name": "juju-with-key",
}
+@asynctest.mock.patch("juju.controller.Controller.connect")
class TestJujuPaasConnector(asynctest.TestCase):
def setUp(self):
self.db = Mock()
- self.env = ActivityEnvironment()
self.juju_paas_connector = JujuPaasConnector(self.db)
- self.controller = AsyncMock(spec=Controller)
-
- async def get_controller(_: str):
- return self.controller
-
- self.juju_paas_connector._get_controller = get_controller
- self._api_endpoints = []
- self.controller.api_endpoints = self.api_endpoints
+ self.juju_paas_connector._decrypt_password = Mock()
+ self.juju_paas_connector._decrypt_password.side_effect = ["password"]
- @property
- async def api_endpoints(self):
- return self._api_endpoints
+ async def test_get_controller_nominal_case(self, mock_connect):
+ self.db.get_one.side_effect = [vim_content]
+ controller = await self.juju_paas_connector._get_controller(vim_content["_id"])
+ self.assertIsInstance(controller, Controller)
+ mock_connect.assert_called_once_with(
+ endpoint=vim_content["vim_url"],
+ username=vim_content["vim_user"],
+ password="password",
+ cacert=vim_content["config"]["ca_cert_content"],
+ )
- async def test_create_model_nominal_case(self):
+ async def test_get_controller_raises_juju_error(self, mock_connect):
self.db.get_one.side_effect = [vim_content]
- self.juju_paas_connector._decrypt_password = Mock()
- self.juju_paas_connector._decrypt_password.side_effect = ["password"]
- await self.env.run(self.juju_paas_connector.create_model, model_info)
- self.controller.add_model.assert_called_once_with(
- model_info.model_name,
- config=ANY,
- cloud_name=vim_content["config"]["cloud"],
- credential_name=vim_content["config"]["cloud_credentials"],
+ mock_connect.side_effect = JujuError()
+ with self.assertRaises(JujuError):
+ await self.juju_paas_connector._get_controller(vim_content["_id"])
+ mock_connect.assert_called_once_with(
+ endpoint=vim_content["vim_url"],
+ username=vim_content["vim_user"],
+ password="password",
+ cacert=vim_content["config"]["ca_cert_content"],
)
- async def test_create_model_already_exists(self):
- self.controller.list_models.return_value = [model_info.model_name]
- await self.env.run(self.juju_paas_connector.create_model, model_info)
- self.controller.add_model.assert_not_called()
+ async def test_get_controller_raises_db_error(self, mock_connect):
+ self.db.get_one.side_effect = DbException("DB Exception")
+ with self.assertRaises(DbException):
+ await self.juju_paas_connector._get_controller(vim_content["_id"])
+ mock_connect.assert_not_called()
class TestJujuPaasActivitiesBase(asynctest.TestCase):
def setUp(self) -> None:
self.db = Mock()
self.env = ActivityEnvironment()
- self.controller = AsyncMock()
- self.model = Mock(spec=Model)
- self.model.applications = {}
- self.controller.get_model.return_value = self.model
+ self.controller = AsyncMock(spec=Controller)
+ self.mock_model()
async def get_controller(_: str):
return self.controller
self.juju_paas = JujuPaasConnector(self.db)
self.juju_paas._get_controller = get_controller
+ def mock_model(self) -> None:
+ self.model = Mock(spec=Model)
+ self.model.applications = {}
+ self.controller.get_model.return_value = self.model
+
def add_application(self, application_name) -> None:
self.application_name = application_name
self.application = Mock(spec=Application)
self.model.applications = {self.application_name: self.application}
+class TestCreateModel(TestJujuPaasActivitiesBase):
+ model_info = ModelInfo(vim_content["_id"], namespace)
+
+ def setUp(self):
+ super().setUp()
+ self._api_endpoints = []
+ self.controller.api_endpoints = self.api_endpoints
+
+ @property
+ async def api_endpoints(self):
+ return self._api_endpoints
+
+ async def test_create_model_nominal_case(self):
+ self.db.get_one.side_effect = [vim_content]
+ self.juju_paas._decrypt_password = Mock()
+ self.juju_paas._decrypt_password.side_effect = ["password"]
+ await self.env.run(self.juju_paas.create_model, self.model_info)
+ self.controller.add_model.assert_called_once_with(
+ self.model_info.model_name,
+ config=ANY,
+ cloud_name=vim_content["config"]["cloud"],
+ credential_name=vim_content["config"]["cloud_credentials"],
+ )
+
+ async def test_create_model_already_exists(self):
+ self.controller.list_models.return_value = [self.model_info.model_name]
+ await self.env.run(self.juju_paas.create_model, self.model_info)
+ self.controller.add_model.assert_not_called()
+
+
class TestCheckCharmStatus(TestJujuPaasActivitiesBase):
def setUp(self) -> None:
super().setUp()
channel = "latest"
entity_url = "ch:my-charm"
charm_info = CharmInfo(app_name, channel, entity_url)
- vdu_instantiate_input = VduInstantiateInput(vim_id, namespace, charm_info)
+ constraints = VduComputeConstraints(1, 2, "")
+ vdu_instantiate_input = VduInstantiateInput(
+ vim_content["_id"], namespace, charm_info, constraints
+ )
async def test_deploy_charm_nominal_case(self):
await self.env.run(self.juju_paas.deploy_charm, self.vdu_instantiate_input)
with self.assertRaises(JujuError):
await self.env.run(self.juju_paas.deploy_charm, self.vdu_instantiate_input)
self.model.deploy.assert_not_called()
+
+
+class TestTestVimConnectivity(TestJujuPaasActivitiesBase):
+ test_vim_connectivity_input = TestVimConnectivityInput(vim_content["_id"])
+
+ async def test_connectivity_nominal_case(self):
+ await self.env.run(
+ self.juju_paas.test_vim_connectivity, self.test_vim_connectivity_input
+ )
+
+ async def test_connectivity_raises_exception(self):
+ self.juju_paas._get_controller = Mock()
+ self.juju_paas._get_controller.side_effect = JujuError()
+ with self.assertRaises(JujuError):
+ await self.env.run(
+ self.juju_paas.test_vim_connectivity, self.test_vim_connectivity_input
+ )
projects / osm / LCM.git / commitdiff