diff options
Diffstat (limited to 'components/pm-subscription-handler/tests/test_pmsh_utils.py')
| -rw-r--r-- | components/pm-subscription-handler/tests/test_pmsh_utils.py | 105 |
1 files changed, 39 insertions, 66 deletions
diff --git a/components/pm-subscription-handler/tests/test_pmsh_utils.py b/components/pm-subscription-handler/tests/test_pmsh_utils.py index 6e5585f5..1bc039d2 100644 --- a/components/pm-subscription-handler/tests/test_pmsh_utils.py +++ b/components/pm-subscription-handler/tests/test_pmsh_utils.py @@ -15,88 +15,67 @@ # # SPDX-License-Identifier: Apache-2.0 # ============LICENSE_END===================================================== -import json -import os from test.support import EnvironmentVarGuard -from unittest import TestCase -from unittest.mock import patch +from unittest.mock import patch, Mock import responses from requests import Session from tenacity import RetryError -from mod import db, get_db_connection_url, create_app +from mod import get_db_connection_url +from mod.network_function import NetworkFunction from mod.pmsh_utils import AppConfig +from tests.base_setup import BaseClassSetup +from tests.base_setup import get_pmsh_config -class PmshUtilsTestCase(TestCase): +class PmshUtilsTestCase(BaseClassSetup): - @patch('mod.update_logging_config') - @patch('mod.create_app') - @patch('mod.get_db_connection_url') - def setUp(self, mock_get_db_url, mock_app, mock_update_config): - mock_get_db_url.return_value = 'sqlite://' - with open(os.path.join(os.path.dirname(__file__), 'data/cbs_data_1.json'), 'r') as data: - self.cbs_data = json.load(data) - self.env = EnvironmentVarGuard() - self.env.set('LOGGER_CONFIG', os.path.join(os.path.dirname(__file__), 'log_config.yaml')) - self.mock_app = mock_app - self.app = create_app() - self.app_context = self.app.app_context() - self.app_context.push() - db.create_all() - - @patch('mod.pmsh_utils.AppConfig._get_pmsh_config') - def test_utils_get_mr_sub(self, mock_get_pmsh_config): - mock_get_pmsh_config.return_value = self.cbs_data - self.app_conf = AppConfig() + @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') - @patch('mod.pmsh_utils.AppConfig._get_pmsh_config') - def test_utils_get_mr_sub_fails_with_invalid_name(self, mock_get_pmsh_config): - mock_get_pmsh_config.return_value = self.cbs_data - self.app_conf = AppConfig() + def test_utils_get_mr_sub_fails_with_invalid_name(self): with self.assertRaises(KeyError): self.app_conf.get_mr_sub('invalid_sub') - @patch('mod.pmsh_utils.AppConfig._get_pmsh_config') - def test_utils_get_mr_pub(self, mock_get_pmsh_config): - mock_get_pmsh_config.return_value = self.cbs_data - self.app_conf = AppConfig() + 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!') - @patch('mod.pmsh_utils.AppConfig._get_pmsh_config') - def test_utils_get_mr_pub_fails_with_invalid_name(self, mock_get_pmsh_config): - mock_get_pmsh_config.return_value = self.cbs_data - self.app_conf = AppConfig() + def test_utils_get_mr_pub_fails_with_invalid_name(self): with self.assertRaises(KeyError): self.app_conf.get_mr_pub('invalid_pub') - @patch('mod.pmsh_utils.AppConfig._get_pmsh_config') - def test_utils_get_cert_data(self, mock_get_pmsh_config): - mock_get_pmsh_config.return_value = self.cbs_data - self.app_conf = AppConfig() + 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('mod.pmsh_utils.AppConfig._get_pmsh_config') @patch.object(Session, 'post') - def test_mr_pub_publish_to_topic_success(self, mock_session, mock_get_pmsh_config): - mock_get_pmsh_config.return_value = self.cbs_data - self.app_conf = AppConfig() + 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() - @patch('mod.pmsh_utils.AppConfig._get_pmsh_config') @responses.activate - def test_mr_pub_publish_to_topic_fail(self, mock_get_pmsh_config): - mock_get_pmsh_config.return_value = self.cbs_data - self.app_conf = AppConfig() + 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) @@ -104,21 +83,19 @@ class PmshUtilsTestCase(TestCase): with self.assertRaises(Exception): mr_policy_pub.publish_to_topic({"dummy_val": "43c4ee19-6b8d-4279-a80f-c507850aae47"}) - @patch('mod.pmsh_utils.AppConfig._get_pmsh_config') - def test_mr_pub_publish_sub_event_data_success(self, mock_get_pmsh_config): - mock_get_pmsh_config.return_value = self.cbs_data - self.app_conf = AppConfig() + 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, 'pnf201', - self.app_conf) + 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'), + self.app_conf) pub_to_topic_call.assert_called_once() - @patch('mod.pmsh_utils.AppConfig._get_pmsh_config') @responses.activate - def test_mr_sub_get_from_topic_success(self, mock_get_pmsh_config): - mock_get_pmsh_config.return_value = self.cbs_data - self.app_conf = AppConfig() + 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/' @@ -127,11 +104,8 @@ class PmshUtilsTestCase(TestCase): mr_topic_data = policy_mr_sub.get_from_topic(1) self.assertIsNotNone(mr_topic_data) - @patch('mod.pmsh_utils.AppConfig._get_pmsh_config') @responses.activate - def test_mr_sub_get_from_topic_fail(self, mock_get_pmsh_config): - mock_get_pmsh_config.return_value = self.cbs_data - self.app_conf = AppConfig() + 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/' @@ -158,15 +132,14 @@ class PmshUtilsTestCase(TestCase): @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 = self.cbs_data - self.app_conf = AppConfig() + 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.return_value = self.cbs_data + mock_cbs_client_get_all.return_value = get_pmsh_config() self.app_conf = AppConfig() mock_cbs_client_get_all.side_effect = Exception with self.assertRaises(RetryError): |