diff options
Diffstat (limited to 'components/pm-subscription-handler/tests')
| -rw-r--r-- | components/pm-subscription-handler/tests/services/test_measurement_group_service.py | 51 | ||||
| -rwxr-xr-x | components/pm-subscription-handler/tests/test_controller.py | 37 |
2 files changed, 62 insertions, 26 deletions
diff --git a/components/pm-subscription-handler/tests/services/test_measurement_group_service.py b/components/pm-subscription-handler/tests/services/test_measurement_group_service.py index 25ab2581..7190069e 100644 --- a/components/pm-subscription-handler/tests/services/test_measurement_group_service.py +++ b/components/pm-subscription-handler/tests/services/test_measurement_group_service.py @@ -20,7 +20,8 @@ import json import os from unittest.mock import patch -from mod.api.custom_exception import InvalidDataException, DataConflictException +from mod.api.custom_exception import InvalidDataException, \ + DataConflictException, DuplicateDataException from mod.api.services.measurement_group_service import MgNfState from mod.network_function import NetworkFunction, NetworkFunctionFilter from mod.pmsh_config import AppConfig @@ -146,13 +147,9 @@ class MeasurementGroupServiceTestCase(BaseClassSetup): NfMeasureGroupRelationalModel.measurement_grp_name == 'measure_grp_name2', NfMeasureGroupRelationalModel.nf_name == 'pnf_test2').one_or_none()) self.assertIsNone(measurement_grp_rel) - network_function = (NetworkFunctionModel.query.filter( - NetworkFunctionModel.nf_name == 'pnf_test2').one_or_none()) - self.assertIsNone(network_function) @patch.object(NetworkFunction, 'delete') - @patch('mod.logger.error') - def test_delete_nf_to_measurement_group_failure(self, mock_logger, nf_delete_func): + def test_delete_nf_to_measurement_group_failure(self, nf_delete_func): nf = NetworkFunction(nf_name='pnf_test2') nf_service.save_nf(nf) db.session.commit() @@ -165,11 +162,6 @@ class MeasurementGroupServiceTestCase(BaseClassSetup): NfMeasureGroupRelationalModel.measurement_grp_name == 'measure_grp_name2', NfMeasureGroupRelationalModel.nf_name == 'pnf_test2').one_or_none()) self.assertIsNone(measurement_grp_rel) - network_function = (NetworkFunctionModel.query.filter( - NetworkFunctionModel.nf_name == 'pnf_test2').one_or_none()) - self.assertIsNotNone(network_function) - mock_logger.assert_called_with('Failed to delete nf: pnf_test2 for ' - 'measurement group: measure_grp_name2 due to: delete failed') @patch.object(db.session, 'commit') @patch('mod.logger.error') @@ -311,9 +303,6 @@ class MeasurementGroupServiceTestCase(BaseClassSetup): NfMeasureGroupRelationalModel.measurement_grp_name == 'MG2', NfMeasureGroupRelationalModel.nf_name == 'pnf_101').one_or_none()) self.assertIsNone(measurement_grp_rel) - network_function = (NetworkFunctionModel.query.filter( - NetworkFunctionModel.nf_name == 'pnf_101').one_or_none()) - self.assertIsNone(network_function) meas_grp = measurement_group_service.query_meas_group_by_name('sub', 'MG2') self.assertEqual(meas_grp.subscription_name, 'sub') self.assertEqual(meas_grp.measurement_group_name, 'MG2') @@ -337,13 +326,28 @@ class MeasurementGroupServiceTestCase(BaseClassSetup): NfMeasureGroupRelationalModel.measurement_grp_name == 'MG2', NfMeasureGroupRelationalModel.nf_name == 'pnf_101').one_or_none()) self.assertIsNone(measurement_grp_rel) - network_function = (NetworkFunctionModel.query.filter( - NetworkFunctionModel.nf_name == 'pnf_101').one_or_none()) - self.assertIsNone(network_function) - meas_grp = measurement_group_service.query_meas_group_by_name('sub', 'MG2') - self.assertEqual(meas_grp.subscription_name, 'sub') - self.assertEqual(meas_grp.measurement_group_name, 'MG2') - self.assertEqual(meas_grp.administrative_state, 'LOCKING') + + def test_check_duplication_exception(self): + sub = create_subscription_data('sub') + db.session.add(sub) + try: + measurement_group_service.check_duplication('sub', 'MG1') + except DuplicateDataException as e: + self.assertEqual(e.args[0], 'Measurement Group Name: MG1 already exists.') + + def test_check_measurement_group_names_comply(self): + mg = {'subscription_name': 'sub', + 'measurementGroupName': 'MG2', + 'administrativeState': 'UNLOCKED', + 'fileBasedGP': 15, + 'fileLocation': '/pm/pm.xml', + 'measurementTypes': '[{ "measurementType": "countera" }, ' + '{ "measurementType": "counterb" }]', + 'managedObjectDNsBasic': '[{ "DN":"dna"},{"DN":"dnb"}]'} + try: + measurement_group_service.check_measurement_group_names_comply('MG1', mg) + except InvalidDataException as e: + self.assertEqual(e.args[0], 'Measurement Group Name in body does not match with URI') def test_filter_nf_to_meas_grp_for_delete(self): sub = create_subscription_data('sub') @@ -355,13 +359,12 @@ class MeasurementGroupServiceTestCase(BaseClassSetup): db.session.commit() measurement_group_service.filter_nf_to_meas_grp( "pnf_test2", "MG2", MgNfState.DELETED.value) + measurement_group_service.filter_nf_to_meas_grp("pnf_test2", "MG2", + MgNfState.DELETED.value) measurement_grp_rel = (NfMeasureGroupRelationalModel.query.filter( NfMeasureGroupRelationalModel.measurement_grp_name == 'MG2', NfMeasureGroupRelationalModel.nf_name == 'pnf_test2').one_or_none()) self.assertIsNone(measurement_grp_rel) - network_function = (NetworkFunctionModel.query.filter( - NetworkFunctionModel.nf_name == 'pnf_test2').one_or_none()) - self.assertIsNone(network_function) meas_grp = measurement_group_service.query_meas_group_by_name('sub', 'MG2') self.assertEqual(meas_grp.subscription_name, 'sub') self.assertEqual(meas_grp.measurement_group_name, 'MG2') diff --git a/components/pm-subscription-handler/tests/test_controller.py b/components/pm-subscription-handler/tests/test_controller.py index 797666d9..07c17be7 100755 --- a/components/pm-subscription-handler/tests/test_controller.py +++ b/components/pm-subscription-handler/tests/test_controller.py @@ -23,11 +23,12 @@ from http import HTTPStatus from mod import aai_client, db from mod.api.controller import status, post_subscription, get_subscription_by_name, \ get_subscriptions, get_meas_group_with_nfs, delete_subscription_by_name, update_admin_state, \ - delete_meas_group_by_name, put_nf_filter + delete_meas_group_by_name, post_meas_group, put_nf_filter from mod.api.services.measurement_group_service import query_meas_group_by_name from tests.base_setup import BaseClassSetup from mod.api.custom_exception import InvalidDataException, DataConflictException -from mod.api.db_models import SubscriptionModel, NfMeasureGroupRelationalModel +from mod.api.db_models import SubscriptionModel, \ + NfMeasureGroupRelationalModel, MeasurementGroupModel from mod.network_function import NetworkFunctionFilter from tests.base_setup import create_subscription_data, create_multiple_subscription_data, \ create_multiple_network_function_data @@ -197,6 +198,38 @@ class ControllerTestCase(BaseClassSetup): error, status_code = get_meas_group_with_nfs('sub1', 'MG1') self.assertEqual(status_code, HTTPStatus.INTERNAL_SERVER_ERROR.value) + @patch.object(aai_client, '_get_all_aai_nf_data') + @patch.object(aai_client, 'get_aai_model_data') + def test_post_meas_group(self, mock_model_aai, mock_aai): + mock_aai.return_value = json.loads(self.aai_response_data) + mock_model_aai.return_value = json.loads(self.good_model_info) + subscription_data = create_subscription_data('Post_MG') + measurement_grp = {'subscription_name': 'sub', + 'measurementGroupName': 'MG2', + 'administrativeState': 'UNLOCKED', + 'fileBasedGP': 15, + 'fileLocation': '/pm/pm.xml', + 'measurementTypes': '[{ "measurementType": "countera" }, ' + '{ "measurementType": "counterb" }]', + 'managedObjectDNsBasic': '[{ "DN":"dna"},{"DN":"dnb"}]'} + db.session.add(subscription_data) + db.session.commit() + db.session.remove() + _, status_code = post_meas_group('Post_MG', 'MG3', measurement_grp) + self.assertEqual(status_code, 201) + + def test_post_meas_group_with_duplicate(self): + subscription_data = create_subscription_data('Post_MG') + measurement_grp = MeasurementGroupModel('Post_MG', 'MG1', 'UNLOCKED', 15, '/pm/pm.xml', + '[{ "measurementType": "countera" }, ' + '{ "measurementType": "counterb" }]', + '[{ "DN":"dna"},{"DN":"dnb"}]') + db.session.add(subscription_data) + db.session.commit() + db.session.remove() + _, status_code = post_meas_group('Post_MG', 'MG1', measurement_grp) + self.assertEqual(status_code, 409) + def test_delete_sub_when_state_unlocked(self): subscription_unlocked_data = create_subscription_data('MG_unlocked') subscription_unlocked_data.measurement_groups[0].measurement_group_name = 'unlock' |