from connexion import NoContent
from mod.api.custom_exception import InvalidDataException, DuplicateDataException, \
DataConflictException
+from mod.subscription import AdministrativeState
def status():
f'{exception}'}, HTTPStatus.INTERNAL_SERVER_ERROR.value
+def delete_meas_group_by_name(subscription_name, measurement_group_name):
+ """Deletes the measurement group by name
+
+ Args:
+ subscription_name (String): Name of the subscription
+ measurement_group_name (String): Name of measurement group
+
+ Returns:
+ NoneType, HTTPStatus: None, 204
+ dict, HTTPStatus: measurement group not defined, 404
+ dict, HTTPStatus: Reason for not deleting measurement group, 409
+ dict, HTTPStatus: Exception details of failure, 500
+ """
+ logger.info(f'API call received to delete measurement group: {measurement_group_name}')
+ try:
+ measurement_group_administrative_status = \
+ measurement_group_service.query_get_meas_group_admin_status(subscription_name, measurement_group_name)
+ if measurement_group_administrative_status == AdministrativeState.LOCKED.value:
+ if measurement_group_service.query_to_delete_meas_group(subscription_name, measurement_group_name) == 1:
+ return None, HTTPStatus.NO_CONTENT
+ else:
+ logger.error(f'Measurement Group not found with name {measurement_group_name}')
+ return {'error': f'Measurement Group not found with name {measurement_group_name}'}, \
+ HTTPStatus.NOT_FOUND.value
+ else:
+ logger.error('Measurement Group was not deleted because the Administrative State '
+ f'was {measurement_group_administrative_status}')
+ return {'error': 'Measurement Group was not deleted because the Administrative State '
+ 'was {measurement_group_administrative_status}'}, \
+ HTTPStatus.CONFLICT.value
+ except Exception as e:
+ logger.error(f'Try again, measurement group {measurement_group_name} was not'
+ f'deleted due to exception: {e}')
+ return {'error': f'Try again, measurement group {measurement_group_name} was not '
+ f'deleted due to exception: {e}'}, HTTPStatus.INTERNAL_SERVER_ERROR.value
+
+
def delete_subscription_by_name(subscription_name):
""" Deletes the subscription by name
return meas_group
+def query_to_delete_meas_group(subscription_name, measurement_group_name):
+ """
+ Deletes a measurement group by name
+
+ Args:
+ subscription_name (String): Name of the Subscription
+ measurement_group_name (String): Name of MG
+
+ Returns:
+ int: Returns '1' if subscription exists and deleted successfully else '0'
+ """
+ affected_rows = db.session.query(MeasurementGroupModel) \
+ .filter_by(subscription_name=subscription_name, measurement_group_name=measurement_group_name).delete()
+ db.session.commit()
+ return affected_rows
+
+
+def query_get_meas_group_admin_status(subscription_name, measurement_group_name):
+ """
+ Queries the administrative state by using subscription name and measurement group name
+
+ Args:
+ subscription_name (String): Name of the subscription.
+ measurement_group_name (String): Name of the measurement group
+
+ Returns:
+ administrative_state (String): Admin State (LOCKED, UNLOCKED, LOCKING)
+ """
+ meas_group = query_meas_group_by_name(subscription_name, measurement_group_name)
+ return meas_group.administrative_state
+
+
def lock_nf_to_meas_grp(nf_name, measurement_group_name, status):
""" Deletes a particular nf related to a measurement group name and
if no more relations of nf exist to measurement group then delete nf from PMSH
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
+ get_subscriptions, get_meas_group_with_nfs, delete_subscription_by_name, update_admin_state, \
+ delete_meas_group_by_name
+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
error, status_code = get_meas_group_with_nfs('sub1', 'MG1')
self.assertEqual(status_code, HTTPStatus.INTERNAL_SERVER_ERROR.value)
- def test_delete_when_state_unlocked(self):
+ 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'
subscription_unlocked_data.measurement_groups[0].administrative_state = 'UNLOCKED'
self.assertEqual(subscription_service.query_subscription_by_name('MG_unlocked')
.subscription_name, 'MG_unlocked')
- def test_delete_when_state_locked(self):
+ def test_delete_mg_when_state_unlocked(self):
+ subscription_unlocked_data = create_subscription_data('MG_unlocked')
+ db.session.add(subscription_unlocked_data)
+ db.session.commit()
+ db.session.remove()
+ message, status_code = delete_meas_group_by_name('MG_unlocked', 'MG1')
+ self.assertEqual(status_code, HTTPStatus.CONFLICT.value)
+ self.assertEqual(query_meas_group_by_name('MG_unlocked', 'MG1').measurement_group_name,
+ 'MG1')
+
+
+ def test_delete_sub_when_state_locked(self):
subscription_unlocked_data = create_subscription_data('MG_locked')
subscription_unlocked_data.measurement_groups[0].measurement_group_name = 'lock'
subscription_unlocked_data.measurement_groups[0].administrative_state = 'LOCKED'
self.assertEqual(status_code, HTTPStatus.NO_CONTENT.value)
self.assertEqual(subscription_service.query_subscription_by_name('MG_locked'), None)
- def test_delete_when_state_locking(self):
+ def test_delete_mg_when_state_locked(self):
+ subscription_locked_data = create_subscription_data('MG_locked')
+ subscription_locked_data.measurement_groups[0].administrative_state = 'LOCKED'
+ db.session.add(subscription_locked_data)
+ db.session.add(subscription_locked_data.measurement_groups[0])
+ db.session.commit()
+ db.session.remove()
+ non_type, status_code = delete_meas_group_by_name('MG_locked', 'MG1')
+ self.assertEqual(non_type, None)
+ self.assertEqual(status_code, HTTPStatus.NO_CONTENT.value)
+ self.assertEqual(query_meas_group_by_name('MG_locked', 'MG1'), None)
+
+ def test_delete_sub_when_state_locking(self):
subscription_locking_data = create_subscription_data('MG_locking')
subscription_locking_data.measurement_groups[0].measurement_group_name = 'locking'
subscription_locking_data.measurement_groups[0].administrative_state = 'LOCKING'
self.assertEqual(subscription_service.query_subscription_by_name('MG_locking')
.subscription_name, 'MG_locking')
+ def test_delete_mg_when_state_locking(self):
+ subscription_locking_data = create_subscription_data('MG_locking')
+ subscription_locking_data.measurement_groups[0].administrative_state = 'LOCKING'
+ db.session.add(subscription_locking_data)
+ db.session.add(subscription_locking_data.measurement_groups[0])
+ db.session.commit()
+ db.session.remove()
+ message, status_code = delete_meas_group_by_name('MG_locking', 'MG1')
+ self.assertEqual(status_code, HTTPStatus.CONFLICT.value)
+ self.assertEqual(query_meas_group_by_name('MG_locking', 'MG1').measurement_group_name,
+ 'MG1')
+
def test_delete_sub_none(self):
message, status_code = delete_subscription_by_name('None')
self.assertEqual(message['error'], 'Subscription is not defined with name None')
self.assertEqual(status_code, HTTPStatus.NOT_FOUND.value)
+ def test_delete_mg_exception(self):
+ subscription_locking_data = create_subscription_data('MG_locking')
+ message, status_code = delete_meas_group_by_name(subscription_locking_data, 'None')
+ self.assertEqual(status_code, HTTPStatus.INTERNAL_SERVER_ERROR.value)
+
@patch('mod.api.services.subscription_service.query_to_delete_subscription_by_name',
MagicMock(side_effect=Exception('something failed')))
def test_delete_sub_exception(self):
Code Review / dcaegen2 / services.git / commitdiff