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
8 # http://www.apache.org/licenses/LICENSE-2.0
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.
16 # SPDX-License-Identifier: Apache-2.0
17 # ============LICENSE_END=====================================================
20 from unittest.mock import patch, Mock
22 from requests import Session
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
30 class SubscriptionTest(BaseClassSetup):
36 @patch.object(Session, 'get')
37 @patch.object(Session, 'put')
38 def setUp(self, mock_session_put, mock_session_get):
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()
57 def tearDownClass(cls):
58 super().tearDownClass()
60 def test_sub_measurement_group(self):
61 self.assertEqual(len(self.app_conf.subscription.measurementGroups), 2)
63 def test_sub_file_location(self):
64 self.assertEqual(self.app_conf.subscription.fileLocation, '/pm/pm.xml')
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)
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],
76 self.app_conf.subscription.add_network_function_to_subscription(list(self.xnfs)[1],
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()))
85 def test_add_network_functions_per_subscription(self):
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))
95 def test_add_duplicate_network_functions_per_subscription(self):
96 self.app_conf.subscription.add_network_function_to_subscription(list(self.xnfs)[0],
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],
102 nf_subs = Subscription.get_all_nfs_subscription_relations()
103 self.assertEqual(1, len(nf_subs))
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()
111 self.assertEqual('new_status', sub.status)
113 def test_update_sub_nf_status(self):
114 sub_name = 'ExtraPM-All-gNB-R2B'
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)
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)
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,
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)
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)
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')
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)
163 def test_get_network_functions(self):
165 self.app_conf.subscription.add_network_function_to_subscription(nf, self.sub_model)
166 nfs = self.app_conf.subscription.get_network_functions()
168 self.assertEqual(3, len(nfs))
169 self.assertIsInstance(nfs[0], NetworkFunction)