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):
@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
assert autoscaling_mon_client.called
assert alarming_lcm_client.called
assert alarming_mon_client.called
+ assert healing_lcm_client.called
+ assert healing_mon_client.called
content = {
"nslcmop_id": "test_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
get_nslcmop.return_value = nslcmop_completed
@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
assert autoscaling_mon_client.called
assert alarming_lcm_client.called
assert alarming_mon_client.called
+ assert healing_lcm_client.called
+ assert healing_mon_client.called
content = {
"notify_details": {
"alarm_uuid": "test_alarm_uuid",
}
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)
+ 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")