## [2.2.0]
### Changed
* Update Filter API (DCAEGEN2-2922)
+* Cleaning up old App Config, subscription handler and it's subsequent calls (DCAEGEN2-3085)
## [2.1.1]
### Changed
# ============LICENSE_START===================================================
-# Copyright (C) 2019-2021 Nordix Foundation.
+# Copyright (C) 2019-2022 Nordix Foundation.
# ============================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
#
# SPDX-License-Identifier: Apache-2.0
# ============LICENSE_END=====================================================
-import logging as logging
+import logging
import os
import pathlib
import ssl
from onaplogging import monkey
from onaplogging.mdcContext import MDC
from ruamel.yaml import YAML
+import uuid
+from functools import wraps
+from os import getenv
db = SQLAlchemy()
basedir = os.path.abspath(os.path.dirname(__file__))
logger = logging.getLogger('onap_logger')
+def mdc_handler(func):
+ @wraps(func)
+ def wrapper(*args, **kwargs):
+ request_id = str(uuid.uuid1())
+ invocation_id = str(uuid.uuid1())
+ MDC.put('ServiceName', getenv('HOSTNAME'))
+ MDC.put('RequestID', request_id)
+ MDC.put('InvocationID', invocation_id)
+
+ kwargs['request_id'] = request_id
+ kwargs['invocation_id'] = invocation_id
+ return func(*args, **kwargs)
+
+ return wrapper
+
+
def _get_app():
global _connexion_app
if not _connexion_app:
# ============LICENSE_START===================================================
-# Copyright (C) 2019-2021 Nordix Foundation.
+# Copyright (C) 2019-2022 Nordix Foundation.
# ============================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
import requests
from requests.auth import HTTPBasicAuth
import mod.network_function
-import mod.pmsh_utils
-from mod import logger
+from mod import logger, mdc_handler
def get_pmsh_nfs_from_aai(app_conf, nf_filter):
}"""
params = {'format': 'simple', 'nodesOnly': 'true'}
response = session.put(aai_named_query_endpoint, headers=headers,
- auth=HTTPBasicAuth(app_conf.aaf_creds.get('aaf_id'),
- app_conf.aaf_creds.get('aaf_pass')),
+ auth=HTTPBasicAuth(app_conf.aaf_id,
+ app_conf.aaf_pass),
data=data, params=params,
verify=(app_conf.ca_cert_path if app_conf.enable_tls else False),
cert=((app_conf.cert_path,
raise
-@mod.pmsh_utils.mdc_handler
+@mdc_handler
def _get_aai_request_headers(**kwargs):
return {'accept': 'application/json',
'content-type': 'application/json',
Returns:
json (dict): the sdnc_model json object.
-
- Raises:
- Exception: if AAI model data cannot be retrieved.
"""
- try:
- session = requests.Session()
- aai_model_ver_endpoint = \
- f'{_get_aai_service_url()}/service-design-and-creation/models/model/' \
- f'{model_invariant_id}/model-vers/model-ver/{model_version_id}'
-
- logger.info(f'Fetching sdnc-model info for xNF: {nf_name} from AAI.')
- headers = _get_aai_request_headers()
- response = session.get(aai_model_ver_endpoint, headers=headers,
- auth=HTTPBasicAuth(app_conf.aaf_creds.get('aaf_id'),
- app_conf.aaf_creds.get('aaf_pass')),
- verify=(app_conf.ca_cert_path if app_conf.enable_tls else False),
- cert=((app_conf.cert_path,
- app_conf.key_path) if app_conf.enable_tls else None))
- response.raise_for_status()
- if response.ok:
- data = json.loads(response.text)
- logger.debug(f'Successfully fetched sdnc-model info from AAI: {data}')
- return data
- except Exception:
- raise
+ session = requests.Session()
+ aai_model_ver_endpoint = \
+ f'{_get_aai_service_url()}/service-design-and-creation/models/model/' \
+ f'{model_invariant_id}/model-vers/model-ver/{model_version_id}'
+
+ logger.info(f'Fetching sdnc-model info for xNF: {nf_name} from AAI.')
+ headers = _get_aai_request_headers()
+ response = session.get(aai_model_ver_endpoint, headers=headers,
+ auth=HTTPBasicAuth(app_conf.aaf_id,
+ app_conf.aaf_pass),
+ verify=(app_conf.ca_cert_path if app_conf.enable_tls else False),
+ cert=((app_conf.cert_path,
+ app_conf.key_path) if app_conf.enable_tls else None))
+ response.raise_for_status()
+ if response.ok:
+ data = json.loads(response.text)
+ logger.debug(f'Successfully fetched sdnc-model info from AAI: {data}')
+ return data
from connexion import NoContent
from mod.api.custom_exception import InvalidDataException, DuplicateDataException, \
DataConflictException
-from mod.subscription import AdministrativeState
+from mod.api.services.measurement_group_service import AdministrativeState
def status():
from mod.api.services import nf_service, subscription_service
from mod.network_function import NetworkFunction
from mod.pmsh_config import MRTopic, AppConfig
-from mod.subscription import AdministrativeState, SubNfState
from sqlalchemy import or_
+from enum import Enum
+
+
+class MgNfState(Enum):
+ PENDING_CREATE = 'PENDING_CREATE'
+ CREATE_FAILED = 'CREATE_FAILED'
+ CREATED = 'CREATED'
+ PENDING_DELETE = 'PENDING_DELETE'
+ DELETE_FAILED = 'DELETE_FAILED'
+ DELETED = 'DELETED'
+
+
+class AdministrativeState(Enum):
+ UNLOCKED = 'UNLOCKED'
+ LOCKING = 'LOCKING'
+ LOCKED = 'LOCKED'
+ FILTERING = 'FILTERING'
+
+
+mg_nf_states = {
+ AdministrativeState.LOCKED.value: {
+ 'success': MgNfState.DELETED,
+ 'failed': MgNfState.DELETE_FAILED
+ },
+ AdministrativeState.UNLOCKED.value: {
+ 'success': MgNfState.CREATED,
+ 'failed': MgNfState.CREATE_FAILED
+ },
+ AdministrativeState.LOCKING.value: {
+ 'success': MgNfState.DELETED,
+ 'failed': MgNfState.DELETE_FAILED
+ }
+}
def save_measurement_group(measurement_group, subscription_name):
logger.info(f'Saving measurement group to nf name, measure_grp_name: {nf.nf_name},'
f'{measurement_group.measurement_group_name} with DELETE request')
update_measurement_group_nf_status(measurement_group.measurement_group_name,
- SubNfState.PENDING_DELETE.value, nf.nf_name)
+ MgNfState.PENDING_DELETE.value, nf.nf_name)
try:
network_function = NetworkFunction(**nf.serialize_meas_group_nfs())
logger.info(f'Publishing event for nf name, measure_grp_name: {nf.nf_name},'
apply_nf_status_to_measurement_group(nf.nf_name,
measurement_group.measurement_group_name,
- SubNfState.PENDING_CREATE.value)
+ MgNfState.PENDING_CREATE.value)
db.session.commit()
try:
network_function = NetworkFunction(**nf.serialize_nf())
status (string): status of the network function for measurement group
"""
try:
- if status == SubNfState.DELETED.value:
+ if status == MgNfState.DELETED.value:
delete_nf_to_measurement_group(nf_name, measurement_group_name,
- SubNfState.DELETED.value)
- elif status == SubNfState.CREATED.value:
+ MgNfState.DELETED.value)
+ elif status == MgNfState.CREATED.value:
update_measurement_group_nf_status(measurement_group_name,
- SubNfState.CREATED.value, nf_name)
+ MgNfState.CREATED.value, nf_name)
nf_measurement_group_rels = NfMeasureGroupRelationalModel.query.filter(
NfMeasureGroupRelationalModel.measurement_grp_name == measurement_group_name,
or_(NfMeasureGroupRelationalModel.nf_measure_grp_status.like('PENDING_%'),
from mod.api.services import measurement_group_service, nf_service
from mod.api.custom_exception import InvalidDataException, DuplicateDataException, \
DataConflictException
-from mod.subscription import AdministrativeState, SubNfState
+from mod.api.services.measurement_group_service import MgNfState, AdministrativeState
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import joinedload
f'{measurement_group.measurement_group_name}')
measurement_group_service.apply_nf_status_to_measurement_group(
nf.nf_name, measurement_group.measurement_group_name,
- SubNfState.PENDING_CREATE.value)
+ MgNfState.PENDING_CREATE.value)
def check_missing_data(subscription):
from mod import logger, db
from mod.api.services import subscription_service, measurement_group_service
-from mod.subscription import AdministrativeState
+from mod.api.services.measurement_group_service import AdministrativeState
class ExitHandler:
# ============LICENSE_START===================================================
-# Copyright (C) 2021 Nordix Foundation.
+# Copyright (C) 2021-2022 Nordix Foundation.
# ============================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
from requests.auth import HTTPBasicAuth
from tenacity import wait_fixed, stop_after_attempt, retry, retry_if_exception_type
-from mod import logger
-from mod.pmsh_utils import mdc_handler
+from mod import logger, mdc_handler
@unique
self.aaf_pass = app_config['config'].get('aaf_password')
self.streams_publishes = app_config['config'].get('streams_publishes')
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}
@staticmethod
def get_instance():
return config
except Exception as e:
logger.error(f'Failed to get config from CBS: {e}', exc_info=True)
- raise ValueError(e)
+ raise ValueError(e) from e
@mdc_handler
def publish_to_topic(self, mr_topic, event_json, **kwargs):
Args:
mr_topic (enum) : Message Router topic to publish.
event_json (dict): the json data to be published.
-
- Raises:
- Exception: if post request fails.
"""
- try:
- session = requests.Session()
- topic_url = self.streams_publishes[mr_topic].get('dmaap_info').get('topic_url')
- headers = {'content-type': 'application/json', 'x-transactionid': kwargs['request_id'],
- 'InvocationID': kwargs['invocation_id'], 'RequestID': kwargs['request_id']}
- logger.info(f'Publishing event to MR topic: {topic_url}')
- response = session.post(topic_url, headers=headers,
- auth=HTTPBasicAuth(self.aaf_id, self.aaf_pass), json=event_json,
- verify=(self.ca_cert_path if self.enable_tls else False))
- response.raise_for_status()
- except Exception as e:
- raise e
+ session = requests.Session()
+ topic_url = self.streams_publishes[mr_topic].get('dmaap_info').get('topic_url')
+ headers = {'content-type': 'application/json', 'x-transactionid': kwargs['request_id'],
+ 'InvocationID': kwargs['invocation_id'], 'RequestID': kwargs['request_id']}
+ logger.info(f'Publishing event to MR topic: {topic_url}')
+ response = session.post(topic_url, headers=headers,
+ auth=HTTPBasicAuth(self.aaf_id, self.aaf_pass), json=event_json,
+ verify=(self.ca_cert_path if self.enable_tls else False))
+ response.raise_for_status()
@mdc_handler
def get_from_topic(self, mr_topic, consumer_id, consumer_group='dcae_pmsh_cg', timeout=5000,
verify=(self.ca_cert_path if self.enable_tls else False))
if response.status_code == 503:
logger.error(f'MR Service is unavailable at present: {response.content}')
- pass
response.raise_for_status()
if response.ok:
return response.json()
+++ /dev/null
-# ============LICENSE_START===================================================
-# 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.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-# SPDX-License-Identifier: Apache-2.0
-# ============LICENSE_END=====================================================
-import json
-import os
-import uuid
-from functools import wraps
-from json import JSONDecodeError
-from os import getenv
-from threading import Timer
-
-import requests
-from jsonschema import validate, ValidationError
-from onap_dcae_cbs_docker_client.client import get_all
-from onaplogging.mdcContext import MDC
-from requests.auth import HTTPBasicAuth
-from tenacity import wait_fixed, stop_after_attempt, retry, retry_if_exception_type
-
-from mod import logger
-from mod.subscription import Subscription
-
-
-def mdc_handler(func):
- @wraps(func)
- def wrapper(*args, **kwargs):
- request_id = str(uuid.uuid1())
- invocation_id = str(uuid.uuid1())
- MDC.put('ServiceName', getenv('HOSTNAME'))
- MDC.put('RequestID', request_id)
- MDC.put('InvocationID', invocation_id)
-
- kwargs['request_id'] = request_id
- kwargs['invocation_id'] = invocation_id
- return func(*args, **kwargs)
-
- return wrapper
-
-
-def _load_sub_schema_from_file():
- try:
- with open(os.path.join(os.path.dirname(__file__), 'sub_schema.json')) as sub:
- return json.load(sub)
- except OSError as err:
- logger.error(f'Failed to read sub schema file: {err}', exc_info=True)
- except JSONDecodeError as json_err:
- logger.error(f'sub schema file is not a valid JSON file: {json_err}', exc_info=True)
-
-
-class AppConfig:
- INSTANCE = None
-
- def __init__(self):
- conf = self._get_pmsh_config()
- self.aaf_creds = {'aaf_id': conf['config'].get('aaf_identity'),
- 'aaf_pass': conf['config'].get('aaf_password')}
- self.enable_tls = conf['config'].get('enable_tls')
- self.ca_cert_path = conf['config'].get('ca_cert_path')
- self.cert_path = conf['config'].get('cert_path')
- self.key_path = conf['config'].get('key_path')
- self.streams_subscribes = conf['config'].get('streams_subscribes')
- self.streams_publishes = conf['config'].get('streams_publishes')
- 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(self.control_loop_name,
- self.operational_policy_name,
- **conf['config']['pmsh_policy']['subscription'])
- self.nf_filter = None
-
- def __new__(cls, *args, **kwargs):
- if AppConfig.INSTANCE is None:
- AppConfig.INSTANCE = super().__new__(cls, *args, **kwargs)
- return AppConfig.INSTANCE
-
- @mdc_handler
- @retry(wait=wait_fixed(5), stop=stop_after_attempt(5),
- retry=retry_if_exception_type(ValueError))
- def _get_pmsh_config(self, **kwargs):
- """ Retrieves PMSH's configuration from Config binding service. If a non-2xx response
- is received, it retries after 2 seconds for 5 times before raising an exception.
-
- Returns:
- dict: Dictionary representation of the the service configuration
-
- Raises:
- Exception: If any error occurred pulling configuration from Config binding service.
- """
- try:
- logger.info('Attempting to fetch PMSH Configuration from CBS.')
- config = get_all()
- logger.info(f'Successfully fetched PMSH config from CBS: {config}')
- return config
- except Exception as e:
- logger.error(f'Failed to get config from CBS: {e}', exc_info=True)
- raise ValueError(e)
-
- def validate_sub_schema(self):
- """
- Validates schema of PMSH subscription
-
- Raises:
- ValidationError: If the PMSH subscription schema is invalid
- """
- sub_data = self.subscription.__dict__
- validate(instance=sub_data, schema=self.sub_schema)
- nf_filter = sub_data["nfFilter"]
- if not [filter_name for filter_name, val in nf_filter.items() if len(val) > 0]:
- raise ValidationError("At least one filter within nfFilter must not be empty")
- logger.debug("Subscription schema is valid.")
-
- def refresh_config(self):
- """
- Update the relevant attributes of the AppConfig object.
-
- Raises:
- Exception: if cbs request fails.
- """
- try:
- app_conf = self._get_pmsh_config()
- 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')
- raise
-
- def get_mr_sub(self, sub_name):
- """
- Returns the MrSub object requested.
-
- Args:
- sub_name: the key of the subscriber object.
-
- Returns:
- MrSub: an Instance of an `MrSub` <MrSub> Object.
-
- Raises:
- KeyError: if the sub_name is not found.
- """
- try:
- return _MrSub(sub_name, self.aaf_creds, self.ca_cert_path,
- self.enable_tls, self.cert_params, **self.streams_subscribes[sub_name])
- except KeyError as e:
- logger.error(f'Failed to get MrSub {sub_name}: {e}', exc_info=True)
- raise
-
- def get_mr_pub(self, pub_name):
- """
- Returns the MrPub object requested.
-
- Args:
- pub_name: the key of the publisher object.
-
- Returns:
- MrPub: an Instance of an `MrPub` <MrPub> Object.
-
- Raises:
- KeyError: if the sub_name is not found.
- """
- try:
- return _MrPub(pub_name, self.aaf_creds, self.ca_cert_path,
- self.enable_tls, self.cert_params, **self.streams_publishes[pub_name])
- except KeyError as e:
- logger.error(f'Failed to get MrPub {pub_name}: {e}', exc_info=True)
- raise
-
- @property
- def cert_params(self):
- """
- Returns the tls artifact paths.
-
- Returns:
- cert_path, key_path (tuple): the path to tls cert and key.
- """
- return self.cert_path, self.key_path
-
-
-class _DmaapMrClient:
- def __init__(self, aaf_creds, ca_cert_path, enable_tls, cert_params, **kwargs):
- """
- A DMaaP Message Router utility class.
- Sub classes should be invoked via the AppConfig.get_mr_{pub|sub} only.
- Args:
- aaf_creds (dict): a dict of aaf secure credentials.
- ca_cert_path (str): path to the ca certificate.
- enable_tls (bool): TLS if True, else False
- cert_params (tuple): client side (cert, key) tuple.
- **kwargs: a dict of streams_{subscribes|publishes} data.
- """
- self.topic_url = kwargs.get('dmaap_info').get('topic_url')
- self.aaf_id = aaf_creds.get('aaf_id')
- self.aaf_pass = aaf_creds.get('aaf_pass')
- self.ca_cert_path = ca_cert_path
- self.enable_tls = enable_tls
- self.cert_params = cert_params
-
-
-class _MrPub(_DmaapMrClient):
- def __init__(self, pub_name, aaf_creds, ca_cert_path, enable_tls, cert_params, **kwargs):
- self.pub_name = pub_name
- super().__init__(aaf_creds, ca_cert_path, enable_tls, cert_params, **kwargs)
-
- @mdc_handler
- def publish_to_topic(self, event_json, **kwargs):
- """
- Publish the event to the DMaaP Message Router topic.
-
- Args:
- event_json (dict): the json data to be published.
-
- Raises:
- Exception: if post request fails.
- """
- try:
- session = requests.Session()
- headers = {'content-type': 'application/json', 'x-transactionid': kwargs['request_id'],
- 'InvocationID': kwargs['invocation_id'],
- 'RequestID': kwargs['request_id']
- }
- logger.info(f'Publishing event to {self.topic_url}')
- response = session.post(self.topic_url, headers=headers,
- auth=HTTPBasicAuth(self.aaf_id, self.aaf_pass), json=event_json,
- verify=(self.ca_cert_path if self.enable_tls else False))
- response.raise_for_status()
- except Exception as e:
- raise e
-
- 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.
- """
- try:
- 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)
- raise e
-
-
-class _MrSub(_DmaapMrClient):
- def __init__(self, sub_name, aaf_creds, ca_cert_path, enable_tls, cert_params, **kwargs):
- self.sub_name = sub_name
- super().__init__(aaf_creds, ca_cert_path, enable_tls, cert_params, **kwargs)
-
- @mdc_handler
- def get_from_topic(self, consumer_id, consumer_group='dcae_pmsh_cg', timeout=1000, **kwargs):
- """
- Returns the json data from the MrTopic.
-
- Args:
- consumer_id (str): Within your subscribers group, a name that uniquely
- identifies your subscribers process.
- consumer_group (str): A name that uniquely identifies your subscribers.
- timeout (int): The request timeout value in mSec.
-
- Returns:
- list[str]: the json response from DMaaP Message Router topic.
- """
- try:
- session = requests.Session()
- headers = {'accept': 'application/json', 'content-type': 'application/json',
- 'InvocationID': kwargs['invocation_id'],
- 'RequestID': kwargs['request_id']}
- logger.info(f'Fetching messages from MR topic: {self.topic_url}')
- response = session.get(f'{self.topic_url}/{consumer_group}/{consumer_id}'
- f'?timeout={timeout}',
- auth=HTTPBasicAuth(self.aaf_id, self.aaf_pass), headers=headers,
- verify=(self.ca_cert_path if self.enable_tls else False))
- if response.status_code == 503:
- logger.error(f'MR Service is unavailable at present: {response.content}')
- pass
- response.raise_for_status()
- if response.ok:
- return response.json()
- except Exception as e:
- logger.error(f'Failed to fetch message from MR: {e}', exc_info=True)
- raise
-
-
-class PeriodicTask(Timer):
- """
- See :class:`Timer`.
- """
-
- def run(self):
- self.function(*self.args, **self.kwargs)
- while not self.finished.wait(self.interval):
- try:
- self.function(*self.args, **self.kwargs)
- except Exception as e:
- logger.error(f'Exception in thread: {self.name}: {e}', exc_info=True)
# SPDX-License-Identifier: Apache-2.0
# ============LICENSE_END=====================================================
import json
+
+from mod.api.services.measurement_group_service import mg_nf_states, \
+ AdministrativeState, MgNfState
from mod.pmsh_config import MRTopic, AppConfig
from mod import logger
-from mod.subscription import AdministrativeState, subscription_nf_states, SubNfState
from mod.api.db_models import MeasurementGroupModel, NfMeasureGroupRelationalModel
from mod.api.services import measurement_group_service
NfMeasureGroupRelationalModel.measurement_grp_name == measurement_group_name,
NfMeasureGroupRelationalModel.nf_name == nf_name
).one_or_none()
- if nf_msg_rel.nf_measure_grp_status == SubNfState.PENDING_DELETE.value:
+ if nf_msg_rel.nf_measure_grp_status == MgNfState.PENDING_DELETE.value:
administrative_state = AdministrativeState.LOCKING.value
- elif nf_msg_rel.nf_measure_grp_status == SubNfState.PENDING_CREATE.value:
+ elif nf_msg_rel.nf_measure_grp_status == MgNfState.PENDING_CREATE.value:
administrative_state = AdministrativeState.UNLOCKED.value
- nf_measure_grp_status = (subscription_nf_states[administrative_state]
+ nf_measure_grp_status = (mg_nf_states[administrative_state]
[response_message]).value
policy_response_handle_functions[administrative_state][response_message](
measurement_group_name=measurement_group_name, status=nf_measure_grp_status,
+++ /dev/null
-# ============LICENSE_START===================================================
-# Copyright (C) 2019-2022 Nordix Foundation.
-# ============================================================================
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-# SPDX-License-Identifier: Apache-2.0
-# ============LICENSE_END=====================================================
-from enum import Enum
-
-from mod import db, logger
-from mod.api.db_models import SubscriptionModel, NfSubRelationalModel, NetworkFunctionModel
-
-
-class SubNfState(Enum):
- PENDING_CREATE = 'PENDING_CREATE'
- CREATE_FAILED = 'CREATE_FAILED'
- CREATED = 'CREATED'
- PENDING_DELETE = 'PENDING_DELETE'
- DELETE_FAILED = 'DELETE_FAILED'
- DELETED = 'DELETED'
-
-
-class AdministrativeState(Enum):
- UNLOCKED = 'UNLOCKED'
- LOCKING = 'LOCKING'
- LOCKED = 'LOCKED'
- FILTERING = 'FILTERING'
-
-
-subscription_nf_states = {
- AdministrativeState.LOCKED.value: {
- 'success': SubNfState.DELETED,
- 'failed': SubNfState.DELETE_FAILED
- },
- AdministrativeState.UNLOCKED.value: {
- 'success': SubNfState.CREATED,
- 'failed': SubNfState.CREATE_FAILED
- },
- AdministrativeState.LOCKING.value: {
- 'success': SubNfState.DELETED,
- 'failed': SubNfState.DELETE_FAILED
- }
-}
-
-
-def _get_nf_objects(nf_sub_relationships):
- nfs = []
- for nf_sub_entry in nf_sub_relationships:
- nf_model_object = NetworkFunctionModel.query.filter(
- NetworkFunctionModel.nf_name == nf_sub_entry.nf_name).one_or_none()
- nfs.append(nf_model_object.to_nf())
- return nfs
-
-
-class Subscription:
- 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):
- self.administrativeState = admin_state
- self.fileBasedGP = file_based_gp
- self.fileLocation = file_location
- self.measurementGroups = meas_groups
-
- def create(self):
- """ Creates a subscription database entry
-
- Returns:
- Subscription object
- """
- try:
- 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,
- 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
- else:
- logger.debug(f'Subscription {self.subscriptionName} already exists,'
- f' returning this subscription..')
- return existing_subscription
- except Exception as e:
- logger.error(f'Failed to create subscription {self.subscriptionName} in the DB: {e}',
- exc_info=True)
- finally:
- db.session.remove()
-
- def update_subscription_status(self):
- """ Updates the status of subscription in subscription table """
- try:
- SubscriptionModel.query.filter(
- SubscriptionModel.subscription_name == self.subscriptionName) \
- .update({SubscriptionModel.status: self.administrativeState},
- synchronize_session='evaluate')
-
- db.session.commit()
- except Exception as e:
- logger.error(f'Failed to update status of subscription: {self.subscriptionName}: {e}',
- exc_info=True)
- finally:
- db.session.remove()
-
- def prepare_subscription_event(self, nf):
- """Prepare the sub event for publishing
-
- Args:
- nf (NetworkFunction): the AAI nf.
-
- Returns:
- dict: the Subscription event to be published.
- """
- try:
- 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:
- change_type = 'CREATE'
-
- sub_event = {
- '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,
- 'operationalPolicyName': self.operational_policy_name,
- 'changeType': change_type,
- 'controlLoopName': self.control_loop_name,
- 'subscription': clean_sub}
- return sub_event
- except Exception as e:
- logger.error(f'Failed to prep Sub event for xNF {nf.nf_name}: {e}', exc_info=True)
- raise
-
- def add_network_function_to_subscription(self, nf, sub_model):
- """ Associates a network function to a Subscription
-
- Args:
- sub_model(SubscriptionModel): The SubscriptionModel from the DB.
- nf(NetworkFunction): A NetworkFunction object.
- """
- try:
- current_nf = nf.create()
- existing_entry = NfSubRelationalModel.query.filter(
- NfSubRelationalModel.subscription_name == self.subscriptionName,
- NfSubRelationalModel.nf_name == current_nf.nf_name).one_or_none()
- if existing_entry is None:
- new_nf_sub = NfSubRelationalModel(self.subscriptionName,
- nf.nf_name, SubNfState.PENDING_CREATE.value)
- sub_model.nfs.append(new_nf_sub)
- db.session.add(sub_model)
- db.session.commit()
- logger.info(f'Network function {current_nf.nf_name} added to Subscription '
- f'{self.subscriptionName}')
- except Exception as e:
- logger.error(f'Failed to add nf {nf.nf_name} to subscription '
- f'{self.subscriptionName}: {e}', exc_info=True)
- logger.debug(f'Subscription {self.subscriptionName} now contains these XNFs:'
- f'{[nf.nf_name for nf.nf_name in self.get_network_functions()]}')
-
- def get(self):
- """ Retrieves a SubscriptionModel object
-
- Returns:
- SubscriptionModel object else None
- """
- sub_model = SubscriptionModel.query.filter(
- SubscriptionModel.subscription_name == self.subscriptionName).one_or_none()
- return sub_model
-
- def get_local_sub_admin_state(self):
- """ Retrieves the subscription admin state
-
- Returns:
- str: The admin state of the SubscriptionModel
- """
- sub_model = SubscriptionModel.query.filter(
- SubscriptionModel.subscription_name == self.subscriptionName).one_or_none()
- db.session.remove()
- return sub_model.status
-
- 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
- """
- try:
- existing_nfs = self.get_network_functions()
- sub_model = self.get()
- 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)
- 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):
- """ Publishes an event to delete a Subscription from an nf
-
- Args:
- nfs(list[NetworkFunction]): A list of NetworkFunction Objects.
- mr_pub (_MrPub): MR publisher
- """
- 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)
- self.update_sub_nf_status(self.subscriptionName,
- SubNfState.PENDING_DELETE.value,
- nf.nf_name)
- except Exception as err:
- raise Exception(f'Error publishing delete event to MR: {err}')
-
- @staticmethod
- def get_all_nfs_subscription_relations():
- """ Retrieves all network function to subscription relations
-
- Returns:
- list(NfSubRelationalModel): NetworkFunctions per Subscription list else empty
- """
- nf_per_subscriptions = NfSubRelationalModel.query.all()
- db.session.remove()
- return nf_per_subscriptions
-
- @staticmethod
- def update_sub_nf_status(subscription_name, status, nf_name):
- """ Updates the status of the subscription for a particular nf
-
- Args:
- subscription_name (str): The subscription name
- nf_name (str): The network function name
- status (str): Status of the subscription
- """
- try:
- NfSubRelationalModel.query.filter(
- NfSubRelationalModel.subscription_name == subscription_name,
- NfSubRelationalModel.nf_name == nf_name). \
- update({NfSubRelationalModel.nf_sub_status: status}, synchronize_session='evaluate')
- db.session.commit()
- except Exception as e:
- logger.error(f'Failed to update status of nf: {nf_name} for subscription: '
- f'{subscription_name}: {e}', exc_info=True)
-
- def get_network_functions(self):
- nf_sub_relationships = NfSubRelationalModel.query.filter(
- NfSubRelationalModel.subscription_name == self.subscriptionName)
- nfs = _get_nf_objects(nf_sub_relationships)
- db.session.remove()
- return nfs
-
- def get_delete_failed_nfs(self):
- nf_sub_relationships = NfSubRelationalModel.query.filter(
- NfSubRelationalModel.subscription_name == self.subscriptionName,
- NfSubRelationalModel.nf_sub_status == SubNfState.DELETE_FAILED.value)
- nfs = _get_nf_objects(nf_sub_relationships)
- db.session.remove()
- return nfs
-
- def get_delete_pending_nfs(self):
- nf_sub_relationships = NfSubRelationalModel.query.filter(
- NfSubRelationalModel.subscription_name == self.subscriptionName,
- NfSubRelationalModel.nf_sub_status == SubNfState.PENDING_DELETE.value)
- nfs = _get_nf_objects(nf_sub_relationships)
- db.session.remove()
- return nfs
+++ /dev/null
-# ============LICENSE_START===================================================
-# 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.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-# SPDX-License-Identifier: Apache-2.0
-# ============LICENSE_END=====================================================
-from jsonschema import ValidationError
-
-from mod import logger, aai_client
-from mod.network_function import NetworkFunctionFilter
-from mod.subscription import AdministrativeState
-
-
-class SubscriptionHandler:
- def __init__(self, mr_pub, aai_sub, app, app_conf):
- self.mr_pub = mr_pub
- self.aai_sub = aai_sub
- self.app = app
- self.app_conf = app_conf
- self.aai_event_thread = None
-
- def execute(self):
- """
- Checks for changes of administrative state in config and proceeds to process
- the Subscription if a change has occurred
- """
- self.app.app_context().push()
- try:
- local_admin_state = self.app_conf.subscription.get_local_sub_admin_state()
- if local_admin_state == AdministrativeState.LOCKING.value:
- self._check_for_failed_nfs()
- else:
- self.app_conf.refresh_config()
- self.app_conf.validate_sub_schema()
- new_administrative_state = self.app_conf.subscription.administrativeState
- if local_admin_state == new_administrative_state:
- logger.info(f'Administrative State did not change in the app config: '
- f'{new_administrative_state}')
- else:
- self._check_state_change(local_admin_state, new_administrative_state)
- except (ValidationError, TypeError) as err:
- logger.error(f'Error occurred during validation of subscription schema {err}',
- exc_info=True)
- except Exception as err:
- logger.error(f'Error occurred during the activation/deactivation process {err}',
- exc_info=True)
-
- def _check_state_change(self, local_admin_state, new_administrative_state):
- if new_administrative_state == AdministrativeState.UNLOCKED.value:
- logger.info(f'Administrative State has changed from {local_admin_state} '
- f'to {new_administrative_state}.')
- self._activate(new_administrative_state)
- elif new_administrative_state == AdministrativeState.LOCKED.value:
- logger.info(f'Administrative State has changed from {local_admin_state} '
- f'to {new_administrative_state}.')
- self._deactivate()
- else:
- raise Exception(f'Invalid AdministrativeState: {new_administrative_state}')
-
- def _activate(self, new_administrative_state):
- if not self.app_conf.nf_filter:
- self.app_conf.nf_filter = NetworkFunctionFilter(**self.app_conf.subscription.nfFilter)
- self.app_conf.subscription.update_sub_params(new_administrative_state,
- self.app_conf.subscription.fileBasedGP,
- 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.subscription.update_subscription_status()
-
- def _deactivate(self):
- nfs = self.app_conf.subscription.get_network_functions()
- if nfs:
- 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.subscription.update_subscription_status()
-
- def stop_aai_event_thread(self):
- if self.aai_event_thread is not None:
- self.aai_event_thread.cancel()
- self.aai_event_thread = None
- logger.info('Stopping polling for NFs events on AAI-EVENT topic in MR.')
-
- def _check_for_failed_nfs(self):
- logger.info('Checking for DELETE_FAILED NFs before LOCKING Subscription.')
- del_failed_nfs = self.app_conf.subscription.get_delete_failed_nfs()
- if del_failed_nfs or self.app_conf.subscription.get_delete_pending_nfs():
- for nf in del_failed_nfs:
- nf_model = nf.get(nf.nf_name)
- if nf_model.retry_count < 3:
- 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)
- nf.increment_retry_count()
- else:
- logger.error(f'Failed to delete the subscription '
- f'{self.app_conf.subscription.subscriptionName} '
- f'from NF: {nf.nf_name} after {nf_model.retry_count} '
- f'attempts. Removing NF from DB')
- nf.delete(nf_name=nf.nf_name)
- else:
- logger.info('Proceeding to LOCKED adminState.')
- self.app_conf.subscription.administrativeState = AdministrativeState.LOCKED.value
- self.app_conf.subscription.update_subscription_status()
from mod import db, create_app, launch_api_server, logger
from mod.exit_handler import ExitHandler
from mod.pmsh_config import AppConfig
-from mod.pmsh_utils import PeriodicTask
from mod.policy_response_handler import PolicyResponseHandler
+from threading import Timer
+
+
+class PeriodicTask(Timer):
+ """
+ See :class:`Timer`.
+ """
+
+ def run(self):
+ self.function(*self.args, **self.kwargs)
+ while not self.finished.wait(self.interval):
+ try:
+ self.function(*self.args, **self.kwargs)
+ except Exception as e:
+ logger.error(f'Exception in thread: {self.name}: {e}', exc_info=True)
def main():
from mod import create_app, db
from mod.api.db_models import NetworkFunctionFilterModel, MeasurementGroupModel, \
SubscriptionModel, NetworkFunctionModel, NfSubRelationalModel
-from mod.network_function import NetworkFunctionFilter
-from mod.pmsh_utils import AppConfig
-from mod.pmsh_config import AppConfig as NewAppConfig
+from mod.pmsh_config import AppConfig
def get_pmsh_config(file_path='data/cbs_data_1.json'):
cls.app_context = cls.app.app_context()
cls.app_context.push()
- @patch('mod.pmsh_utils.AppConfig._get_pmsh_config', MagicMock(return_value=get_pmsh_config()))
+ @patch('mod.pmsh_config.AppConfig._get_config', MagicMock(return_value=get_pmsh_config()))
def setUp(self):
os.environ['AAI_SERVICE_PORT'] = '8443'
db.create_all()
self.app_conf = AppConfig()
- self.app_conf.nf_filter = NetworkFunctionFilter(**self.app_conf.subscription.nfFilter)
-
- @patch('mod.pmsh_config.AppConfig._get_config', MagicMock(return_value=get_pmsh_config()))
- def setUpAppConf(self):
- self.pmsh_app_conf = NewAppConfig()
def tearDown(self):
db.drop_all()
"key_path":"/opt/app/pmsh/etc/certs/key.pem",
"ca_cert_path":"/opt/app/pmsh/etc/certs/cacert.pem",
"enable_tls":"true",
- "pmsh_policy":{
- "subscription":{
- "subscriptionName":"ExtraPM-All-gNB-R2B",
- "administrativeState":"UNLOCKED",
- "fileBasedGP":15,
- "fileLocation":"\/pm\/pm.xml",
- "nfFilter":{
- "nfNames":[
- "^pnf.*",
- "^vnf.*"
- ],
- "modelInvariantIDs":[
-
- ],
- "modelVersionIDs":[
-
- ],
- "modelNames":[
-
- ]
- },
- "measurementGroups":[
- {
- "measurementGroup":{
- "measurementTypes":[
- {
- "measurementType":"countera"
- },
- {
- "measurementType":"counterb"
- }
- ],
- "managedObjectDNsBasic":[
- {
- "DN":"dna"
- },
- {
- "DN":"dnb"
- }
- ]
- }
- },
- {
- "measurementGroup":{
- "measurementTypes":[
- {
- "measurementType":"counterc"
- },
- {
- "measurementType":"counterd"
- }
- ],
- "managedObjectDNsBasic":[
- {
- "DN":"dnc"
- },
- {
- "DN":"dnd"
- }
- ]
- }
- }
- ]
- }
- },
"streams_subscribes":{
"aai_subscriber":{
"type":"message_router",
}
}
}
-}
\ No newline at end of file
+}
+++ /dev/null
-{
- "config":{
- "control_loop_name":"pmsh-control-loop",
- "operational_policy_name":"pmsh-operational-policy",
- "aaf_password":"demo123456!",
- "aaf_identity":"dcae@dcae.onap.org",
- "cert_path":"/opt/app/pmsh/etc/certs/cert.pem",
- "key_path":"/opt/app/pmsh/etc/certs/key.pem",
- "ca_cert_path":"/opt/app/pmsh/etc/certs/cacert.pem",
- "enable_tls":"true",
- "pmsh_policy":{
- "subscription":{
- "subscriptionName":"ExtraPM-All-gNB-R2B",
- "administrativeState":"UNLOCKED",
- "fileBasedGP":15,
- "fileLocation":"\/pm\/pm.xml",
- "nfFilter":{
- "nfNames":[
-
- ],
- "modelInvariantIDs":[
-
- ],
- "modelVersionIDs":[
-
- ],
- "modelNames":[
-
- ]
- },
- "measurementGroups":[
- {
- "measurementGroup":{
- "measurementTypes":[
- {
- "measurementType":"countera"
- },
- {
- "measurementType":"counterb"
- }
- ],
- "managedObjectDNsBasic":[
- {
- "DN":"dna"
- },
- {
- "DN":"dnb"
- }
- ]
- }
- },
- {
- "measurementGroup":{
- "measurementTypes":[
- {
- "measurementType":"counterc"
- },
- {
- "measurementType":"counterd"
- }
- ],
- "managedObjectDNsBasic":[
- {
- "DN":"dnc"
- },
- {
- "DN":"dnd"
- }
- ]
- }
- }
- ]
- }
- },
- "streams_subscribes":{
- "aai_subscriber":{
- "type":"message_router",
- "dmaap_info":{
- "topic_url":"https://message-router:3905/events/AAI_EVENT",
- "client_role":"org.onap.dcae.aaiSub",
- "location":"san-francisco",
- "client_id":"1575976809466"
- }
- },
- "policy_pm_subscriber":{
- "type":"message_router",
- "dmaap_info":{
- "topic_url":"https://message-router:3905/events/org.onap.dmaap.mr.PM_SUBSCRIPTIONS",
- "client_role":"org.onap.dcae.pmSubscriber",
- "location":"san-francisco",
- "client_id":"1575876809456"
- }
- }
- },
- "streams_publishes":{
- "policy_pm_publisher":{
- "type":"message_router",
- "dmaap_info":{
- "topic_url":"https://message-router:3905/events/org.onap.dmaap.mr.PM_SUBSCRIPTIONS",
- "client_role":"org.onap.dcae.pmPublisher",
- "location":"san-francisco",
- "client_id":"1475976809466"
- }
- },
- "other_publisher":{
- "type":"message_router",
- "dmaap_info":{
- "topic_url":"https://message-router:3905/events/org.onap.dmaap.mr.SOME_OTHER_TOPIC",
- "client_role":"org.onap.dcae.pmControlPub",
- "location":"san-francisco",
- "client_id":"1875976809466"
- }
- }
- }
- }
-}
\ No newline at end of file
+++ /dev/null
-{\r
- "nfName":"pnf_1",\r
- "blueprintName": "some-name",\r
- "blueprintVersion": "some-version",\r
- "operationalPolicyName":"pmsh-operational-policy",\r
- "changeType":"CREATE",\r
- "controlLoopName":"pmsh-control-loop",\r
- "ipAddress": "204.120.0.15",\r
- "subscription":{\r
- "subscriptionName":"ExtraPM-All-gNB-R2B",\r
- "administrativeState":"UNLOCKED",\r
- "fileBasedGP":15,\r
- "fileLocation":"/pm/pm.xml",\r
- "measurementGroups":[\r
- {\r
- "measurementGroup":{\r
- "measurementTypes":[\r
- {\r
- "measurementType":"countera"\r
- },\r
- {\r
- "measurementType":"counterb"\r
- }\r
- ],\r
- "managedObjectDNsBasic":[\r
- {\r
- "DN":"dna"\r
- },\r
- {\r
- "DN":"dnb"\r
- }\r
- ]\r
- }\r
- },\r
- {\r
- "measurementGroup":{\r
- "measurementTypes":[\r
- {\r
- "measurementType":"counterc"\r
- },\r
- {\r
- "measurementType":"counterd"\r
- }\r
- ],\r
- "managedObjectDNsBasic":[\r
- {\r
- "DN":"dnc"\r
- },\r
- {\r
- "DN":"dnd"\r
- }\r
- ]\r
- }\r
- }\r
- ]\r
- }\r
-}\r
from unittest.mock import patch
from mod.api.custom_exception import InvalidDataException, DataConflictException
+from mod.api.services.measurement_group_service import MgNfState
from mod.network_function import NetworkFunction, NetworkFunctionFilter
from mod.pmsh_config import AppConfig
from mod import db, aai_client
from mod.api.services import measurement_group_service, nf_service
from mod.api.db_models import MeasurementGroupModel, NfMeasureGroupRelationalModel, \
SubscriptionModel, NetworkFunctionModel
-from mod.subscription import SubNfState
class MeasurementGroupServiceTestCase(BaseClassSetup):
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
def setUp(self):
super().setUp()
'r') as data:
self.good_model_info = data.read()
- def tearDown(self):
- super().tearDown()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
@patch.object(AppConfig, 'publish_to_topic')
def test_publish_measurement_group(self, mock_mr):
- super().setUpAppConf()
+ super().setUp()
nf_1 = NetworkFunction(**{'nf_name': 'pnf_1',
'ipv4_address': '204.120.0.15',
'ipv6_address': '2001:db8:3333:4444:5555:6666:7777:8888',
def test_apply_nf_to_measurement_group_status(self):
measurement_group_service.apply_nf_status_to_measurement_group(
- "pnf_test", "measure_grp_name", SubNfState.PENDING_CREATE.value)
+ "pnf_test", "measure_grp_name", MgNfState.PENDING_CREATE.value)
db.session.commit()
measurement_grp_rel = (NfMeasureGroupRelationalModel.query.filter(
NfMeasureGroupRelationalModel.measurement_grp_name == 'measure_grp_name',
NfMeasureGroupRelationalModel.nf_name == 'pnf_test').one_or_none())
self.assertIsNotNone(measurement_grp_rel)
self.assertEqual(measurement_grp_rel.nf_measure_grp_status,
- SubNfState.PENDING_CREATE.value)
+ MgNfState.PENDING_CREATE.value)
def test_update_measurement_group_nf_status(self):
measurement_group_service.apply_nf_status_to_measurement_group(
- "pnf_test", "measure_grp_name", SubNfState.PENDING_CREATE.value)
+ "pnf_test", "measure_grp_name", MgNfState.PENDING_CREATE.value)
measurement_group_service.update_measurement_group_nf_status(
- "measure_grp_name", SubNfState.CREATED.value, "pnf_test")
+ "measure_grp_name", MgNfState.CREATED.value, "pnf_test")
db.session.commit()
measurement_grp_rel = (NfMeasureGroupRelationalModel.query.filter(
NfMeasureGroupRelationalModel.measurement_grp_name == 'measure_grp_name',
NfMeasureGroupRelationalModel.nf_name == 'pnf_test').one_or_none())
self.assertIsNotNone(measurement_grp_rel)
self.assertEqual(measurement_grp_rel.nf_measure_grp_status,
- SubNfState.CREATED.value)
+ MgNfState.CREATED.value)
def test_delete_nf_to_measurement_group_without_nf_delete(self):
nf = NetworkFunction(nf_name='pnf_test1')
nf_service.save_nf(nf)
db.session.commit()
measurement_group_service.apply_nf_status_to_measurement_group(
- "pnf_test1", "measure_grp_name1", SubNfState.PENDING_CREATE.value)
+ "pnf_test1", "measure_grp_name1", MgNfState.PENDING_CREATE.value)
measurement_group_service.apply_nf_status_to_measurement_group(
- "pnf_test1", "measure_grp_name2", SubNfState.PENDING_CREATE.value)
+ "pnf_test1", "measure_grp_name2", MgNfState.PENDING_CREATE.value)
measurement_group_service.delete_nf_to_measurement_group(
- "pnf_test1", "measure_grp_name1", SubNfState.DELETED.value)
+ "pnf_test1", "measure_grp_name1", MgNfState.DELETED.value)
measurement_grp_rel = (NfMeasureGroupRelationalModel.query.filter(
NfMeasureGroupRelationalModel.measurement_grp_name == 'measure_grp_name1',
NfMeasureGroupRelationalModel.nf_name == 'pnf_test1').one_or_none())
nf_service.save_nf(nf)
db.session.commit()
measurement_group_service.apply_nf_status_to_measurement_group(
- "pnf_test2", "measure_grp_name2", SubNfState.PENDING_CREATE.value)
+ "pnf_test2", "measure_grp_name2", MgNfState.PENDING_CREATE.value)
measurement_group_service.delete_nf_to_measurement_group(
- "pnf_test2", "measure_grp_name2", SubNfState.DELETED.value)
+ "pnf_test2", "measure_grp_name2", MgNfState.DELETED.value)
measurement_grp_rel = (NfMeasureGroupRelationalModel.query.filter(
NfMeasureGroupRelationalModel.measurement_grp_name == 'measure_grp_name2',
NfMeasureGroupRelationalModel.nf_name == 'pnf_test2').one_or_none())
nf_service.save_nf(nf)
db.session.commit()
measurement_group_service.apply_nf_status_to_measurement_group(
- "pnf_test2", "measure_grp_name2", SubNfState.PENDING_CREATE.value)
+ "pnf_test2", "measure_grp_name2", MgNfState.PENDING_CREATE.value)
nf_delete_func.side_effect = Exception('delete failed')
measurement_group_service.delete_nf_to_measurement_group(
- "pnf_test2", "measure_grp_name2", SubNfState.DELETED.value)
+ "pnf_test2", "measure_grp_name2", MgNfState.DELETED.value)
measurement_grp_rel = (NfMeasureGroupRelationalModel.query.filter(
NfMeasureGroupRelationalModel.measurement_grp_name == 'measure_grp_name2',
NfMeasureGroupRelationalModel.nf_name == 'pnf_test2').one_or_none())
def test_update_nf_to_measurement_group_failure(self, mock_logger, db_commit_call):
db_commit_call.side_effect = Exception('update failed')
measurement_group_service.update_measurement_group_nf_status(
- "measure_grp_name2", SubNfState.CREATE_FAILED.value, "pnf_test2")
+ "measure_grp_name2", MgNfState.CREATE_FAILED.value, "pnf_test2")
mock_logger.assert_called_with('Failed to update nf: pnf_test2 for '
'measurement group: measure_grp_name2 due to: update failed')
subscription = subscription.replace('msrmt_grp_name', new_msrmt_grp_name)
return subscription
- @patch.object(AppConfig, 'publish_to_topic')
- def test_update_admin_status_to_locking(self, mock_mr):
- super().setUpAppConf()
+ def test_update_admin_status_to_locking(self):
+ super().setUp()
sub = create_subscription_data('sub')
nf_list = create_multiple_network_function_data(['pnf_101', 'pnf_102'])
db.session.add(sub)
measurement_group_service. \
apply_nf_status_to_measurement_group(nf.nf_name, sub.measurement_groups[0].
measurement_group_name,
- SubNfState.CREATED.value)
+ MgNfState.CREATED.value)
db.session.commit()
measurement_group_service.update_admin_status(sub.measurement_groups[0], 'LOCKED')
meas_grp = measurement_group_service.query_meas_group_by_name('sub', 'MG1')
NfMeasureGroupRelationalModel.measurement_grp_name == meas_grp.measurement_group_name)\
.all()
for nf in meas_group_nfs:
- self.assertEqual(nf.nf_measure_grp_status, SubNfState.PENDING_DELETE.value)
+ self.assertEqual(nf.nf_measure_grp_status, MgNfState.PENDING_DELETE.value)
- @patch.object(AppConfig, 'publish_to_topic')
- def test_update_admin_status_to_locked(self, mock_mr):
- super().setUpAppConf()
+ def test_update_admin_status_to_locked(self):
+ super().setUp()
sub = create_subscription_data('sub')
db.session.add(sub)
measurement_group_service.update_admin_status(sub.measurement_groups[0], 'LOCKED')
self.assertEqual(meas_grp.measurement_group_name, 'MG1')
self.assertEqual(meas_grp.administrative_state, 'LOCKED')
- @patch.object(AppConfig, 'publish_to_topic')
@patch.object(aai_client, '_get_all_aai_nf_data')
@patch.object(aai_client, 'get_aai_model_data')
@patch.object(NetworkFunctionFilter, 'get_network_function_filter')
def test_update_admin_status_to_unlocked_with_no_nfs(self, mock_filter_call,
- mock_model_aai, mock_aai, mock_mr):
+ 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)
- super().setUpAppConf()
+ super().setUp()
sub = create_subscription_data('sub')
sub.nfs = []
db.session.add(sub)
NfMeasureGroupRelationalModel.measurement_grp_name == meas_grp.measurement_group_name)\
.all()
for nf in meas_group_nfs:
- self.assertEqual(nf.nf_measure_grp_status, SubNfState.PENDING_CREATE.value)
+ self.assertEqual(nf.nf_measure_grp_status, MgNfState.PENDING_CREATE.value)
- @patch.object(AppConfig, 'publish_to_topic')
@patch.object(aai_client, '_get_all_aai_nf_data')
@patch.object(aai_client, 'get_aai_model_data')
@patch.object(NetworkFunctionFilter, 'get_network_function_filter')
def test_update_admin_status_to_unlocking(self, mock_filter_call,
- mock_model_aai, mock_aai, mock_mr):
+ 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)
- super().setUpAppConf()
+ super().setUp()
sub = create_subscription_data('sub')
db.session.add(sub)
db.session.commit()
NfMeasureGroupRelationalModel.measurement_grp_name == meas_grp.measurement_group_name)\
.all()
for nf in meas_group_nfs:
- self.assertEqual(nf.nf_measure_grp_status, SubNfState.PENDING_CREATE.value)
+ self.assertEqual(nf.nf_measure_grp_status, MgNfState.PENDING_CREATE.value)
def test_update_admin_status_for_missing_measurement_group(self):
try:
'for admin status update')
def test_update_admin_status_for_data_conflict(self):
- super().setUpAppConf()
+ super().setUp()
sub = create_subscription_data('sub1')
sub.measurement_groups[0].administrative_state = 'LOCKING'
try:
'meas group name: MG1')
def test_update_admin_status_for_same_state(self):
- super().setUpAppConf()
+ super().setUp()
sub = create_subscription_data('sub1')
try:
measurement_group_service.update_admin_status(sub.measurement_groups[0], 'UNLOCKED')
measurement_group_service. \
apply_nf_status_to_measurement_group(nf.nf_name, sub.measurement_groups[1].
measurement_group_name,
- SubNfState.PENDING_DELETE.value)
+ MgNfState.PENDING_DELETE.value)
db.session.commit()
measurement_group_service.lock_nf_to_meas_grp(
- "pnf_101", "MG2", SubNfState.DELETED.value)
+ "pnf_101", "MG2", MgNfState.DELETED.value)
measurement_grp_rel = (NfMeasureGroupRelationalModel.query.filter(
NfMeasureGroupRelationalModel.measurement_grp_name == 'MG2',
NfMeasureGroupRelationalModel.nf_name == 'pnf_101').one_or_none())
measurement_group_service. \
apply_nf_status_to_measurement_group(nf.nf_name, sub.measurement_groups[1].
measurement_group_name,
- SubNfState.PENDING_DELETE.value)
+ MgNfState.PENDING_DELETE.value)
db.session.commit()
measurement_group_service.lock_nf_to_meas_grp(
- "pnf_101", "MG2", SubNfState.DELETED.value)
+ "pnf_101", "MG2", MgNfState.DELETED.value)
measurement_grp_rel = (NfMeasureGroupRelationalModel.query.filter(
NfMeasureGroupRelationalModel.measurement_grp_name == 'MG2',
NfMeasureGroupRelationalModel.nf_name == 'pnf_101').one_or_none())
nf = NetworkFunction(nf_name='pnf_test2')
nf_service.save_nf(nf)
measurement_group_service.apply_nf_status_to_measurement_group(
- "pnf_test2", "MG2", SubNfState.PENDING_DELETE.value)
+ "pnf_test2", "MG2", MgNfState.PENDING_DELETE.value)
db.session.commit()
measurement_group_service.filter_nf_to_meas_grp(
- "pnf_test2", "MG2", SubNfState.DELETED.value)
+ "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())
nf = NetworkFunction(nf_name='pnf_test2')
nf_service.save_nf(nf)
measurement_group_service.apply_nf_status_to_measurement_group(
- "pnf_test2", "MG2", SubNfState.PENDING_CREATE.value)
+ "pnf_test2", "MG2", MgNfState.PENDING_CREATE.value)
db.session.commit()
measurement_group_service.filter_nf_to_meas_grp(
- "pnf_test2", "MG2", SubNfState.CREATED.value)
+ "pnf_test2", "MG2", MgNfState.CREATED.value)
measurement_grp_rel = (NfMeasureGroupRelationalModel.query.filter(
NfMeasureGroupRelationalModel.measurement_grp_name == 'MG2',
NfMeasureGroupRelationalModel.nf_name == 'pnf_test2').one_or_none())
# ============LICENSE_START===================================================
-# Copyright (C) 2021 Nordix Foundation.
+# Copyright (C) 2021-2022 Nordix Foundation.
# ============================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
class NetworkFunctionServiceTestCase(BaseClassSetup):
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
def setUp(self):
super().setUp()
'r') as data:
self.good_model_info = data.read()
- def tearDown(self):
- super().tearDown()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
def create_test_subs(self, new_sub_name, new_msrmt_grp_name):
subscription = self.subscription_request.replace('ExtraPM-All-gNB-R2B', new_sub_name)
subscription = subscription.replace('msrmt_grp_name', new_msrmt_grp_name)
from mod.api.db_models import SubscriptionModel, MeasurementGroupModel, \
NfMeasureGroupRelationalModel, NetworkFunctionModel, NfSubRelationalModel, \
convert_db_string_to_list, NetworkFunctionFilterModel
+from mod.api.services.measurement_group_service import MgNfState
from mod.network_function import NetworkFunctionFilter
-from mod.subscription import SubNfState
from mod import aai_client, db
from mod.api.custom_exception import DuplicateDataException, InvalidDataException, \
DataConflictException
class SubscriptionServiceTestCase(BaseClassSetup):
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
def setUp(self):
super().setUp()
'r') as data:
self.good_model_info = data.read()
- def tearDown(self):
- super().tearDown()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
def create_test_subs(self, new_sub_name, new_msrmt_grp_name):
subscription = self.subscription_request.replace('ExtraPM-All-gNB-R2B', new_sub_name)
subscription = subscription.replace('msrmt_grp_name', new_msrmt_grp_name)
NfMeasureGroupRelationalModel.measurement_grp_name == 'msrmt_grp_name-new')).all()
for pubslished_event in msr_grp_nf_rel:
self.assertEqual(pubslished_event.nf_measure_grp_status,
- SubNfState.PENDING_CREATE.value)
+ MgNfState.PENDING_CREATE.value)
@patch('mod.api.services.subscription_service.save_nf_filter', MagicMock(return_value=None))
@patch.object(AppConfig, 'publish_to_topic')
def test_perform_validation_existing_sub(self):
try:
+ subscription = create_subscription_data('ExtraPM-All-gNB-R2B')
+ db.session.add(subscription)
+ db.session.commit()
subscription_service.create_subscription(json.loads(self.subscription_request)
['subscription'])
except DuplicateDataException as exception:
"No value provided for measurement group name")
def test_validate_nf_filter_with_no_filter_values(self):
- nfFilter = '{"nfNames": [],"modelInvariantIDs": [], ' \
- '"modelVersionIDs": [],"modelNames": []}'
+ nf_filter = '{"nfNames": [],"modelInvariantIDs": [], ' \
+ '"modelVersionIDs": [],"modelNames": []}'
try:
- subscription_service.validate_nf_filter(json.loads(nfFilter))
+ subscription_service.validate_nf_filter(json.loads(nf_filter))
except InvalidDataException as invalidEx:
self.assertEqual(invalidEx.args[0],
"At least one filter within nfFilter must not be empty")
self.assertEqual(len(meas_group_nfs), 2)
self.assertEqual(meas_group_nfs[0].nf_name, 'pnf201')
self.assertEqual(meas_group_nfs[0].nf_measure_grp_status,
- SubNfState.PENDING_CREATE.value)
+ MgNfState.PENDING_CREATE.value)
self.assertEqual(meas_group_nfs[1].nf_name, 'pnf_33_ericsson')
self.assertEqual(meas_group_nfs[1].nf_measure_grp_status,
- SubNfState.PENDING_CREATE.value)
+ MgNfState.PENDING_CREATE.value)
meas_grp = measurement_group_service.query_meas_group_by_name('sub_01', 'msg_01')
self.assertEqual(meas_grp.administrative_state, 'UNLOCKED')
# Creating test data for update filter function
.all()
self.assertEqual(meas_group_nfs[0].nf_name, 'pnf201')
self.assertEqual(meas_group_nfs[0].nf_measure_grp_status,
- SubNfState.PENDING_DELETE.value)
+ MgNfState.PENDING_DELETE.value)
self.assertEqual(meas_group_nfs[1].nf_name, 'pnf_33_ericsson')
self.assertEqual(meas_group_nfs[1].nf_measure_grp_status,
- SubNfState.PENDING_DELETE.value)
+ MgNfState.PENDING_DELETE.value)
self.assertEqual(meas_group_nfs[2].nf_name, 'xnf111')
self.assertEqual(meas_group_nfs[2].nf_measure_grp_status,
- SubNfState.PENDING_CREATE.value)
+ MgNfState.PENDING_CREATE.value)
meas_grp = measurement_group_service.query_meas_group_by_name('sub_01', 'msg_01')
self.assertEqual(meas_grp.administrative_state, 'FILTERING')
from os import path
from unittest.mock import patch, MagicMock
from mod.aai_event_handler import AAIEventHandler
-from mod.api.db_models import NetworkFunctionModel, NetworkFunctionFilterModel, \
- MeasurementGroupModel, SubscriptionModel
-from mod.subscription import AdministrativeState
-from tests.base_setup import BaseClassSetup
+from mod.api.db_models import NetworkFunctionModel
+from tests.base_setup import BaseClassSetup, create_subscription_data
from mod import db
class AAIEventHandlerTest(BaseClassSetup):
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
def setUp(self):
super().setUp()
- super().setUpAppConf()
+ subscription = create_subscription_data('aai_event_handler')
+ subscription.measurement_groups[1].administravtive_sate = 'UNLOCKED'
+ db.session.add(subscription)
+ db.session.add(subscription.measurement_groups[0])
+ db.session.add(subscription.measurement_groups[1])
+ db.session.add(subscription.network_filter)
+ db.session.add(subscription.nfs[0])
+ db.session.commit()
with open(path.join(path.dirname(__file__), 'data/mr_aai_events.json'), 'r') as data:
self.mr_aai_events = json.load(data)["mr_response"]
- def tearDown(self):
- super().tearDown()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
@patch('mod.pmsh_config.AppConfig.get_from_topic')
@patch('mod.network_function.NetworkFunction.set_nf_model_params')
@patch('mod.network_function.NetworkFunction.delete')
apply_nfs_to_measure_grp):
mock_set_sdnc_params.return_value = True
mr_aai_mock.return_value = self.mr_aai_events
- aai_handler = AAIEventHandler(self.app)
- subscription = SubscriptionModel(subscription_name='ExtraPM-All-gNB-R2B2',
- operational_policy_name='operation_policy',
- control_loop_name="control-loop",
- status=AdministrativeState.UNLOCKED.value)
- db.session.add(subscription)
- db.session.commit()
- generate_nf_filter_measure_grp('ExtraPM-All-gNB-R2B', 'msr_grp_name')
- generate_nf_filter_measure_grp('ExtraPM-All-gNB-R2B2', 'msr_grp_name2')
+ nf_to_subscription = create_subscription_data('nf_to_subscription')
+ nf_to_subscription.measurement_groups[0].measurement_group_name = 'NF_MG_ONE'
+ nf_to_subscription.measurement_groups[1].measurement_group_name = 'NF_MG_TWO'
+ db.session.add(nf_to_subscription)
+ db.session.add(nf_to_subscription.measurement_groups[0])
+ db.session.add(nf_to_subscription.network_filter)
db.session.commit()
+ aai_handler = AAIEventHandler(self.app)
aai_handler.execute()
self.assertEqual(apply_nfs_to_measure_grp.call_count, 2)
mr_aai_mock.return_value = self.mr_aai_events
apply_nfs_to_measure_grp.side_effect = Exception("publish failed")
aai_handler = AAIEventHandler(self.app)
- generate_nf_filter_measure_grp('ExtraPM-All-gNB-R2B', 'msr_grp_name3')
- db.session.commit()
aai_handler.execute()
mock_logger.assert_called_with('Failed to process AAI event for '
- 'subscription: ExtraPM-All-gNB-R2B '
+ 'subscription: aai_event_handler '
'due to: publish failed')
@patch('mod.pmsh_config.AppConfig.publish_to_topic', MagicMock(return_value=None))
mock_set_sdnc_params.return_value = True
mr_aai_mock.side_effect = Exception("AAI failure")
aai_handler = AAIEventHandler(self.app)
- generate_nf_filter_measure_grp('ExtraPM-All-gNB-R2B', 'msr_grp_name3')
- db.session.commit()
aai_handler.execute()
mock_logger.assert_called_with('Failed to process AAI event due to: AAI failure')
-
-
-def generate_nf_filter_measure_grp(sub_name, msg_name):
- nf_filter = NetworkFunctionFilterModel(
- subscription_name=sub_name, nf_names='{^pnf.*, ^vnf.*}',
- model_invariant_ids='{}',
- model_version_ids='{}',
- model_names='{}')
- measurement_group = MeasurementGroupModel(
- subscription_name=sub_name, measurement_group_name=msg_name,
- administrative_state='UNLOCKED', file_based_gp=15, file_location='pm.xml',
- measurement_type=[], managed_object_dns_basic=[])
- db.session.add(nf_filter)
- db.session.add(measurement_group)
# ============LICENSE_START===================================================
-# Copyright (C) 2019-2021 Nordix Foundation.
+# Copyright (C) 2019-2022 Nordix Foundation.
# ============================================================================
# Licensed under the Apache License, Version 2.0 (the 'License');
# you may not use this file except in compliance with the License.
import responses
from requests import Session, HTTPError
-import mod.aai_client as aai_client
-from tests.base_setup import BaseClassSetup
+from mod import aai_client
+from mod.network_function import NetworkFunctionFilter
+from tests.base_setup import BaseClassSetup, create_subscription_data
class AaiClientTestCase(BaseClassSetup):
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
-
def setUp(self):
super().setUp()
+ self.subscription = create_subscription_data('ExtraPM-All-gNB-R2B')
with open(os.path.join(os.path.dirname(__file__), 'data/aai_xnfs.json'), 'r') as data:
self.aai_response_data = data.read()
with open(os.path.join(os.path.dirname(__file__), 'data/aai_model_info.json'), 'r') as data:
self.good_model_info = data.read()
- def tearDown(self):
- super().tearDown()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
@patch('mod.network_function.NetworkFunction.set_nf_model_params')
@patch.object(Session, 'get')
@patch.object(Session, 'put')
mock_get_session.return_value.status_code = 200
mock_get_session.return_value.text = self.good_model_info
mock_get_sdnc_params.return_value = True
- xnfs = aai_client.get_pmsh_nfs_from_aai(self.app_conf, self.app_conf.nf_filter)
- self.assertEqual(self.app_conf.subscription.subscriptionName, 'ExtraPM-All-gNB-R2B')
- self.assertEqual(self.app_conf.subscription.administrativeState, 'UNLOCKED')
+ nf_filter = NetworkFunctionFilter(**self.subscription.network_filter.serialize())
+ xnfs = aai_client.get_pmsh_nfs_from_aai(self.app_conf, nf_filter)
+ self.assertEqual(self.subscription.subscription_name, 'ExtraPM-All-gNB-R2B')
+ self.assertEqual(self.subscription.measurement_groups[0].administrative_state, 'UNLOCKED')
self.assertEqual(len(xnfs), 3)
@patch.object(Session, 'put')
mock_session.return_value.status_code = 404
with mock.patch('mod.aai_client._get_all_aai_nf_data', return_value=None):
with self.assertRaises(RuntimeError):
- aai_client.get_pmsh_nfs_from_aai(self.app_conf, self.app_conf.nf_filter)
+ aai_client.get_pmsh_nfs_from_aai(self.app_conf, self.subscription.network_filter)
@responses.activate
def test_aai_client_get_all_aai_xnf_data_not_found(self):
from tests.base_setup import BaseClassSetup
from mod.api.custom_exception import InvalidDataException, DataConflictException
from mod.api.db_models import SubscriptionModel, NfMeasureGroupRelationalModel
-from mod.subscription import SubNfState
from mod.network_function import NetworkFunctionFilter
from tests.base_setup import create_subscription_data, create_multiple_subscription_data, \
create_multiple_network_function_data
from mod.api.services import measurement_group_service, nf_service, subscription_service
+from mod.api.services.measurement_group_service import MgNfState
class ControllerTestCase(BaseClassSetup):
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
-
def setUp(self):
super().setUp()
- super().setUpAppConf()
with open(os.path.join(os.path.dirname(__file__), 'data/aai_xnfs.json'), 'r') as data:
self.aai_response_data = data.read()
with open(os.path.join(os.path.dirname(__file__), 'data/aai_model_info.json'), 'r') as data:
'data/create_subscription_request.json'), 'r') as data:
self.subscription_request = data.read()
- def tearDown(self):
- super().tearDown()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
def test_status_response_healthy(self):
self.assertEqual(status()['status'], 'healthy')
NfMeasureGroupRelationalModel.measurement_grp_name == mes_grp_name)).all()
for published_event in msr_grp_nf_rel:
self.assertEqual(published_event.nf_measure_grp_status,
- SubNfState.PENDING_CREATE.value)
+ MgNfState.PENDING_CREATE.value)
self.assertEqual(response[1], 201)
def test_post_subscription_duplicate_sub(self):
# Posting the same subscription request stored in previous test to get duplicate response
+ subscription = create_subscription_data('ExtraPM-All-gNB-R2B')
+ db.session.add(subscription)
+ db.session.commit()
response = post_subscription(json.loads(self.subscription_request))
self.assertEqual(response[1], 409)
self.assertEqual(response[0], 'subscription Name: ExtraPM-All-gNB-R2B already exists.')
measurement_group_service. \
apply_nf_status_to_measurement_group(nf.nf_name, sub.measurement_groups[0].
measurement_group_name,
- SubNfState.PENDING_CREATE.value)
+ MgNfState.PENDING_CREATE.value)
db.session.commit()
mg_with_nfs, status_code = get_meas_group_with_nfs('sub1', 'MG1')
self.assertEqual(status_code, HTTPStatus.OK.value)
measurement_group_service. \
apply_nf_status_to_measurement_group(nf.nf_name, sub.measurement_groups[0].
measurement_group_name,
- SubNfState.CREATED.value)
+ MgNfState.CREATED.value)
db.session.commit()
response = update_admin_state('sub1', 'MG1', {'administrativeState': 'LOCKED'})
self.assertEqual(response[1], HTTPStatus.OK.value)
self.assertEqual(mg_with_nfs['measurementGroupName'], 'MG1')
self.assertEqual(mg_with_nfs['administrativeState'], 'LOCKING')
for nf in mg_with_nfs['networkFunctions']:
- self.assertEqual(nf['nfMgStatus'], SubNfState.PENDING_DELETE.value)
+ self.assertEqual(nf['nfMgStatus'], MgNfState.PENDING_DELETE.value)
@patch('mod.pmsh_config.AppConfig.publish_to_topic', MagicMock(return_value=None))
@patch.object(aai_client, '_get_all_aai_nf_data')
self.assertEqual(mg_with_nfs['measurementGroupName'], 'MG2')
self.assertEqual(mg_with_nfs['administrativeState'], 'UNLOCKED')
for nf in mg_with_nfs['networkFunctions']:
- self.assertEqual(nf['nfMgStatus'], SubNfState.PENDING_CREATE.value)
+ self.assertEqual(nf['nfMgStatus'], MgNfState.PENDING_CREATE.value)
@patch('mod.api.services.measurement_group_service.update_admin_status',
MagicMock(side_effect=InvalidDataException('Bad request')))
# ============LICENSE_END=====================================================
import os
from signal import SIGTERM, signal
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
+from mod import db
from mod.exit_handler import ExitHandler
-from tests.base_setup import BaseClassSetup
+from tests.base_setup import BaseClassSetup, create_subscription_data
class ExitHandlerTests(BaseClassSetup):
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
-
- @patch('mod.pmsh_utils.PeriodicTask')
- @patch('mod.pmsh_utils.PeriodicTask')
+ @patch('pmsh_service_main.PeriodicTask')
+ @patch('pmsh_service_main.PeriodicTask')
def setUp(self, mock_periodic_task_aai, mock_periodic_task_policy):
super().setUp()
+ subscription = create_subscription_data('aai_event_handler')
+ subscription.measurement_groups[1].administravtive_sate = 'UNLOCKED'
+ db.session.add(subscription)
+ db.session.add(subscription.measurement_groups[0])
+ db.session.add(subscription.measurement_groups[1])
+ db.session.add(subscription.network_filter)
+ db.session.add(subscription.nfs[0])
+ db.session.commit()
self.mock_aai_event_thread = mock_periodic_task_aai
self.mock_policy_resp_handler_thread = mock_periodic_task_policy
- def tearDown(self):
- super().tearDown()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
+ @patch('mod.pmsh_config.AppConfig.publish_to_topic', MagicMock(return_value=None))
def test_terminate_signal_successful(self):
handler = ExitHandler(periodic_tasks=[self.mock_aai_event_thread,
self.mock_policy_resp_handler_thread])
# ============LICENSE_START===================================================
-# Copyright (C) 2019-2021 Nordix Foundation.
+# Copyright (C) 2019-2022 Nordix Foundation.
# ============================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# ============LICENSE_END=====================================================
import json
import os
-from unittest.mock import patch, Mock
+from unittest.mock import patch
from mod.network_function import NetworkFunction
from tests.base_setup import BaseClassSetup
class NetworkFunctionTests(BaseClassSetup):
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
-
def setUp(self):
super().setUp()
self.nf_1 = NetworkFunction(sdnc_model_name='blah', sdnc_model_version=1.0,
'data/aai_model_info_no_sdnc.json'), 'r') as data:
self.bad_model_info = json.loads(data.read())
- def tearDown(self):
- super().tearDown()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
def test_get_network_function(self):
self.nf_1.create()
nf = NetworkFunction.get('pnf_1')
def test_delete_network_function(self):
for nf in [self.nf_1, self.nf_2]:
- self.app_conf.subscription.add_network_function_to_subscription(nf, Mock())
+ nf.create()
nfs = NetworkFunction.get_all()
self.assertEqual(2, len(nfs))
NetworkFunction.delete(nf_name=self.nf_1.nf_name)
# ============LICENSE_START===================================================
-# Copyright (C) 2021 Nordix Foundation.
+# Copyright (C) 2021-2022 Nordix Foundation.
# ============================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
#
# SPDX-License-Identifier: Apache-2.0
# ============LICENSE_END=====================================================
-from unittest.mock import Mock, patch
+from unittest.mock import patch
import responses
from requests import Session
class PmshConfigTestCase(BaseClassSetup):
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
-
- def setUp(self):
- super().setUpAppConf()
- self.mock_app = Mock()
-
- def tearDown(self):
- super().tearDown()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
def test_config_get_aaf_creds(self):
- self.assertEqual(self.pmsh_app_conf.enable_tls, 'true')
- self.assertEqual(self.pmsh_app_conf.aaf_id, 'dcae@dcae.onap.org')
- self.assertEqual(self.pmsh_app_conf.aaf_pass, 'demo123456!')
+ self.assertEqual(self.app_conf.enable_tls, 'true')
+ self.assertEqual(self.app_conf.aaf_id, 'dcae@dcae.onap.org')
+ self.assertEqual(self.app_conf.aaf_pass, 'demo123456!')
def test_config_get_cert_data(self):
- self.assertEqual(self.pmsh_app_conf.key_path, '/opt/app/pmsh/etc/certs/key.pem')
- self.assertEqual(self.pmsh_app_conf.cert_path, '/opt/app/pmsh/etc/certs/cert.pem')
- self.assertEqual(self.pmsh_app_conf.ca_cert_path, '/opt/app/pmsh/etc/certs/cacert.pem')
+ self.assertEqual(self.app_conf.key_path, '/opt/app/pmsh/etc/certs/key.pem')
+ self.assertEqual(self.app_conf.cert_path, '/opt/app/pmsh/etc/certs/cert.pem')
+ self.assertEqual(self.app_conf.ca_cert_path, '/opt/app/pmsh/etc/certs/cacert.pem')
def test_singleton_instance_is_accessible_using_class_method(self):
my_singleton_instance = AppConfig.get_instance()
def test_mr_pub_publish_to_topic_success(self, mock_session):
mock_session.return_value.status_code = 200
with patch('requests.Session.post') as session_post_call:
- self.pmsh_app_conf.publish_to_topic(MRTopic.POLICY_PM_PUBLISHER.value,
- {"key": "43c4ee19-6b8d-4279-a80f-c507850aae47"})
+ self.app_conf.publish_to_topic(MRTopic.POLICY_PM_PUBLISHER.value,
+ {"key": "43c4ee19-6b8d-4279-a80f-c507850aae47"})
session_post_call.assert_called_once()
@responses.activate
'https://message-router:3905/events/org.onap.dmaap.mr.PM_SUBSCRIPTIONS',
json={"error": "Client Error"}, status=400)
with self.assertRaises(Exception):
- self.pmsh_app_conf.publish_to_topic(MRTopic.POLICY_PM_PUBLISHER.value,
- {"key": "43c4ee19-6b8d-4279-a80f-c507850aae47"})
+ self.app_conf.publish_to_topic(MRTopic.POLICY_PM_PUBLISHER.value,
+ {"key": "43c4ee19-6b8d-4279-a80f-c507850aae47"})
@responses.activate
def test_mr_sub_get_from_topic_success(self):
'https://message-router:3905/events/org.onap.dmaap.mr.PM_SUBSCRIPTIONS/'
'dcae_pmsh_cg/1?timeout=5000',
json={"key": "43c4ee19-6b8d-4279-a80f-c507850aae47"}, status=200)
- mr_topic_data = self.pmsh_app_conf.get_from_topic(MRTopic.POLICY_PM_SUBSCRIBER.value, 1)
+ mr_topic_data = self.app_conf.get_from_topic(MRTopic.POLICY_PM_SUBSCRIBER.value, 1)
self.assertIsNotNone(mr_topic_data)
@responses.activate
'dcae_pmsh_cg/1?timeout=5000',
json={"key": "43c4ee19-6b8d-4279-a80f-c507850aae47"}, status=400)
with self.assertRaises(Exception):
- self.pmsh_app_conf.get_from_topic(MRTopic.POLICY_PM_SUBSCRIBER.value, 1)
+ self.app_conf.get_from_topic(MRTopic.POLICY_PM_SUBSCRIBER.value, 1)
+++ /dev/null
-# ============LICENSE_START===================================================
-# 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.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-# SPDX-License-Identifier: Apache-2.0
-# ============LICENSE_END=====================================================
-from test.support import EnvironmentVarGuard
-from unittest.mock import patch, Mock
-
-import responses
-from jsonschema import ValidationError
-from requests import Session
-from tenacity import RetryError
-
-from mod import get_db_connection_url
-from mod.network_function import NetworkFunction
-from tests.base_setup import BaseClassSetup
-from tests.base_setup import get_pmsh_config
-
-
-class PmshUtilsTestCase(BaseClassSetup):
-
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
-
- def setUp(self):
- super().setUp()
- self.mock_app = Mock()
-
- def tearDown(self):
- super().tearDown()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
- def test_utils_get_mr_sub(self):
- mr_policy_sub = self.app_conf.get_mr_sub('policy_pm_subscriber')
- self.assertTrue(mr_policy_sub.aaf_id, 'dcae@dcae.onap.org')
-
- def test_utils_get_mr_sub_fails_with_invalid_name(self):
- with self.assertRaises(KeyError):
- self.app_conf.get_mr_sub('invalid_sub')
-
- def test_utils_get_mr_pub(self):
- mr_policy_pub = self.app_conf.get_mr_pub('policy_pm_publisher')
- self.assertTrue(mr_policy_pub.aaf_pass, 'demo123456!')
-
- def test_utils_get_mr_pub_fails_with_invalid_name(self):
- with self.assertRaises(KeyError):
- self.app_conf.get_mr_pub('invalid_pub')
-
- def test_utils_get_cert_data(self):
- self.assertEqual(self.app_conf.cert_params, ('/opt/app/pmsh/etc/certs/cert.pem',
- '/opt/app/pmsh/etc/certs/key.pem'))
-
- @patch.object(Session, 'post')
- def test_mr_pub_publish_to_topic_success(self, mock_session):
- mock_session.return_value.status_code = 200
- mr_policy_pub = self.app_conf.get_mr_pub('policy_pm_publisher')
- with patch('requests.Session.post') as session_post_call:
- mr_policy_pub.publish_to_topic({"dummy_val": "43c4ee19-6b8d-4279-a80f-c507850aae47"})
- session_post_call.assert_called_once()
-
- @responses.activate
- def test_mr_pub_publish_to_topic_fail(self):
- responses.add(responses.POST,
- 'https://message-router:3905/events/org.onap.dmaap.mr.PM_SUBSCRIPTIONS',
- json={'error': 'Client Error'}, status=400)
- mr_policy_pub = self.app_conf.get_mr_pub('policy_pm_publisher')
- with self.assertRaises(Exception):
- mr_policy_pub.publish_to_topic({"dummy_val": "43c4ee19-6b8d-4279-a80f-c507850aae47"})
-
- def test_mr_pub_publish_sub_event_data_success(self):
- mr_policy_pub = self.app_conf.get_mr_pub('policy_pm_publisher')
- with patch('mod.pmsh_utils._MrPub.publish_to_topic') as pub_to_topic_call:
- mr_policy_pub.publish_subscription_event_data(
- self.app_conf.subscription,
- NetworkFunction(nf_name='pnf_1',
- model_invariant_id='some-id',
- model_version_id='some-id'))
- pub_to_topic_call.assert_called_once()
-
- @responses.activate
- def test_mr_sub_get_from_topic_success(self):
- policy_mr_sub = self.app_conf.get_mr_sub('policy_pm_subscriber')
- responses.add(responses.GET,
- 'https://message-router:3905/events/org.onap.dmaap.mr.PM_SUBSCRIPTIONS/'
- 'dcae_pmsh_cg/1?timeout=1000',
- json={"dummy_val": "43c4ee19-6b8d-4279-a80f-c507850aae47"}, status=200)
- mr_topic_data = policy_mr_sub.get_from_topic(1)
- self.assertIsNotNone(mr_topic_data)
-
- @responses.activate
- def test_mr_sub_get_from_topic_fail(self):
- policy_mr_sub = self.app_conf.get_mr_sub('policy_pm_subscriber')
- responses.add(responses.GET,
- 'https://message-router:3905/events/org.onap.dmaap.mr.PM_SUBSCRIPTIONS/'
- 'dcae_pmsh_cg/1?timeout=1000',
- json={"dummy_val": "43c4ee19-6b8d-4279-a80f-c507850aae47"}, status=400)
- with self.assertRaises(Exception):
- policy_mr_sub.get_from_topic(1)
-
- def test_get_db_connection_url_success(self):
- self.env = EnvironmentVarGuard()
- self.env.set('PMSH_PG_URL', '1.2.3.4')
- self.env.set('PMSH_PG_USERNAME', 'pmsh')
- self.env.set('PMSH_PG_PASSWORD', 'pass')
- db_url = get_db_connection_url()
- self.assertEqual(db_url, 'postgresql+psycopg2://pmsh:pass@1.2.3.4:5432/pmsh')
-
- def test_get_db_connection_url_fail(self):
- self.env = EnvironmentVarGuard()
- self.env.set('PMSH_PG_USERNAME', 'pmsh')
- self.env.set('PMSH_PG_PASSWORD', 'pass')
- with self.assertRaises(Exception):
- get_db_connection_url()
-
- @patch('mod.logger.info')
- @patch('mod.pmsh_utils.get_all')
- def test_refresh_config_success(self, mock_cbs_client_get_all, mock_logger):
- mock_cbs_client_get_all.return_value = get_pmsh_config()
- self.app_conf.refresh_config()
- mock_logger.assert_called_with('AppConfig data has been refreshed')
-
- @patch('mod.logger.error')
- @patch('mod.pmsh_utils.get_all')
- def test_refresh_config_fail(self, mock_cbs_client_get_all, mock_logger):
- mock_cbs_client_get_all.side_effect = ValueError
- with self.assertRaises(RetryError):
- self.app_conf.refresh_config()
- mock_logger.assert_called_with('Failed to refresh PMSH AppConfig')
-
- @patch('mod.logger.debug')
- def test_utils_validate_config_subscription(self, mock_logger):
- self.app_conf.validate_sub_schema()
- mock_logger.assert_called_with("Subscription schema is valid.")
-
- @patch('mod.logger.debug')
- def test_utils_validate_config_subscription_administrativeState_locked(self, mock_logger):
- self.app_conf.subscription.administrativeState = "LOCKED"
- self.app_conf.validate_sub_schema()
- mock_logger.assert_called_with("Subscription schema is valid.")
-
- def test_utils_validate_config_subscription_administrativeState_invalid_value(self):
- self.app_conf.subscription.administrativeState = "FAILED"
- with self.assertRaises(ValidationError):
- self.app_conf.validate_sub_schema()
-
- def test_utils_validate_config_subscription_nfFilter_failed(self):
- self.app_conf.subscription.nfFilter = {}
- with self.assertRaises(ValidationError):
- self.app_conf.validate_sub_schema()
-
- def test_utils_validate_config_subscription_nfFilter_not_empty(self):
- self.app_conf.subscription.nfFilter = {
- "nfNames": [
-
- ],
- "modelInvariantIDs": [
-
- ],
- "modelVersionIDs": [
-
- ],
- "modelNames": [
-
- ]
- }
- with self.assertRaises(ValidationError):
- self.app_conf.validate_sub_schema()
-
- @patch('mod.logger.debug')
- def test_utils_validate_config_subscription_nfFilter_with_empty_property(self, mock_logger):
- self.app_conf.subscription.nfFilter = {
- "nfNames": [
- "^pnf.*",
- "^vnf.*"
- ],
- "modelInvariantIDs": [
- "7129e420-d396-4efb-af02-6b83499b12f8"
- ],
- "modelVersionIDs": [
-
- ],
- "modelNames": [
- "pnf102"
- ]
- }
- self.app_conf.validate_sub_schema()
- mock_logger.assert_called_with("Subscription schema is valid.")
-
- def test_utils_validate_config_subscription_where_measurementTypes_is_empty(self):
- self.app_conf.subscription.measurementGroups = [{
- "measurementGroup": {
- "measurementTypes": [
- ],
- "managedObjectDNsBasic": [
- {
- "DN": "dna"
- },
- {
- "DN": "dnb"
- }
- ]
- }
- }]
- with self.assertRaises(ValidationError):
- self.app_conf.validate_sub_schema()
-
- def test_utils_validate_config_subscription_where_managedObjectDNsBasic_is_empty(self):
- self.app_conf.subscription.measurementGroups = [{
- "measurementGroup": {
- "measurementTypes": [
- {
- "measurementType": "countera"
- },
- {
- "measurementType": "counterb"
- }
- ],
- "managedObjectDNsBasic": [
-
- ]
- }
- }]
- with self.assertRaises(ValidationError):
- self.app_conf.validate_sub_schema()
-
- def test_utils_validate_config_subscription_where_measurementGroups_is_empty(self):
- self.app_conf.subscription.measurementGroups = []
- with self.assertRaises(ValidationError):
- self.app_conf.validate_sub_schema()
from unittest.mock import patch, MagicMock
from mod import db
+from mod.api.services.measurement_group_service import MgNfState, AdministrativeState
from mod.api.services import measurement_group_service
from mod.network_function import NetworkFunction
from mod.policy_response_handler import PolicyResponseHandler, policy_response_handle_functions
-from mod.subscription import AdministrativeState, SubNfState
from tests.base_setup import BaseClassSetup, create_subscription_data
class PolicyResponseHandlerTest(BaseClassSetup):
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
@patch('mod.create_app')
def setUp(self, mock_app):
super().setUp()
- super().setUpAppConf()
self.nf = NetworkFunction(nf_name='nf1')
self.policy_response_handler = PolicyResponseHandler(mock_app)
- def tearDown(self):
- super().tearDown()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
@patch('mod.network_function.NetworkFunction.delete')
def test_handle_response_locked_success(self, mock_delete):
with patch.dict(policy_response_handle_functions,
self.nf.nf_name, 'failed')
mock_update_sub_nf.assert_called_with(
measurement_group_name='msr_grp_name',
- status=SubNfState.DELETE_FAILED.value, nf_name=self.nf.nf_name)
+ status=MgNfState.DELETE_FAILED.value, nf_name=self.nf.nf_name)
@patch('mod.network_function.NetworkFunction.delete')
def test_handle_response_locking_success(self, mock_delete):
self.nf.nf_name, 'failed')
mock_update_sub_nf.assert_called_with(
measurement_group_name='msr_grp_name',
- status=SubNfState.DELETE_FAILED.value, nf_name=self.nf.nf_name)
+ status=MgNfState.DELETE_FAILED.value, nf_name=self.nf.nf_name)
@patch('mod.api.services.measurement_group_service.update_measurement_group_nf_status')
def test_handle_response_unlocked_success(self, mock_update_sub_nf):
self.nf.nf_name, 'success')
mock_update_sub_nf.assert_called_with(
measurement_group_name='msr_grp_name',
- status=SubNfState.CREATED.value, nf_name=self.nf.nf_name)
+ status=MgNfState.CREATED.value, nf_name=self.nf.nf_name)
@patch('mod.api.services.measurement_group_service.update_measurement_group_nf_status')
def test_handle_response_unlocked_success_filtering(self, mock_update_sub_nf):
sub = create_subscription_data('sub')
db.session.add(sub)
measurement_group_service.apply_nf_status_to_measurement_group(
- self.nf.nf_name, "MG2", SubNfState.PENDING_CREATE.value)
+ self.nf.nf_name, "MG2", MgNfState.PENDING_CREATE.value)
db.session.commit()
self.policy_response_handler._handle_response(
'MG2',
self.nf.nf_name, 'success')
mock_update_sub_nf.assert_called_with(
measurement_group_name='MG2',
- status=SubNfState.CREATED.value, nf_name=self.nf.nf_name)
+ status=MgNfState.CREATED.value, nf_name=self.nf.nf_name)
@patch('mod.api.services.measurement_group_service.update_measurement_group_nf_status')
def test_handle_response_locking_success_filtering(self, mock_update_sub_nf):
sub = create_subscription_data('sub')
db.session.add(sub)
measurement_group_service.apply_nf_status_to_measurement_group(
- self.nf.nf_name, "MG2", SubNfState.PENDING_DELETE.value)
+ self.nf.nf_name, "MG2", MgNfState.PENDING_DELETE.value)
db.session.commit()
self.policy_response_handler._handle_response(
'MG2',
self.nf.nf_name, 'success')
mock_update_sub_nf.assert_called_with(
measurement_group_name='MG2',
- status=SubNfState.DELETED.value, nf_name=self.nf.nf_name)
+ status=MgNfState.DELETED.value, nf_name=self.nf.nf_name)
@patch('mod.api.services.measurement_group_service.update_measurement_group_nf_status')
def test_handle_response_unlocked_failed(self, mock_update_sub_nf):
self.nf.nf_name, 'failed')
mock_update_sub_nf.assert_called_with(
measurement_group_name='msr_grp_name',
- status=SubNfState.CREATE_FAILED.value, nf_name=self.nf.nf_name)
+ status=MgNfState.CREATE_FAILED.value, nf_name=self.nf.nf_name)
@patch('mod.api.services.measurement_group_service.update_measurement_group_nf_status')
def test_handle_response_create_failed_filtering(self, mock_update_sub_nf):
sub = create_subscription_data('sub')
db.session.add(sub)
measurement_group_service.apply_nf_status_to_measurement_group(
- self.nf.nf_name, "MG2", SubNfState.PENDING_CREATE.value)
+ self.nf.nf_name, "MG2", MgNfState.PENDING_CREATE.value)
db.session.commit()
self.policy_response_handler._handle_response(
'MG2',
self.nf.nf_name, 'failed')
mock_update_sub_nf.assert_called_with(
measurement_group_name='MG2',
- status=SubNfState.CREATE_FAILED.value, nf_name=self.nf.nf_name)
+ status=MgNfState.CREATE_FAILED.value, nf_name=self.nf.nf_name)
@patch('mod.api.services.measurement_group_service.update_measurement_group_nf_status')
def test_handle_response_delete_failed_filtering(self, mock_update_sub_nf):
sub = create_subscription_data('sub')
db.session.add(sub)
measurement_group_service.apply_nf_status_to_measurement_group(
- self.nf.nf_name, "MG2", SubNfState.PENDING_DELETE.value)
+ self.nf.nf_name, "MG2", MgNfState.PENDING_DELETE.value)
db.session.commit()
self.policy_response_handler._handle_response(
'MG2',
self.nf.nf_name, 'failed')
mock_update_sub_nf.assert_called_with(
measurement_group_name='MG2',
- status=SubNfState.DELETE_FAILED.value, nf_name=self.nf.nf_name)
+ status=MgNfState.DELETE_FAILED.value, nf_name=self.nf.nf_name)
def test_handle_response_exception(self):
self.assertRaises(Exception, self.policy_response_handler._handle_response, 'sub1',
+++ /dev/null
-# ============LICENSE_START===================================================
-# 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.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-# SPDX-License-Identifier: Apache-2.0
-# ============LICENSE_END=====================================================
-import json
-import os
-from unittest.mock import patch, Mock
-
-from requests import Session
-
-import mod.aai_client as aai_client
-from mod.network_function import NetworkFunction
-from mod.subscription import Subscription
-from tests.base_setup import BaseClassSetup
-
-
-class SubscriptionTest(BaseClassSetup):
-
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
-
- @patch.object(Session, 'get')
- @patch.object(Session, 'put')
- def setUp(self, mock_session_put, mock_session_get):
- super().setUp()
- with open(os.path.join(os.path.dirname(__file__), 'data/aai_xnfs.json'), 'r') as data:
- self.aai_response_data = data.read()
- mock_session_put.return_value.status_code = 200
- mock_session_put.return_value.text = self.aai_response_data
- with open(os.path.join(os.path.dirname(__file__), 'data/aai_model_info.json'), 'r') as data:
- self.aai_model_data = data.read()
- mock_session_get.return_value.status_code = 200
- mock_session_get.return_value.text = self.aai_model_data
- self.mock_mr_sub = Mock()
- self.mock_mr_pub = Mock()
- self.app_conf.subscription.create()
- self.xnfs = aai_client.get_pmsh_nfs_from_aai(self.app_conf, self.app_conf.nf_filter)
- self.sub_model = self.app_conf.subscription.get()
-
- def tearDown(self):
- super().tearDown()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
- def test_sub_measurement_group(self):
- self.assertEqual(len(self.app_conf.subscription.measurementGroups), 2)
-
- def test_sub_file_location(self):
- self.assertEqual(self.app_conf.subscription.fileLocation, '/pm/pm.xml')
-
- def test_get_subscription(self):
- sub_name = 'ExtraPM-All-gNB-R2B'
- new_sub = self.app_conf.subscription.get()
- self.assertEqual(sub_name, new_sub.subscription_name)
-
- def test_get_nf_names_per_sub(self):
- self.app_conf.subscription.add_network_function_to_subscription(list(self.xnfs)[0],
- self.sub_model)
- self.app_conf.subscription.add_network_function_to_subscription(list(self.xnfs)[1],
- self.sub_model)
-
- def test_add_duplicate_network_functions_per_subscription(self):
- self.app_conf.subscription.add_network_function_to_subscription(list(self.xnfs)[0],
- self.sub_model)
- nf_subs = Subscription.get_all_nfs_subscription_relations()
- self.assertEqual(1, len(nf_subs))
- self.app_conf.subscription.add_network_function_to_subscription(list(self.xnfs)[0],
- self.sub_model)
- nf_subs = Subscription.get_all_nfs_subscription_relations()
- self.assertEqual(1, len(nf_subs))
-
- def test_update_subscription_status(self):
- self.app_conf.subscription.administrativeState = 'new_status'
- self.app_conf.subscription.update_subscription_status()
- sub = self.app_conf.subscription.get()
-
- self.assertEqual('new_status', sub.status)
-
- def test_update_sub_nf_status(self):
- sub_name = 'ExtraPM-All-gNB-R2B'
- for nf in self.xnfs:
- self.app_conf.subscription.add_network_function_to_subscription(nf, self.sub_model)
- sub_nfs = Subscription.get_all_nfs_subscription_relations()
- self.assertEqual('PENDING_CREATE', sub_nfs[0].nf_sub_status)
-
- Subscription.update_sub_nf_status(sub_name, 'Active', 'pnf_23')
- sub_nfs = Subscription.get_all_nfs_subscription_relations()
- self.assertEqual('PENDING_CREATE', sub_nfs[0].nf_sub_status)
- self.assertEqual('PENDING_CREATE', sub_nfs[1].nf_sub_status)
-
- @patch('mod.subscription.Subscription.add_network_function_to_subscription')
- @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)
-
- mock_add_nfs.assert_called()
- self.assertTrue(self.mock_mr_pub.publish_subscription_event_data.called)
- mock_update_sub_nf.assert_called_with(self.app_conf.subscription.subscriptionName,
- 'PENDING_CREATE', list(self.xnfs)[0].nf_name)
-
- @patch('mod.subscription.Subscription.get_network_functions')
- @patch('mod.subscription.Subscription.update_sub_nf_status')
- 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.assertTrue(self.mock_mr_pub.publish_subscription_event_data.called)
- self.assertEqual(mock_update_sub_nf.call_count, 3)
-
- def test_activate_subscription_exception(self):
- self.assertRaises(Exception, self.app_conf.subscription.create_subscription_on_nfs,
- [list(self.xnfs)[0]], 'not_mr_pub', 'app_config')
-
- def test_prepare_subscription_event(self):
- with open(os.path.join(os.path.dirname(__file__),
- 'data/pm_subscription_event.json'), 'r') as data:
- expected_sub_event = json.load(data)
- nf = NetworkFunction(nf_name='pnf_1',
- ipv4_address='204.120.0.15',
- ipv6_address='',
- model_invariant_id='some-id',
- 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)
- print(actual_sub_event)
- self.assertEqual(expected_sub_event, actual_sub_event)
-
- def test_prepare_subscription_event_with_ipv6(self):
- with open(os.path.join(os.path.dirname(__file__),
- 'data/pm_subscription_event.json'), 'r') as data:
- expected_sub_event = json.load(data)
- expected_sub_event['ipAddress'] = '2001:db8:3333:4444:5555:6666:7777:8888'
- nf = NetworkFunction(nf_name='pnf_1',
- ipv4_address='204.120.0.15',
- ipv6_address='2001:db8:3333:4444:5555:6666:7777:8888',
- model_invariant_id='some-id',
- 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)
- print(actual_sub_event)
- self.assertEqual(expected_sub_event, actual_sub_event)
-
- def test_get_network_functions(self):
- for nf in self.xnfs:
- self.app_conf.subscription.add_network_function_to_subscription(nf, self.sub_model)
- nfs = self.app_conf.subscription.get_network_functions()
-
- self.assertEqual(3, len(nfs))
- self.assertIsInstance(nfs[0], NetworkFunction)
+++ /dev/null
-# ============LICENSE_START===================================================
-# 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.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-# SPDX-License-Identifier: Apache-2.0
-# ============LICENSE_END=====================================================
-from unittest.mock import patch, Mock, MagicMock
-
-from mod.api.db_models import NetworkFunctionModel
-from mod.network_function import NetworkFunction
-from mod.subscription import AdministrativeState
-from mod.subscription_handler import SubscriptionHandler
-from tests.base_setup import BaseClassSetup, get_pmsh_config
-
-
-class SubscriptionHandlerTest(BaseClassSetup):
- nfs = [
- NetworkFunction(nf_name='pnf_1', model_invariant_id='some-id', model_version_id='some-id'),
- NetworkFunction(nf_name='pnf_2', model_invariant_id='some-id', model_version_id='some-id')]
-
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
-
- @patch('mod.pmsh_utils._MrSub')
- @patch('mod.pmsh_utils._MrPub')
- def setUp(self, mock_mr_pub, mock_mr_sub):
- super().setUp()
- self.mock_mr_pub = mock_mr_pub
- self.mock_mr_sub = mock_mr_sub
- self.mock_policy_event_thread = Mock()
-
- def tearDown(self):
- super().tearDown()
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
- @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
- @patch('mod.subscription.Subscription.get_local_sub_admin_state')
- @patch('mod.logger.info')
- @patch('mod.aai_client.get_pmsh_nfs_from_aai')
- def test_execute_no_change_of_state(self, mock_get_aai, mock_logger, mock_get_sub_status):
- mock_get_sub_status.return_value = AdministrativeState.UNLOCKED.value
- mock_get_aai.return_value = self.nfs
- sub_handler = SubscriptionHandler(self.mock_mr_pub,
- self.mock_mr_sub, self.app, self.app_conf)
- sub_handler.execute()
- mock_logger.assert_called_with('Administrative State did not change '
- 'in the app config: UNLOCKED')
-
- @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
- @patch('mod.subscription.Subscription.get_local_sub_admin_state')
- @patch('mod.subscription.Subscription.create_subscription_on_nfs')
- @patch('mod.aai_client.get_pmsh_nfs_from_aai')
- def test_execute_change_of_state_to_unlocked(self, mock_get_aai, mock_activate_sub,
- mock_get_sub_status):
- mock_get_aai.return_value = self.nfs
- mock_get_sub_status.return_value = AdministrativeState.LOCKED.value
- sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
- self.app_conf)
- 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)
-
- @patch('mod.subscription.Subscription.get_network_functions', MagicMock(return_value=nfs))
- @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
- @patch('mod.subscription.Subscription.get_local_sub_admin_state')
- @patch('mod.subscription.Subscription.delete_subscription_from_nfs')
- def test_execute_change_of_state_to_locked(self, mock_deactivate_sub, mock_get_sub_status):
- mock_get_sub_status.return_value = AdministrativeState.UNLOCKED.value
- self.app_conf.subscription.administrativeState = AdministrativeState.LOCKED.value
- self.app_conf.subscription.update_subscription_status()
- 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)
-
- @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
- @patch('mod.subscription.Subscription.create_subscription_on_nfs')
- @patch('mod.logger.error')
- @patch('mod.aai_client.get_pmsh_nfs_from_aai')
- def test_execute_exception(self, mock_get_aai, mock_logger, mock_activate_sub):
- mock_get_aai.return_value = self.nfs
- mock_activate_sub.side_effect = Exception
- sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
- self.app_conf)
- sub_handler.execute()
- mock_logger.assert_called_with('Error occurred during the activation/deactivation process ',
- exc_info=True)
-
- @patch('mod.network_function.NetworkFunction.get',
- MagicMock(return_value=NetworkFunctionModel(nf_name='pnf_1',
- model_invariant_id='some-id',
- model_version_id='some-id',
- ipv4_address='ip_address4',
- ipv6_address='ip_address6',
- model_name='model_name',
- sdnc_model_name='sdnc_model_name',
- sdnc_model_version='sdnc_model_version')))
- @patch('mod.subscription.Subscription.get_delete_failed_nfs', MagicMock(return_value=nfs))
- @patch('mod.subscription.Subscription.get_network_functions', MagicMock(return_value=nfs))
- @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
- @patch('mod.subscription.Subscription.get_local_sub_admin_state')
- @patch('mod.subscription.Subscription.delete_subscription_from_nfs')
- @patch('mod.network_function.NetworkFunction.increment_retry_count')
- def test_execute_change_of_state_to_locking_retry_delete(self, mock_retry_inc, mock_delete_sub,
- mock_get_sub_status):
- mock_get_sub_status.return_value = AdministrativeState.LOCKING.value
- sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
- self.app_conf)
- sub_handler.execute()
- self.assertEqual(mock_delete_sub.call_count, 2)
- self.assertEqual(mock_retry_inc.call_count, 2)
-
- @patch('mod.subscription.Subscription.get_delete_failed_nfs', MagicMock(return_value=[]))
- @patch('mod.subscription.Subscription.get_network_functions', MagicMock(return_value=nfs))
- @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
- @patch('mod.subscription.Subscription.get_local_sub_admin_state')
- @patch('mod.subscription.Subscription.update_subscription_status')
- def test_execute_change_of_state_to_locking_success(self, mock_update_sub,
- mock_get_sub_status):
- mock_get_sub_status.return_value = AdministrativeState.LOCKING.value
- sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
- self.app_conf)
- sub_handler.execute()
- mock_update_sub.assert_called_once()
-
- @patch('mod.network_function.NetworkFunction.get',
- MagicMock(return_value=NetworkFunctionModel(nf_name='pnf_1',
- model_invariant_id='some-id',
- model_version_id='some-id',
- ipv4_address='ip_address4',
- ipv6_address='ip_address6',
- model_name='model_name',
- sdnc_model_name='sdnc_model_name',
- sdnc_model_version='sdnc_model_version',
- retry_count=3)))
- @patch('mod.subscription.Subscription.get_delete_failed_nfs', MagicMock(return_value=[nfs[0]]))
- @patch('mod.subscription.Subscription.get_network_functions', MagicMock(return_value=nfs[0]))
- @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
- @patch('mod.subscription.Subscription.get_local_sub_admin_state')
- @patch('mod.network_function.NetworkFunction.delete')
- def test_execute_change_of_state_to_locking_retry_failed(self, mock_nf_del,
- mock_get_sub_status):
- mock_get_sub_status.return_value = AdministrativeState.LOCKING.value
- sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
- self.app_conf)
- sub_handler.execute()
- mock_nf_del.assert_called_once()
-
- @patch('mod.pmsh_utils.AppConfig._get_pmsh_config',
- MagicMock(return_value=get_pmsh_config('data/cbs_invalid_data.json')))
- @patch('mod.subscription_handler.SubscriptionHandler._check_state_change')
- def test_execute_invalid_schema(self, mock_change_state_check):
- sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
- self.app_conf)
- sub_handler.execute()
- mock_change_state_check.assert_not_called()
-
- @patch('mod.pmsh_utils.AppConfig._get_pmsh_config',
- MagicMock(return_value=get_pmsh_config()))
- @patch('mod.subscription_handler.SubscriptionHandler._check_state_change')
- def test_execute_valid_schema(self, mock_change_state_check):
- sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
- self.app_conf)
- sub_handler.execute()
- mock_change_state_check.assert_called_once()
Code Review / dcaegen2 / services.git / commitdiff