components/pm-subscription-handler/tests/test_pmsh_utils.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 test.support import EnvironmentVarGuard
  21 from unittest import TestCase
  22 from unittest.mock import patch
  23
  24 import responses
  25 from requests import Session
  26
  27 from mod import db, get_db_connection_url, create_app
  28 from mod.pmsh_utils import AppConfig
  29 from mod.subscription import Subscription
  30
  31
  32 class PmshUtilsTestCase(TestCase):
  33
  34     @patch('mod.create_app')
  35     @patch('mod.get_db_connection_url')
  36     def setUp(self, mock_get_db_url, mock_app):
  37         mock_get_db_url.return_value = 'sqlite://'
  38         with open(os.path.join(os.path.dirname(__file__), 'data/cbs_data_1.json'), 'r') as data:
  39             self.cbs_data = json.load(data)
  40         self.app_conf = AppConfig(**self.cbs_data['config'])
  41         self.sub = Subscription(**self.cbs_data['policy']['subscription'])
  42         self.env = EnvironmentVarGuard()
  43         self.env.set('LOGS_PATH', './unit_test_logs')
  44         self.policy_mr_sub = self.app_conf.get_mr_sub('policy_pm_subscriber')
  45         self.mock_app = mock_app
  46         self.app = create_app()
  47         self.app_context = self.app.app_context()
  48         self.app_context.push()
  49         db.create_all()
  50
  51     def test_utils_get_mr_sub(self):
  52         mr_policy_sub = self.app_conf.get_mr_sub('policy_pm_subscriber')
  53         self.assertTrue(mr_policy_sub.aaf_id, 'dcae@dcae.onap.org')
  54
  55     def test_utils_get_mr_sub_fails_with_invalid_name(self):
  56         with self.assertRaises(KeyError):
  57             self.app_conf.get_mr_sub('invalid_sub')
  58
  59     def test_utils_get_mr_pub(self):
  60         mr_policy_pub = self.app_conf.get_mr_pub('policy_pm_publisher')
  61         self.assertTrue(mr_policy_pub.aaf_pass, 'demo123456!')
  62
  63     def test_utils_get_mr_pub_fails_with_invalid_name(self):
  64         with self.assertRaises(KeyError):
  65             self.app_conf.get_mr_pub('invalid_pub')
  66
  67     def test_utils_get_cert_data(self):
  68         self.assertTrue(self.app_conf.cert_params, ('/opt/app/pm-mapper/etc/certs/cert.pem',
  69                                                     '/opt/app/pm-mapper/etc/certs/key.pem'))
  70
  71     @patch.object(Session, 'post')
  72     def test_mr_pub_publish_to_topic_success(self, mock_session):
  73         mock_session.return_value.status_code = 200
  74         mr_policy_pub = self.app_conf.get_mr_pub('policy_pm_publisher')
  75         with patch('requests.Session.post') as session_post_call:
  76             mr_policy_pub.publish_to_topic({"dummy_val": "43c4ee19-6b8d-4279-a80f-c507850aae47"})
  77             session_post_call.assert_called_once()
  78
  79     @responses.activate
  80     def test_mr_pub_publish_to_topic_fail(self):
  81         responses.add(responses.POST,
  82                       'https://node:30226/events/org.onap.dmaap.mr.PM_SUBSCRIPTIONS',
  83                       json={'error': 'Client Error'}, status=400)
  84         mr_policy_pub = self.app_conf.get_mr_pub('policy_pm_publisher')
  85         with self.assertRaises(Exception):
  86             mr_policy_pub.publish_to_topic({"dummy_val": "43c4ee19-6b8d-4279-a80f-c507850aae47"})
  87
  88     def test_mr_pub_publish_sub_event_data_success(self):
  89         mr_policy_pub = self.app_conf.get_mr_pub('policy_pm_publisher')
  90         with patch('mod.pmsh_utils._MrPub.publish_to_topic') as pub_to_topic_call:
  91             mr_policy_pub.publish_subscription_event_data(self.sub, 'pnf201', self.app_conf)
  92             pub_to_topic_call.assert_called_once()
  93
  94     @responses.activate
  95     def test_mr_sub_get_from_topic_success(self):
  96         responses.add(responses.GET,
  97                       'https://node:30226/events/org.onap.dmaap.mr.PM_SUBSCRIPTIONS/'
  98                       'dcae_pmsh_cg/1?timeout=1000',
  99                       json={"dummy_val": "43c4ee19-6b8d-4279-a80f-c507850aae47"}, status=200)
 100         mr_topic_data = self.policy_mr_sub.get_from_topic(1)
 101         self.assertIsNotNone(mr_topic_data)
 102
 103     @responses.activate
 104     def test_mr_sub_get_from_topic_fail(self):
 105         responses.add(responses.GET,
 106                       'https://node:30226/events/org.onap.dmaap.mr.PM_SUBSCRIPTIONS/'
 107                       'dcae_pmsh_cg/1?timeout=1000',
 108                       json={"dummy_val": "43c4ee19-6b8d-4279-a80f-c507850aae47"}, status=400)
 109         mr_topic_data = self.policy_mr_sub.get_from_topic(1)
 110         self.assertIsNone(mr_topic_data)
 111
 112     def test_get_db_connection_url_success(self):
 113         self.env = EnvironmentVarGuard()
 114         self.env.set('PMSH_PG_URL', '1.2.3.4')
 115         self.env.set('PMSH_PG_USERNAME', 'pmsh')
 116         self.env.set('PMSH_PG_PASSWORD', 'pass')
 117         db_url = get_db_connection_url()
 118         self.assertEqual(db_url, 'postgres+psycopg2://pmsh:pass@1.2.3.4:5432/pmsh')
 119
 120     def test_get_db_connection_url_fail(self):
 121         self.env = EnvironmentVarGuard()
 122         self.env.set('PMSH_PG_USERNAME', 'pmsh')
 123         self.env.set('PMSH_PG_PASSWORD', 'pass')
 124         with self.assertRaises(Exception):
 125             get_db_connection_url()