Raises (Retryable):
ApplicationError If Juju controller is not reachable
+ If application already exists
Activity Lifecycle:
This activity should complete relatively quickly (in a few seconds).
and wait on each connection attempt.
"""
model_name = deploy_charm_input.model_name
- libjuju = self._get_libjuju(deploy_charm_input.vim_uuid)
charm_info = deploy_charm_input.charm_info
- await libjuju.deploy_charm(
- application_name=charm_info.app_name,
- path=charm_info.entity_url,
- model_name=model_name,
+ application_name = charm_info.app_name
+ controller = await self._get_controller(deploy_charm_input.vim_uuid)
+ model = await controller.get_model(model_name)
+ if application_name in model.applications:
+ raise Exception("Application {} already exists".format(application_name))
+ await model.deploy(
+ entity_url=charm_info.entity_url,
+ application_name=application_name,
channel=charm_info.channel,
)
import unittest.mock as mock
from juju.application import Application
+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 CheckCharmStatusInput, ModelInfo
+from osm_common.dataclasses.temporal_dataclasses import (
+ CharmInfo,
+ CheckCharmStatusInput,
+ ModelInfo,
+ 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 AsyncMock, Mock
-
vim_id = "some-vim-uuid"
namespace = "some-namespace"
model_info = ModelInfo(vim_id, namespace)
mock_add_model.assert_not_called()
-class JujuPaasActivitiesTest(asynctest.TestCase):
+class TestJujuPaasActivitiesBase(asynctest.TestCase):
def setUp(self) -> None:
self.db = Mock()
self.env = ActivityEnvironment()
- self.env.on_heartbeat = self.on_heartbeat
- self.heartbeat_count = 0
- self.heartbeat_maximum = 5
-
self.controller = AsyncMock()
- self.model = Mock()
+ self.model = Mock(spec=Model)
+ self.model.applications = {}
self.controller.get_model.return_value = self.model
async def get_controller(_: str):
self.juju_paas = JujuPaasConnector(self.db)
self.juju_paas._get_controller = get_controller
- self.application_name = "application"
+ def add_application(self, application_name) -> None:
+ self.application_name = application_name
self.application = Mock(spec=Application)
self.application.name = self.application_name
self.model.applications = {self.application_name: self.application}
+
+class TestCheckCharmStatus(TestJujuPaasActivitiesBase):
+ def setUp(self) -> None:
+ super().setUp()
+ self.env.on_heartbeat = self.on_heartbeat
+ self.heartbeat_count = 0
+ self.heartbeat_maximum = 5
+ self.add_application("application")
+
def on_heartbeat(self, *args, **kwargs):
self.heartbeat_count += 1
if self.heartbeat_count > self.heartbeat_maximum:
)
await self.env.run(self.juju_paas.check_charm_status, arg)
+
+
+class TestDeployCharm(TestJujuPaasActivitiesBase):
+ app_name = "my_app_name"
+ channel = "latest"
+ entity_url = "ch:my-charm"
+ charm_info = CharmInfo(app_name, channel, entity_url)
+ vdu_instantiate_input = VduInstantiateInput(vim_id, namespace, charm_info)
+
+ async def test_deploy_charm_nominal_case(self):
+ await self.env.run(self.juju_paas.deploy_charm, self.vdu_instantiate_input)
+ self.model.deploy.assert_called_once_with(
+ entity_url=self.entity_url,
+ application_name=self.app_name,
+ channel=self.channel,
+ )
+
+ async def test_deploy_charm_app_already_exists(self):
+ self.add_application(self.app_name)
+ with self.assertRaises(Exception) as err:
+ await self.env.run(self.juju_paas.deploy_charm, self.vdu_instantiate_input)
+ self.model.deploy.assert_not_called()
+ self.assertEqual(
+ str(err.exception.args[0]),
+ "Application {} already exists".format(self.app_name),
+ )
+
+ async def test_deploy_charm_raises_exception(self):
+ self.controller.get_model.side_effect = JujuError()
+ with self.assertRaises(JujuError):
+ await self.env.run(self.juju_paas.deploy_charm, self.vdu_instantiate_input)
+ self.model.deploy.assert_not_called()
--- /dev/null
+#######################################################################################
+# Copyright ETSI Contributors and Others.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asynctest
+
+from osm_common.dataclasses.temporal_dataclasses import CharmInfo, VduInstantiateInput
+from osm_common.temporal_constants import (
+ ACTIVITY_DEPLOY_CHARM,
+ ACTIVITY_CHECK_CHARM_STATUS,
+ LCM_TASK_QUEUE,
+)
+from osm_lcm.temporal.vdu_workflows import CheckCharmStatusInput, VduInstantiateWorkflow
+from temporalio import activity
+from temporalio.client import WorkflowFailureError
+from temporalio.testing import WorkflowEnvironment
+from temporalio.worker import Worker
+
+
+@activity.defn(name=ACTIVITY_DEPLOY_CHARM)
+async def deploy_charm_mocked(deploy_charm_input: VduInstantiateInput) -> None:
+ pass
+
+
+@activity.defn(name=ACTIVITY_DEPLOY_CHARM)
+async def deploy_charm_mocked_raises(deploy_charm_input: VduInstantiateInput) -> None:
+ raise Exception()
+
+
+@activity.defn(name=ACTIVITY_CHECK_CHARM_STATUS)
+async def check_charm_status_mocked(check_charm_status: CheckCharmStatusInput) -> None:
+ pass
+
+
+@activity.defn(name=ACTIVITY_CHECK_CHARM_STATUS)
+async def check_charm_status_mocked_raises(
+ check_charm_status: CheckCharmStatusInput,
+) -> None:
+ raise Exception()
+
+
+class TestVduWorkflows(asynctest.TestCase):
+ task_queue_name = LCM_TASK_QUEUE
+ vim_id = "some-vim-uuid"
+ namespace = "some-namespace"
+ app_name = "my_app_name"
+ channel = "latest"
+ entity_url = "ch:my-charm"
+ charm_info = CharmInfo(app_name, channel, entity_url)
+ vdu_instantiate_input = VduInstantiateInput(vim_id, namespace, charm_info)
+ worflow_id = namespace + "-" + app_name
+
+ async def setUp(self):
+ self.env = await WorkflowEnvironment.start_time_skipping()
+ self.client = self.env.client
+
+ def get_worker(self, activities: list) -> Worker:
+ return Worker(
+ self.client,
+ task_queue=self.task_queue_name,
+ workflows=[VduInstantiateWorkflow],
+ activities=activities,
+ debug_mode=True,
+ )
+
+ async def test_vdu_instantiate(self):
+ activities = [deploy_charm_mocked, check_charm_status_mocked]
+ async with self.env, self.get_worker(activities):
+ result = await self.client.execute_workflow(
+ VduInstantiateWorkflow.run,
+ arg=self.vdu_instantiate_input,
+ id=self.worflow_id,
+ task_queue=self.task_queue_name,
+ )
+ self.assertIsNone(result)
+
+ async def test_vdu_instantiate_deploy_raises(self):
+ activities = [deploy_charm_mocked_raises, check_charm_status_mocked]
+ async with self.env, self.get_worker(activities):
+ with self.assertRaises(WorkflowFailureError):
+ result = await self.client.execute_workflow(
+ VduInstantiateWorkflow.run,
+ arg=self.vdu_instantiate_input,
+ id=self.worflow_id,
+ task_queue=self.task_queue_name,
+ )
+ self.assertIsNone(result)
+
+ async def test_vdu_instantiate_check_status_raises(self):
+ activities = [deploy_charm_mocked, check_charm_status_mocked_raises]
+ async with self.env, self.get_worker(activities):
+ with self.assertRaises(WorkflowFailureError):
+ result = await self.client.execute_workflow(
+ VduInstantiateWorkflow.run,
+ arg=self.vdu_instantiate_input,
+ id=self.worflow_id,
+ task_queue=self.task_queue_name,
+ )
+ self.assertIsNone(result)
projects / osm / LCM.git / commitdiff