import asyncio
import logging
+from unittest.mock import Mock
+from unittest.mock import patch
+
+
import asynctest
+from n2vc.definitions import Offer, RelationEndpoint
from n2vc.n2vc_juju_conn import N2VCJujuConnector
from osm_common import fslocal
from n2vc.exceptions import (
- JujuK8sProxycharmNotSupported,
N2VCBadArgumentsException,
N2VCException,
)
+from n2vc.tests.unit.utils import AsyncMock
+from n2vc.vca.connection_data import ConnectionData
class N2VCJujuConnTestCase(asynctest.TestCase):
- @asynctest.mock.patch("n2vc.libjuju.Libjuju._create_health_check_task")
- @asynctest.mock.patch("juju.controller.Controller.update_endpoints")
- @asynctest.mock.patch("juju.client.connector.Connector.connect")
- @asynctest.mock.patch("juju.controller.Controller.connection")
- @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
+ @asynctest.mock.patch("n2vc.n2vc_juju_conn.MotorStore")
+ @asynctest.mock.patch("n2vc.n2vc_juju_conn.get_connection")
+ @asynctest.mock.patch("n2vc.vca.connection_data.base64_to_cacert")
def setUp(
self,
- mock__get_api_endpoints_db=None,
- mock_connection=None,
- mock_connect=None,
- mock_update_endpoints=None,
- mock__create_health_check_task=None,
+ mock_base64_to_cacert=None,
+ mock_get_connection=None,
+ mock_store=None,
):
- mock__get_api_endpoints_db.return_value = ["2.2.2.2:17070"]
- loop = asyncio.get_event_loop()
- db = {}
- vca_config = {
- "secret": "secret",
- "api_proxy": "api_proxy",
- "cloud": "cloud",
- "k8s_cloud": "k8s_cloud",
- }
-
+ self.loop = asyncio.get_event_loop()
+ self.db = Mock()
+ mock_base64_to_cacert.return_value = """
+ -----BEGIN CERTIFICATE-----
+ SOMECERT
+ -----END CERTIFICATE-----"""
+ mock_store.return_value = AsyncMock()
+ mock_vca_connection = Mock()
+ mock_get_connection.return_value = mock_vca_connection
+ mock_vca_connection.data.return_value = ConnectionData(
+ **{
+ "endpoints": ["1.2.3.4:17070"],
+ "user": "user",
+ "secret": "secret",
+ "cacert": "cacert",
+ "pubkey": "pubkey",
+ "lxd-cloud": "cloud",
+ "lxd-credentials": "credentials",
+ "k8s-cloud": "k8s_cloud",
+ "k8s-credentials": "k8s_credentials",
+ "model-config": {},
+ "api-proxy": "api_proxy",
+ }
+ )
logging.disable(logging.CRITICAL)
+ N2VCJujuConnector.get_public_key = Mock()
self.n2vc = N2VCJujuConnector(
- db=db,
+ db=self.db,
fs=fslocal.FsLocal(),
log=None,
- loop=loop,
- url="2.2.2.2:17070",
- username="admin",
- vca_config=vca_config,
+ loop=self.loop,
on_update_db=None,
)
+ N2VCJujuConnector.get_public_key.assert_not_called()
+ self.n2vc.libjuju = Mock()
-@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_metrics")
class GetMetricssTest(N2VCJujuConnTestCase):
def setUp(self):
super(GetMetricssTest, self).setUp()
+ self.n2vc.libjuju.get_metrics = AsyncMock()
- def test_success(self, mock_get_metrics):
+ def test_success(self):
_ = self.loop.run_until_complete(self.n2vc.get_metrics("model", "application"))
- mock_get_metrics.assert_called_once()
+ self.n2vc.libjuju.get_metrics.assert_called_once()
- def test_except(self, mock_get_metrics):
- mock_get_metrics.side_effect = Exception()
+ def test_except(self):
+ self.n2vc.libjuju.get_metrics.side_effect = Exception()
with self.assertRaises(Exception):
_ = self.loop.run_until_complete(
self.n2vc.get_metrics("model", "application")
)
- mock_get_metrics.assert_called_once()
+ self.n2vc.libjuju.get_metrics.assert_called_once()
-@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
-@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
-@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_executed_actions")
-@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_actions")
-@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_application_configs")
-@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
class UpdateVcaStatusTest(N2VCJujuConnTestCase):
def setUp(self):
super(UpdateVcaStatusTest, self).setUp()
+ self.n2vc.libjuju.get_controller = AsyncMock()
+ self.n2vc.libjuju.get_model = AsyncMock()
+ self.n2vc.libjuju.get_executed_actions = AsyncMock()
+ self.n2vc.libjuju.get_actions = AsyncMock()
+ self.n2vc.libjuju.get_application_configs = AsyncMock()
+ self.n2vc.libjuju._get_application = AsyncMock()
def test_success(
self,
- mock_get_application,
- mock_get_application_configs,
- mock_get_actions,
- mock_get_executed_actions,
- mock_get_model,
- mock_get_controller,
):
- self.loop.run_until_complete(self.n2vc.update_vca_status(
- {"model": {"applications": {"app": {"actions": {}}}}}))
- mock_get_executed_actions.assert_called_once()
- mock_get_actions.assert_called_once()
- mock_get_application_configs.assert_called_once()
+ self.loop.run_until_complete(
+ self.n2vc.update_vca_status(
+ {"model": {"applications": {"app": {"actions": {}}}}}
+ )
+ )
+ self.n2vc.libjuju.get_executed_actions.assert_called_once()
+ self.n2vc.libjuju.get_actions.assert_called_once()
+ self.n2vc.libjuju.get_application_configs.assert_called_once()
- def test_exception(
- self,
- mock_get_application,
- mock_get_application_configs,
- mock_get_actions,
- mock_get_executed_actions,
- mock_get_model,
- mock_get_controller,
- ):
- mock_get_model.return_value = None
- mock_get_executed_actions.side_effect = Exception()
+ def test_exception(self):
+ self.n2vc.libjuju.get_model.return_value = None
+ self.n2vc.libjuju.get_executed_actions.side_effect = Exception()
with self.assertRaises(Exception):
- self.loop.run_until_complete(self.n2vc.update_vca_status(
- {"model": {"applications": {"app": {"actions": {}}}}}))
- mock_get_executed_actions.assert_not_called()
- mock_get_actions.assert_not_called_once()
- mock_get_application_configs.assert_not_called_once()
+ self.loop.run_until_complete(
+ self.n2vc.update_vca_status(
+ {"model": {"applications": {"app": {"actions": {}}}}}
+ )
+ )
+ self.n2vc.libjuju.get_executed_actions.assert_not_called()
+ self.n2vc.libjuju.get_actions.assert_not_called_once()
+ self.n2vc.libjuju.get_application_configs.assert_not_called_once()
-@asynctest.mock.patch("n2vc.libjuju.Libjuju.model_exists")
@asynctest.mock.patch("osm_common.fslocal.FsLocal.file_exists")
@asynctest.mock.patch(
"osm_common.fslocal.FsLocal.path", new_callable=asynctest.PropertyMock, create=True
)
-@asynctest.mock.patch("n2vc.libjuju.Libjuju.deploy_charm")
-@asynctest.mock.patch("n2vc.libjuju.Libjuju.add_model")
class K8sProxyCharmsTest(N2VCJujuConnTestCase):
def setUp(self):
super(K8sProxyCharmsTest, self).setUp()
+ self.n2vc.libjuju.model_exists = AsyncMock()
+ self.n2vc.libjuju.add_model = AsyncMock()
+ self.n2vc.libjuju.deploy_charm = AsyncMock()
+ self.n2vc.libjuju.model_exists.return_value = False
+ @patch(
+ "n2vc.n2vc_juju_conn.generate_random_alfanum_string",
+ **{"return_value": "random"}
+ )
def test_success(
- self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists
+ self,
+ mock_generate_random_alfanum_string,
+ mock_path,
+ mock_file_exists,
):
- mock_model_exists.return_value = None
mock_file_exists.return_value = True
mock_path.return_value = "/path"
ee_id = self.loop.run_until_complete(
self.n2vc.install_k8s_proxy_charm(
- "charm", "nsi-id.ns-id.vnf-id.vdu", "////path/", {},
+ "charm",
+ "nsi-id.ns-id.vnf-id.vdu",
+ "////path/",
+ {},
)
)
- mock_add_model.assert_called_once_with(
- "ns-id-k8s",
- cloud_name=self.n2vc.k8s_cloud,
- credential_name=self.n2vc.k8s_cloud
- )
- mock_deploy_charm.assert_called_once_with(
+ self.n2vc.libjuju.add_model.assert_called_once()
+ self.n2vc.libjuju.deploy_charm.assert_called_once_with(
model_name="ns-id-k8s",
- application_name="app-vnf-vnf-id-vdu-vdu",
+ application_name="app-vnf-vnf-id-vdu-vdu-random",
path="/path/path/",
machine_id=None,
db_dict={},
total_timeout=None,
config=None,
)
- self.assertEqual(ee_id, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu.k8s")
+ self.assertEqual(ee_id, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu-random.k8s")
- @asynctest.mock.patch(
- "n2vc.n2vc_juju_conn.N2VCJujuConnector.k8s_cloud",
- new_callable=asynctest.PropertyMock,
- create=True,
- )
- def test_no_k8s_cloud(
+ def test_no_artifact_path(
self,
- mock_k8s_cloud,
- mock_add_model,
- mock_deploy_charm,
mock_path,
mock_file_exists,
- mock_model_exists,
- ):
- mock_k8s_cloud.return_value = None
- with self.assertRaises(JujuK8sProxycharmNotSupported):
- ee_id = self.loop.run_until_complete(
- self.n2vc.install_k8s_proxy_charm(
- "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {},
- )
- )
- self.assertIsNone(ee_id)
-
- def test_no_artifact_path(
- self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
):
with self.assertRaises(N2VCBadArgumentsException):
ee_id = self.loop.run_until_complete(
self.n2vc.install_k8s_proxy_charm(
- "charm", "nsi-id.ns-id.vnf-id.vdu", "", {},
+ "charm",
+ "nsi-id.ns-id.vnf-id.vdu",
+ "",
+ {},
)
)
self.assertIsNone(ee_id)
def test_no_db(
- self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
+ self,
+ mock_path,
+ mock_file_exists,
):
with self.assertRaises(N2VCBadArgumentsException):
ee_id = self.loop.run_until_complete(
self.n2vc.install_k8s_proxy_charm(
- "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", None,
+ "charm",
+ "nsi-id.ns-id.vnf-id.vdu",
+ "/path/",
+ None,
)
)
self.assertIsNone(ee_id)
def test_file_not_exists(
- self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
+ self,
+ mock_path,
+ mock_file_exists,
):
mock_file_exists.return_value = False
with self.assertRaises(N2VCBadArgumentsException):
ee_id = self.loop.run_until_complete(
self.n2vc.install_k8s_proxy_charm(
- "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {},
+ "charm",
+ "nsi-id.ns-id.vnf-id.vdu",
+ "/path/",
+ {},
)
)
self.assertIsNone(ee_id)
def test_exception(
- self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
+ self,
+ mock_path,
+ mock_file_exists,
):
- mock_model_exists.return_value = None
mock_file_exists.return_value = True
mock_path.return_value = "/path"
- mock_deploy_charm.side_effect = Exception()
+ self.n2vc.libjuju.deploy_charm.side_effect = Exception()
with self.assertRaises(N2VCException):
ee_id = self.loop.run_until_complete(
self.n2vc.install_k8s_proxy_charm(
- "charm", "nsi-id.ns-id.vnf-id.vdu", "path/", {},
+ "charm",
+ "nsi-id.ns-id.vnf-id.vdu",
+ "path/",
+ {},
)
)
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)
+ )
projects / osm / N2VC.git / blobdiff