Fix Bug 2199 Fixing ns update operation for KNF instances
This patch enables ns update operations on KNF instances which are deployed with Helm chart
and using .charm zipped files.
Change-Id: I6c598868ae81243ef598ae5b7df88e26a5b870c0
Signed-off-by: aticig <gulsum.atici@canonical.com>
diff --git a/osm_lcm/tests/test_ns.py b/osm_lcm/tests/test_ns.py
index 2004701..56d7ffa 100644
--- a/osm_lcm/tests/test_ns.py
+++ b/osm_lcm/tests/test_ns.py
@@ -18,6 +18,7 @@
import asynctest # pip3 install asynctest --user
import asyncio
+from copy import deepcopy
import yaml
import copy
from os import getenv
@@ -75,6 +76,18 @@
"ng": True,
},
}
+nsr_id = descriptors.test_ids["TEST-A"]["ns"]
+nslcmop_id = descriptors.test_ids["TEST-A"]["update"]
+vnfr_id = "6421c7c9-d865-4fb4-9a13-d4275d243e01"
+vnfd_id = "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"
+update_fs = Mock(autospec=True)
+update_fs.path.__add__ = Mock()
+update_fs.path.side_effect = ["/", "/", "/", "/"]
+update_fs.sync.side_effect = [None, None]
+
+
+def callable(a):
+ return a
class TestMyNS(asynctest.TestCase):
@@ -87,7 +100,7 @@
params={},
machine_spec={},
callback=None,
- *callback_args
+ *callback_args,
):
if callback:
for status, message in (
@@ -943,747 +956,570 @@
# self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
# self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
- # Test update method
+ @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
+ @patch(
+ "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable=asynctest.Mock(autospec=True)
+ )
+ @patch("osm_lcm.data_utils.vnfd.find_software_version")
+ @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
+ async def test_update_change_vnfpkg_sw_version_not_changed(
+ self,
+ mock_juju_bundle,
+ mock_software_version,
+ mock_charm_upgrade,
+ mock_charm_hash,
+ ):
+ """Update type: CHANGE_VNFPKG, latest_vnfd revision changed,
+ Charm package changed, sw-version is not changed"""
+ self.db.set_one(
+ "vnfds",
+ q_filter={"_id": vnfd_id},
+ update_dict={"_admin.revision": 3, "kdu": []},
+ )
- async def test_update(self):
+ self.db.set_one(
+ "vnfds_revisions",
+ q_filter={"_id": vnfd_id + ":1"},
+ update_dict={"_admin.revision": 1, "kdu": []},
+ )
- nsr_id = descriptors.test_ids["TEST-A"]["ns"]
- nslcmop_id = descriptors.test_ids["TEST-A"]["update"]
- vnfr_id = "6421c7c9-d865-4fb4-9a13-d4275d243e01"
- vnfd_id = "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"
+ self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1})
- def mock_reset():
- mock_charm_hash.reset_mock()
- mock_juju_bundle.reset_mock()
- fs.sync.reset_mock()
- mock_charm_upgrade.reset_mock()
- mock_software_version.reset_mock()
+ mock_charm_hash.return_value = True
+ mock_software_version.side_effect = ["1.0", "1.0"]
- with self.subTest(
- i=1,
- t="Update type: CHANGE_VNFPKG, latest_vnfd revision changed,"
- "Charm package changed, sw-version is not changed.",
- ):
+ task = asyncio.Future()
+ task.set_result(("COMPLETED", "some_output"))
+ mock_charm_upgrade.return_value = task
- self.db.set_one(
- "vnfds",
- q_filter={"_id": vnfd_id},
- update_dict={"_admin.revision": 3, "kdu": []},
- )
+ instance = self.my_ns
+ fs = deepcopy(update_fs)
+ instance.fs = fs
- self.db.set_one(
- "vnfds_revisions",
- q_filter={"_id": vnfd_id + ":1"},
- update_dict={"_admin.revision": 1, "kdu": []},
- )
+ expected_operation_state = "COMPLETED"
+ expected_operation_error = ""
+ expected_vnfr_revision = 3
+ expected_ns_state = "INSTANTIATED"
+ expected_ns_operational_state = "running"
- self.db.set_one(
- "vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1}
- )
+ await instance.update(nsr_id, nslcmop_id)
+ return_operation_state = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
+ "operationState"
+ )
+ return_operation_error = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
+ "errorMessage"
+ )
+ return_ns_operational_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
+ "operational-status"
+ )
+ return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
+ "revision"
+ )
+ return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get("nsState")
+ mock_charm_hash.assert_called_with(
+ f"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
+ f"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
+ )
+ self.assertEqual(fs.sync.call_count, 2)
+ self.assertEqual(return_ns_state, expected_ns_state)
+ self.assertEqual(return_operation_state, expected_operation_state)
+ self.assertEqual(return_operation_error, expected_operation_error)
+ self.assertEqual(return_ns_operational_state, expected_ns_operational_state)
+ self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
- mock_charm_hash = Mock(autospec=True)
- mock_charm_hash.return_value = True
+ @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
+ @patch(
+ "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable=asynctest.Mock(autospec=True)
+ )
+ @patch("osm_lcm.data_utils.vnfd.find_software_version")
+ @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
+ async def test_update_change_vnfpkg_vnfd_revision_not_changed(
+ self,
+ mock_juju_bundle,
+ mock_software_version,
+ mock_charm_upgrade,
+ mock_charm_hash,
+ ):
+ """Update type: CHANGE_VNFPKG, latest_vnfd revision not changed"""
+ self.db.set_one(
+ "vnfds", q_filter={"_id": vnfd_id}, update_dict={"_admin.revision": 1}
+ )
+ self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1})
- mock_juju_bundle = Mock(return_value=None)
+ mock_charm_hash.return_value = True
- mock_software_version = Mock(autospec=True)
- mock_software_version.side_effect = ["1.0", "1.0"]
+ task = asyncio.Future()
+ task.set_result(("COMPLETED", "some_output"))
+ mock_charm_upgrade.return_value = task
- mock_charm_upgrade = asynctest.Mock(autospec=True)
- task = asyncio.Future()
- task.set_result(("COMPLETED", "some_output"))
- mock_charm_upgrade.return_value = task
+ instance = self.my_ns
- fs = Mock(autospec=True)
- fs.path.__add__ = Mock()
- fs.path.side_effect = ["/", "/", "/", "/"]
- fs.sync.side_effect = [None, None]
+ expected_operation_state = "COMPLETED"
+ expected_operation_error = ""
+ expected_vnfr_revision = 1
+ expected_ns_state = "INSTANTIATED"
+ expected_ns_operational_state = "running"
- instance = self.my_ns
+ await instance.update(nsr_id, nslcmop_id)
+ return_operation_state = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
+ "operationState"
+ )
+ return_operation_error = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
+ "errorMessage"
+ )
+ return_ns_operational_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
+ "operational-status"
+ )
+ return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get("nsState")
+ return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
+ "revision"
+ )
+ mock_charm_hash.assert_not_called()
+ mock_software_version.assert_not_called()
+ mock_juju_bundle.assert_not_called()
+ mock_charm_upgrade.assert_not_called()
+ update_fs.sync.assert_not_called()
+ self.assertEqual(return_ns_state, expected_ns_state)
+ self.assertEqual(return_operation_state, expected_operation_state)
+ self.assertEqual(return_operation_error, expected_operation_error)
+ self.assertEqual(return_ns_operational_state, expected_ns_operational_state)
+ self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
- expected_operation_state = "COMPLETED"
- expected_operation_error = ""
- expected_vnfr_revision = 3
- expected_ns_state = "INSTANTIATED"
- expected_ns_operational_state = "running"
+ @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
+ @patch(
+ "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable=asynctest.Mock(autospec=True)
+ )
+ @patch("osm_lcm.data_utils.vnfd.find_software_version")
+ @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
+ async def test_update_change_vnfpkg_charm_is_not_changed(
+ self,
+ mock_juju_bundle,
+ mock_software_version,
+ mock_charm_upgrade,
+ mock_charm_hash,
+ ):
+ """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
+ Charm package is not changed, sw-version is not changed"""
+ self.db.set_one(
+ "vnfds",
+ q_filter={"_id": vnfd_id},
+ update_dict={"_admin.revision": 3, "kdu": []},
+ )
+ self.db.set_one(
+ "vnfds_revisions",
+ q_filter={"_id": vnfd_id + ":1"},
+ update_dict={"_admin.revision": 1, "kdu": []},
+ )
+ self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1})
- with patch.object(instance, "fs", fs), patch(
- "osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed", mock_charm_hash
- ), patch("osm_lcm.ns.NsLcm._ns_charm_upgrade", mock_charm_upgrade), patch(
- "osm_lcm.data_utils.vnfd.find_software_version", mock_software_version
- ), patch(
- "osm_lcm.lcm_utils.check_juju_bundle_existence", mock_juju_bundle
- ):
+ mock_charm_hash.return_value = False
+ mock_software_version.side_effect = ["1.0", "1.0"]
- await instance.update(nsr_id, nslcmop_id)
- return_operation_state = self.db.get_one(
- "nslcmops", {"_id": nslcmop_id}
- ).get("operationState")
- return_operation_error = self.db.get_one(
- "nslcmops", {"_id": nslcmop_id}
- ).get("errorMessage")
- return_ns_operational_state = self.db.get_one(
- "nsrs", {"_id": nsr_id}
- ).get("operational-status")
+ instance = self.my_ns
+ fs = deepcopy(update_fs)
+ instance.fs = fs
+ expected_operation_state = "COMPLETED"
+ expected_operation_error = ""
+ expected_vnfr_revision = 3
+ expected_ns_state = "INSTANTIATED"
+ expected_ns_operational_state = "running"
- return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
- "revision"
- )
+ await instance.update(nsr_id, nslcmop_id)
+ return_operation_state = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
+ "operationState"
+ )
+ return_operation_error = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
+ "errorMessage"
+ )
+ return_ns_operational_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
+ "operational-status"
+ )
+ return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
+ "revision"
+ )
+ return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get("nsState")
+ mock_charm_hash.assert_called_with(
+ f"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
+ f"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
+ )
+ self.assertEqual(fs.sync.call_count, 2)
+ self.assertEqual(mock_charm_hash.call_count, 1)
+ mock_juju_bundle.assert_not_called()
+ mock_charm_upgrade.assert_not_called()
+ self.assertEqual(return_ns_state, expected_ns_state)
+ self.assertEqual(return_operation_state, expected_operation_state)
+ self.assertEqual(return_operation_error, expected_operation_error)
+ self.assertEqual(return_ns_operational_state, expected_ns_operational_state)
+ self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
- return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
- "nsState"
- )
+ @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
+ @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
+ @patch(
+ "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable=asynctest.Mock(autospec=True)
+ )
+ @patch("osm_lcm.lcm_utils.get_charm_artifact_path")
+ async def test_update_change_vnfpkg_sw_version_changed(
+ self, mock_charm_artifact, mock_charm_upgrade, mock_charm_hash, mock_juju_bundle
+ ):
+ """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
+ Charm package exists, sw-version changed."""
+ self.db.set_one(
+ "vnfds",
+ q_filter={"_id": vnfd_id},
+ update_dict={"_admin.revision": 3, "software-version": "3.0", "kdu": []},
+ )
+ self.db.set_one(
+ "vnfds_revisions",
+ q_filter={"_id": vnfd_id + ":1"},
+ update_dict={"_admin.revision": 1, "kdu": []},
+ )
+ self.db.set_one(
+ "vnfrs",
+ q_filter={"_id": vnfr_id},
+ update_dict={"revision": 1},
+ )
+ mock_charm_hash.return_value = False
- mock_charm_hash.assert_called_with(
- "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:1/hackfest_3charmed_vnfd/charms/simple",
- "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:3/hackfest_3charmed_vnfd/charms/simple",
- )
+ mock_charm_artifact.side_effect = [
+ f"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
+ f"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
+ ]
- self.assertEqual(fs.sync.call_count, 2)
- self.assertEqual(return_ns_state, expected_ns_state)
- self.assertEqual(return_operation_state, expected_operation_state)
- self.assertEqual(return_operation_error, expected_operation_error)
- self.assertEqual(
- return_ns_operational_state, expected_ns_operational_state
- )
- self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
+ instance = self.my_ns
+ fs = deepcopy(update_fs)
+ instance.fs = fs
+ expected_operation_state = "FAILED"
+ expected_operation_error = "FAILED Checking if existing VNF has charm: Software version change is not supported as VNF instance 6421c7c9-d865-4fb4-9a13-d4275d243e01 has charm."
+ expected_vnfr_revision = 1
+ expected_ns_state = "INSTANTIATED"
+ expected_ns_operational_state = "running"
- mock_reset()
+ await instance.update(nsr_id, nslcmop_id)
+ return_operation_state = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
+ "operationState"
+ )
+ return_operation_error = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
+ "errorMessage"
+ )
+ return_ns_operational_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
+ "operational-status"
+ )
+ return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
+ "revision"
+ )
+ return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get("nsState")
+ self.assertEqual(fs.sync.call_count, 2)
+ mock_charm_hash.assert_not_called()
+ mock_juju_bundle.assert_not_called()
+ mock_charm_upgrade.assert_not_called()
+ self.assertEqual(return_ns_state, expected_ns_state)
+ self.assertEqual(return_operation_state, expected_operation_state)
+ self.assertEqual(return_operation_error, expected_operation_error)
+ self.assertEqual(return_ns_operational_state, expected_ns_operational_state)
+ self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
- with self.subTest(
- i=2, t="Update type: CHANGE_VNFPKG, latest_vnfd revision not changed"
- ):
+ @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
+ @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
+ @patch(
+ "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable=asynctest.Mock(autospec=True)
+ )
+ @patch("osm_lcm.data_utils.vnfd.find_software_version")
+ async def test_update_change_vnfpkg_juju_bundle_exists(
+ self,
+ mock_software_version,
+ mock_charm_upgrade,
+ mock_charm_hash,
+ mock_juju_bundle,
+ ):
+ """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
+ Charm package exists, sw-version not changed, juju-bundle exists"""
+ # Upgrade is not allowed with juju bundles, this will cause TypeError
+ self.db.set_one(
+ "vnfds",
+ q_filter={"_id": vnfd_id},
+ update_dict={
+ "_admin.revision": 5,
+ "software-version": "1.0",
+ "kdu": [{"kdu_name": "native-kdu", "juju-bundle": "stable/native-kdu"}],
+ },
+ )
+ self.db.set_one(
+ "vnfds_revisions",
+ q_filter={"_id": vnfd_id + ":1"},
+ update_dict={
+ "_admin.revision": 1,
+ "software-version": "1.0",
+ "kdu": [{"kdu_name": "native-kdu", "juju-bundle": "stable/native-kdu"}],
+ },
+ )
+ self.db.set_one(
+ "nsrs",
+ q_filter={"_id": nsr_id},
+ update_dict={
+ "_admin.deployed.VCA.0.kdu_name": "native-kdu",
+ },
+ )
+ self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1})
- self.db.set_one(
- "vnfds", q_filter={"_id": vnfd_id}, update_dict={"_admin.revision": 1}
- )
+ mock_charm_hash.side_effect = [True]
+ mock_software_version.side_effect = ["1.0", "1.0"]
+ mock_juju_bundle.return_value = True
+ instance = self.my_ns
+ fs = deepcopy(update_fs)
+ instance.fs = fs
- self.db.set_one(
- "vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1}
- )
+ expected_vnfr_revision = 1
+ expected_ns_state = "INSTANTIATED"
+ expected_ns_operational_state = "running"
- mock_charm_hash = Mock(autospec=True)
- mock_charm_hash.return_value = True
+ await instance.update(nsr_id, nslcmop_id)
+ return_ns_operational_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
+ "operational-status"
+ )
+ return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
+ "revision"
+ )
+ return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get("nsState")
+ self.assertEqual(fs.sync.call_count, 2)
+ mock_charm_upgrade.assert_not_called()
+ mock_charm_hash.assert_not_called()
+ self.assertEqual(return_ns_state, expected_ns_state)
+ self.assertEqual(return_ns_operational_state, expected_ns_operational_state)
+ self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
- mock_juju_bundle = Mock(return_value=None)
- mock_software_version = Mock(autospec=True)
+ @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
+ @patch(
+ "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable=asynctest.Mock(autospec=True)
+ )
+ async def test_update_change_vnfpkg_charm_upgrade_failed(
+ self, mock_charm_upgrade, mock_charm_hash
+ ):
+ """ "Update type: CHANGE_VNFPKG, latest_vnfd revision changed"
+ Charm package exists, sw-version not changed, charm-upgrade failed"""
+ self.db.set_one(
+ "vnfds",
+ q_filter={"_id": vnfd_id},
+ update_dict={
+ "_admin.revision": 3,
+ "software-version": "1.0",
+ "kdu": [],
+ },
+ )
+ self.db.set_one(
+ "vnfds_revisions",
+ q_filter={"_id": vnfd_id + ":1"},
+ update_dict={
+ "_admin.revision": 1,
+ "software-version": "1.0",
+ "kdu": [],
+ },
+ )
+ self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1})
- mock_charm_upgrade = asynctest.Mock(autospec=True)
- task = asyncio.Future()
- task.set_result(("COMPLETED", "some_output"))
- mock_charm_upgrade.return_value = task
+ mock_charm_hash.return_value = True
- fs = Mock(autospec=True)
- fs.path.__add__ = Mock()
- fs.path.side_effect = ["/", "/", "/", "/"]
- fs.sync.side_effect = [None, None]
+ task = asyncio.Future()
+ task.set_result(("FAILED", "some_error"))
+ mock_charm_upgrade.return_value = task
- instance = self.my_ns
+ instance = self.my_ns
+ fs = deepcopy(update_fs)
+ instance.fs = fs
+ expected_operation_state = "FAILED"
+ expected_operation_error = "some_error"
+ expected_vnfr_revision = 1
+ expected_ns_state = "INSTANTIATED"
+ expected_ns_operational_state = "running"
- expected_operation_state = "COMPLETED"
- expected_operation_error = ""
- expected_vnfr_revision = 1
- expected_ns_state = "INSTANTIATED"
- expected_ns_operational_state = "running"
+ await instance.update(nsr_id, nslcmop_id)
+ return_operation_state = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
+ "operationState"
+ )
+ return_operation_error = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
+ "errorMessage"
+ )
+ return_ns_operational_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
+ "operational-status"
+ )
+ return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
+ "revision"
+ )
+ return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get("nsState")
+ self.assertEqual(fs.sync.call_count, 2)
+ self.assertEqual(mock_charm_hash.call_count, 1)
+ self.assertEqual(mock_charm_upgrade.call_count, 1)
+ self.assertEqual(return_ns_state, expected_ns_state)
+ self.assertEqual(return_operation_state, expected_operation_state)
+ self.assertEqual(return_operation_error, expected_operation_error)
+ self.assertEqual(return_ns_operational_state, expected_ns_operational_state)
+ self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
- with patch.object(instance, "fs", fs), patch(
- "osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed", mock_charm_hash
- ), patch("osm_lcm.ns.NsLcm._ns_charm_upgrade", mock_charm_upgrade), patch(
- "osm_lcm.lcm_utils.check_juju_bundle_existence", mock_juju_bundle
- ):
+ def test_ns_update_find_sw_version_vnfd_not_includes(self):
+ """Find software version, VNFD does not have software version"""
- await instance.update(nsr_id, nslcmop_id)
+ db_vnfd = self.db.get_one("vnfds", {"_id": vnfd_id})
+ expected_result = "1.0"
+ result = find_software_version(db_vnfd)
+ self.assertEqual(result, expected_result, "Default sw version should be 1.0")
- return_operation_state = self.db.get_one(
- "nslcmops", {"_id": nslcmop_id}
- ).get("operationState")
+ def test_ns_update_find_sw_version_vnfd_includes(self):
+ """Find software version, VNFD includes software version"""
- return_operation_error = self.db.get_one(
- "nslcmops", {"_id": nslcmop_id}
- ).get("errorMessage")
+ db_vnfd = self.db.get_one("vnfds", {"_id": vnfd_id})
+ db_vnfd["software-version"] = "3.1"
+ expected_result = "3.1"
+ result = find_software_version(db_vnfd)
+ self.assertEqual(result, expected_result, "VNFD software version is wrong")
- return_ns_operational_state = self.db.get_one(
- "nsrs", {"_id": nsr_id}
- ).get("operational-status")
+ @patch("os.path.exists")
+ @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
+ @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
+ def test_ns_update_check_charm_hash_not_changed(
+ self, mock_compare_charm_hash, mock_compare_charmdir_hash, mock_path_exists
+ ):
+ """Check charm hash, Hash did not change"""
- return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
- "nsState"
- )
+ current_path, target_path = "/tmp/charm1", "/tmp/charm1"
- return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
- "revision"
- )
+ fs = Mock()
+ fs.path.__add__ = Mock()
+ fs.path.side_effect = [current_path, target_path]
+ fs.path.__add__.side_effect = [current_path, target_path]
- mock_charm_hash.assert_not_called()
- mock_software_version.assert_not_called()
- mock_juju_bundle.assert_not_called()
- mock_charm_upgrade.assert_not_called()
- fs.sync.assert_not_called()
+ mock_path_exists.side_effect = [True, True]
- self.assertEqual(return_ns_state, expected_ns_state)
- self.assertEqual(return_operation_state, expected_operation_state)
- self.assertEqual(return_operation_error, expected_operation_error)
- self.assertEqual(
- return_ns_operational_state, expected_ns_operational_state
- )
- self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
+ mock_compare_charmdir_hash.return_value = callable(False)
+ mock_compare_charm_hash.return_value = callable(False)
- mock_reset()
+ instance = self.my_ns
+ instance.fs = fs
+ expected_result = False
- with self.subTest(
- i=3,
- t="Update type: CHANGE_VNFPKG, latest_vnfd revision changed, "
- "Charm package is not changed, sw-version is not changed.",
- ):
+ result = instance.check_charm_hash_changed(current_path, target_path)
+ self.assertEqual(result, expected_result, "Wrong charm hash control value")
+ self.assertEqual(mock_path_exists.call_count, 2)
+ self.assertEqual(mock_compare_charmdir_hash.call_count, 1)
+ self.assertEqual(mock_compare_charm_hash.call_count, 0)
- self.db.set_one(
- "vnfds", q_filter={"_id": vnfd_id}, update_dict={"_admin.revision": 3}
- )
+ @patch("os.path.exists")
+ @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
+ @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
+ def test_ns_update_check_charm_hash_changed(
+ self, mock_compare_charm_hash, mock_compare_charmdir_hash, mock_path_exists
+ ):
+ """Check charm hash, Hash has changed"""
- self.db.set_one(
- "vnfds_revisions",
- q_filter={"_id": vnfd_id + ":1"},
- update_dict={"_admin.revision": 1},
- )
+ current_path, target_path = "/tmp/charm1", "/tmp/charm2"
- self.db.set_one(
- "vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1}
- )
+ fs = Mock()
+ fs.path.__add__ = Mock()
+ fs.path.side_effect = [current_path, target_path, current_path, target_path]
+ fs.path.__add__.side_effect = [
+ current_path,
+ target_path,
+ current_path,
+ target_path,
+ ]
- mock_charm_hash = Mock(autospec=True)
- mock_charm_hash.return_value = False
+ mock_path_exists.side_effect = [True, True]
+ mock_compare_charmdir_hash.return_value = callable(True)
+ mock_compare_charm_hash.return_value = callable(True)
- mock_juju_bundle = Mock(return_value=None)
+ instance = self.my_ns
+ instance.fs = fs
+ expected_result = True
- mock_software_version = Mock(autospec=True)
-
- mock_charm_upgrade = asynctest.Mock(autospec=True)
- task = asyncio.Future()
- task.set_result(("COMPLETED", "some_output"))
- mock_charm_upgrade.return_value = task
- mock_software_version.side_effect = ["1.0", "1.0"]
-
- fs = Mock(autospec=True)
- fs.path.__add__ = Mock()
- fs.path.side_effect = ["/", "/", "/", "/"]
- fs.sync.side_effect = [None, None]
-
- instance = self.my_ns
-
- expected_operation_state = "COMPLETED"
- expected_operation_error = ""
- expected_vnfr_revision = 3
- expected_ns_state = "INSTANTIATED"
- expected_ns_operational_state = "running"
-
- with patch.object(instance, "fs", fs), patch(
- "osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed", mock_charm_hash
- ), patch("osm_lcm.ns.NsLcm._ns_charm_upgrade", mock_charm_upgrade), patch(
- "osm_lcm.lcm_utils.check_juju_bundle_existence", mock_juju_bundle
- ):
-
- await instance.update(nsr_id, nslcmop_id)
-
- return_operation_state = self.db.get_one(
- "nslcmops", {"_id": nslcmop_id}
- ).get("operationState")
-
- return_operation_error = self.db.get_one(
- "nslcmops", {"_id": nslcmop_id}
- ).get("errorMessage")
-
- return_ns_operational_state = self.db.get_one(
- "nsrs", {"_id": nsr_id}
- ).get("operational-status")
-
- return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
- "revision"
- )
-
- return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
- "nsState"
- )
-
- mock_charm_hash.assert_called_with(
- "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:1/hackfest_3charmed_vnfd/charms/simple",
- "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:3/hackfest_3charmed_vnfd/charms/simple",
- )
-
- self.assertEqual(fs.sync.call_count, 2)
- self.assertEqual(mock_charm_hash.call_count, 1)
-
- mock_juju_bundle.assert_not_called()
- mock_charm_upgrade.assert_not_called()
-
- self.assertEqual(return_ns_state, expected_ns_state)
- self.assertEqual(return_operation_state, expected_operation_state)
- self.assertEqual(return_operation_error, expected_operation_error)
- self.assertEqual(
- return_ns_operational_state, expected_ns_operational_state
- )
- self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
-
- mock_reset()
-
- with self.subTest(
- i=4,
- t="Update type: CHANGE_VNFPKG, latest_vnfd revision changed, "
- "Charm package exists, sw-version changed.",
- ):
-
- self.db.set_one(
- "vnfds",
- q_filter={"_id": vnfd_id},
- update_dict={"_admin.revision": 3, "software-version": "3.0"},
- )
-
- self.db.set_one(
- "vnfds_revisions",
- q_filter={"_id": vnfd_id + ":1"},
- update_dict={"_admin.revision": 1},
- )
-
- self.db.set_one(
- "vnfrs",
- q_filter={"_id": vnfr_id},
- update_dict={"revision": 1},
- )
-
- mock_charm_hash = Mock(autospec=True)
- mock_charm_hash.return_value = False
-
- mock_juju_bundle = Mock(return_value=None)
-
- mock_charm_upgrade = asynctest.Mock(autospec=True)
- task = asyncio.Future()
- task.set_result(("COMPLETED", "some_output"))
- mock_charm_upgrade.return_value = task
-
- mock_charm_artifact = Mock(autospec=True)
- mock_charm_artifact.side_effect = [
- "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:1/hackfest_3charmed_vnfd/charms/simple",
- "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77/hackfest_3charmed_vnfd/charms/simple",
- ]
-
- fs = Mock(autospec=True)
- fs.path.__add__ = Mock()
- fs.path.side_effect = ["/", "/", "/", "/"]
- fs.sync.side_effect = [None, None]
-
- instance = self.my_ns
-
- expected_operation_state = "FAILED"
- expected_operation_error = "FAILED Checking if existing VNF has charm: Software version change is not supported as VNF instance 6421c7c9-d865-4fb4-9a13-d4275d243e01 has charm."
- expected_vnfr_revision = 1
- expected_ns_state = "INSTANTIATED"
- expected_ns_operational_state = "running"
-
- with patch.object(instance, "fs", fs), patch(
- "osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed", mock_charm_hash
- ), patch("osm_lcm.ns.NsLcm._ns_charm_upgrade", mock_charm_upgrade), patch(
- "osm_lcm.lcm_utils.get_charm_artifact_path", mock_charm_artifact
- ):
-
- await instance.update(nsr_id, nslcmop_id)
-
- return_operation_state = self.db.get_one(
- "nslcmops", {"_id": nslcmop_id}
- ).get("operationState")
-
- return_operation_error = self.db.get_one(
- "nslcmops", {"_id": nslcmop_id}
- ).get("errorMessage")
-
- return_ns_operational_state = self.db.get_one(
- "nsrs", {"_id": nsr_id}
- ).get("operational-status")
-
- return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
- "revision"
- )
-
- return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
- "nsState"
- )
-
- self.assertEqual(fs.sync.call_count, 2)
- mock_charm_hash.assert_not_called()
-
- mock_juju_bundle.assert_not_called()
- mock_charm_upgrade.assert_not_called()
-
- self.assertEqual(return_ns_state, expected_ns_state)
- self.assertEqual(return_operation_state, expected_operation_state)
- self.assertEqual(return_operation_error, expected_operation_error)
- self.assertEqual(
- return_ns_operational_state, expected_ns_operational_state
- )
- self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
-
- mock_reset()
-
- with self.subTest(
- i=5,
- t="Update type: CHANGE_VNFPKG, latest_vnfd revision changed,"
- "Charm package exists, sw-version not changed, juju-bundle exists",
- ):
-
- self.db.set_one(
- "vnfds",
- q_filter={"_id": vnfd_id},
- update_dict={
- "_admin.revision": 3,
- "software-version": "1.0",
- "kdu.0.juju-bundle": "stable/native-kdu",
- },
- )
-
- self.db.set_one(
- "vnfds_revisions",
- q_filter={"_id": vnfd_id + ":1"},
- update_dict={
- "_admin.revision": 1,
- "software-version": "1.0",
- "kdu.0.juju-bundle": "stable/native-kdu",
- },
- )
-
- self.db.set_one(
- "vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1}
- )
-
- mock_charm_hash = Mock(autospec=True)
- mock_charm_hash.return_value = True
-
- mock_charm_artifact = Mock(autospec=True)
- mock_charm_artifact.side_effect = [
- "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:1/hackfest_3charmed_vnfd/charms/simple",
- "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77/hackfest_3charmed_vnfd/charms/simple",
- ]
-
- fs = Mock(autospec=True)
- fs.path.__add__ = Mock()
- fs.path.side_effect = ["/", "/", "/", "/"]
- fs.sync.side_effect = [None, None]
-
- instance = self.my_ns
-
- expected_operation_state = "FAILED"
- expected_operation_error = "FAILED Checking whether VNF uses juju bundle: Charm upgrade is not supported for the instance which uses juju-bundle: stable/native-kdu"
- expected_vnfr_revision = 1
- expected_ns_state = "INSTANTIATED"
- expected_ns_operational_state = "running"
-
- with patch.object(instance, "fs", fs), patch(
- "osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed", mock_charm_hash
- ), patch("osm_lcm.lcm_utils.get_charm_artifact_path", mock_charm_artifact):
-
- await instance.update(nsr_id, nslcmop_id)
-
- return_operation_state = self.db.get_one(
- "nslcmops", {"_id": nslcmop_id}
- ).get("operationState")
-
- return_operation_error = self.db.get_one(
- "nslcmops", {"_id": nslcmop_id}
- ).get("errorMessage")
-
- return_ns_operational_state = self.db.get_one(
- "nsrs", {"_id": nsr_id}
- ).get("operational-status")
-
- return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
- "revision"
- )
-
- return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
- "nsState"
- )
-
- self.assertEqual(fs.sync.call_count, 2)
- self.assertEqual(mock_charm_hash.call_count, 1)
- self.assertEqual(mock_charm_hash.call_count, 1)
-
- mock_charm_upgrade.assert_not_called()
-
- self.assertEqual(return_ns_state, expected_ns_state)
- self.assertEqual(return_operation_state, expected_operation_state)
- self.assertEqual(return_operation_error, expected_operation_error)
- self.assertEqual(
- return_ns_operational_state, expected_ns_operational_state
- )
- self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
-
- mock_reset()
-
- with self.subTest(
- i=6,
- t="Update type: CHANGE_VNFPKG, latest_vnfd revision changed,"
- "Charm package exists, sw-version not changed, charm-upgrade failed",
- ):
-
- self.db.set_one(
- "vnfds",
- q_filter={"_id": vnfd_id},
- update_dict={
- "_admin.revision": 3,
- "software-version": "1.0",
- "kdu": [],
- },
- )
-
- self.db.set_one(
- "vnfds_revisions",
- q_filter={"_id": vnfd_id + ":1"},
- update_dict={
- "_admin.revision": 1,
- "software-version": "1.0",
- "kdu": [],
- },
- )
-
- self.db.set_one(
- "vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1}
- )
-
- mock_charm_hash = Mock(autospec=True)
- mock_charm_hash.return_value = True
-
- mock_charm_upgrade = asynctest.Mock(autospec=True)
- task = asyncio.Future()
- task.set_result(("FAILED", "some_error"))
- mock_charm_upgrade.return_value = task
-
- mock_charm_artifact = Mock(autospec=True)
- mock_charm_artifact.side_effect = [
- "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:1/hackfest_3charmed_vnfd/charms/simple",
- "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77/hackfest_3charmed_vnfd/charms/simple",
- ]
-
- fs = Mock(autospec=True)
- fs.path.__add__ = Mock()
- fs.path.side_effect = ["/", "/", "/", "/"]
- fs.sync.side_effect = [None, None]
-
- instance = self.my_ns
-
- expected_operation_state = "FAILED"
- expected_operation_error = "some_error"
- expected_vnfr_revision = 1
- expected_ns_state = "INSTANTIATED"
- expected_ns_operational_state = "running"
-
- with patch.object(instance, "fs", fs), patch(
- "osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed", mock_charm_hash
- ), patch("osm_lcm.ns.NsLcm._ns_charm_upgrade", mock_charm_upgrade):
-
- await instance.update(nsr_id, nslcmop_id)
-
- return_operation_state = self.db.get_one(
- "nslcmops", {"_id": nslcmop_id}
- ).get("operationState")
-
- return_operation_error = self.db.get_one(
- "nslcmops", {"_id": nslcmop_id}
- ).get("errorMessage")
-
- return_ns_operational_state = self.db.get_one(
- "nsrs", {"_id": nsr_id}
- ).get("operational-status")
-
- return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
- "revision"
- )
-
- return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
- "nsState"
- )
-
- self.assertEqual(fs.sync.call_count, 2)
- self.assertEqual(mock_charm_hash.call_count, 1)
- self.assertEqual(mock_charm_upgrade.call_count, 1)
-
- self.assertEqual(return_ns_state, expected_ns_state)
- self.assertEqual(return_operation_state, expected_operation_state)
- self.assertEqual(return_operation_error, expected_operation_error)
- self.assertEqual(
- return_ns_operational_state, expected_ns_operational_state
- )
- self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
-
- mock_reset()
-
- def test_ns_update_helper_methods(self):
- def mock_reset():
- fs.mock_reset()
- mock_path.mock_reset()
- mock_checksumdir.mock_reset()
-
- with self.subTest(
- i=1, t="Find software version, VNFD does not have have software version"
- ):
- # Testing method find_software_version
-
- db_vnfd = self.db.get_one(
- "vnfds", {"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"}
- )
- expected_result = "1.0"
- result = find_software_version(db_vnfd)
- self.assertEqual(
- result, expected_result, "Default sw version should be 1.0"
- )
-
- with self.subTest(
- i=2, t="Find software version, VNFD includes software version"
- ):
- # Testing method find_software_version
-
- db_vnfd = self.db.get_one(
- "vnfds", {"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"}
- )
- db_vnfd["software-version"] = "3.1"
- expected_result = "3.1"
- result = find_software_version(db_vnfd)
- self.assertEqual(result, expected_result, "VNFD software version is wrong")
-
- with self.subTest(i=3, t="Check charm hash, Hash did not change"):
- # Testing method check_charm_hash_changed
-
- current_path, target_path = "/tmp/charm1", "/tmp/charm1"
- fs = Mock(autospec=True)
- fs.path.__add__ = Mock()
- fs.path.side_effect = ["/", "/", "/", "/"]
-
- mock_path = Mock(autospec=True)
- mock_path.exists.side_effect = [True, True]
-
- mock_checksumdir = Mock(autospec=True)
- mock_checksumdir.dirhash.side_effect = ["hash_value", "hash_value"]
-
- instance = self.my_ns
- expected_result = False
-
- with patch.object(instance, "fs", fs), patch(
- "checksumdir.dirhash", mock_checksumdir.dirhash
- ), patch("os.path.exists", mock_path.exists):
-
- result = instance.check_charm_hash_changed(current_path, target_path)
- self.assertEqual(
- result, expected_result, "Wrong charm hash control value"
- )
- self.assertEqual(mock_path.exists.call_count, 2)
- self.assertEqual(mock_checksumdir.dirhash.call_count, 2)
-
- mock_reset()
-
- with self.subTest(i=4, t="Check charm hash, Hash has changed"):
- # Testing method check_charm_hash_changed
-
- current_path, target_path = "/tmp/charm1", "/tmp/charm2"
- fs = Mock(autospec=True)
- fs.path.__add__ = Mock()
- fs.path.side_effect = ["/", "/", "/", "/"]
-
- mock_path = Mock(autospec=True)
- mock_path.exists.side_effect = [True, True]
-
- mock_checksumdir = Mock(autospec=True)
- mock_checksumdir.dirhash.side_effect = ["hash_value", "another_hash_value"]
-
- instance = self.my_ns
- expected_result = True
-
- with patch.object(instance, "fs", fs), patch(
- "checksumdir.dirhash", mock_checksumdir.dirhash
- ), patch("os.path.exists", mock_path.exists):
-
- result = instance.check_charm_hash_changed(current_path, target_path)
- self.assertEqual(
- result, expected_result, "Wrong charm hash control value"
- )
- self.assertEqual(mock_path.exists.call_count, 2)
- self.assertEqual(mock_checksumdir.dirhash.call_count, 2)
-
- mock_reset()
-
- with self.subTest(i=5, t="Check charm hash, Charm path does not exists"):
- # Testing method check_charm_hash_changed
-
- current_path, target_path = "/tmp/charm1", "/tmp/charm2"
- fs = Mock(autospec=True)
- fs.path.__add__ = Mock()
- fs.path.side_effect = ["/", "/", "/", "/"]
-
- mock_path = Mock(autospec=True)
- mock_path.exists.side_effect = [True, False]
-
- mock_checksumdir = Mock(autospec=True)
- mock_checksumdir.dirhash.side_effect = ["hash_value", "hash_value"]
-
- instance = self.my_ns
-
- with patch.object(instance, "fs", fs), patch(
- "checksumdir.dirhash", mock_checksumdir.dirhash
- ), patch("os.path.exists", mock_path.exists):
+ result = instance.check_charm_hash_changed(current_path, target_path)
+ self.assertEqual(result, expected_result, "Wrong charm hash control value")
+ self.assertEqual(mock_path_exists.call_count, 2)
+ self.assertEqual(mock_compare_charmdir_hash.call_count, 1)
+ self.assertEqual(mock_compare_charm_hash.call_count, 0)
- with self.assertRaises(LcmException):
+ @patch("os.path.exists")
+ @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
+ @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
+ def test_ns_update_check_no_charm_path(
+ self, mock_compare_charm_hash, mock_compare_charmdir_hash, mock_path_exists
+ ):
+ """Check charm hash, Charm path does not exist"""
- instance.check_charm_hash_changed(current_path, target_path)
- self.assertEqual(mock_path.exists.call_count, 2)
- self.assertEqual(mock_checksumdir.dirhash.call_count, 0)
+ current_path, target_path = "/tmp/charm1", "/tmp/charm2"
- mock_reset()
+ fs = Mock()
+ fs.path.__add__ = Mock()
+ fs.path.side_effect = [current_path, target_path, current_path, target_path]
+ fs.path.__add__.side_effect = [
+ current_path,
+ target_path,
+ current_path,
+ target_path,
+ ]
- with self.subTest(i=6, t="Check juju bundle existence"):
- # Testing method check_juju_bundle_existence
+ mock_path_exists.side_effect = [True, False]
- test_vnfd1 = self.db.get_one(
- "vnfds", {"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"}
- )
- test_vnfd2 = self.db.get_one(
- "vnfds", {"_id": "d96b1cdf-5ad6-49f7-bf65-907ada989293"}
- )
+ mock_compare_charmdir_hash.return_value = callable(False)
+ mock_compare_charm_hash.return_value = callable(False)
+ instance = self.my_ns
+ instance.fs = fs
- expected_result = None
- result = check_juju_bundle_existence(test_vnfd1)
- self.assertEqual(result, expected_result, "Wrong juju bundle name")
+ with self.assertRaises(LcmException):
+ instance.check_charm_hash_changed(current_path, target_path)
+ self.assertEqual(mock_path_exists.call_count, 2)
+ self.assertEqual(mock_compare_charmdir_hash.call_count, 0)
+ self.assertEqual(mock_compare_charm_hash.call_count, 0)
- expected_result = "stable/native-kdu"
- result = check_juju_bundle_existence(test_vnfd2)
- self.assertEqual(result, expected_result, "Wrong juju bundle name")
+ def test_ns_update_check_juju_bundle_existence_bundle_exists(self):
+ """Check juju bundle existence"""
+ test_vnfd2 = self.db.get_one(
+ "vnfds", {"_id": "d96b1cdf-5ad6-49f7-bf65-907ada989293"}
+ )
+ expected_result = "stable/native-kdu"
+ result = check_juju_bundle_existence(test_vnfd2)
+ self.assertEqual(result, expected_result, "Wrong juju bundle name")
- with self.subTest(i=7, t="Check charm artifacts"):
- # Testing method check_juju_bundle_existence
+ def test_ns_update_check_juju_bundle_existence_bundle_does_not_exist(self):
+ """Check juju bundle existence"""
+ test_vnfd1 = self.db.get_one("vnfds", {"_id": vnfd_id})
+ expected_result = None
+ result = check_juju_bundle_existence(test_vnfd1)
+ self.assertEqual(result, expected_result, "Wrong juju bundle name")
- base_folder = {
- "folder": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
- "pkg-dir": "hackfest_3charmed_vnfd",
- }
- charm_name = "simple"
- charm_type = "lxc_proxy_charm"
- revision = 3
+ def test_ns_update_check_juju_bundle_existence_empty_vnfd(self):
+ """Check juju bundle existence"""
+ test_vnfd1 = {}
+ expected_result = None
+ result = check_juju_bundle_existence(test_vnfd1)
+ self.assertEqual(result, expected_result, "Wrong juju bundle name")
- expected_result = "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:3/hackfest_3charmed_vnfd/charms/simple"
- result = get_charm_artifact_path(
- base_folder, charm_name, charm_type, revision
- )
- self.assertEqual(result, expected_result, "Wrong charm artifact path")
+ def test_ns_update_check_juju_bundle_existence_invalid_vnfd(self):
+ """Check juju bundle existence"""
+ test_vnfd1 = [{"_id": vnfd_id}]
+ with self.assertRaises(AttributeError):
+ check_juju_bundle_existence(test_vnfd1)
- # SOL004 packages
- base_folder = {
- "folder": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
- }
- charm_name = "basic"
- charm_type = ""
- revision = ""
+ def test_ns_update_check_juju_charm_artifacts_base_folder_wth_pkgdir(self):
+ """Check charm artifacts"""
+ base_folder = {
+ "folder": vnfd_id,
+ "pkg-dir": "hackfest_3charmed_vnfd",
+ }
+ charm_name = "simple"
+ charm_type = "lxc_proxy_charm"
+ revision = 3
+ expected_result = f"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple"
+ result = get_charm_artifact_path(base_folder, charm_name, charm_type, revision)
+ self.assertEqual(result, expected_result, "Wrong charm artifact path")
- expected_result = (
- "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77/Scripts/helm-charts/basic"
- )
- result = get_charm_artifact_path(
- base_folder, charm_name, charm_type, revision
- )
- self.assertEqual(result, expected_result, "Wrong charm artifact path")
+ def test_ns_update_check_juju_charm_artifacts_base_folder_wthout_pkgdir(self):
+ """Check charm artifacts, SOL004 packages"""
+ base_folder = {
+ "folder": vnfd_id,
+ }
+ charm_name = "basic"
+ charm_type, revision = "", ""
+ expected_result = f"{vnfd_id}/Scripts/helm-charts/basic"
+ result = get_charm_artifact_path(base_folder, charm_name, charm_type, revision)
+ self.assertEqual(result, expected_result, "Wrong charm artifact path")
if __name__ == "__main__":