components/pm-subscription-handler/tests/test_subscription.py

   1 # ============LICENSE_START===================================================
   2 #  Copyright (C) 2019-2020 Nordix Foundation.
   3 # ============================================================================
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #      http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 #
  16 # SPDX-License-Identifier: Apache-2.0
  17 # ============LICENSE_END=====================================================
  18 import json
  19 import os
  20 from unittest.mock import patch, Mock
  21
  22 from requests import Session
  23
  24 import mod.aai_client as aai_client
  25 from mod.network_function import NetworkFunction
  26 from mod.subscription import Subscription
  27 from tests.base_setup import BaseClassSetup
  28
  29
  30 class SubscriptionTest(BaseClassSetup):
  31
  32     @classmethod
  33     def setUpClass(cls):
  34         super().setUpClass()
  35
  36     @patch.object(Session, 'get')
  37     @patch.object(Session, 'put')
  38     def setUp(self, mock_session_put, mock_session_get):
  39         super().setUp()
  40         with open(os.path.join(os.path.dirname(__file__), 'data/aai_xnfs.json'), 'r') as data:
  41             self.aai_response_data = data.read()
  42         mock_session_put.return_value.status_code = 200
  43         mock_session_put.return_value.text = self.aai_response_data
  44         with open(os.path.join(os.path.dirname(__file__), 'data/aai_model_info.json'), 'r') as data:
  45             self.aai_model_data = data.read()
  46         mock_session_get.return_value.status_code = 200
  47         mock_session_get.return_value.text = self.aai_model_data
  48         self.mock_mr_sub = Mock()
  49         self.mock_mr_pub = Mock()
  50         self.xnfs = aai_client.get_pmsh_nfs_from_aai(self.app_conf)
  51         self.sub_model = self.app_conf.subscription.get()
  52
  53     def tearDown(self):
  54         super().tearDown()
  55
  56     @classmethod
  57     def tearDownClass(cls):
  58         super().tearDownClass()
  59
  60     def test_sub_measurement_group(self):
  61         self.assertEqual(len(self.app_conf.subscription.measurementGroups), 2)
  62
  63     def test_sub_file_location(self):
  64         self.assertEqual(self.app_conf.subscription.fileLocation, '/pm/pm.xml')
  65
  66     def test_get_subscription(self):
  67         sub_name = 'ExtraPM-All-gNB-R2B'
  68         self.app_conf.subscription.create()
  69         new_sub = self.app_conf.subscription.get()
  70         self.assertEqual(sub_name, new_sub.subscription_name)
  71
  72     def test_get_nf_names_per_sub(self):
  73         self.app_conf.subscription.create()
  74         self.app_conf.subscription.add_network_function_to_subscription(list(self.xnfs)[0],
  75                                                                         self.sub_model)
  76         self.app_conf.subscription.add_network_function_to_subscription(list(self.xnfs)[1],
  77                                                                         self.sub_model)
  78
  79     def test_create_existing_subscription(self):
  80         sub1 = self.app_conf.subscription.create()
  81         same_sub1 = self.app_conf.subscription.create()
  82         self.assertEqual(sub1, same_sub1)
  83         self.assertEqual(1, len(self.app_conf.subscription.get_all()))
  84
  85     def test_add_network_functions_per_subscription(self):
  86         for nf in self.xnfs:
  87             self.app_conf.subscription.add_network_function_to_subscription(nf, self.sub_model)
  88         nfs_for_sub_1 = Subscription.get_all_nfs_subscription_relations()
  89         self.assertEqual(3, len(nfs_for_sub_1))
  90         new_nf = NetworkFunction(nf_name='vnf_3', orchestration_status='Active')
  91         self.app_conf.subscription.add_network_function_to_subscription(new_nf, self.sub_model)
  92         nf_subs = Subscription.get_all_nfs_subscription_relations()
  93         self.assertEqual(4, len(nf_subs))
  94
  95     def test_add_duplicate_network_functions_per_subscription(self):
  96         self.app_conf.subscription.add_network_function_to_subscription(list(self.xnfs)[0],
  97                                                                         self.sub_model)
  98         nf_subs = Subscription.get_all_nfs_subscription_relations()
  99         self.assertEqual(1, len(nf_subs))
 100         self.app_conf.subscription.add_network_function_to_subscription(list(self.xnfs)[0],
 101                                                                         self.sub_model)
 102         nf_subs = Subscription.get_all_nfs_subscription_relations()
 103         self.assertEqual(1, len(nf_subs))
 104
 105     def test_update_subscription_status(self):
 106         self.app_conf.subscription.create()
 107         self.app_conf.subscription.administrativeState = 'new_status'
 108         self.app_conf.subscription.update_subscription_status()
 109         sub = self.app_conf.subscription.get()
 110
 111         self.assertEqual('new_status', sub.status)
 112
 113     def test_update_sub_nf_status(self):
 114         sub_name = 'ExtraPM-All-gNB-R2B'
 115         for nf in self.xnfs:
 116             self.app_conf.subscription.add_network_function_to_subscription(nf, self.sub_model)
 117         sub_nfs = Subscription.get_all_nfs_subscription_relations()
 118         self.assertEqual('PENDING_CREATE', sub_nfs[0].nf_sub_status)
 119
 120         Subscription.update_sub_nf_status(sub_name, 'Active', 'pnf_23')
 121         sub_nfs = Subscription.get_all_nfs_subscription_relations()
 122         self.assertEqual('PENDING_CREATE', sub_nfs[0].nf_sub_status)
 123         self.assertEqual('PENDING_CREATE', sub_nfs[1].nf_sub_status)
 124
 125     @patch('mod.subscription.Subscription.add_network_function_to_subscription')
 126     @patch('mod.subscription.Subscription.update_sub_nf_status')
 127     def test_process_activate_subscription(self, mock_update_sub_nf, mock_add_nfs):
 128         self.app_conf.subscription.activate_subscription([list(self.xnfs)[0]], self.mock_mr_pub,
 129                                                          self.app_conf)
 130
 131         mock_add_nfs.assert_called()
 132         self.assertTrue(self.mock_mr_pub.publish_subscription_event_data.called)
 133         mock_update_sub_nf.assert_called_with(self.app_conf.subscription.subscriptionName,
 134                                               'PENDING_CREATE', list(self.xnfs)[0].nf_name)
 135
 136     @patch('mod.subscription.Subscription.get_network_functions')
 137     @patch('mod.subscription.Subscription.update_sub_nf_status')
 138     def test_process_deactivate_subscription(self, mock_update_sub_nf, mock_get_nfs):
 139         self.app_conf.subscription.administrativeState = 'LOCKED'
 140         mock_get_nfs.return_value = [list(self.xnfs)[0]]
 141         self.app_conf.subscription.deactivate_subscription(self.mock_mr_pub, self.app_conf)
 142         self.assertTrue(self.mock_mr_pub.publish_subscription_event_data.called)
 143         mock_update_sub_nf.assert_called_with(self.app_conf.subscription.subscriptionName,
 144                                               'PENDING_DELETE', list(self.xnfs)[0].nf_name)
 145
 146     def test_activate_subscription_exception(self):
 147         self.assertRaises(Exception, self.app_conf.subscription.activate_subscription,
 148                           [list(self.xnfs)[0]], 'not_mr_pub', 'app_config')
 149
 150     def test_prepare_subscription_event(self):
 151         with open(os.path.join(os.path.dirname(__file__),
 152                                'data/pm_subscription_event.json'), 'r') as data:
 153             expected_sub_event = json.load(data)
 154         nf = NetworkFunction(nf_name='pnf_1',
 155                              model_invariant_id='some-id',
 156                              model_version_id='some-id')
 157         nf.sdnc_model_name = 'some-name'
 158         nf.sdnc_model_version = 'some-version'
 159         actual_sub_event = self.app_conf.subscription.prepare_subscription_event(nf, self.app_conf)
 160         print(actual_sub_event)
 161         self.assertEqual(expected_sub_event, actual_sub_event)
 162
 163     def test_get_network_functions(self):
 164         for nf in self.xnfs:
 165             self.app_conf.subscription.add_network_function_to_subscription(nf, self.sub_model)
 166         nfs = self.app_conf.subscription.get_network_functions()
 167
 168         self.assertEqual(3, len(nfs))
 169         self.assertIsInstance(nfs[0], NetworkFunction)