diff options
Diffstat (limited to 'tests/test_config_notif.py')
| -rw-r--r-- | tests/test_config_notif.py | 436 |
1 files changed, 0 insertions, 436 deletions
diff --git a/tests/test_config_notif.py b/tests/test_config_notif.py deleted file mode 100644 index 01ea737..0000000 --- a/tests/test_config_notif.py +++ /dev/null @@ -1,436 +0,0 @@ -# ============LICENSE_START======================================================= -# Copyright (c) 2020 AT&T Intellectual Property. All rights reserved. -# Copyright 2020 Deutsche Telekom. All rights reserved. -# Copyright 2021 Fujitsu Ltd. -# ================================================================================ -# 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 config_notif -import mod.trapd_get_cbs_config -import mod.trapd_settings - -from . import monkey_psycopg2 -import psycopg2 -import tempfile, json, os - -def assert_default_values(ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval): - """ - used in the test_read_hb_properties*() tests - """ - assert(str(port_num) == "5432") - assert(str(user_name) == "postgres") - assert(str(db_name) == "postgres") - assert(str(password) == "postgres") - -def test_read_hb_properties_default(): - """ - run read_hb_properties_default() - """ - ( ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval ) = config_notif.read_hb_properties_default() - assert_default_values(ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval) - -def test_read_hb_properties_success(): - """ - run read_hb_properties() to read properties from a file - """ - tmp = tempfile.NamedTemporaryFile(mode="w+") - testdata = { - "pg_ipAddress": "10.0.0.99", - "pg_portNum": 65432, - "pg_dbName": "dbname", - "pg_userName": "pguser", - "pg_passwd": "pgpswd", - "CBS_polling_allowed": True, - "CBS_polling_interval": 30, - "SERVICE_NAME": "service_name" - } - json.dump(testdata, tmp) - tmp.flush() - ( ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval ) = config_notif.read_hb_properties(tmp.name) - assert(str(ip_address) == str(testdata["pg_ipAddress"])) - assert(str(port_num) == str(testdata["pg_portNum"])) - assert(str(user_name) == str(testdata["pg_userName"])) - assert(str(password) == str(testdata["pg_passwd"])) - assert(str(db_name) == str(testdata["pg_dbName"])) - assert(str(cbs_polling_required) == str(testdata["CBS_polling_allowed"])) - assert(str(cbs_polling_interval) == str(testdata["CBS_polling_interval"])) - assert(str(os.environ['SERVICE_NAME']) == str(testdata["SERVICE_NAME"])) - -def test_read_hb_properties_fail_bad_json(): - """ - failure case for read_hb_properties: bad json in the file - """ - tmp = tempfile.NamedTemporaryFile(mode="w+") - print("bad json", file=tmp) - tmp.flush() - ( ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval ) = config_notif.read_hb_properties(tmp.name) - assert_default_values(ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval) - -def test_read_hb_properties_fail_missing_parameter(): - """ - failure case for read_hb_properties: CBS_polling_allowed is missing - """ - tmp = tempfile.NamedTemporaryFile(mode="w+") - testdata = { - "pg_ipAddress": "10.0.0.99", - "pg_portNum": 65432, - "pg_dbName": "dbname", - "pg_userName": "pguser", - "pg_passwd": "pgpswd", - # "CBS_polling_allowed": True, # missing CBS_polling_allowed - "CBS_polling_interval": 30, - "SERVICE_NAME": "service_name" - } - json.dump(testdata, tmp) - tmp.flush() - ( ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval ) = config_notif.read_hb_properties(tmp.name) - assert_default_values(ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval) - -def test_postgres_db_open(monkeypatch): - """ - test postgres_db_open() - """ - monkeypatch.setattr(psycopg2, 'connect', monkey_psycopg2.monkey_connect) - monkey_psycopg2.monkey_reset_forces() - dbconn = config_notif.postgres_db_open("test", "testpswd", "testsite", 5432, "dbname") - assert(type(dbconn) is monkey_psycopg2.MockConn) - -def test_postgres_db_open_fail(monkeypatch): - """ - failure ase for postgres_db_open() - """ - monkeypatch.setattr(psycopg2, 'connect', monkey_psycopg2.monkey_connect) - monkey_psycopg2.monkey_reset_forces(connect=True) - dbconn = config_notif.postgres_db_open("test", "badpassword", "testsite", 5432, "dbname") - assert(type(dbconn) is not monkey_psycopg2.MockConn) - -def test_db_table_creation_check(monkeypatch): - """ - test db_table_creation_check() - """ - monkeypatch.setattr(psycopg2, 'connect', monkey_psycopg2.monkey_connect) - monkey_psycopg2.monkey_reset_forces() - dbconn = config_notif.postgres_db_open("test", "testpswd", "testsite", 5432, "dbname") - dbconn.monkey_setDbInfo({ "select * from information_schema.tables": [ [ "testtable" ] ] }) - assert(type(dbconn) is monkey_psycopg2.MockConn) - ret = config_notif.db_table_creation_check(dbconn, "testtable") - assert(ret == True) - ret2 = config_notif.db_table_creation_check(dbconn, "missingtable") - monkey_psycopg2.monkey_reset_forces(cursor=True) - ret3 = config_notif.db_table_creation_check(dbconn, "testtable") - assert(ret3 is None) - -def test_commit_and_close_db(monkeypatch): - """ - test commit_and_close_db() - """ - monkeypatch.setattr(psycopg2, 'connect', monkey_psycopg2.monkey_connect) - monkey_psycopg2.monkey_reset_forces() - dbconn = config_notif.postgres_db_open("test", "testpswd", "testsite", 5432, "dbname") - assert(type(dbconn) is monkey_psycopg2.MockConn) - print("commit_and_close_db(): no forced failures") - ret = config_notif.commit_and_close_db(dbconn) - assert(ret == True) - -def test_commit_and_close_db_fail1(monkeypatch): - """ - failure case for commit_and_close_db(): dbconn.close() fails - """ - monkeypatch.setattr(psycopg2, 'connect', monkey_psycopg2.monkey_connect) - monkey_psycopg2.monkey_reset_forces() - dbconn = config_notif.postgres_db_open("test", "testpswd", "testsite", 5432, "dbname") - assert(type(dbconn) is monkey_psycopg2.MockConn) - print("commit_and_close_db() - close failure") - monkey_psycopg2.monkey_reset_forces(close=True) - ret = config_notif.commit_and_close_db(dbconn) - assert(ret == False) - -def test_commit_and_close_db_fail2(monkeypatch): - """ - failure case for commit_and_close_db(): dbconn.commit() fails - """ - monkeypatch.setattr(psycopg2, 'connect', monkey_psycopg2.monkey_connect) - monkey_psycopg2.monkey_reset_forces() - dbconn = config_notif.postgres_db_open("test", "testpswd", "testsite", 5432, "dbname") - assert(type(dbconn) is monkey_psycopg2.MockConn) - print("commit_and_close_db() - commit failure") - monkey_psycopg2.monkey_reset_forces(commit=True) - ret = config_notif.commit_and_close_db(dbconn) - assert(ret == False) - -def test_read_hb_properties_default(monkeypatch): - """ - test read_hb_properties_default() - """ - monkeypatch.setattr(psycopg2, 'connect', monkey_psycopg2.monkey_connect) - monkey_psycopg2.monkey_reset_forces() - monkey_psycopg2.monkey_set_defaults({ - "testdb1": { - "hb_common": [ - [ 1, "sn1", 31, "st1" ], - [ 2, "sn2", 32, "st2" ] - ] - } - }) - - output = config_notif.read_hb_common("test", "testpswd", "testsite", 5432, "testdb1") - assert(output[0] == 1) - assert(output[1] == "st1") - assert(output[2] == "sn1") - assert(output[3] == 31) - -def test_update_hb_common(monkeypatch): - """ - test update_hb_common() - """ - monkeypatch.setattr(psycopg2, 'connect', monkey_psycopg2.monkey_connect) - monkey_psycopg2.monkey_reset_forces() - output = config_notif.update_hb_common(None, 1234, "st1234", "test", "testpswd", "testsite", 5432, "testdb1") - assert(output == True) - -def monkeypatch_get_cbs_config_False(): - """ - monkeypatch for get_cbs_config() to force it to return False - Required side affect: c_config is set to a json value - """ - print("monkeypatch_get_cbs_config_False()") - mod.trapd_settings.c_config = { "patch": "false" } - return False - -def monkeypatch_get_cbs_config_True(): - """ - monkeypatch for get_cbs_config() to force it to return False - Required side affect: c_config is set to a json value - """ - print("monkeypatch_get_cbs_config_True()") - mod.trapd_settings.c_config = { "patch": "true" } - return True - -def test_fetch_json_file_get_cbs_config_is_true(monkeypatch): - """ - test fetch_json_file() with get_cbs_config() returning True - """ - monkeypatch.setattr(mod.trapd_get_cbs_config, 'get_cbs_config', monkeypatch_get_cbs_config_True) - tmp1 = tempfile.NamedTemporaryFile(mode="w+") - tmp2 = tempfile.NamedTemporaryFile(mode="w+") - output = config_notif.fetch_json_file(download_json = tmp1.name, config_json = tmp2.name) - assert(output == tmp1.name) - with open(tmp1.name, "r") as fp: - j1 = json.load(fp) - print(f"j1={j1}") - assert("patch" in j1 and j1["patch"] == "true") - -def test_fetch_json_file_get_cbs_config_is_false(monkeypatch): - """ - test fetch_json_file() with get_cbs_config() returning False - """ - monkeypatch.setattr(mod.trapd_get_cbs_config, 'get_cbs_config', monkeypatch_get_cbs_config_False) - tmp1 = tempfile.NamedTemporaryFile(mode="w+") - tmp2 = tempfile.NamedTemporaryFile(mode="w+") - output = config_notif.fetch_json_file(download_json = tmp1.name, config_json = tmp2.name) - assert(output == tmp2.name) - -FETCH_JSON_FILE = None - -def monkeypatch_fetch_json_file(): - """ - Monkeypatch for fetch_json_file() to test config_notif_run() - """ - print("monkeypatch_fetch_json_file()") - return FETCH_JSON_FILE - -def monkeypatch_return_False(*args, **kwargs): - """ - Monkeypatch that can be used to force a function to return False - """ - print("monkeypatch_return_False()") - return False - - -def test_config_notif_run_good(monkeypatch): - """ - test config_notif_run() - everything good: "dbname" found (from below JSON info), "hb_common" listed in tables - and hb_common has data. - """ - monkeypatch.setattr(config_notif, 'fetch_json_file', monkeypatch_fetch_json_file) - - tmp = tempfile.NamedTemporaryFile(mode="w+") - global FETCH_JSON_FILE - FETCH_JSON_FILE = tmp.name - - monkeypatch.setattr(psycopg2, 'connect', monkey_psycopg2.monkey_connect) - monkey_psycopg2.monkey_reset_forces() - - monkey_psycopg2.monkey_set_defaults({ - "dbname": { - "from information_schema.tables": [ - [ "hb_common" ] - ], - "from hb_common": [ - [ 1, "sn1", 31, "st1" ], - [ 2, "sn2", 32, "st2" ] - ] - } - }) - - testdata = { - "pg_ipAddress": "10.0.0.99", - "pg_portNum": 65432, - "pg_dbName": "dbname", - "pg_userName": "pguser", - "pg_passwd": "pgpswd", - "CBS_polling_allowed": True, - "CBS_polling_interval": 30, - "SERVICE_NAME": "service_name" - } - json.dump(testdata, tmp) - tmp.flush() - - output = config_notif.config_notif_run() - print(f"output={output}") - assert(output == True) - -def test_config_notif_run_fail1(monkeypatch): - """ - test config_notif_run() - Failure case 1: "dbname" NOT found (from below JSON info), "hb_common" listed in tables - and hb_common has data. - """ - monkeypatch.setattr(config_notif, 'fetch_json_file', monkeypatch_fetch_json_file) - - tmp = tempfile.NamedTemporaryFile(mode="w+") - global FETCH_JSON_FILE - FETCH_JSON_FILE = tmp.name - - monkeypatch.setattr(psycopg2, 'connect', monkey_psycopg2.monkey_connect) - monkey_psycopg2.monkey_reset_forces() - - monkey_psycopg2.monkey_set_defaults({ - "dbnameNOTHERE": { - "from information_schema.tables": [ - [ "hb_common" ] - ], - "from hb_common": [ - [ 1, "sn1", 31, "st1" ], - [ 2, "sn2", 32, "st2" ] - ] - } - }) - - testdata = { - "pg_ipAddress": "10.0.0.99", - "pg_portNum": 65432, - "pg_dbName": "dbname", - "pg_userName": "pguser", - "pg_passwd": "pgpswd", - "CBS_polling_allowed": True, - "CBS_polling_interval": 30, - "SERVICE_NAME": "service_name" - } - json.dump(testdata, tmp) - tmp.flush() - - output = config_notif.config_notif_run() - print(f"output={output}") - assert(output is None) - -def test_config_notif_run_fail2(monkeypatch): - """ - test config_notif_run() - Failure case 2: "dbname" found (from below JSON info), "hb_common" NOT listed in tables - and hb_common has data. - """ - monkeypatch.setattr(config_notif, 'fetch_json_file', monkeypatch_fetch_json_file) - - tmp = tempfile.NamedTemporaryFile(mode="w+") - global FETCH_JSON_FILE - FETCH_JSON_FILE = tmp.name - - monkeypatch.setattr(psycopg2, 'connect', monkey_psycopg2.monkey_connect) - monkey_psycopg2.monkey_reset_forces() - - monkey_psycopg2.monkey_set_defaults({ - "dbname": { - "from information_schema.tables": [ - [ "hb_commonNOTHERE" ] - ], - "from hb_common": [ - [ 1, "sn1", 31, "st1" ], - [ 2, "sn2", 32, "st2" ] - ] - } - }) - - testdata = { - "pg_ipAddress": "10.0.0.99", - "pg_portNum": 65432, - "pg_dbName": "dbname", - "pg_userName": "pguser", - "pg_passwd": "pgpswd", - "CBS_polling_allowed": True, - "CBS_polling_interval": 30, - "SERVICE_NAME": "service_name" - } - json.dump(testdata, tmp) - tmp.flush() - - output = config_notif.config_notif_run() - print(f"output={output}") - assert(output is None) - -def test_config_notif_run_fail3(monkeypatch): - """ - test config_notif_run() - Failure case 3: "dbname" found (from below JSON info), "hb_common" listed in tables - and update_hb_common() fails - """ - monkeypatch.setattr(config_notif, 'fetch_json_file', monkeypatch_fetch_json_file) - monkeypatch.setattr(config_notif, 'update_hb_common', monkeypatch_return_False) - - tmp = tempfile.NamedTemporaryFile(mode="w+") - global FETCH_JSON_FILE - FETCH_JSON_FILE = tmp.name - - monkeypatch.setattr(psycopg2, 'connect', monkey_psycopg2.monkey_connect) - monkey_psycopg2.monkey_reset_forces() - - monkey_psycopg2.monkey_set_defaults({ - "dbname": { - "from information_schema.tables": [ - [ "hb_common" ] - ], - "from hb_common": [ - [ 1, "sn1", 31, "st1" ], - [ 2, "sn2", 32, "st2" ] - ] - } - }) - - testdata = { - "pg_ipAddress": "10.0.0.99", - "pg_portNum": 65432, - "pg_dbName": "dbname", - "pg_userName": "pguser", - "pg_passwd": "pgpswd", - "CBS_polling_allowed": True, - "CBS_polling_interval": 30, - "SERVICE_NAME": "service_name" - } - json.dump(testdata, tmp) - tmp.flush() - - output = config_notif.config_notif_run() - print(f"output={output}") - assert(output == False) |