--- /dev/null
+# Copyright 2021 Canonical Ltd.
+#
+# 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.
+
+from typing import NoReturn
+
+from n2vc.utils import get_ee_id_components
+
+
+class RelationEndpoint:
+ """Represents an endpoint of an application"""
+
+ def __init__(self, ee_id: str, vca_id: str, endpoint_name: str) -> NoReturn:
+ """
+ Args:
+ ee_id: Execution environment id.
+ Format: "<model>.<application_name>.<machine_id>".
+ vca_id: Id of the VCA. Identifies the Juju Controller
+ where the application is deployed
+ endpoint_name: Name of the endpoint for the relation
+ """
+ ee_components = get_ee_id_components(ee_id)
+ self._model_name = ee_components[0]
+ self._application_name = ee_components[1]
+ self._vca_id = vca_id
+ self._endpoint_name = endpoint_name
+
+ @property
+ def application_name(self) -> str:
+ """Returns the application name"""
+ return self._application_name
+
+ @property
+ def endpoint(self) -> str:
+ """Returns the application name and the endpoint. Format: <application>:<endpoint>"""
+ return f"{self.application_name}:{self._endpoint_name}"
+
+ @property
+ def endpoint_name(self) -> str:
+ """Returns the endpoint name"""
+ return self._endpoint_name
+
+ @property
+ def model_name(self) -> str:
+ """Returns the model name"""
+ return self._model_name
+
+ @property
+ def vca_id(self) -> str:
+ """Returns the vca id"""
+ return self._vca_id
+
+ def __str__(self) -> str:
+ app = self.application_name
+ endpoint = self.endpoint_name
+ model = self.model_name
+ vca = self.vca_id
+ return f"{app}:{endpoint} (model: {model}, vca: {vca})"
+
+
+class Offer:
+ """Represents a juju offer"""
+
+ def __init__(self, url: str, vca_id: str = None) -> NoReturn:
+ """
+ Args:
+ url: Offer url. Format: <user>/<model>.<offer-name>.
+ """
+ self._url = url
+ self._username = url.split(".")[0].split("/")[0]
+ self._model_name = url.split(".")[0].split("/")[1]
+ self._name = url.split(".")[1]
+ self._vca_id = vca_id
+
+ @property
+ def model_name(self) -> str:
+ """Returns the model name"""
+ return self._model_name
+
+ @property
+ def name(self) -> str:
+ """Returns the offer name"""
+ return self._name
+
+ @property
+ def username(self) -> str:
+ """Returns the username"""
+ return self._username
+
+ @property
+ def url(self) -> str:
+ """Returns the offer url"""
+ return self._url
+
+ @property
+ def vca_id(self) -> str:
+ """Returns the vca id"""
+ return self._vca_id
import binascii
from n2vc.config import EnvironConfig
+from n2vc.definitions import RelationEndpoint
from n2vc.exceptions import K8sException
from n2vc.k8s_conn import K8sConnector
from n2vc.kubectl import Kubectl
return status
+ async def add_relation(
+ self,
+ provider: RelationEndpoint,
+ requirer: RelationEndpoint,
+ ):
+ """
+ Add relation between two charmed endpoints
+
+ :param: provider: Provider relation endpoint
+ :param: requirer: Requirer relation endpoint
+ """
+ self.log.debug(f"adding new relation between {provider} and {requirer}")
+ cross_model_relation = (
+ provider.model_name != requirer.model_name
+ or requirer.vca_id != requirer.vca_id
+ )
+ try:
+ if cross_model_relation:
+ # Cross-model relation
+ provider_libjuju = await self._get_libjuju(provider.vca_id)
+ requirer_libjuju = await self._get_libjuju(requirer.vca_id)
+ offer = await provider_libjuju.offer(provider)
+ if offer:
+ saas_name = await requirer_libjuju.consume(
+ requirer.model_name, offer, provider_libjuju
+ )
+ await requirer_libjuju.add_relation(
+ requirer.model_name,
+ requirer.endpoint,
+ saas_name,
+ )
+ else:
+ # Standard relation
+ vca_id = provider.vca_id
+ model = provider.model_name
+ libjuju = await self._get_libjuju(vca_id)
+ # add juju relations between two applications
+ await libjuju.add_relation(
+ model_name=model,
+ endpoint_1=provider.endpoint,
+ endpoint_2=requirer.endpoint,
+ )
+ except Exception as e:
+ message = f"Error adding relation between {provider} and {requirer}: {e}"
+ self.log.error(message)
+ raise Exception(message=message)
+
async def update_vca_status(self, vcastatus: dict, kdu_instance: str, **kwargs):
"""
Add all configs, actions, executed actions of all applications in a model to vcastatus dict
from juju.client import client
from juju import tag
+from n2vc.definitions import Offer, RelationEndpoint
from n2vc.juju_watcher import JujuModelWatcher
from n2vc.provisioner import AsyncSSHProvisioner
from n2vc.n2vc_conn import N2VCConnector
await self.disconnect_model(model)
await self.disconnect_controller(controller)
+ async def offer(self, endpoint: RelationEndpoint) -> Offer:
+ """
+ Create an offer from a RelationEndpoint
+
+ :param: endpoint: Relation endpoint
+
+ :return: Offer object
+ """
+ model_name = endpoint.model_name
+ offer_name = f"{endpoint.application_name}-{endpoint.endpoint_name}"
+ controller = await self.get_controller()
+ model = None
+ try:
+ model = await self.get_model(controller, model_name)
+ await model.create_offer(endpoint.endpoint, offer_name=offer_name)
+ offer_list = await self._list_offers(model_name, offer_name=offer_name)
+ if offer_list:
+ return Offer(offer_list[0].offer_url)
+ else:
+ raise Exception("offer was not created")
+ except juju.errors.JujuError as e:
+ if "application offer already exists" not in e.message:
+ raise e
+ finally:
+ if model:
+ self.disconnect_model(model)
+ self.disconnect_controller(controller)
+
async def consume(
self,
- offer_url: str,
model_name: str,
- ):
+ offer: Offer,
+ provider_libjuju: "Libjuju",
+ ) -> str:
"""
- Adds a remote offer to the model. Relations can be created later using "juju relate".
+ Consumes a remote offer in the model. Relations can be created later using "juju relate".
- :param: offer_url: Offer Url
- :param: model_name: Model name
+ :param: model_name: Model name
+ :param: offer: Offer object to consume
+ :param: provider_libjuju: Libjuju object of the provider endpoint
:raises ParseError if there's a problem parsing the offer_url
:raises JujuError if remote offer includes and endpoint
:raises JujuAPIError if the operation is not successful
+
+ :returns: Saas name. It is the application name in the model that reference the remote application.
"""
+ saas_name = f'{offer.name}-{offer.model_name.replace("-", "")}'
+ if offer.vca_id:
+ saas_name = f"{saas_name}-{offer.vca_id}"
controller = await self.get_controller()
- model = await controller.get_model(model_name)
-
+ model = None
+ provider_controller = None
try:
- await model.consume(offer_url)
+ model = await controller.get_model(model_name)
+ provider_controller = await provider_libjuju.get_controller()
+ await model.consume(
+ offer.url, application_alias=saas_name, controller=provider_controller
+ )
+ return saas_name
finally:
- await self.disconnect_model(model)
+ if model:
+ await self.disconnect_model(model)
+ if provider_controller:
+ await provider_libjuju.disconnect_controller(provider_controller)
await self.disconnect_controller(controller)
async def destroy_model(self, model_name: str, total_timeout: float = 1800):
finally:
await self.disconnect_controller(controller)
- async def list_offers(self, model_name: str) -> QueryApplicationOffersResults:
- """List models with certain names
+ async def _list_offers(
+ self, model_name: str, offer_name: str = None
+ ) -> QueryApplicationOffersResults:
+ """
+ List offers within a model
:param: model_name: Model name
+ :param: offer_name: Offer name to filter.
- :return: Returns list of offers
+ :return: Returns application offers results in the model
"""
controller = await self.get_controller()
try:
- return await controller.list_offers(model_name)
+ offers = (await controller.list_offers(model_name)).results
+ if offer_name:
+ matching_offer = []
+ for offer in offers:
+ if offer.offer_name == offer_name:
+ matching_offer.append(offer)
+ break
+ offers = matching_offer
+ return offers
finally:
await self.disconnect_controller(controller)
import logging
from n2vc.config import EnvironConfig
+from n2vc.definitions import RelationEndpoint
from n2vc.exceptions import (
N2VCBadArgumentsException,
N2VCException,
from n2vc.n2vc_conn import obj_to_dict, obj_to_yaml
from n2vc.libjuju import Libjuju
from n2vc.store import MotorStore
+from n2vc.utils import get_ee_id_components
from n2vc.vca.connection import get_connection
from retrying_async import retry
async def add_relation(
self,
- ee_id_1: str,
- ee_id_2: str,
- endpoint_1: str,
- endpoint_2: str,
- vca_id: str = None,
+ provider: RelationEndpoint,
+ requirer: RelationEndpoint,
):
"""
Add relation between two charmed endpoints
- :param: ee_id_1: The id of the first execution environment
- :param: ee_id_2: The id of the second execution environment
- :param: endpoint_1: The endpoint in the first execution environment
- :param: endpoint_2: The endpoint in the second execution environment
- :param: vca_id: VCA ID
+ :param: provider: Provider relation endpoint
+ :param: requirer: Requirer relation endpoint
"""
- self.log.debug(
- "adding new relation between {} and {}, endpoints: {}, {}".format(
- ee_id_1, ee_id_2, endpoint_1, endpoint_2
- )
+ self.log.debug(f"adding new relation between {provider} and {requirer}")
+ cross_model_relation = (
+ provider.model_name != requirer.model_name
+ or requirer.vca_id != requirer.vca_id
)
- libjuju = await self._get_libjuju(vca_id)
-
- # check arguments
- if not ee_id_1:
- message = "EE 1 is mandatory"
- self.log.error(message)
- raise N2VCBadArgumentsException(message=message, bad_args=["ee_id_1"])
- if not ee_id_2:
- message = "EE 2 is mandatory"
- self.log.error(message)
- raise N2VCBadArgumentsException(message=message, bad_args=["ee_id_2"])
- if not endpoint_1:
- message = "endpoint 1 is mandatory"
- self.log.error(message)
- raise N2VCBadArgumentsException(message=message, bad_args=["endpoint_1"])
- if not endpoint_2:
- message = "endpoint 2 is mandatory"
- self.log.error(message)
- raise N2VCBadArgumentsException(message=message, bad_args=["endpoint_2"])
-
- # get the model, the applications and the machines from the ee_id's
- model_1, app_1, _machine_1 = self._get_ee_id_components(ee_id_1)
- model_2, app_2, _machine_2 = self._get_ee_id_components(ee_id_2)
-
- # model must be the same
- if model_1 != model_2:
- message = "EE models are not the same: {} vs {}".format(ee_id_1, ee_id_2)
- self.log.error(message)
- raise N2VCBadArgumentsException(
- message=message, bad_args=["ee_id_1", "ee_id_2"]
- )
-
- # add juju relations between two applications
try:
- await libjuju.add_relation(
- model_name=model_1,
- endpoint_1="{}:{}".format(app_1, endpoint_1),
- endpoint_2="{}:{}".format(app_2, endpoint_2),
- )
+ if cross_model_relation:
+ # Cross-model relation
+ provider_libjuju = await self._get_libjuju(provider.vca_id)
+ requirer_libjuju = await self._get_libjuju(requirer.vca_id)
+ offer = await provider_libjuju.offer(provider)
+ if offer:
+ saas_name = await requirer_libjuju.consume(
+ requirer.model_name, offer, provider_libjuju
+ )
+ await requirer_libjuju.add_relation(
+ requirer.model_name,
+ requirer.endpoint,
+ saas_name,
+ )
+ else:
+ # Standard relation
+ vca_id = provider.vca_id
+ model = provider.model_name
+ libjuju = await self._get_libjuju(vca_id)
+ # add juju relations between two applications
+ await libjuju.add_relation(
+ model_name=model,
+ endpoint_1=provider.endpoint,
+ endpoint_2=requirer.endpoint,
+ )
except Exception as e:
- message = "Error adding relation between {} and {}: {}".format(
- ee_id_1, ee_id_2, e
- )
+ message = f"Error adding relation between {provider} and {requirer}: {e}"
self.log.error(message)
raise N2VCException(message=message)
machine_id=machine_id,
progress_timeout=progress_timeout,
total_timeout=total_timeout,
- **params_dict
+ **params_dict,
)
if status == "completed":
return output
:return: model_name, application_name, machine_id
"""
- if ee_id is None:
- return None, None, None
-
- # split components of id
- parts = ee_id.split(".")
- model_name = parts[0]
- application_name = parts[1]
- machine_id = parts[2]
- return model_name, application_name, machine_id
+ return get_ee_id_components(ee_id)
def _get_application_name(self, namespace: str) -> str:
"""
--- /dev/null
+# Copyright 2021 Canonical Ltd.
+#
+# 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.
+
+from typing import NoReturn
+from unittest import TestCase
+from unittest.mock import patch
+
+from n2vc.definitions import Offer, RelationEndpoint
+
+
+@patch("n2vc.definitions.get_ee_id_components")
+class RelationEndpointTest(TestCase):
+ def test_success(self, mock_get_ee_id_components) -> NoReturn:
+ mock_get_ee_id_components.return_value = ("model", "application", "machine_id")
+ relation_endpoint = RelationEndpoint(
+ "model.application.machine_id",
+ "vca",
+ "endpoint",
+ )
+ self.assertEqual(relation_endpoint.model_name, "model")
+ self.assertEqual(relation_endpoint.application_name, "application")
+ self.assertEqual(relation_endpoint.vca_id, "vca")
+ self.assertEqual(relation_endpoint.endpoint, "application:endpoint")
+ self.assertEqual(relation_endpoint.endpoint_name, "endpoint")
+ self.assertEqual(
+ str(relation_endpoint), "application:endpoint (model: model, vca: vca)"
+ )
+
+
+class OfferTest(TestCase):
+ def test_success(self) -> NoReturn:
+ url = "admin/test-model.my-offer"
+ offer = Offer(url)
+ self.assertEqual(offer.model_name, "test-model")
+ self.assertEqual(offer.name, "my-offer")
+ self.assertEqual(offer.username, "admin")
+ self.assertEqual(offer.url, url)
import logging
import asynctest
from unittest.mock import Mock
+from n2vc.definitions import Offer, RelationEndpoint
from n2vc.k8s_juju_conn import K8sJujuConnector, RBAC_LABEL_KEY_NAME
from osm_common import fslocal
from .utils import kubeconfig, FakeModel, FakeFileWrapper, AsyncMock, FakeApplication
)
self.assertIsNone(status)
self.k8s_juju_conn.libjuju.get_model_status.assert_called_once()
+
+
+class AddRelationTest(K8sJujuConnTestCase):
+ def setUp(self):
+ super(AddRelationTest, self).setUp()
+ self.k8s_juju_conn.libjuju.add_relation = AsyncMock()
+ self.k8s_juju_conn.libjuju.offer = AsyncMock()
+ self.k8s_juju_conn.libjuju.get_controller = AsyncMock()
+ self.k8s_juju_conn.libjuju.consume = AsyncMock()
+
+ def test_standard_relation(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-1.app2.1", None, "endpoint")
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.add_relation(relation_endpoint_1, relation_endpoint_2)
+ )
+ self.k8s_juju_conn.libjuju.add_relation.assert_called_once_with(
+ model_name="model-1", endpoint_1="app1:endpoint", endpoint_2="app2:endpoint"
+ )
+ self.k8s_juju_conn.libjuju.offer.assert_not_called()
+ self.k8s_juju_conn.libjuju.consume.assert_not_called()
+
+ def test_cmr_relation_same_controller(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
+ offer = Offer("admin/model-1.app1")
+ self.k8s_juju_conn.libjuju.offer.return_value = offer
+ self.k8s_juju_conn.libjuju.consume.return_value = "saas"
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.add_relation(relation_endpoint_1, relation_endpoint_2)
+ )
+ self.k8s_juju_conn.libjuju.offer.assert_called_once_with(relation_endpoint_1)
+ self.k8s_juju_conn.libjuju.consume.assert_called_once()
+ self.k8s_juju_conn.libjuju.add_relation.assert_called_once_with(
+ "model-2", "app2:endpoint", "saas"
+ )
+
+ def test_relation_exception(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
+ self.k8s_juju_conn.libjuju.offer.side_effect = Exception()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.add_relation(
+ relation_endpoint_1, relation_endpoint_2
+ )
+ )
import kubernetes
from juju.errors import JujuAPIError
import logging
+
+from n2vc.definitions import Offer, RelationEndpoint
from .utils import (
FakeApplication,
FakeMachine,
mock_get_controller.return_value = juju.controller.Controller()
mock_list_offers.side_effect = Exception()
with self.assertRaises(Exception):
- self.loop.run_until_complete(self.libjuju.list_offers("model"))
+ self.loop.run_until_complete(self.libjuju._list_offers("model"))
mock_disconnect_controller.assert_called_once()
def test_empty_list(
mock_get_controller,
):
mock_get_controller.return_value = juju.controller.Controller()
- mock_list_offers.return_value = []
- offers = self.loop.run_until_complete(self.libjuju.list_offers("model"))
+ offer_results = Mock()
+ offer_results.results = []
+ mock_list_offers.return_value = offer_results
+ offers = self.loop.run_until_complete(self.libjuju._list_offers("model"))
self.assertEqual(offers, [])
mock_disconnect_controller.assert_called_once()
mock_get_controller,
):
mock_get_controller.return_value = juju.controller.Controller()
- mock_list_offers.return_value = ["offer"]
- offers = self.loop.run_until_complete(self.libjuju.list_offers("model"))
- self.assertEqual(offers, ["offer"])
+ offer = Mock()
+ offer_results = Mock()
+ offer_results.results = [offer]
+ mock_list_offers.return_value = offer_results
+ offers = self.loop.run_until_complete(self.libjuju._list_offers("model"))
+ self.assertEqual(offers, [offer])
+ mock_disconnect_controller.assert_called_once()
+
+ def test_matching_offer_name(
+ self,
+ mock_list_offers,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ offer_1 = Mock()
+ offer_1.offer_name = "offer1"
+ offer_2 = Mock()
+ offer_2.offer_name = "offer2"
+ offer_results = Mock()
+ offer_results.results = [offer_1, offer_2]
+ mock_list_offers.return_value = offer_results
+ offers = self.loop.run_until_complete(
+ self.libjuju._list_offers("model", offer_name="offer2")
+ )
+ self.assertEqual(offers, [offer_2])
+ mock_disconnect_controller.assert_called_once()
+
+ def test_not_matching_offer_name(
+ self,
+ mock_list_offers,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ offer_1 = Mock()
+ offer_1.offer_name = "offer1"
+ offer_2 = Mock()
+ offer_2.offer_name = "offer2"
+ offer_results = Mock()
+ offer_results.results = [offer_1, offer_2]
+ mock_list_offers.return_value = offer_results
+ offers = self.loop.run_until_complete(
+ self.libjuju._list_offers("model", offer_name="offer3")
+ )
+ self.assertEqual(offers, [])
mock_disconnect_controller.assert_called_once()
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("juju.controller.Controller.get_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._list_offers")
+@asynctest.mock.patch("juju.model.Model.create_offer")
+class OfferTest(LibjujuTestCase):
+ def setUp(self):
+ super(OfferTest, self).setUp()
+
+ def test_offer(
+ self,
+ mock_create_offer,
+ mock__list_offers,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ controller = juju.controller.Controller()
+ model = juju.model.Model()
+ mock_get_controller.return_value = controller
+ mock_get_model.return_value = model
+ endpoint = RelationEndpoint("model.app-name.0", "vca", "endpoint")
+ self.loop.run_until_complete(self.libjuju.offer(endpoint))
+ mock_create_offer.assert_called_with(
+ "app-name:endpoint", offer_name="app-name-endpoint"
+ )
+ mock_disconnect_model.assert_called_once_with(model)
+ mock_disconnect_controller.assert_called_once_with(controller)
+
+ def test_offer_exception(
+ self,
+ mock_create_offer,
+ mock__list_offers,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ controller = juju.controller.Controller()
+ model = juju.model.Model()
+ mock_get_controller.return_value = controller
+ mock_get_model.return_value = model
+ mock__list_offers.return_value = []
+ endpoint = RelationEndpoint("model.app-name.0", "vca", "endpoint")
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(self.libjuju.offer(endpoint))
+ mock_create_offer.assert_called_with(
+ "app-name:endpoint", offer_name="app-name-endpoint"
+ )
+ mock_disconnect_model.assert_called_once_with(model)
+ mock_disconnect_controller.assert_called_once_with(controller)
+
+
@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
@asynctest.mock.patch("juju.controller.Controller.get_model")
@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
@asynctest.mock.patch("juju.model.Model.consume")
class ConsumeTest(LibjujuTestCase):
def setUp(self):
+ self.offer_url = "admin/model.offer_name"
super(ConsumeTest, self).setUp()
+ self.provider_libjuju = self.libjuju
def test_consume(
self,
mock_get_model,
mock_get_controller,
):
- mock_get_controller.return_value = juju.controller.Controller()
+ self_controller = juju.controller.Controller()
+ provider_controller = juju.controller.Controller()
+ mock_get_controller.side_effect = [self_controller, provider_controller]
mock_get_model.return_value = juju.model.Model()
- self.loop.run_until_complete(self.libjuju.consume("offer_url", "model_name"))
- mock_consume.assert_called_once()
+ self.loop.run_until_complete(
+ self.libjuju.consume(
+ "model_name",
+ Offer(self.offer_url, vca_id="vca-id"),
+ self.provider_libjuju,
+ )
+ )
+ mock_consume.assert_called_once_with(
+ "admin/model.offer_name",
+ application_alias="offer_name-model-vca-id",
+ controller=provider_controller,
+ )
mock_disconnect_model.assert_called_once()
- mock_disconnect_controller.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
def test_parsing_error_exception(
self,
with self.assertRaises(juju.offerendpoints.ParseError):
self.loop.run_until_complete(
- self.libjuju.consume("offer_url", "model_name")
+ self.libjuju.consume(
+ "model_name", Offer(self.offer_url), self.provider_libjuju
+ )
)
mock_consume.assert_called_once()
mock_disconnect_model.assert_called_once()
- mock_disconnect_controller.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
def test_juju_error_exception(
self,
with self.assertRaises(juju.errors.JujuError):
self.loop.run_until_complete(
- self.libjuju.consume("offer_url", "model_name")
+ self.libjuju.consume(
+ "model_name", Offer(self.offer_url), self.provider_libjuju
+ )
)
mock_consume.assert_called_once()
mock_disconnect_model.assert_called_once()
- mock_disconnect_controller.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
def test_juju_api_error_exception(
self,
with self.assertRaises(juju.errors.JujuAPIError):
self.loop.run_until_complete(
- self.libjuju.consume("offer_url", "model_name")
+ self.libjuju.consume(
+ "model_name", Offer(self.offer_url), self.provider_libjuju
+ )
)
mock_consume.assert_called_once()
mock_disconnect_model.assert_called_once()
- mock_disconnect_controller.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
import asynctest
+from n2vc.definitions import Offer, RelationEndpoint
from n2vc.n2vc_juju_conn import N2VCJujuConnector
from osm_common import fslocal
from n2vc.exceptions import (
)
)
self.assertIsNone(ee_id)
+
+
+class AddRelationTest(N2VCJujuConnTestCase):
+ def setUp(self):
+ super(AddRelationTest, self).setUp()
+ self.n2vc.libjuju.add_relation = AsyncMock()
+ self.n2vc.libjuju.offer = AsyncMock()
+ self.n2vc.libjuju.get_controller = AsyncMock()
+ self.n2vc.libjuju.consume = AsyncMock()
+
+ def test_standard_relation(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-1.app2.1", None, "endpoint")
+ self.loop.run_until_complete(
+ self.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2)
+ )
+ self.n2vc.libjuju.add_relation.assert_called_once_with(
+ model_name="model-1", endpoint_1="app1:endpoint", endpoint_2="app2:endpoint"
+ )
+ self.n2vc.libjuju.offer.assert_not_called()
+ self.n2vc.libjuju.consume.assert_not_called()
+
+ def test_cmr_relation_same_controller(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
+ offer = Offer("admin/model-1.app1")
+ self.n2vc.libjuju.offer.return_value = offer
+ self.n2vc.libjuju.consume.return_value = "saas"
+ self.loop.run_until_complete(
+ self.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2)
+ )
+ self.n2vc.libjuju.offer.assert_called_once_with(relation_endpoint_1)
+ self.n2vc.libjuju.consume.assert_called_once()
+ self.n2vc.libjuju.add_relation.assert_called_once_with(
+ "model-2", "app2:endpoint", "saas"
+ )
+
+ def test_relation_exception(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
+ self.n2vc.libjuju.offer.side_effect = Exception()
+ with self.assertRaises(N2VCException):
+ self.loop.run_until_complete(
+ self.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2)
+ )
from unittest import TestCase
-from n2vc.utils import Dict, EntityType, JujuStatusToOSM, N2VCDeploymentStatus
+from n2vc.utils import (
+ Dict,
+ EntityType,
+ JujuStatusToOSM,
+ N2VCDeploymentStatus,
+ get_ee_id_components,
+)
from juju.machine import Machine
from juju.application import Application
from juju.action import Action
osm_status = status["osm"]
self.assertTrue(juju_status in JujuStatusToOSM[entity_type])
self.assertEqual(osm_status, JujuStatusToOSM[entity_type][juju_status])
+
+
+class GetEEComponentTest(TestCase):
+ def test_valid(self):
+ model, application, machine = get_ee_id_components("model.application.machine")
+ self.assertEqual(model, "model")
+ self.assertEqual(application, "application")
+ self.assertEqual(machine, "machine")
+
+ def test_invalid(self):
+ with self.assertRaises(Exception):
+ get_ee_id_components("model.application.machine.1")
+ with self.assertRaises(Exception):
+ get_ee_id_components("model.application")
from juju.action import Action
from juju.unit import Unit
from n2vc.exceptions import N2VCInvalidCertificate
+from typing import Tuple
def base64_to_cacert(b64string):
yaml_text = obj_to_yaml(obj)
# parse to dict
return yaml.load(yaml_text, Loader=yaml.Loader)
+
+
+def get_ee_id_components(ee_id: str) -> Tuple[str, str, str]:
+ """
+ Get model, application and machine components from an execution environment id
+ :param ee_id:
+ :return: model_name, application_name, machine_id
+ """
+ parts = ee_id.split(".")
+ if len(parts) != 3:
+ raise Exception("invalid ee id.")
+ model_name = parts[0]
+ application_name = parts[1]
+ machine_id = parts[2]
+ return model_name, application_name, machine_id
projects / osm / N2VC.git / commitdiff