* Updated PMSH app configuration, simplified existing config (DCAEGEN2-2814)
* Created Schema definitions in swagger file according to the new structure (DCAEGEN2-2889)
* Implemented Create Subscription public API (DCAEGEN2-2819)
+* Added 2 new attributes to the subscription model (DCAEGEN2-2913)
## [1.3.2]
### Changed
# ============LICENSE_START===================================================
-# Copyright (C) 2020 Nordix Foundation.
+# Copyright (C) 2020-2021 Nordix Foundation.
# ============================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
def _process_event(action, nf, mr_pub, app_conf):
if action == AAIEvent.UPDATE.value:
logger.info(f'Update event found for network function {nf.nf_name}')
- app_conf.subscription.create_subscription_on_nfs([nf], mr_pub, app_conf)
+ app_conf.subscription.create_subscription_on_nfs([nf], mr_pub)
elif action == AAIEvent.DELETE.value:
logger.info(f'Delete event found for network function {nf.nf_name}')
NetworkFunction.delete(nf_name=nf.nf_name)
__tablename__ = 'subscriptions'
id = Column(Integer, primary_key=True, autoincrement=True)
subscription_name = Column(String(100), unique=True, nullable=False)
+ operational_policy_name = Column(String(80), nullable=False)
+ control_loop_name = Column(String(80))
status = Column(String(20))
nfs = relationship(
cascade='all, delete-orphan',
backref='subscription')
- def __init__(self, subscription_name, status):
+ def __init__(self, subscription_name, operational_policy_name, control_loop_name, status):
self.subscription_name = subscription_name
+ self.operational_policy_name = operational_policy_name
+ self.control_loop_name = control_loop_name
self.status = status
def __repr__(self):
- return f'subscription_name: {self.subscription_name}, status: {self.status}'
+ return f'subscription_name: {self.subscription_name},' \
+ f'operational_policy_name: {self.operational_policy_name},' \
+ f'control_loop_name: {self.control_loop_name},' \
+ f'status: {self.status}'
def __eq__(self, other):
if isinstance(self, other.__class__):
sub_nfs = NfSubRelationalModel.query.filter(
NfSubRelationalModel.subscription_name == self.subscription_name).all()
db.session.remove()
- return {'subscription_name': self.subscription_name, 'subscription_status': self.status,
+ return {'subscription_name': self.subscription_name,
+ 'operational_policy_name': self.operational_policy_name,
+ 'control_loop_name': self.control_loop_name,
+ 'subscription_status': self.status,
'network_functions': [sub_nf.serialize_nf() for sub_nf in sub_nfs]}
db.session.add(new_nf_measure_grp_rel)
-def publish_measurement_group(subscription_name, measurement_group, nf):
+def publish_measurement_group(sub_model, measurement_group, nf):
"""
Publishes an event for measurement group against nfs to MR
Args:
- subscription_name (string): Subscription name to publish against nf
+ sub_model(SubscriptionModel): Subscription model object
measurement_group (MeasurementGroupModel): Measurement group to publish
nf (NetworkFunction): Network function to publish.
"""
- event_body = nf_service.create_nf_event_body(nf, 'CREATE')
+ event_body = nf_service.create_nf_event_body(nf, 'CREATE', sub_model)
event_body['subscription'] = {
"administrativeState": measurement_group.administrative_state,
- "subscriptionName": subscription_name,
+ "subscriptionName": sub_model.subscription_name,
"fileBasedGP": measurement_group.file_based_gp,
"fileLocation": measurement_group.file_location,
"measurementGroup": {
"managedObjectDNsBasic": measurement_group.managed_object_dns_basic
}
}
+ logger.debug(f'Event Body: {event_body}')
AppConfig.get_instance().publish_to_topic(MRTopic.POLICY_PM_PUBLISHER.value, event_body)
return aai_client.get_pmsh_nfs_from_aai(AppConfig.get_instance(), nf_filter)
-def create_nf_event_body(nf, change_type):
+def create_nf_event_body(nf, change_type, sub_model):
"""
Creates a network function event body to publish on MR
Args:
nf (NetworkFunction): the Network function to include in the event.
change_type (string): define the change type to be applied on node
+ sub_model(SubscriptionModel): Subscription model object
Returns:
dict: network function event body to publish on MR.
"""
- app_conf = AppConfig.get_instance()
return {'nfName': nf.nf_name,
'ipAddress': nf.ipv4_address if nf.ipv6_address in (None, '')
else nf.ipv6_address,
'blueprintName': nf.sdnc_model_name,
'blueprintVersion': nf.sdnc_model_version,
- 'policyName': app_conf.operational_policy_name,
+ 'operationalPolicyName': sub_model.operational_policy_name,
'changeType': change_type,
- 'closedLoopControlName': app_conf.control_loop_name}
+ 'controlLoopName': sub_model.control_loop_name}
def save_nf(nf):
logger.info(f'Initiating create subscription for: {subscription["subscriptionName"]}')
perform_validation(subscription)
try:
- sub_name, measurement_groups = save_subscription_request(subscription)
+ sub_model, measurement_groups = save_subscription_request(subscription)
db.session.commit()
logger.info(f'Successfully saved subscription request for: '
f'{subscription["subscriptionName"]}')
- filtered_nfs = nf_service.capture_filtered_nfs(sub_name)
+ filtered_nfs = nf_service.capture_filtered_nfs(sub_model.subscription_name)
if filtered_nfs:
- logger.info(f'Applying the filtered nfs for subscription: {sub_name}')
+ logger.info(f'Applying the filtered nfs for subscription: '
+ f'{sub_model.subscription_name}')
save_filtered_nfs(filtered_nfs)
- apply_subscription_to_nfs(filtered_nfs, sub_name)
+ apply_subscription_to_nfs(filtered_nfs, sub_model.subscription_name)
unlocked_msmt_groups = apply_measurement_grp_to_nfs(filtered_nfs, measurement_groups)
db.session.commit()
if unlocked_msmt_groups:
- publish_measurement_grp_to_nfs(sub_name, filtered_nfs, unlocked_msmt_groups)
+ publish_measurement_grp_to_nfs(sub_model, filtered_nfs,
+ unlocked_msmt_groups)
else:
- logger.error(f'All measurement groups are locked for subscription: {sub_name}, '
+ logger.error(f'All measurement groups are locked for subscription: '
+ f'{sub_model.subscription_name}, '
f'please verify/check measurement groups.')
else:
- logger.error(f'No network functions found for subscription: {sub_name}, '
+ logger.error(f'No network functions found for subscription: '
+ f'{sub_model.subscription_name}, '
f'please verify/check NetworkFunctionFilter.')
except IntegrityError as e:
db.session.rollback()
db.session.remove()
-def publish_measurement_grp_to_nfs(subscription_name, filtered_nfs, measurement_groups):
+def publish_measurement_grp_to_nfs(sub_model, filtered_nfs,
+ measurement_groups):
"""
Publishes an event for measurement groups against nfs
Args:
- subscription_name (string): subscription name against nfs
+ sub_model(SubscriptionModel): Subscription model object
filtered_nfs (list[NetworkFunction])): list of filtered network functions
measurement_groups (list[MeasurementGroupModel]): list of unlocked measurement group
"""
logger.info(f'Publishing event for nf name, measure_grp_name: {nf.nf_name},'
f'{measurement_group.measurement_group_name}')
measurement_group_service.publish_measurement_group(
- subscription_name, measurement_group, nf)
+ sub_model, measurement_group, nf)
except Exception as ex:
logger.error(f'Publish event failed for nf name, measure_grp_name, sub_name: '
f'{nf.nf_name},{measurement_group.measurement_group_name}, '
- f'{subscription_name} with error: {ex}')
+ f'{sub_model.subscription_name} with error: {ex}')
def save_filtered_nfs(filtered_nfs):
"""
if subscription['subscriptionName'].strip() in (None, ''):
raise InvalidDataException("No value provided in subscription name")
+ if subscription['operationalPolicyName'].strip() in (None, ''):
+ raise InvalidDataException("Value required for operational Policy Name")
for measurement_group in subscription.get('measurementGroups'):
measurement_group_details = measurement_group['measurementGroup']
list[MeasurementGroupModel]: list of measurement groups
"""
logger.info(f'Saving subscription request for: {subscription["subscriptionName"]}')
- sub_name = save_subscription(subscription).subscription_name
+ sub_model = save_subscription(subscription)
save_nf_filter(subscription["nfFilter"], subscription["subscriptionName"])
measurement_groups = []
for measurement_group in subscription['measurementGroups']:
measurement_group_service.save_measurement_group(
measurement_group['measurementGroup'],
subscription["subscriptionName"]))
- return sub_name, measurement_groups
+ return sub_model, measurement_groups
def check_duplicate_fields(subscription_name):
Args:
subscription (dict): subscription model to be saved.
+ Returns:
+ subscription_model(SubscriptionModel): subscription model
+ which is added to the session
"""
- subscription_model = SubscriptionModel(subscription_name=subscription["subscriptionName"],
- status=AdministrativeState.LOCKED.value)
+ control_loop_name = ""
+ if 'controlLoopName' in subscription:
+ control_loop_name = subscription['controlLoopName']
+ subscription_model = \
+ SubscriptionModel(subscription_name=subscription["subscriptionName"],
+ operational_policy_name=subscription["operationalPolicyName"],
+ control_loop_name=control_loop_name,
+ status=AdministrativeState.LOCKED.value)
db.session.add(subscription_model)
return subscription_model
self.streams_subscribes = app_config['config'].get('streams_subscribes')
# TODO: aaf_creds variable should be removed on code cleanup
self.aaf_creds = {'aaf_id': self.aaf_id, 'aaf_pass': self.aaf_pass}
- # TODO: changes under discussion once resolve is confirmed will be removed
- self.operational_policy_name = 'pmsh-operational-policy'
- self.control_loop_name = 'pmsh-control-loop'
@staticmethod
def get_instance():
self.operational_policy_name = conf['config'].get('operational_policy_name')
self.control_loop_name = conf['config'].get('control_loop_name')
self.sub_schema = _load_sub_schema_from_file()
- self.subscription = Subscription(**conf['config']['pmsh_policy']['subscription'])
+ self.subscription = Subscription(self.control_loop_name,
+ self.operational_policy_name,
+ **conf['config']['pmsh_policy']['subscription'])
self.nf_filter = None
def __new__(cls, *args, **kwargs):
"""
try:
app_conf = self._get_pmsh_config()
- self.subscription = Subscription(**app_conf['config']['pmsh_policy']['subscription'])
+ self.operational_policy_name = app_conf['config'].get('operational_policy_name')
+ self.control_loop_name = app_conf['config'].get('control_loop_name')
+ self.subscription = Subscription(self.control_loop_name,
+ self.operational_policy_name,
+ **app_conf['config']['pmsh_policy']['subscription'])
logger.info("AppConfig data has been refreshed")
except Exception:
logger.error('Failed to refresh PMSH AppConfig')
except Exception as e:
raise e
- def publish_subscription_event_data(self, subscription, nf, app_conf):
+ def publish_subscription_event_data(self, subscription, nf):
"""
Update the Subscription dict with xnf and policy name then publish to DMaaP MR topic.
Args:
subscription (Subscription): the `Subscription` <Subscription> object.
nf (NetworkFunction): the NetworkFunction to include in the event.
- app_conf (AppConfig): the application configuration.
"""
try:
- subscription_event = subscription.prepare_subscription_event(nf, app_conf)
+ subscription_event = subscription.prepare_subscription_event(nf)
+ logger.debug(f'Subscription event: {subscription_event}')
self.publish_to_topic(subscription_event)
except Exception as e:
logger.error(f'Failed to publish to topic {self.topic_url}: {e}', exc_info=True)
class Subscription:
- def __init__(self, **kwargs):
+ def __init__(self, control_loop_name, operational_policy_name, **kwargs):
self.subscriptionName = kwargs.get('subscriptionName')
self.administrativeState = kwargs.get('administrativeState')
self.fileBasedGP = kwargs.get('fileBasedGP')
self.fileLocation = kwargs.get('fileLocation')
self.nfFilter = kwargs.get('nfFilter')
self.measurementGroups = kwargs.get('measurementGroups')
+ self.control_loop_name = control_loop_name
+ self.operational_policy_name = operational_policy_name
self.create()
def update_sub_params(self, admin_state, file_based_gp, file_location, meas_groups):
existing_subscription = (SubscriptionModel.query.filter(
SubscriptionModel.subscription_name == self.subscriptionName).one_or_none())
if existing_subscription is None:
- new_subscription = SubscriptionModel(subscription_name=self.subscriptionName,
- status=AdministrativeState.LOCKED.value)
+ new_subscription = \
+ SubscriptionModel(subscription_name=self.subscriptionName,
+ operational_policy_name=self.operational_policy_name,
+ control_loop_name=self.control_loop_name,
+ status=AdministrativeState.LOCKED.value)
+
db.session.add(new_subscription)
db.session.commit()
return new_subscription
finally:
db.session.remove()
- def prepare_subscription_event(self, nf, app_conf):
+ def prepare_subscription_event(self, nf):
"""Prepare the sub event for publishing
Args:
nf (NetworkFunction): the AAI nf.
- app_conf (AppConfig): the application configuration.
Returns:
dict: the Subscription event to be published.
"""
try:
- clean_sub = {k: v for k, v in self.__dict__.items() if k != 'nfFilter'}
+ clean_sub = \
+ {k: v for k, v in self.__dict__.items() if
+ (k != 'nfFilter' and k != 'control_loop_name' and k != 'operational_policy_name')}
if self.administrativeState == AdministrativeState.LOCKING.value:
change_type = 'DELETE'
else:
'ipAddress': nf.ipv4_address if nf.ipv6_address in (None, '') else nf.ipv6_address,
'blueprintName': nf.sdnc_model_name,
'blueprintVersion': nf.sdnc_model_version,
- 'policyName': app_conf.operational_policy_name,
+ 'operationalPolicyName': self.operational_policy_name,
'changeType': change_type,
- 'closedLoopControlName': app_conf.control_loop_name,
+ 'controlLoopName': self.control_loop_name,
'subscription': clean_sub}
return sub_event
except Exception as e:
db.session.remove()
return sub_models
- def create_subscription_on_nfs(self, nfs, mr_pub, app_conf):
+ def create_subscription_on_nfs(self, nfs, mr_pub):
""" Publishes an event to create a Subscription on an nf
Args:
nfs(list[NetworkFunction]): A list of NetworkFunction Objects.
mr_pub (_MrPub): MR publisher
- app_conf (AppConfig): the application configuration.
"""
try:
existing_nfs = self.get_network_functions()
for nf in [new_nf for new_nf in nfs if new_nf not in existing_nfs]:
logger.info(f'Publishing event to create '
f'Sub: {self.subscriptionName} on nf: {nf.nf_name}')
- mr_pub.publish_subscription_event_data(self, nf, app_conf)
+ mr_pub.publish_subscription_event_data(self, nf)
self.add_network_function_to_subscription(nf, sub_model)
self.update_sub_nf_status(self.subscriptionName, SubNfState.PENDING_CREATE.value,
nf.nf_name)
except Exception as err:
raise Exception(f'Error publishing create event to MR: {err}')
- def delete_subscription_from_nfs(self, nfs, mr_pub, app_conf):
+ def delete_subscription_from_nfs(self, nfs, mr_pub):
""" Publishes an event to delete a Subscription from an nf
Args:
nfs(list[NetworkFunction]): A list of NetworkFunction Objects.
mr_pub (_MrPub): MR publisher
- app_conf (AppConfig): the application configuration.
"""
try:
for nf in nfs:
logger.debug(f'Publishing Event to delete '
f'Sub: {self.subscriptionName} from the nf: {nf.nf_name}')
- mr_pub.publish_subscription_event_data(self, nf, app_conf)
+ mr_pub.publish_subscription_event_data(self, nf)
self.update_sub_nf_status(self.subscriptionName,
SubNfState.PENDING_DELETE.value,
nf.nf_name)
self.app_conf.subscription.fileLocation,
self.app_conf.subscription.measurementGroups)
nfs_in_aai = aai_client.get_pmsh_nfs_from_aai(self.app_conf, self.app_conf.nf_filter)
- self.app_conf.subscription.create_subscription_on_nfs(nfs_in_aai, self.mr_pub,
- self.app_conf)
+ self.app_conf.subscription.create_subscription_on_nfs(nfs_in_aai, self.mr_pub)
self.app_conf.subscription.update_subscription_status()
def _deactivate(self):
self.stop_aai_event_thread()
self.app_conf.subscription.administrativeState = AdministrativeState.LOCKING.value
logger.info('Subscription is now LOCKING/DEACTIVATING.')
- self.app_conf.subscription.delete_subscription_from_nfs(nfs, self.mr_pub, self.app_conf)
+ self.app_conf.subscription.delete_subscription_from_nfs(nfs, self.mr_pub)
self.app_conf.subscription.update_subscription_status()
def _start_aai_event_thread(self):
logger.info(f'Retry deletion of subscription '
f'{self.app_conf.subscription.subscriptionName} '
f'from NF: {nf.nf_name}')
- self.app_conf.subscription.delete_subscription_from_nfs([nf], self.mr_pub,
- self.app_conf)
+ self.app_conf.subscription.delete_subscription_from_nfs([nf], self.mr_pub)
nf.increment_retry_count()
else:
logger.error(f'Failed to delete the subscription '
{
"subscription": {
"subscriptionName": "ExtraPM-All-gNB-R2B",
+ "operationalPolicyName":"pmsh-operational-policy",
+ "controlLoopName":"pmsh-control-loop",
"nfFilter": {
"nfNames": [
"^pnf.*",
"nfName":"pnf_1",\r
"blueprintName": "some-name",\r
"blueprintVersion": "some-version",\r
- "policyName":"pmsh-operational-policy",\r
+ "operationalPolicyName":"pmsh-operational-policy",\r
"changeType":"CREATE",\r
- "closedLoopControlName":"pmsh-control-loop",\r
+ "controlLoopName":"pmsh-control-loop",\r
"ipAddress": "204.120.0.15",\r
"subscription":{\r
"subscriptionName":"ExtraPM-All-gNB-R2B",\r
}\r
]\r
}\r
-}
\ No newline at end of file
+}\r
from mod import db
from tests.base_setup import BaseClassSetup
from mod.api.services import measurement_group_service
-from mod.api.db_models import MeasurementGroupModel, NfMeasureGroupRelationalModel
+from mod.api.db_models import MeasurementGroupModel, NfMeasureGroupRelationalModel,\
+ SubscriptionModel
from mod.subscription import SubNfState
'msg_publish', 'UNLOCKED',
15, 'pm.xml', [{"measurementType": "counter_a"}],
[{"DN": "string"}])
+ sub_model = SubscriptionModel('sub_publish', 'pmsh-operational-policy',
+ 'pmsh-control-loop', 'LOCKED')
measurement_group_service.publish_measurement_group(
- 'sub_publish', measurement_grp, nf_1)
+ sub_model, measurement_grp, nf_1)
mock_mr.assert_called_once_with('policy_pm_publisher',
{'nfName': 'pnf_1',
'ipAddress': '2001:db8:3333:4444:5555:6666:7777:8888',
'blueprintName': 'blah',
'blueprintVersion': 1.0,
- 'policyName': 'pmsh-operational-policy',
+ 'operationalPolicyName': 'pmsh-operational-policy',
'changeType': 'CREATE',
- 'closedLoopControlName': 'pmsh-control-loop',
+ 'controlLoopName': 'pmsh-control-loop',
'subscription':
{'administrativeState': 'UNLOCKED',
'subscriptionName': 'sub_publish',
import os
from unittest.mock import patch
from flask import current_app
-from mod.api.db_models import NetworkFunctionModel
+from mod.api.db_models import NetworkFunctionModel, SubscriptionModel
from mod import aai_client
from tests.base_setup import BaseClassSetup
from mod.api.services import nf_service
subscription = json.loads(self.subscription_request)['subscription']
mock_filter_call.return_value = NetworkFunctionFilter(**subscription["nfFilter"])
nf = nf_service.capture_filtered_nfs(subscription["subscriptionName"])[0]
- event_body = nf_service.create_nf_event_body(nf, 'CREATE')
+ sub_model = SubscriptionModel(subscription["subscriptionName"],
+ subscription['operationalPolicyName'],
+ subscription['controlLoopName'], 'LOCKED')
+ event_body = nf_service.create_nf_event_body(nf, 'CREATE', sub_model)
self.assertEqual(event_body['nfName'], nf.nf_name)
self.assertEqual(event_body['ipAddress'], nf.ipv6_address)
self.assertEqual(event_body['blueprintName'], nf.sdnc_model_name)
self.assertEqual(event_body['blueprintVersion'], nf.sdnc_model_version)
- self.assertEqual(event_body['policyName'],
- self.app_conf.operational_policy_name)
+ self.assertEqual(event_body['operationalPolicyName'],
+ sub_model.operational_policy_name)
self.assertEqual(event_body['changeType'], 'CREATE')
- self.assertEqual(event_body['closedLoopControlName'],
- self.app_conf.control_loop_name)
+ self.assertEqual(event_body['controlLoopName'],
+ sub_model.control_loop_name)
@patch.object(aai_client, '_get_all_aai_nf_data')
@patch.object(aai_client, 'get_aai_model_data')
mock_publish.return_value = None
subscription = self.create_test_subs('xtraPM-All-gNB-R2B-new2', 'msrmt_grp_name-new2')
subscription = json.loads(subscription)['subscription']
+ sub_model = SubscriptionModel(subscription["subscriptionName"],
+ subscription['operationalPolicyName'],
+ subscription['controlLoopName'], 'LOCKED')
measurement_grp = MeasurementGroupModel('subscription_name_1',
'msrmt_grp_name', 'UNLOCKED',
15, 'pm.xml', [], [])
mock_filter_call.return_value = NetworkFunctionFilter(**subscription["nfFilter"])
filtered_nfs = nf_service.capture_filtered_nfs(subscription["subscriptionName"])
subscription_service.publish_measurement_grp_to_nfs(
- subscription["subscriptionName"], filtered_nfs, measurement_grps)
+ sub_model, filtered_nfs, measurement_grps)
# Two unlocked measurement Group published
self.assertEqual(mock_publish.call_count, 6)
mock_publish.side_effect = Exception('Publish failed')
subscription = self.create_test_subs('xtraPM-All-gNB-R2B-new2', 'msrmt_grp_name-new2')
subscription = json.loads(subscription)['subscription']
+ sub_model = SubscriptionModel(subscription["subscriptionName"],
+ subscription['operationalPolicyName'],
+ subscription['controlLoopName'], 'LOCKED')
measurement_grp = MeasurementGroupModel('subscription_name_1',
'msrmt_grp_name', 'UNLOCKED',
15, 'pm.xml', [], [])
mock_filter_call.return_value = NetworkFunctionFilter(**subscription["nfFilter"])
filtered_nfs = nf_service.capture_filtered_nfs(subscription["subscriptionName"])
subscription_service.publish_measurement_grp_to_nfs(
- subscription["subscriptionName"], filtered_nfs, measurement_grps)
+ sub_model, filtered_nfs, measurement_grps)
mock_logger.assert_called_with('Publish event failed for nf name, measure_grp_name, '
'sub_name: pnf_33_ericsson,meas2, xtraPM-All-gNB-R2B-new2 '
'with error: Publish failed')
self.app_conf.subscription,
NetworkFunction(nf_name='pnf_1',
model_invariant_id='some-id',
- model_version_id='some-id'),
- self.app_conf)
+ model_version_id='some-id'))
pub_to_topic_call.assert_called_once()
@responses.activate
# ============LICENSE_START===================================================
-# Copyright (C) 2019-2020 Nordix Foundation.
+# Copyright (C) 2019-2021 Nordix Foundation.
# ============================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
'"ExtraPM-All-gNB-R2B", "nfName": "pnf300", "message": "success" } }']
self.mock_policy_mr_sub.get_from_topic.return_value = response_data
mock_get_sub.return_value = SubscriptionModel(subscription_name='ExtraPM-All-gNB-R2B',
+ operational_policy_name='policy-name',
+ control_loop_name='control-loop-name',
status=AdministrativeState.UNLOCKED.value)
self.policy_response_handler.poll_policy_topic()
self.mock_policy_mr_sub.get_from_topic.assert_called()
'"Different_Subscription", "nfName": "pnf300", "message": "success" } }']
self.mock_policy_mr_sub.get_from_topic.return_value = response_data
mock_get_sub.return_value = SubscriptionModel(subscription_name='ExtraPM-All-gNB-R2B',
+ operational_policy_name='policy-name',
+ control_loop_name='control-loop-name',
status=AdministrativeState.UNLOCKED.value)
self.policy_response_handler.poll_policy_topic()
def test_poll_policy_topic_exception(self, mock_get_sub, mock_logger):
self.mock_policy_mr_sub.get_from_topic.return_value = 'wrong_return'
mock_get_sub.return_value = SubscriptionModel(subscription_name='ExtraPM-All-gNB-R2B',
+ operational_policy_name='policy-name',
+ control_loop_name='control-loop-name',
status=AdministrativeState.UNLOCKED.value)
self.policy_response_handler.poll_policy_topic()
mock_logger.assert_called()
@patch('mod.subscription.Subscription.update_sub_nf_status')
def test_process_activate_subscription(self, mock_update_sub_nf, mock_add_nfs):
self.app_conf.subscription.create_subscription_on_nfs([list(self.xnfs)[0]],
- self.mock_mr_pub, self.app_conf)
+ self.mock_mr_pub)
mock_add_nfs.assert_called()
self.assertTrue(self.mock_mr_pub.publish_subscription_event_data.called)
def test_process_deactivate_subscription(self, mock_update_sub_nf, mock_get_nfs):
self.app_conf.subscription.administrativeState = 'LOCKED'
mock_get_nfs.return_value = [list(self.xnfs)[0]]
- self.app_conf.subscription.delete_subscription_from_nfs(self.xnfs, self.mock_mr_pub,
- self.app_conf)
+ self.app_conf.subscription.delete_subscription_from_nfs(self.xnfs, self.mock_mr_pub)
self.assertTrue(self.mock_mr_pub.publish_subscription_event_data.called)
self.assertEqual(mock_update_sub_nf.call_count, 3)
model_version_id='some-id')
nf.sdnc_model_name = 'some-name'
nf.sdnc_model_version = 'some-version'
- actual_sub_event = self.app_conf.subscription.prepare_subscription_event(nf, self.app_conf)
+ actual_sub_event = self.app_conf.subscription.prepare_subscription_event(nf)
print(actual_sub_event)
self.assertEqual(expected_sub_event, actual_sub_event)
model_version_id='some-id')
nf.sdnc_model_name = 'some-name'
nf.sdnc_model_version = 'some-version'
- actual_sub_event = self.app_conf.subscription.prepare_subscription_event(nf, self.app_conf)
+ actual_sub_event = self.app_conf.subscription.prepare_subscription_event(nf)
print(actual_sub_event)
self.assertEqual(expected_sub_event, actual_sub_event)
sub_handler.execute()
self.assertEqual(AdministrativeState.UNLOCKED.value,
self.app_conf.subscription.administrativeState)
- mock_activate_sub.assert_called_with(self.nfs, self.mock_mr_pub, self.app_conf)
+ mock_activate_sub.assert_called_with(self.nfs, self.mock_mr_pub)
@patch('mod.subscription.Subscription.get_network_functions', MagicMock(return_value=nfs))
@patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
self.app_conf)
sub_handler.execute()
- mock_deactivate_sub.assert_called_with(self.nfs, self.mock_mr_pub, self.app_conf)
+ mock_deactivate_sub.assert_called_with(self.nfs, self.mock_mr_pub)
@patch('mod.subscription_handler.SubscriptionHandler._start_aai_event_thread', MagicMock())
@patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
Code Review / dcaegen2 / services.git / commitdiff