[PMSH] Improve CBS data handling
# AppConfog object fetch priodically
# AAI client to only fetch AAI data
Signed-off-by: efiacor <fiachra.corcoran@est.tech>
Change-Id: I78315f141c3bb7e8b0d9efa818d294415fa79918
Issue-ID: DCAEGEN2-2146
diff --git a/components/pm-subscription-handler/tests/test_controller.py b/components/pm-subscription-handler/tests/test_controller.py
index 8ca208a..5723042 100755
--- a/components/pm-subscription-handler/tests/test_controller.py
+++ b/components/pm-subscription-handler/tests/test_controller.py
@@ -26,13 +26,16 @@
from mod import aai_client, create_app, db
from mod.api.controller import status, get_all_sub_to_nf_relations
from mod.network_function import NetworkFunction
+from mod.pmsh_utils import AppConfig
class ControllerTestCase(unittest.TestCase):
- @patch('mod.update_config')
+
+ @patch('mod.pmsh_utils.AppConfig._get_pmsh_config')
+ @patch('mod.update_logging_config')
@patch('mod.get_db_connection_url')
@patch.object(Session, 'put')
- def setUp(self, mock_session, mock_get_db_url, mock_update_config):
+ def setUp(self, mock_session, mock_get_db_url, mock_update_config, mock_get_pmsh_config):
mock_get_db_url.return_value = 'sqlite://'
with open(os.path.join(os.path.dirname(__file__), 'data/aai_xnfs.json'), 'r') as data:
self.aai_response_data = data.read()
@@ -43,8 +46,10 @@
self.env.set('AAI_SERVICE_PORT', '8443')
self.env.set('LOGGER_CONFIG', os.path.join(os.path.dirname(__file__), 'log_config.yaml'))
with open(os.path.join(os.path.dirname(__file__), 'data/cbs_data_1.json'), 'r') as data:
- self.cbs_data_1 = json.load(data)
- self.sub_1, self.xnfs = aai_client.get_pmsh_subscription_data(self.cbs_data_1)
+ self.cbs_data = json.load(data)
+ mock_get_pmsh_config.return_value = self.cbs_data
+ self.app_conf = AppConfig()
+ self.xnfs = aai_client.get_pmsh_nfs_from_aai(self.app_conf)
self.nf_1 = NetworkFunction(nf_name='pnf_1', orchestration_status='Inventoried')
self.nf_2 = NetworkFunction(nf_name='pnf_2', orchestration_status='Active')
self.app = create_app()
@@ -61,9 +66,9 @@
self.assertEqual(status()['status'], 'healthy')
def test_get_all_sub_to_nf_relations(self):
- self.sub_1.create()
+ self.app_conf.subscription.create()
for nf in [self.nf_1, self.nf_2]:
- self.sub_1.add_network_function_to_subscription(nf)
+ self.app_conf.subscription.add_network_function_to_subscription(nf)
all_subs = get_all_sub_to_nf_relations()
self.assertEqual(len(all_subs[0]['network_functions']), 2)
self.assertEqual(all_subs[0]['subscription_name'], 'ExtraPM-All-gNB-R2B')