aboutsummaryrefslogtreecommitdiffstats
path: root/tests/test_misshtbtd.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_misshtbtd.py')
-rw-r--r--tests/test_misshtbtd.py215
1 files changed, 215 insertions, 0 deletions
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()