From 341b5bb2347c30344662675936b90b325efe5520 Mon Sep 17 00:00:00 2001 From: Vijay Venkatesh Kumar Date: Thu, 17 Nov 2022 14:51:38 -0500 Subject: Heartbeat code refactoring code optimization & test improvement Issue-ID: DCAEGEN2-2953 Signed-off-by: Vijay Venkatesh Kumar Change-Id: I99229d966c13ad666ac994ab5a582aeeaa306639 Signed-off-by: Vijay Venkatesh Kumar --- tests/test_misshtbtd.py | 215 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 215 insertions(+) create mode 100644 tests/test_misshtbtd.py (limited to 'tests/test_misshtbtd.py') diff --git a/tests/test_misshtbtd.py b/tests/test_misshtbtd.py new file mode 100644 index 0000000..0fc8a24 --- /dev/null +++ b/tests/test_misshtbtd.py @@ -0,0 +1,215 @@ +# ============LICENSE_START======================================================= +# Copyright (c) 2020-2023 AT&T Intellectual Property. All rights reserved. +# ================================================================================ +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ============LICENSE_END========================================================= + +import htbtworker +import misshtbtd +import time +import psycopg2 +import os +import tempfile +import json +import unittest +from unittest.mock import * +from _pytest.outcomes import skip +from pickle import FALSE + +class Test_misshtbtd(unittest.TestCase): + + def setUp(self): + htbtworker.configjsonfile = (os.path.dirname(__file__))+"/test-config.json" + + @patch ('psycopg2.connect') + def test_create_database(self, mock1): + status = True + mock_cursor = MagicMock() + mock_cursor.configure_mock( + **{ + "fetchone.return_value": [("1")] + } + ) + mock1.return_value = mock_cursor + misshtbtd.create_database("vnf_table_1",htbtworker.configjsonfile, "10.0.0.0","1234","testuser","testpwd","heartbeatdb") + self.assertEqual(status, True) + + @patch ('htbtworker.postgres_db_open') + def test_read_hb_common(self, mock1): + + mock_cursor = Mock() + mock_cursor.configure_mock( + **{ + "cursor.return_value.fetchall.return_value": [["1", "AA", "123456789", "RUNNING"]] + } + ) + mock1.return_value = mock_cursor + self.assertEqual(('1', 'RUNNING', 'AA', '123456789'), misshtbtd.read_hb_common("testuser","testpwd","10.0.0.0","1234","heartbeatdb")) + + @patch ('misshtbtd.db_table_creation_check') + @patch ('htbtworker.postgres_db_open') + def test_create_update_hb_common(self, mock1, mock2): + ''' + TODO: argument ordering TBD + ''' + mock_cursor = Mock() + mock1.return_value = mock_cursor + mock2.return_value = True + status = True + misshtbtd.create_update_hb_common (0,111, "RUNNING", "testuser","testpwd","10.0.0.0","1234","testdb") + self.assertEqual(status, True) + mock2.return_value = False + misshtbtd.create_update_hb_common (1,111, "RUNNING", "testuser","testpwd","10.0.0.0","1234","testdb") + self.assertEqual(status, True) + + def test_db_table_creation_check (self): + """ + Test to verify existence of given table + """ + mock_cursor = Mock() + mock_cursor.configure_mock( + **{ + "cursor.return_value.fetchone.return_value": ("vnf_table_2") + } + ) + status = misshtbtd.db_table_creation_check (mock_cursor,"vnf_table_2") + self.assertEqual(status, True) + + @patch ('misshtbtd.db_table_creation_check') + def test_create_update_vnf_table_1 (self, mock1): + """ + TBD + """ + status = True + mock_cursor = Mock() + mock_cursor.configure_mock( + **{ + "cursor.return_value.fetchall.return_value": [["Heartbeat_vDNS"],["Heartbeat_vFW"]] + } + ) + misshtbtd.create_update_vnf_table_1 (htbtworker.configjsonfile,1, mock_cursor) + self.assertEqual(status, True) + misshtbtd.create_update_vnf_table_1 (htbtworker.configjsonfile,0, mock_cursor) + self.assertEqual(status, True) + + def test_read_hb_properties_default_from_file (self): + """ + TBD + """ + return_val = ("10.0.0.0","1234", "postgres-test", "postgres-test", "postgres", "True", "300") + misshtbtd.hb_properties_file = (os.path.dirname(__file__)) + "/hbproperties-test.yaml" + (ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval) = misshtbtd.read_hb_properties_default() + self.assertEqual (return_val,(ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval) ) + + @patch.dict(os.environ, {"pg_ipAddress": "10.0.0.10", "pg_portNum":"1234", "pg_userName": "test","pg_passwd":"test"}) + def test_read_hb_properties_default_from_env (self): + """ + TBD + """ + return_val = ("10.0.0.10","1234", "test", "test", "postgres", "True", "300") + misshtbtd.hb_properties_file = (os.path.dirname(__file__)) + "/hbproperties-test.yaml" + (ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval) = misshtbtd.read_hb_properties_default() + self.assertEqual (return_val,(ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval) ) + + def test_read_hb_properties_from_file (self): + """ + TBD + """ + htbtworker.configjsonfile = (os.path.dirname(__file__))+"/test-config.json" + misshtbtd.hb_properties_file = (os.path.dirname(__file__)) + "/hbproperties-test.yaml" + + return_val = ("10.0.4.1","5432", "postgres", "postgres", "postgres", "True", "300") + (ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval) = misshtbtd.read_hb_properties(htbtworker.configjsonfile) + self.assertEqual (return_val,(ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval) ) + + @patch.dict(os.environ, {"pg_ipAddress": "10.0.0.10", "pg_portNum":"1234", "pg_userName": "test","pg_passwd":"test"}) + def test_read_hb_properties_exception_handling (self): + """ + TBD + """ + htbtworker.configjsonfile = (os.path.dirname(__file__))+"/aa-config.json" + misshtbtd.hb_properties_file = (os.path.dirname(__file__)) + "/hbproperties-test.yaml" + + return_val = ("10.0.0.10","1234", "test", "test", "postgres", "True", "300") + (ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval) = misshtbtd.read_hb_properties(htbtworker.configjsonfile) + self.assertEqual (return_val,(ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval) ) + + + @patch ('mod.htbt_get_cbs_config') + def test_fetch_json_file (self, mock1): + """ + TBD + """ + #mock.return_value.is_file.return_value = True + mock1.return_value = True + misshtbtd.CONFIG_PATH = "test-config.json" + filename = misshtbtd.fetch_json_file() + self.assertEqual (misshtbtd.CONFIG_PATH, filename) + # mock1.return_value = False + # filename = misshtbtd.fetch_json_file() + # self.assertEqual ("./etc/config.json", filename) + + @patch ('htbtworker.postgres_db_open') + @patch ('misshtbtd.db_table_creation_check') + @patch ('misshtbtd.create_update_vnf_table_1') + def test_create_update_db(self, mock1, mock2, mock3): + ''' + TODO: argument ordering TBD + ''' + mock_cursor = Mock() + mock3.return_value = mock_cursor + mock1.return_value = True + mock2.return_value = True + status = True + misshtbtd.create_update_db (0,htbtworker.configjsonfile, "10.0.0.0", "1234", "test","test", "testdb") + self.assertEqual(status, True) + mock1.return_value = False + misshtbtd.create_update_db (0,htbtworker.configjsonfile, "10.0.0.0", "1234", "test","test", "testdb") + self.assertEqual(status, True) + + @patch ('multiprocessing.Process') + @patch ('misshtbtd.create_update_db') + @patch('misshtbtd.create_update_hb_common') + def test_create_process(self, mock1, mock2, mock3): + job_list = [] + mock_process = Mock() + mock_process.configure_mock( + **{ + "start.return_value": "1" + } + ) + mock1.return_value = mock_process + job_list = misshtbtd.create_process ([],htbtworker.configjsonfile, 1) + self.assertTrue(job_list) + + @patch ('multiprocessing.Process') + @patch ('misshtbtd.read_hb_common') + @patch ('misshtbtd.create_update_db') + @patch ('misshtbtd.create_update_hb_common') + def test_main(self, mock1, mock2, mock3, mock4): + status = True + mock_process = Mock() + mock_process.configure_mock( + **{ + "start.return_value": "1", + "pid.return_Value":"1111" + } + ) + mock1.return_value = mock_process + current_time = round(time.time()) + mock2.return_value = ('1', 'RUNNING', 'AA', current_time) + misshtbtd.main() + self.assertEqual(status, True) + +if __name__ == "__main__": # pragma: no cover + unittest.main() -- cgit 1.2.3-korg