from n2vc.k8s_juju_conn import K8sJujuConnector, RBAC_LABEL_KEY_NAME
from osm_common import fslocal
from .utils import kubeconfig, FakeModel, FakeFileWrapper, AsyncMock, FakeApplication
-from n2vc.exceptions import (
- MethodNotImplemented,
- K8sException,
-)
+from n2vc.exceptions import MethodNotImplemented, K8sException
from n2vc.vca.connection_data import ConnectionData
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
- self.cs_bundle,
- model_name=self.default_namespace,
- wait=True,
- timeout=1800,
+ self.cs_bundle, model_name=self.default_namespace, wait=True, timeout=1800
)
def test_success_http(self, mock_chdir):
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
- self.http_bundle,
- model_name=self.default_namespace,
- wait=True,
- timeout=1800,
+ self.http_bundle, model_name=self.default_namespace, wait=True, timeout=1800
)
def test_success_not_kdu_name(self, mock_chdir):
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
- self.cs_bundle,
- model_name=self.default_namespace,
- wait=True,
- timeout=1800,
+ self.cs_bundle, model_name=self.default_namespace, wait=True, timeout=1800
)
def test_missing_db_dict(self, mock_chdir):
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
- self.cs_bundle,
- model_name=self.default_namespace,
- wait=True,
- timeout=1800,
+ self.cs_bundle, model_name=self.default_namespace, wait=True, timeout=1800
)
def test_missing_bundle(self, mock_chdir):
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")
+ def test_standard_relation_same_model_and_controller(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint1")
+ relation_endpoint_2 = RelationEndpoint("model-1.app2.1", None, "endpoint2")
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"
+ model_name="model-1",
+ endpoint_1="app1:endpoint1",
+ endpoint_2="app2:endpoint2",
)
self.k8s_juju_conn.libjuju.offer.assert_not_called()
self.k8s_juju_conn.libjuju.consume.assert_not_called()
"model-2", "app2:endpoint", "saas"
)
+ def test_cmr_relation_different_controller(self):
+ self.k8s_juju_conn._get_libjuju = AsyncMock(
+ return_value=self.k8s_juju_conn.libjuju
+ )
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", "vca-id-1", "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-1.app2.1", "vca-id-2", "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-1", "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")