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

   1 # ============LICENSE_START===================================================
   2 #  Copyright (C) 2019-2021 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.app_conf.subscription.create()
  51         self.xnfs = aai_client.get_pmsh_nfs_from_aai(self.app_conf, self.app_conf.nf_filter)
  52         self.sub_model = self.app_conf.subscription.get()
  53
  54     def tearDown(self):
  55         super().tearDown()
  56
  57     @classmethod
  58     def tearDownClass(cls):
  59         super().tearDownClass()
  60
  61     def test_sub_measurement_group(self):
  62         self.assertEqual(len(self.app_conf.subscription.measurementGroups), 2)
  63
  64     def test_sub_file_location(self):
  65         self.assertEqual(self.app_conf.subscription.fileLocation, '/pm/pm.xml')
  66
  67     def test_get_subscription(self):
  68         sub_name = 'ExtraPM-All-gNB-R2B'
  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.add_network_function_to_subscription(list(self.xnfs)[0],
  74                                                                         self.sub_model)
  75         self.app_conf.subscription.add_network_function_to_subscription(list(self.xnfs)[1],
  76                                                                         self.sub_model)
  77
  78     def test_add_duplicate_network_functions_per_subscription(self):
  79         self.app_conf.subscription.add_network_function_to_subscription(list(self.xnfs)[0],
  80                                                                         self.sub_model)
  81         nf_subs = Subscription.get_all_nfs_subscription_relations()
  82         self.assertEqual(1, len(nf_subs))
  83         self.app_conf.subscription.add_network_function_to_subscription(list(self.xnfs)[0],
  84                                                                         self.sub_model)
  85         nf_subs = Subscription.get_all_nfs_subscription_relations()
  86         self.assertEqual(1, len(nf_subs))
  87
  88     def test_update_subscription_status(self):
  89         self.app_conf.subscription.administrativeState = 'new_status'
  90         self.app_conf.subscription.update_subscription_status()
  91         sub = self.app_conf.subscription.get()
  92
  93         self.assertEqual('new_status', sub.status)
  94
  95     def test_update_sub_nf_status(self):
  96         sub_name = 'ExtraPM-All-gNB-R2B'
  97         for nf in self.xnfs:
  98             self.app_conf.subscription.add_network_function_to_subscription(nf, self.sub_model)
  99         sub_nfs = Subscription.get_all_nfs_subscription_relations()
 100         self.assertEqual('PENDING_CREATE', sub_nfs[0].nf_sub_status)
 101
 102         Subscription.update_sub_nf_status(sub_name, 'Active', 'pnf_23')
 103         sub_nfs = Subscription.get_all_nfs_subscription_relations()
 104         self.assertEqual('PENDING_CREATE', sub_nfs[0].nf_sub_status)
 105         self.assertEqual('PENDING_CREATE', sub_nfs[1].nf_sub_status)
 106
 107     @patch('mod.subscription.Subscription.add_network_function_to_subscription')
 108     @patch('mod.subscription.Subscription.update_sub_nf_status')
 109     def test_process_activate_subscription(self, mock_update_sub_nf, mock_add_nfs):
 110         self.app_conf.subscription.create_subscription_on_nfs([list(self.xnfs)[0]],
 111                                                               self.mock_mr_pub)
 112
 113         mock_add_nfs.assert_called()
 114         self.assertTrue(self.mock_mr_pub.publish_subscription_event_data.called)
 115         mock_update_sub_nf.assert_called_with(self.app_conf.subscription.subscriptionName,
 116                                               'PENDING_CREATE', list(self.xnfs)[0].nf_name)
 117
 118     @patch('mod.subscription.Subscription.get_network_functions')
 119     @patch('mod.subscription.Subscription.update_sub_nf_status')
 120     def test_process_deactivate_subscription(self, mock_update_sub_nf, mock_get_nfs):
 121         self.app_conf.subscription.administrativeState = 'LOCKED'
 122         mock_get_nfs.return_value = [list(self.xnfs)[0]]
 123         self.app_conf.subscription.delete_subscription_from_nfs(self.xnfs, self.mock_mr_pub)
 124         self.assertTrue(self.mock_mr_pub.publish_subscription_event_data.called)
 125         self.assertEqual(mock_update_sub_nf.call_count, 3)
 126
 127     def test_activate_subscription_exception(self):
 128         self.assertRaises(Exception, self.app_conf.subscription.create_subscription_on_nfs,
 129                           [list(self.xnfs)[0]], 'not_mr_pub', 'app_config')
 130
 131     def test_prepare_subscription_event(self):
 132         with open(os.path.join(os.path.dirname(__file__),
 133                                'data/pm_subscription_event.json'), 'r') as data:
 134             expected_sub_event = json.load(data)
 135         nf = NetworkFunction(nf_name='pnf_1',
 136                              ipv4_address='204.120.0.15',
 137                              ipv6_address='',
 138                              model_invariant_id='some-id',
 139                              model_version_id='some-id')
 140         nf.sdnc_model_name = 'some-name'
 141         nf.sdnc_model_version = 'some-version'
 142         actual_sub_event = self.app_conf.subscription.prepare_subscription_event(nf)
 143         print(actual_sub_event)
 144         self.assertEqual(expected_sub_event, actual_sub_event)
 145
 146     def test_prepare_subscription_event_with_ipv6(self):
 147         with open(os.path.join(os.path.dirname(__file__),
 148                                'data/pm_subscription_event.json'), 'r') as data:
 149             expected_sub_event = json.load(data)
 150         expected_sub_event['ipAddress'] = '2001:db8:3333:4444:5555:6666:7777:8888'
 151         nf = NetworkFunction(nf_name='pnf_1',
 152                              ipv4_address='204.120.0.15',
 153                              ipv6_address='2001:db8:3333:4444:5555:6666:7777:8888',
 154                              model_invariant_id='some-id',
 155                              model_version_id='some-id')
 156         nf.sdnc_model_name = 'some-name'
 157         nf.sdnc_model_version = 'some-version'
 158         actual_sub_event = self.app_conf.subscription.prepare_subscription_event(nf)
 159         print(actual_sub_event)
 160         self.assertEqual(expected_sub_event, actual_sub_event)
 161
 162     def test_get_network_functions(self):
 163         for nf in self.xnfs:
 164             self.app_conf.subscription.add_network_function_to_subscription(nf, self.sub_model)
 165         nfs = self.app_conf.subscription.get_network_functions()
 166
 167         self.assertEqual(3, len(nfs))
 168         self.assertIsInstance(nfs[0], NetworkFunction)