##
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
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))