From be25ac706aebde77865996223e14658d36618248 Mon Sep 17 00:00:00 2001 From: Vijay Venkatesh Kumar Date: Mon, 9 Jan 2023 15:05:56 +0000 Subject: Fix black reported error in rls job Change-Id: I5b3b04214ca2a53a23df170bfbfd25c768dc8ac2 Signed-off-by: Vijay Venkatesh Kumar Issue-ID: DCAEGEN2-3321 Signed-off-by: Vijay Venkatesh Kumar --- tests/test_db_monitoring.py | 60 ++++++---- tests/test_htbt_get_cbs_config.py | 9 +- tests/test_htbt_vnf_table.py | 36 +++--- tests/test_htbtworker.py | 167 ++++++++++++++------------- tests/test_misshtbtd.py | 232 +++++++++++++++++++++----------------- 5 files changed, 277 insertions(+), 227 deletions(-) (limited to 'tests') diff --git a/tests/test_db_monitoring.py b/tests/test_db_monitoring.py index b9a644e..5315a66 100644 --- a/tests/test_db_monitoring.py +++ b/tests/test_db_monitoring.py @@ -21,49 +21,60 @@ import logging import requests import tempfile import os -import json +import json import unittest from unittest.mock import * from _pytest.outcomes import skip _logger = logging.getLogger(__name__) + + class Test_db_monitoring(unittest.TestCase): - def setUp(self): - htbtworker.configjsonfile = (os.path.dirname(__file__))+"/test-config.json" - - @patch('requests.post') + htbtworker.configjsonfile = (os.path.dirname(__file__)) + "/test-config.json" + + @patch("requests.post") def test_sendControlLoopEvent(self, mock1): status = True - mock_resp = Mock() - mock_resp.configure_mock( - **{ - "status_code": 200 - } - ) - mock1.return_value = mock_resp - db_monitoring.sendControlLoopEvent("ONSET", "ABC","1.0","vFW","vFW","VNF","NODE","1234567890","VFW","1.0","DCAE") + mock_resp = Mock() + mock_resp.configure_mock(**{"status_code": 200}) + mock1.return_value = mock_resp + db_monitoring.sendControlLoopEvent( + "ONSET", "ABC", "1.0", "vFW", "vFW", "VNF", "NODE", "1234567890", "VFW", "1.0", "DCAE" + ) self.assertEqual(status, True) - db_monitoring.sendControlLoopEvent("ONSET", "ABC","1.0","vFW","vFW","VM","NODE","1234567890","VFW","1.0","DCAE") + db_monitoring.sendControlLoopEvent( + "ONSET", "ABC", "1.0", "vFW", "vFW", "VM", "NODE", "1234567890", "VFW", "1.0", "DCAE" + ) self.assertEqual(status, True) - db_monitoring.sendControlLoopEvent("ABATED", "ABC","1.0","vFW","vFW","VNF","NODE","1234567890","VFW","1.0","DCAE") + db_monitoring.sendControlLoopEvent( + "ABATED", "ABC", "1.0", "vFW", "vFW", "VNF", "NODE", "1234567890", "VFW", "1.0", "DCAE" + ) self.assertEqual(status, True) - db_monitoring.sendControlLoopEvent("ABATED", "ABC","1.0","vFW","vFW","VM","NODE","1234567890","VFW","1.0","DCAE") + db_monitoring.sendControlLoopEvent( + "ABATED", "ABC", "1.0", "vFW", "vFW", "VM", "NODE", "1234567890", "VFW", "1.0", "DCAE" + ) self.assertEqual(status, True) - @patch('misshtbtd.read_hb_common',return_value = ("1234","RUNNING", "XYZ", 1234)) - @patch('htbtworker.postgres_db_open') + @patch("misshtbtd.read_hb_common", return_value=("1234", "RUNNING", "XYZ", 1234)) + @patch("htbtworker.postgres_db_open") def test_db_monitoring(self, mock1, mock2): status = True - mock_cursor = Mock() + mock_cursor = Mock() mock2.cursor.return_value = mock_cursor - db_monitoring.db_monitoring("111",htbtworker.configjsonfile ,"testuser","testpwd","10.0.0.0","1234","db_name") + db_monitoring.db_monitoring( + "111", htbtworker.configjsonfile, "testuser", "testpwd", "10.0.0.0", "1234", "db_name" + ) self.assertEqual(status, True) - db_monitoring.db_monitoring("1234",htbtworker.configjsonfile ,"testuser","testpwd","10.0.0.0","1234","db_name") + db_monitoring.db_monitoring( + "1234", htbtworker.configjsonfile, "testuser", "testpwd", "10.0.0.0", "1234", "db_name" + ) self.assertEqual(status, True) - mock1.cursor.return_value = ("1234","RECONFIGURATION", "XYZ", 1234) - db_monitoring.db_monitoring("1234",htbtworker.configjsonfile ,"testuser","testpwd","10.0.0.0","1234","db_name") + mock1.cursor.return_value = ("1234", "RECONFIGURATION", "XYZ", 1234) + db_monitoring.db_monitoring( + "1234", htbtworker.configjsonfile, "testuser", "testpwd", "10.0.0.0", "1234", "db_name" + ) self.assertEqual(status, True) def test_db_monitoring_wrapper(self): @@ -71,5 +82,6 @@ class Test_db_monitoring(unittest.TestCase): db_monitoring.db_monitoring_wrapper("111", htbtworker.configjsonfile, number_of_iterations=0) self.assertEqual(status, True) -if __name__ == "__main__": # pragma: no cover + +if __name__ == "__main__": # pragma: no cover unittest.main() diff --git a/tests/test_htbt_get_cbs_config.py b/tests/test_htbt_get_cbs_config.py index 99e347c..2a2b641 100644 --- a/tests/test_htbt_get_cbs_config.py +++ b/tests/test_htbt_get_cbs_config.py @@ -105,7 +105,7 @@ class test_get_cbs_config(unittest.TestCase): ) # create copy of snmphtbt.json for pytest - #pytest_json_config = "/tmp/opt/app/miss_htbt_service/etc/config.json" + # pytest_json_config = "/tmp/opt/app/miss_htbt_service/etc/config.json" pytest_json_config = "test-config.json" with open(pytest_json_config, "w") as outfile: outfile.write(pytest_json_data) @@ -130,19 +130,18 @@ class test_get_cbs_config(unittest.TestCase): print("result: %s" % result) self.assertEqual(result, True) - @patch('misshtbtd.create_update_hb_common') - @patch('misshtbtd.read_hb_common') + @patch("misshtbtd.create_update_hb_common") + @patch("misshtbtd.read_hb_common") def test_poll_cbs(self, mock1, mock2): """ TBD """ status = True current_time = round(time.time()) - mock1.return_value = ('1', 'RUNNING', 'AA', current_time) + mock1.return_value = ("1", "RUNNING", "AA", current_time) # configjsonfile = (os.path.dirname(__file__))+"/test-config.json" configjsonfile = "test-config.json" os.environ.update(CBS_HTBT_JSON=configjsonfile) os.environ["pytest"] = "test" cp.poll_cbs(1) self.assertEqual(status, True) - diff --git a/tests/test_htbt_vnf_table.py b/tests/test_htbt_vnf_table.py index 4b6c3f1..e425766 100644 --- a/tests/test_htbt_vnf_table.py +++ b/tests/test_htbt_vnf_table.py @@ -52,39 +52,39 @@ class test_vnf_tables(unittest.TestCase): global ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval = hb_properties() - @patch('htbtworker.postgres_db_open') - @patch('misshtbtd.db_table_creation_check', return_value=True) + @patch("htbtworker.postgres_db_open") + @patch("misshtbtd.db_table_creation_check", return_value=True) def test_validate_vnf_table_1(self, mock, mock1): result = verify_DB_creation_1(user_name, password, ip_address, port_num, db_name) self.assertEqual(result, True) - @patch('htbtworker.postgres_db_open') - @patch('misshtbtd.db_table_creation_check', return_value=True) + @patch("htbtworker.postgres_db_open") + @patch("misshtbtd.db_table_creation_check", return_value=True) def test_validate_vnf_table_2(self, mock, mock1): result = verify_DB_creation_2(user_name, password, ip_address, port_num, db_name) self.assertEqual(result, True) - @patch('htbtworker.postgres_db_open') - @patch('misshtbtd.db_table_creation_check', return_value=True) + @patch("htbtworker.postgres_db_open") + @patch("misshtbtd.db_table_creation_check", return_value=True) def test_validate_hb_common(self, mock, mock1): result = verify_DB_creation_hb_common() self.assertEqual(result, True) - - @patch('cbs_polling.poll_cbs') + + @patch("cbs_polling.poll_cbs") def test_cbspolling(self, mock): # Check if no exception thrown verify_cbspolling() - @patch('misshtbtd.fetch_json_file') + @patch("misshtbtd.fetch_json_file") def test_fetch_json_file(self, mock1): - configjsonfile = (os.path.dirname(__file__))+"/test-config.json" + configjsonfile = (os.path.dirname(__file__)) + "/test-config.json" mock1.return_value = configjsonfile - + result = verify_fetch_json_file() _logger.info(result) self.assertEqual(result, True) - @patch('misshtbtd.main') + @patch("misshtbtd.main") def test_misshtbtdmain(self, mock): result = verify_misshtbtdmain() _logger.info(result) @@ -95,14 +95,14 @@ class test_vnf_tables(unittest.TestCase): _logger.info(result) self.assertEqual(result, True) - @patch('misshtbtd.fetch_json_file') - @patch('misshtbtd.read_hb_common') - @patch('db_monitoring.db_monitoring') + @patch("misshtbtd.fetch_json_file") + @patch("misshtbtd.read_hb_common") + @patch("db_monitoring.db_monitoring") def test_dbmonitoring(self, mock1, mock2, mock3): - configjsonfile = (os.path.dirname(__file__))+"/test-config.json" + configjsonfile = (os.path.dirname(__file__)) + "/test-config.json" mock1.return_value = configjsonfile - mock2.return_value = ("1234","RUNNING", "XYZ", 1234) - + mock2.return_value = ("1234", "RUNNING", "XYZ", 1234) + result = verify_dbmonitoring() _logger.info(result) self.assertEqual(result, True) diff --git a/tests/test_htbtworker.py b/tests/test_htbtworker.py index ee03ddb..4066a9f 100644 --- a/tests/test_htbtworker.py +++ b/tests/test_htbtworker.py @@ -18,148 +18,159 @@ import htbtworker import os import tempfile -import json +import json import unittest from unittest.mock import * from _pytest.outcomes import skip -class Test_htbtworker(unittest.TestCase): +class Test_htbtworker(unittest.TestCase): def setUp(self): - htbtworker.configjsonfile = (os.path.dirname(__file__))+"/test-config.json" + htbtworker.configjsonfile = (os.path.dirname(__file__)) + "/test-config.json" - @patch('requests.get') - @patch('htbtworker.check_process_reconfiguration', return_value=False) - @patch('htbtworker.get_eventnamelist') - @patch('htbtworker.sql_executor') - def test_process_msg(self, mock1, mock2, mock3, sqlmock1): + @patch("requests.get") + @patch("htbtworker.check_process_reconfiguration", return_value=False) + @patch("htbtworker.get_eventnamelist") + @patch("htbtworker.sql_executor") + def test_process_msg(self, mock1, mock2, mock3, sqlmock1): """ Test to verify event processing using mock TBD - Negative test """ - + status = True - dmaap_data = [{"event":{"commonEventHeader":{"startEpochMicrosec":1544608845841,"sourceId":"VNFB_SRC5","eventId":"mvfs10","nfcNamingCode":"VNFB","timeZoneOffset":"UTC-05:30","reportingEntityId":"cc305d54-75b4-431b-adb2-eb6b9e541234","eventType":"platform","priority":"Normal","version":"4.0.2","reportingEntityName":"ibcx0001vm002oam001","sequence":1000,"domain":"heartbeat","lastEpochMicrosec":1544608845841,"eventName":"Heartbeat_vFW","vesEventListenerVersion":"7.0.2","sourceName":"SOURCE_NAME2","nfNamingCode":"VNFB"},"heartbeatFields":{"heartbeatInterval":20,"heartbeatFieldsVersion":"3.0"}}}] - - mock_resp = Mock() - mock_resp.configure_mock( - **{ - "text": json.dumps(dmaap_data) - } - ) + dmaap_data = [ + { + "event": { + "commonEventHeader": { + "startEpochMicrosec": 1544608845841, + "sourceId": "VNFB_SRC5", + "eventId": "mvfs10", + "nfcNamingCode": "VNFB", + "timeZoneOffset": "UTC-05:30", + "reportingEntityId": "cc305d54-75b4-431b-adb2-eb6b9e541234", + "eventType": "platform", + "priority": "Normal", + "version": "4.0.2", + "reportingEntityName": "ibcx0001vm002oam001", + "sequence": 1000, + "domain": "heartbeat", + "lastEpochMicrosec": 1544608845841, + "eventName": "Heartbeat_vFW", + "vesEventListenerVersion": "7.0.2", + "sourceName": "SOURCE_NAME2", + "nfNamingCode": "VNFB", + }, + "heartbeatFields": {"heartbeatInterval": 20, "heartbeatFieldsVersion": "3.0"}, + } + } + ] + + mock_resp = Mock() + mock_resp.configure_mock(**{"text": json.dumps(dmaap_data)}) mock3.return_value = [("Heartbeat_vDNS", "Heartbeat_vFW")] - mock1.return_value = mock_resp + mock1.return_value = mock_resp filename = "test-config.json" htbtworker.process_msg(filename, number_of_iterations=1) self.assertEqual(status, True) - def test_parse_event(self): """ test_parse_event() opens the file test1.json and returns attributes """ - filename = (os.path.dirname(__file__))+"/test1.json" - with open(filename,"r") as fp: - data = fp.read() - srcname,lastepo,seqnum,event_name = htbtworker.parse_event(data) + filename = (os.path.dirname(__file__)) + "/test1.json" + with open(filename, "r") as fp: + data = fp.read() + srcname, lastepo, seqnum, event_name = htbtworker.parse_event(data) self.assertEqual(srcname, "SOURCE_NAME1") self.assertEqual(event_name, "Heartbeat_vDNS") - filename = (os.path.dirname(__file__))+"/test4.json" - with open(filename,"r") as fp: - data = fp.read() - srcname,lastepo,seqnum,event_name = htbtworker.parse_event(data) + filename = (os.path.dirname(__file__)) + "/test4.json" + with open(filename, "r") as fp: + data = fp.read() + srcname, lastepo, seqnum, event_name = htbtworker.parse_event(data) self.assertEqual(srcname, "zalp1bmdns01cmd010") self.assertEqual(event_name, "Heartbeat_vDNS") - - @patch('htbtworker.sql_executor') - def test_create_and_check_vnf2_table (self, mock_settings): + @patch("htbtworker.sql_executor") + def test_create_and_check_vnf2_table(self, mock_settings): """ Test to verify existence of given table """ - mock_cursor = Mock() - mock_cursor.configure_mock( - **{ - "fetchone.return_value": [("vnf_table_2")] - } - ) + mock_cursor = Mock() + mock_cursor.configure_mock(**{"fetchone.return_value": [("vnf_table_2")]}) mock_settings.return_value = mock_cursor - status = htbtworker.check_and_create_vnf2_table () - self.assertEqual(status, True) - - with patch('htbtworker.sql_executor', new=Mock(side_effect=htbtworker.psycopg2.DatabaseError())): - status = htbtworker.check_and_create_vnf2_table () - self.assertEqual(False, status) - - @patch('htbtworker.sql_executor') - def test_new_vnf_entry (self, sql_mock): + status = htbtworker.check_and_create_vnf2_table() + self.assertEqual(status, True) + + with patch("htbtworker.sql_executor", new=Mock(side_effect=htbtworker.psycopg2.DatabaseError())): + status = htbtworker.check_and_create_vnf2_table() + self.assertEqual(False, status) + + @patch("htbtworker.sql_executor") + def test_new_vnf_entry(self, sql_mock): """ Check to verify if new node entry is made for tracking HB """ status = True - htbtworker.new_vnf_entry ("Heartbeat_vDNS", "TESTNODE", 1548313727714000, "TESTNODE", 1) + htbtworker.new_vnf_entry("Heartbeat_vDNS", "TESTNODE", 1548313727714000, "TESTNODE", 1) self.assertEqual(status, True) - @patch('htbtworker.sql_executor') - def test_get_eventnamelist (self, sql_mock): + @patch("htbtworker.sql_executor") + def test_get_eventnamelist(self, sql_mock): """ - Test to verify eventname list is returned from vnf_table_1 - TBD - List comparison + Test to verify eventname list is returned from vnf_table_1 + TBD - List comparison """ eventname_list = [("Heartbeat_vDNS", "Heartbeat_vFW")] - mock_cursor = Mock() - mock_cursor.configure_mock( - **{ - "fetchall.return_value": eventname_list - } - ) + mock_cursor = Mock() + mock_cursor.configure_mock(**{"fetchall.return_value": eventname_list}) sql_mock.return_value = mock_cursor - return_list = htbtworker.get_eventnamelist () - self.assertIn("Heartbeat_vDNS", return_list) - - @patch('htbtworker.postgres_db_open') - def test_sql_executor (self, db_mock): + return_list = htbtworker.get_eventnamelist() + self.assertIn("Heartbeat_vDNS", return_list) + + @patch("htbtworker.postgres_db_open") + def test_sql_executor(self, db_mock): """ - Test sql executor wrapper method + Test sql executor wrapper method """ - htbtworker.sql_executor ("SELECT * FROM information_schema.tables WHERE table_name = %s", "vnf_table_2") - htbtworker.sql_executor ("INSERT into information_schema.tables,") + htbtworker.sql_executor("SELECT * FROM information_schema.tables WHERE table_name = %s", "vnf_table_2") + htbtworker.sql_executor("INSERT into information_schema.tables,") connection_db = db_mock - with patch('htbtworker.postgres_db_open.commit', new=Mock(side_effect=htbtworker.psycopg2.DatabaseError())): + with patch("htbtworker.postgres_db_open.commit", new=Mock(side_effect=htbtworker.psycopg2.DatabaseError())): flag = htbtworker.commit_and_close_db(connection_db) self.assertEqual(False, flag) - @patch('psycopg2.connect') - def test_postgres_db_open (self, mock): + @patch("psycopg2.connect") + def test_postgres_db_open(self, mock): """ Test wrapper for postgres db connection """ conn = htbtworker.postgres_db_open() self.assertIsNotNone(conn) - @patch('misshtbtd.read_hb_common') - def test_check_process_reconfiguration (self, mock): + @patch("misshtbtd.read_hb_common") + def test_check_process_reconfiguration(self, mock): """ Test if DB is in reconfiguration state """ - mock.return_value = ("1234","RUNNING", "XYZ", 1234) - flag = htbtworker.check_process_reconfiguration("test", "test","x.x.x.x", "1234", "test_db") + mock.return_value = ("1234", "RUNNING", "XYZ", 1234) + flag = htbtworker.check_process_reconfiguration("test", "test", "x.x.x.x", "1234", "test_db") self.assertEqual(False, flag) - - @patch('htbtworker.postgres_db_open') - def test_commit_and_close_db (self, db_mock): + + @patch("htbtworker.postgres_db_open") + def test_commit_and_close_db(self, db_mock): """ Test commit and close db """ connection_db = db_mock flag = htbtworker.commit_and_close_db(connection_db) self.assertEqual(True, flag) - with patch('htbtworker.postgres_db_open.commit', new=Mock(side_effect=htbtworker.psycopg2.DatabaseError())): + with patch("htbtworker.postgres_db_open.commit", new=Mock(side_effect=htbtworker.psycopg2.DatabaseError())): flag = htbtworker.commit_and_close_db(connection_db) self.assertEqual(False, flag) - - -if __name__ == "__main__": # pragma: no cover + + +if __name__ == "__main__": # pragma: no cover unittest.main() diff --git a/tests/test_misshtbtd.py b/tests/test_misshtbtd.py index 0fc8a24..0374a6b 100644 --- a/tests/test_misshtbtd.py +++ b/tests/test_misshtbtd.py @@ -20,196 +20,224 @@ import time import psycopg2 import os import tempfile -import json +import json import unittest from unittest.mock import * from _pytest.outcomes import skip from pickle import FALSE -class Test_misshtbtd(unittest.TestCase): +class Test_misshtbtd(unittest.TestCase): def setUp(self): - htbtworker.configjsonfile = (os.path.dirname(__file__))+"/test-config.json" + htbtworker.configjsonfile = (os.path.dirname(__file__)) + "/test-config.json" - @patch ('psycopg2.connect') + @patch("psycopg2.connect") def test_create_database(self, mock1): status = True - mock_cursor = MagicMock() - mock_cursor.configure_mock( - **{ - "fetchone.return_value": [("1")] - } + 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" ) - 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') + + @patch("htbtworker.postgres_db_open") def test_read_hb_common(self, mock1): - - mock_cursor = Mock() + + mock_cursor = Mock() mock_cursor.configure_mock( - **{ - "cursor.return_value.fetchall.return_value": [["1", "AA", "123456789", "RUNNING"]] - } + **{"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")) + 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') + @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 + """ + 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") + 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") + 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): + 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) + 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): + @patch("misshtbtd.db_table_creation_check") + def test_create_update_vnf_table_1(self, mock1): """ TBD """ status = True - mock_cursor = Mock() + mock_cursor = Mock() mock_cursor.configure_mock( - **{ - "cursor.return_value.fetchall.return_value": [["Heartbeat_vDNS"],["Heartbeat_vFW"]] - } + **{"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) + 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): + def test_read_hb_properties_default_from_file(self): """ TBD """ - return_val = ("10.0.0.0","1234", "postgres-test", "postgres-test", "postgres", "True", "300") + 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) ) + ( + 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): + @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") + 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) ) + ( + 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): + def test_read_hb_properties_from_file(self): """ TBD """ - htbtworker.configjsonfile = (os.path.dirname(__file__))+"/test-config.json" + 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): + 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" + 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) ) + 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): + @patch("mod.htbt_get_cbs_config") + def test_fetch_json_file(self, mock1): """ TBD """ - #mock.return_value.is_file.return_value = True + # 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) + 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') + @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 + """ + 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") + 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") + 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') + @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" - } - ) + mock_process.configure_mock(**{"start.return_value": "1"}) mock1.return_value = mock_process - job_list = misshtbtd.create_process ([],htbtworker.configjsonfile, 1) + 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') + @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" - } - ) + 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) + mock2.return_value = ("1", "RUNNING", "AA", current_time) misshtbtd.main() self.assertEqual(status, True) -if __name__ == "__main__": # pragma: no cover + +if __name__ == "__main__": # pragma: no cover unittest.main() -- cgit 1.2.3-korg