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