X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=tests%2Funit%2Ftest_alarm.py;h=1503d098f1b5d65ae6df4a36c742963d50734a66;hb=HEAD;hp=5bcde82564ac4b0352097d6f6b84262f0ed065eb;hpb=d2f6cc674bf3623caf114a8d7709e70d55ec9340;p=pti%2Fo2.git diff --git a/tests/unit/test_alarm.py b/tests/unit/test_alarm.py index 5bcde82..1503d09 100644 --- a/tests/unit/test_alarm.py +++ b/tests/unit/test_alarm.py @@ -52,11 +52,17 @@ def test_view_alarm_event_records(mock_uow): alarm_event_record1 = MagicMock() alarm_event_record1.serialize.return_value = { "alarmEventRecordId": alarm_event_record_id1} - session.return_value.query.return_value = [alarm_event_record1] - alarm_event_record_list = alarm_view.alarm_event_records(uow) - assert str(alarm_event_record_list[0].get( - "alarmEventRecordId")) == alarm_event_record_id1 + order_by = MagicMock() + order_by.count.return_value = 1 + order_by.limit.return_value.offset.return_value = [alarm_event_record1] + session.return_value.query.return_value.filter.return_value.\ + order_by.return_value = order_by + + result = alarm_view.alarm_event_records(uow) + assert result['count'] == 1 + ret_list = result['results'] + assert str(ret_list[0].get("alarmEventRecordId")) == alarm_event_record_id1 def test_view_alarm_event_record_one(mock_uow): @@ -81,24 +87,58 @@ def test_view_alarm_event_record_one(mock_uow): "alarmEventRecordId")) == alarm_event_record_id1 -def test_alarm_dictionary(mock_uow): +def test_view_alarm_subscriptions(mock_uow): session, uow = mock_uow - alarm_dict1 = alarm_obj.AlarmDictionary('test1') - alarm_dict1.entityType = 'test1' - with uow: - uow.alarm_dictionaries.add(alarm_dict1) - alarm_dict2 = uow.alarm_dictionaries.get('test1') - assert alarm_dict1 == alarm_dict2 + subscription_id1 = str(uuid.uuid4()) + sub1 = MagicMock() + sub1.serialize.return_value = { + "alarmSubscriptionId": subscription_id1, + } + + order_by = MagicMock() + order_by.count.return_value = 1 + order_by.limit.return_value.offset.return_value = [sub1] + session.return_value.query.return_value.filter.return_value.\ + order_by.return_value = order_by + + result = alarm_view.subscriptions(uow) + assert result['count'] == 1 + ret_list = result['results'] + assert str(ret_list[0].get("alarmSubscriptionId")) == subscription_id1 - dict_list = uow.alarm_dictionaries.list() - assert len(dict_list) > 0 + +def test_view_alarm_subscription_one(mock_uow): + session, uow = mock_uow + + subscription_id1 = str(uuid.uuid4()) + session.return_value.query.return_value.filter_by.return_value.first.\ + return_value.serialize.return_value = None + + # Query return None + subscription_res = alarm_view.subscription_one( + subscription_id1, uow) + assert subscription_res is None + + session.return_value.query.return_value.filter_by.return_value.first.\ + return_value.serialize.return_value = { + "alarmSubscriptionId": subscription_id1, + } + + subscription_res = alarm_view.subscription_one( + subscription_id1, uow) + assert str(subscription_res.get( + "alarmSubscriptionId")) == subscription_id1 def test_flask_get_list(mock_flask_uow): session, app = mock_flask_uow - session.query.return_value = [] - apibase = config.get_o2ims_monitoring_api_base() + order_by = MagicMock() + order_by.count.return_value = 0 + order_by.limit.return_value.offset.return_value = [] + session.return_value.query.return_value.filter.return_value.\ + order_by.return_value = order_by + apibase = config.get_o2ims_monitoring_api_base() + '/v1' with app.test_client() as client: # Get list and return empty list @@ -115,7 +155,7 @@ def test_flask_get_one(mock_flask_uow): session.return_value.query.return_value.filter_by.return_value.\ first.return_value = None - apibase = config.get_o2ims_monitoring_api_base() + apibase = config.get_o2ims_monitoring_api_base() + '/v1' with app.test_client() as client: # Get one and return 404 @@ -129,9 +169,15 @@ def test_flask_get_one(mock_flask_uow): assert resp.status_code == 404 -def test_flask_post(mock_flask_uow): +def test_flask_post(mock_flask_uow, mappers): session, app = mock_flask_uow - apibase = config.get_o2ims_monitoring_api_base() + apibase = config.get_o2ims_monitoring_api_base() + '/v1' + + order_by = MagicMock() + order_by.count.return_value = 0 + order_by.limit.return_value.offset.return_value = [] + session.return_value.query.return_value.filter.return_value.\ + order_by.return_value = order_by with app.test_client() as client: session.return_value.execute.return_value = [] @@ -140,7 +186,7 @@ def test_flask_post(mock_flask_uow): resp = client.post(apibase+'/alarmSubscriptions', json={ 'callback': sub_callback, 'consumerSubscriptionId': 'consumerSubId1', - 'filter': 'empty' + 'filter': '(eq,resourceTypeId,xxx)' }) assert resp.status_code == 201 assert 'alarmSubscriptionId' in resp.get_json() @@ -148,19 +194,19 @@ def test_flask_post(mock_flask_uow): def test_flask_delete(mock_flask_uow): session, app = mock_flask_uow - apibase = config.get_o2ims_monitoring_api_base() + apibase = config.get_o2ims_monitoring_api_base() + '/v1' with app.test_client() as client: session.return_value.execute.return_value.first.return_value = {} subscription_id1 = str(uuid.uuid4()) resp = client.delete(apibase+"/alarmSubscriptions/"+subscription_id1) - assert resp.status_code == 204 + assert resp.status_code == 200 def test_flask_not_allowed(mock_flask_uow): _, app = mock_flask_uow - apibase = config.get_o2ims_monitoring_api_base() + apibase = config.get_o2ims_monitoring_api_base() + '/v1' with app.test_client() as client: # Testing resource type not support method @@ -268,7 +314,12 @@ def test_probe_new_alarm(): bus = create_alarm_fake_bus(fakeuow) fakeClient = FakeAlarmClient() alarmwatcher = AlarmWatcher(fakeClient, bus) - cmds = alarmwatcher.probe() + parent = type('obj', (object,), { + 'data': type('obj', (object,), { + 'id': 'test_parent_id', + 'res_pool_id': 'test_res_pool' + })}) + cmds = alarmwatcher.probe(parent) assert cmds is not None assert len(cmds) == 1 assert cmds[0].data.name == "alarm"