Feature 10908: NBI Revisions

[osm/NBI.git] / osm_nbi / tests / test_instance_topics.py

diff --git a/osm_nbi/tests/test_instance_topics.py b/osm_nbi/tests/test_instance_topics.py
index 5e86ae5..cbb80ef 100644 (file)
--- a/osm_nbi/tests/test_instance_topics.py
+++ b/osm_nbi/tests/test_instance_topics.py
@@ -16,12 +16,14 @@
 ##
 
 import unittest
-from unittest.mock import Mock, mock_open  # patch, MagicMock
+from time import time
+from unittest.mock import Mock, mock_open   # patch, MagicMock
 from osm_common.dbbase import DbException
 from osm_nbi.engine import EngineException
 from osm_common.dbmemory import DbMemory
 from osm_common.fsbase import FsBase
 from osm_common.msgbase import MsgBase
+from osm_common import dbbase
 from http import HTTPStatus
 from osm_nbi.instance_topics import NsLcmOpTopic, NsrTopic
 from osm_nbi.tests.test_db_descriptors import (
@@ -226,6 +228,34 @@ class TestNsLcmOpTopic(unittest.TestCase):
             )
 
 
+class TestNsLcmOpTopicWithMock(unittest.TestCase):
+    def setUp(self):
+        self.db = Mock(dbbase.DbBase())
+        self.fs = Mock(FsBase())
+        self.fs.get_params.return_value = {"./fake/folder"}
+        self.fs.file_open = mock_open()
+        self.msg = Mock(MsgBase())
+        # create class
+        self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
+
+    def test_get_vnfd_from_vnf_member_revision(self):
+        test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
+        test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
+        self.db.get_one.side_effect = [test_vnfr, test_vnfd]
+        vnfr = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr['_id'])
+        self.assertEqual(self.db.get_one.call_args_list[0][0][0], 'vnfrs', "Incorrect first DB lookup")
+        self.assertEqual(self.db.get_one.call_args_list[1][0][0], 'vnfds', "Incorrect second DB lookup")
+
+    def test_get_vnfd_from_vnf_member_no_revision(self):
+        test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
+        test_vnfr['revision'] = 3
+        test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
+        self.db.get_one.side_effect = [test_vnfr, test_vnfd]
+        vnfr = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr['_id'])
+        self.assertEqual(self.db.get_one.call_args_list[0][0][0], 'vnfrs', "Incorrect first DB lookup")
+        self.assertEqual(self.db.get_one.call_args_list[1][0][0], 'vnfds_revisions', "Incorrect second DB lookup")
+
+
 class TestNsrTopic(unittest.TestCase):
     def setUp(self):
         self.db = DbMemory()
@@ -375,11 +405,54 @@ class TestNsrTopic(unittest.TestCase):
                 self.assertTrue(e.exception.http_code == expect_code)
             if expect_text_list:
                 for expect_text in expect_text_list:
-                    self.assertIn(
-                        expect_text,
-                        str(e.exception).lower(),
-                        "Expected '{}' at exception text".format(expect_text),
-                    )
+                    self.assertIn(expect_text, str(e.exception).lower(),
+                                  "Expected '{}' at exception text".format(expect_text))
+
+    def test_show_instance(self):
+        session = {"force": False, "admin": False, "public": False, "project_id": [self.nsd_project], "method": "write"}
+        filter_q = {}
+        for refresh_status in ("true", "false"):
+            self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
+            actual_nsr = self.db.get_list("nsrs")[0]
+            nsr_id = actual_nsr["_id"]
+            filter_q['vcaStatus-refresh'] = refresh_status
+            expected_nsr = self.nsr_topic.show(session, nsr_id, filter_q=filter_q)
+            self.nsr_topic.delete(session, nsr_id)
+            actual_nsr.pop("_admin")
+            expected_nsr.pop("_admin")
+            self.assertEqual(expected_nsr, actual_nsr, "Database nsr and show() nsr do not match.")
+
+    def test_vca_status_refresh(self):
+        session = {"force": False, "admin": False, "public": False, "project_id": [self.nsd_project], "method": "write"}
+        filter_q = {'vcaStatus-refresh': 'true'}
+        time_delta = 120
+        self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
+        nsr = self.db.get_list("nsrs")[0]
+
+        # When vcaStatus-refresh is true
+        filter_q['vcaStatus-refresh'] = "true"
+        self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
+        msg_args = self.msg.write.call_args[0]
+        self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action")
+        self.assertGreater(nsr["_admin"]["modified"], time() - time_delta)
+
+        # When vcaStatus-refresh is false but modified time is within threshold
+        filter_q['vcaStatus-refresh'] = "false"
+        time_now = time()
+        nsr["_admin"]["modified"] = time_now
+        self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
+        msg_args = self.msg.write.call_args[1]
+        self.assertEqual(msg_args, {}, "Message should not be sent.")
+        self.assertEqual(nsr["_admin"]["modified"], time_now, "Modified time should not be changed.")
+
+        # When vcaStatus-refresh is false but modified time is less than threshold
+        filter_q['vcaStatus-refresh'] = "false"
+        nsr["_admin"]["modified"] = time() - (2*time_delta)
+        self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
+        msg_args = self.msg.write.call_args[0]
+        self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action")
+        self.nsr_topic.delete(session, nsr["_id"])
+        self.assertGreater(nsr["_admin"]["modified"], time() - time_delta, "Modified time is not changed.")
 
     def test_delete_ns(self):
         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))