X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_policy_module%2Ftests%2Funit%2Fcore%2Ftest_policy_agent.py;h=43bc508e9a7defcdda91498c295630750e9b4801;hb=9ed429081842d22521c7be333e2ab25712316aab;hp=d40b43dc389335ce8c3924d55813749124a310c8;hpb=f7451f8879d188c9fef9c976d82aa39141601938;p=osm%2FPOL.git

diff --git a/osm_policy_module/tests/unit/core/test_policy_agent.py b/osm_policy_module/tests/unit/core/test_policy_agent.py
index d40b43d..43bc508 100644
--- a/osm_policy_module/tests/unit/core/test_policy_agent.py
+++ b/osm_policy_module/tests/unit/core/test_policy_agent.py
@@ -27,39 +27,51 @@ from unittest import mock
 
 from osm_policy_module.alarming.service import AlarmingService
 from osm_policy_module.autoscaling.service import AutoscalingService
+from osm_policy_module.common.common_db_client import CommonDbClient
 from osm_policy_module.core.agent import PolicyModuleAgent
 from osm_policy_module.core.config import Config
+from osm_policy_module.healing.service import HealingService
 
 
 class PolicyAgentTest(unittest.TestCase):
     def setUp(self):
         self.loop = asyncio.new_event_loop()
 
-    @mock.patch('osm_policy_module.alarming.service.CommonDbClient')
-    @mock.patch('osm_policy_module.alarming.service.MonClient')
-    @mock.patch('osm_policy_module.alarming.service.LcmClient')
-    @mock.patch('osm_policy_module.autoscaling.service.CommonDbClient')
-    @mock.patch('osm_policy_module.autoscaling.service.MonClient')
-    @mock.patch('osm_policy_module.autoscaling.service.LcmClient')
-    @mock.patch.object(AutoscalingService, 'configure_scaling_groups')
-    @mock.patch.object(AlarmingService, 'configure_vnf_alarms')
-    @mock.patch.object(AutoscalingService, 'delete_orphaned_alarms')
-    def test_handle_instantiated(self,
-                                 delete_orphaned_alarms,
-                                 configure_vnf_alarms,
-                                 configure_scaling_groups,
-                                 autoscaling_lcm_client,
-                                 autoscaling_mon_client,
-                                 autoscaling_db_client,
-                                 alarming_lcm_client,
-                                 alarming_mon_client,
-                                 alarming_db_client):
+    @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
+    @mock.patch("osm_policy_module.alarming.service.MonClient")
+    @mock.patch("osm_policy_module.alarming.service.LcmClient")
+    @mock.patch("osm_policy_module.autoscaling.service.MonClient")
+    @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
+    @mock.patch("osm_policy_module.healing.service.MonClient")
+    @mock.patch("osm_policy_module.healing.service.LcmClient")
+    @mock.patch.object(AutoscalingService, "configure_scaling_groups")
+    @mock.patch.object(AlarmingService, "configure_vnf_alarms")
+    @mock.patch.object(HealingService, "configure_healing_alarms")
+    @mock.patch.object(AutoscalingService, "delete_orphaned_alarms")
+    @mock.patch.object(CommonDbClient, "get_nslcmop")
+    def test_handle_instantiated(
+        self,
+        get_nslcmop,
+        delete_orphaned_alarms,
+        configure_healing_alarms,
+        configure_vnf_alarms,
+        configure_scaling_groups,
+        autoscaling_lcm_client,
+        autoscaling_mon_client,
+        alarming_lcm_client,
+        alarming_mon_client,
+        healing_lcm_client,
+        healing_mon_client,
+    ):
         async def mock_configure_scaling_groups(nsr_id):
             pass
 
         async def mock_configure_vnf_alarms(nsr_id):
             pass
 
+        async def mock_configure_healing_alarms(nsr_id):
+            pass
+
         async def mock_delete_orphaned_alarms(nsr_id):
             pass
 
@@ -67,51 +79,114 @@ class PolicyAgentTest(unittest.TestCase):
         agent = PolicyModuleAgent(config, self.loop)
         assert autoscaling_lcm_client.called
         assert autoscaling_mon_client.called
-        assert autoscaling_db_client.called
         assert alarming_lcm_client.called
         assert alarming_mon_client.called
-        assert alarming_db_client.called
+        assert healing_lcm_client.called
+        assert healing_mon_client.called
         content = {
-            'nslcmop_id': 'test_id',
+            "nslcmop_id": "test_id",
         }
         nslcmop_completed = {
-            'operationState': 'COMPLETED',
-            'nsInstanceId': 'test_nsr_id'
-        }
-        nslcmop_failed = {
-            'operationState': 'FAILED',
-            'nsInstanceId': 'test_nsr_id'
+            "operationState": "COMPLETED",
+            "nsInstanceId": "test_nsr_id",
         }
+        nslcmop_failed = {"operationState": "FAILED", "nsInstanceId": "test_nsr_id"}
         configure_scaling_groups.side_effect = mock_configure_scaling_groups
         configure_vnf_alarms.side_effect = mock_configure_vnf_alarms
+        configure_healing_alarms.side_effect = mock_configure_healing_alarms
         delete_orphaned_alarms.side_effect = mock_delete_orphaned_alarms
 
-        autoscaling_db_client.return_value.get_nslcmop.return_value = nslcmop_completed
+        get_nslcmop.return_value = nslcmop_completed
         self.loop.run_until_complete(agent._handle_instantiated(content))
-        configure_scaling_groups.assert_called_with('test_nsr_id')
+        configure_scaling_groups.assert_called_with("test_nsr_id")
         configure_scaling_groups.reset_mock()
 
-        autoscaling_db_client.return_value.get_nslcmop.return_value = nslcmop_failed
+        get_nslcmop.return_value = nslcmop_failed
         self.loop.run_until_complete(agent._handle_instantiated(content))
         configure_scaling_groups.assert_not_called()
 
-    @mock.patch('osm_policy_module.autoscaling.service.CommonDbClient')
-    @mock.patch('osm_policy_module.autoscaling.service.MonClient')
-    @mock.patch('osm_policy_module.autoscaling.service.LcmClient')
-    @mock.patch('osm_policy_module.alarming.service.CommonDbClient')
-    @mock.patch('osm_policy_module.alarming.service.MonClient')
-    @mock.patch('osm_policy_module.alarming.service.LcmClient')
-    @mock.patch.object(AutoscalingService, 'handle_alarm')
-    @mock.patch.object(AlarmingService, 'handle_alarm')
-    def test_handle_alarm_notification(self,
-                                       alarming_handle_alarm,
-                                       autoscaling_handle_alarm,
-                                       autoscaling_lcm_client,
-                                       autoscaling_mon_client,
-                                       autoscaling_db_client,
-                                       alarming_lcm_client,
-                                       alarming_mon_client,
-                                       alarming_db_client):
+    @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
+    @mock.patch("osm_policy_module.alarming.service.MonClient")
+    @mock.patch("osm_policy_module.alarming.service.LcmClient")
+    @mock.patch("osm_policy_module.autoscaling.service.MonClient")
+    @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
+    @mock.patch.object(AutoscalingService, "configure_scaling_groups")
+    @mock.patch.object(AlarmingService, "configure_vnf_alarms")
+    @mock.patch.object(AutoscalingService, "delete_scaling_groups")
+    @mock.patch.object(AlarmingService, "delete_vnf_alarms")
+    def test_handle_policy_update(
+        self,
+        delete_vnf_alarms,
+        delete_scaling_groups,
+        configure_vnf_alarms,
+        configure_scaling_groups,
+        autoscaling_lcm_client,
+        autoscaling_mon_client,
+        alarming_lcm_client,
+        alarming_mon_client,
+    ):
+        async def mock_delete_scaling_groups(nsr_id, vnf_member_index):
+            pass
+
+        async def mock_delete_vnf_alarms(nsr_id, vnf_member_index):
+            pass
+
+        async def mock_configure_scaling_groups(nsr_id, vnf_member_index):
+            pass
+
+        async def mock_configure_vnf_alarms(nsr_id, vnf_member_index):
+            pass
+
+        config = Config()
+        agent = PolicyModuleAgent(config, self.loop)
+        assert autoscaling_lcm_client.called
+        assert autoscaling_mon_client.called
+        assert alarming_lcm_client.called
+        assert alarming_mon_client.called
+        content = {
+            "nsr_id": "test_nsr_id",
+            "vnf_member_index": "1",
+            "operationState": "COMPLETED",
+        }
+        failed_content = {
+            "nsr_id": "test_nsr_id",
+            "vnf_member_index": "1",
+            "operationState": "FAILED",
+        }
+        configure_scaling_groups.side_effect = mock_configure_scaling_groups
+        configure_vnf_alarms.side_effect = mock_configure_vnf_alarms
+        delete_scaling_groups.side_effect = mock_delete_scaling_groups
+        delete_vnf_alarms.side_effect = mock_delete_vnf_alarms
+
+        self.loop.run_until_complete(agent._handle_policy_update(content))
+        configure_scaling_groups.assert_called_with("test_nsr_id", "1")
+        configure_scaling_groups.reset_mock()
+
+        self.loop.run_until_complete(agent._handle_policy_update(failed_content))
+        configure_scaling_groups.assert_not_called()
+
+    @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
+    @mock.patch("osm_policy_module.autoscaling.service.MonClient")
+    @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
+    @mock.patch("osm_policy_module.alarming.service.MonClient")
+    @mock.patch("osm_policy_module.alarming.service.LcmClient")
+    @mock.patch("osm_policy_module.healing.service.MonClient")
+    @mock.patch("osm_policy_module.healing.service.LcmClient")
+    @mock.patch.object(AutoscalingService, "handle_alarm")
+    @mock.patch.object(AlarmingService, "handle_alarm")
+    @mock.patch.object(HealingService, "handle_alarm")
+    def test_handle_alarm_notification(
+        self,
+        healing_handle_alarm,
+        alarming_handle_alarm,
+        autoscaling_handle_alarm,
+        autoscaling_lcm_client,
+        autoscaling_mon_client,
+        alarming_lcm_client,
+        alarming_mon_client,
+        healing_lcm_client,
+        healing_mon_client,
+    ):
         async def mock_handle_alarm(alarm_uuid, status, payload=None):
             pass
 
@@ -119,29 +194,79 @@ class PolicyAgentTest(unittest.TestCase):
         agent = PolicyModuleAgent(config, self.loop)
         assert autoscaling_lcm_client.called
         assert autoscaling_mon_client.called
-        assert autoscaling_db_client.called
         assert alarming_lcm_client.called
         assert alarming_mon_client.called
-        assert alarming_db_client.called
+        assert healing_lcm_client.called
+        assert healing_mon_client.called
         content = {
-            'notify_details': {
-                'alarm_uuid': 'test_alarm_uuid',
-                'metric_name': 'test_metric_name',
-                'operation': 'test_operation',
-                'threshold_value': 'test_threshold_value',
-                'vdu_name': 'test_vdu_name',
-                'vnf_member_index': 'test_vnf_member_index',
-                'ns_id': 'test_nsr_id',
-                'status': 'alarm'
+            "notify_details": {
+                "alarm_uuid": "test_alarm_uuid",
+                "metric_name": "test_metric_name",
+                "operation": "test_operation",
+                "threshold_value": "test_threshold_value",
+                "vdu_name": "test_vdu_name",
+                "vnf_member_index": "test_vnf_member_index",
+                "ns_id": "test_nsr_id",
+                "status": "alarm",
             }
         }
         autoscaling_handle_alarm.side_effect = mock_handle_alarm
         alarming_handle_alarm.side_effect = mock_handle_alarm
+        healing_handle_alarm.side_effect = mock_handle_alarm
 
         self.loop.run_until_complete(agent._handle_alarm_notification(content))
-        autoscaling_handle_alarm.assert_called_with('test_alarm_uuid', 'alarm')
-        alarming_handle_alarm.assert_called_with('test_alarm_uuid', 'alarm', content)
+        autoscaling_handle_alarm.assert_called_with("test_alarm_uuid", "alarm")
+        alarming_handle_alarm.assert_called_with("test_alarm_uuid", "alarm", content)
+        healing_handle_alarm.assert_called_with("test_alarm_uuid", "alarm")
+
+    @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
+    @mock.patch("osm_policy_module.alarming.service.MonClient")
+    @mock.patch("osm_policy_module.alarming.service.LcmClient")
+    @mock.patch("osm_policy_module.autoscaling.service.MonClient")
+    @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
+    @mock.patch.object(AutoscalingService, "delete_scaling_groups")
+    @mock.patch.object(AlarmingService, "delete_vnf_alarms")
+    def test_handle_vnf_terminated(
+        self,
+        delete_vnf_alarms,
+        delete_scaling_groups,
+        autoscaling_lcm_client,
+        autoscaling_mon_client,
+        alarming_lcm_client,
+        alarming_mon_client,
+    ):
+        async def mock_delete_scaling_groups(nsr_id, vnf_member_index):
+            pass
+
+        async def mock_delete_vnf_alarms(nsr_id, vnf_member_index):
+            pass
+
+        config = Config()
+        agent = PolicyModuleAgent(config, self.loop)
+        assert autoscaling_lcm_client.called
+        assert autoscaling_mon_client.called
+        assert alarming_lcm_client.called
+        assert alarming_mon_client.called
+        content = {
+            "nsr_id": "test_nsr_id",
+            "vnf_member_index": "1",
+            "operationState": "COMPLETED",
+        }
+        failed_content = {
+            "nsr_id": "test_nsr_id",
+            "vnf_member_index": "1",
+            "operationState": "FAILED",
+        }
+        delete_scaling_groups.side_effect = mock_delete_scaling_groups
+        delete_vnf_alarms.side_effect = mock_delete_vnf_alarms
+
+        self.loop.run_until_complete(agent._handle_vnf_terminated(content))
+        delete_scaling_groups.assert_called_with("test_nsr_id", "1")
+        delete_scaling_groups.reset_mock()
+
+        self.loop.run_until_complete(agent._handle_vnf_terminated(failed_content))
+        delete_scaling_groups.assert_not_called()
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     unittest.main()