diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/monkey_psycopg2.py | 27 | ||||
| -rw-r--r-- | tests/test_binding.py | 85 | ||||
| -rw-r--r-- | tests/test_check_health.py | 34 | ||||
| -rw-r--r-- | tests/test_get_logger.py | 12 | ||||
| -rw-r--r-- | tests/test_htbtworker.py | 15 | ||||
| -rw-r--r-- | tests/test_trapd_exit.py | 15 | ||||
| -rw-r--r-- | tests/test_trapd_get_cbs_config.py | 13 | ||||
| -rw-r--r-- | tests/test_trapd_http_session.py | 5 | ||||
| -rw-r--r-- | tests/test_trapd_runtime_pid.py | 16 | ||||
| -rw-r--r-- | tests/test_trapd_settings.py | 9 | ||||
| -rw-r--r-- | tests/test_trapd_vnf_table.py | 51 |
11 files changed, 176 insertions, 106 deletions
diff --git a/tests/monkey_psycopg2.py b/tests/monkey_psycopg2.py index 05f44b4..e993874 100644 --- a/tests/monkey_psycopg2.py +++ b/tests/monkey_psycopg2.py @@ -1,6 +1,5 @@ # ============LICENSE_START==================================================== -# ============================================================================= -# Copyright (c) 2017-2020 AT&T Intellectual Property. All rights reserved. +# Copyright (c) 2017-2021 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. @@ -34,14 +33,15 @@ FORCE_CLOSE_FAILURE = False # This variable is used by monkey_psycopg2.monkey_set_defaults(multiDbInfo) # to set up default values to be returned by cursors statements. -DEFAULT_MULTI_DBINFO = { } +DEFAULT_MULTI_DBINFO = {} + class MockConn(object): """ mock Connection interface returned by psycopg2.connect() """ - def __init__(self, dbInfo = None): + def __init__(self, dbInfo=None): self.curCmd = None self.curCursor = None self.monkey_setDbInfo(dbInfo) @@ -49,7 +49,7 @@ class MockConn(object): def monkey_setDbInfo(self, dbInfo): """ Set up a set of defaults for the cursors on "select" statements. - The outer scope is a string that will be matched against the currently-active + The outer scope is a string that will be matched against the currently-active select statement being executed. If there is a match, the specified values are returned by the cursor. dbconn.monkey_setDbInfo({ @@ -71,7 +71,7 @@ class MockConn(object): """ if FORCE_COMMIT_FAILURE: raise psycopg2.DatabaseError(f"Unable to commit: force_commit_failure=<{FORCE_COMMIT_FAILURE}>") - + def close(self): """ mock close @@ -110,7 +110,7 @@ class MockConn(object): self.curCursor = None return self - def execute(self, cmd, args = None): + def execute(self, cmd, args=None): """ mock execute @@ -155,6 +155,7 @@ class MockConn(object): print(f"fetchall() returning {self.curCursor}") return self.curCursor + def monkey_reset_forces(connect=False, cursor=False, execute=False, commit=False, close=False): print(f"monkey_reset_forces({connect}, {cursor}, {execute}, {commit}, {close})") global FORCE_CONNECT_FAILURE @@ -168,11 +169,12 @@ def monkey_reset_forces(connect=False, cursor=False, execute=False, commit=False global FORCE_CLOSE_FAILURE FORCE_CLOSE_FAILURE = close + def monkey_set_defaults(multiDbInfo): """ Set up a set of defaults for the cursors on "select" statements. The outer scope gives a database name. - The next level is a string that will be matched against the currently-active + The next level is a string that will be matched against the currently-active select statement being executed. If both match, the specified values are returned by the cursor. monkey_psycopg2.monkey_set_defaults({ @@ -187,10 +189,11 @@ def monkey_set_defaults(multiDbInfo): global DEFAULT_MULTI_DBINFO DEFAULT_MULTI_DBINFO = multiDbInfo + def monkey_connect(database=None, host=None, port=None, user=None, password=None): """ Mock psycopg2 connection. - Returns a mock connection. + Returns a mock connection. Will raise an exception if value of 'FORCE_CONNECT_FAILURE' is true. (Used to force failure for certain code paths.) @@ -199,7 +202,11 @@ def monkey_connect(database=None, host=None, port=None, user=None, password=None (See monkey_set_defaults(), which must have been called prior to this being invoked.) """ if FORCE_CONNECT_FAILURE: - raise Exception("Unable to connect to the database. password=<{}> force_connect_failure=<{}>".format(password, FORCE_CONNECT_FAILURE)) + raise Exception( + "Unable to connect to the database. password=<{}> force_connect_failure=<{}>".format( + password, FORCE_CONNECT_FAILURE + ) + ) if database in DEFAULT_MULTI_DBINFO: return MockConn(DEFAULT_MULTI_DBINFO[database]) diff --git a/tests/test_binding.py b/tests/test_binding.py index 6c8b0ef..5ed6532 100644 --- a/tests/test_binding.py +++ b/tests/test_binding.py @@ -1,8 +1,8 @@ # ============LICENSE_START======================================================= -# Copyright (c) 2017-2020 AT&T Intellectual Property. All rights reserved. +# Copyright (c) 2017-2021 AT&T Intellectual Property. All rights reserved. # Copyright (c) 2019 Pantheon.tech. All rights reserved. -# Copyright 2021 Samsung Electronics. All rights reserved. -# Copyright 2021 Fujitsu Ltd. +# Copyright (c) 2021 Samsung Electronics. All rights reserved. +# Copyright (c) 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. @@ -24,44 +24,85 @@ import httpretty import subprocess import json -MODULE_EXTENSIONS = ('.py', '.pyc', '.pyo') +MODULE_EXTENSIONS = (".py", ".pyc", ".pyo") ##### # MONKEYPATCHES ##### -mr_url = 'http://mrrouter.onap.org:3904' -intopic = 'VESCOLL-VNFNJ-SECHEARTBEAT-OUTPUT' -outopic = 'POLICY-HILOTCA-EVENT-OUTPUT' +mr_url = "http://mrrouter.onap.org:3904" +intopic = "VESCOLL-VNFNJ-SECHEARTBEAT-OUTPUT" +outopic = "POLICY-HILOTCA-EVENT-OUTPUT" @pytest.fixture() def disable_proxy(monkeypatch): - if 'http_proxy' in os.environ: - monkeypatch.delenv('http_proxy') - if 'HTTP_PROXY' in os.environ: - monkeypatch.delenv('HTTP_PROXY') + if "http_proxy" in os.environ: + monkeypatch.delenv("http_proxy") + if "HTTP_PROXY" in os.environ: + monkeypatch.delenv("HTTP_PROXY") @httpretty.activate def test_resolve_all(disable_proxy): htbtmsg = '{"event":{"commonEventHeader":{"startEpochMicrosec":1518616063564475,"sourceId":"587c14b3-72c0-4581-b5cb-6567310b9bb7","eventId":"10048640","reportingEntityId":"587c14b3-72c0-4581-b5cb-6567310b9bb7","priority":"Normal","version":3,"reportingEntityName":"TESTVM","sequence":10048640,"domain":"heartbeat","lastEpochMicrosec":1518616063564476,"eventName":"Heartbeat_vVnf","sourceName":"TESTVM","nfNamingCode":"vVNF"}}}' - send_url = mr_url+'/events/'+intopic+'/DefaultGroup/1?timeout=15000' + send_url = mr_url + "/events/" + intopic + "/DefaultGroup/1?timeout=15000" print(send_url) httpretty.register_uri(httpretty.GET, send_url, body=htbtmsg) - #Use + # Use response = requests.get(send_url) print(response) print(response.text) - assert(response.text == htbtmsg) - htbtmsg = json.dumps({"event":{"commonEventHeader":{"startEpochMicrosec":1518616063564475,"sourceId":"587c14b3-72c0-4581-b5cb-6567310b9bb7","eventId":"10048640","reportingEntityId":"587c14b3-72c0-4581-b5cb-6567310b9bb7","priority":"Normal","version":3,"reportingEntityName":"TESTVM","sequence":10048640,"domain":"heartbeat","lastEpochMicrosec":1518616063564476,"eventName":"Heartbeat_vVnf","sourceName":"TESTVM","nfNamingCode":"vVNF"}}}) - send_url = mr_url+'/events/'+intopic+'/DefaultGroup/1?timeout=15000' - print("Send URL : "+send_url) + assert response.text == htbtmsg + htbtmsg = json.dumps( + { + "event": { + "commonEventHeader": { + "startEpochMicrosec": 1518616063564475, + "sourceId": "587c14b3-72c0-4581-b5cb-6567310b9bb7", + "eventId": "10048640", + "reportingEntityId": "587c14b3-72c0-4581-b5cb-6567310b9bb7", + "priority": "Normal", + "version": 3, + "reportingEntityName": "TESTVM", + "sequence": 10048640, + "domain": "heartbeat", + "lastEpochMicrosec": 1518616063564476, + "eventName": "Heartbeat_vVnf", + "sourceName": "TESTVM", + "nfNamingCode": "vVNF", + } + } + } + ) + send_url = mr_url + "/events/" + intopic + "/DefaultGroup/1?timeout=15000" + print("Send URL : " + send_url) httpretty.register_uri(httpretty.GET, send_url, body=htbtmsg, content_type="application/json") - pol_url = mr_url+'/events/'+outopic+'/DefaultGroup/1?timeout=15000' - pol_body = json.dumps({"event":{"commonEventHeader":{"startEpochMicrosec":1518616063564475,"sourceId":"587c14b3-72c0-4581-b5cb-6567310b9bb7","eventId":"10048640","reportingEntityId":"587c14b3-72c0-4581-b5cb-6567310b9bb7","priority":"Normal","version":3,"reportingEntityName":"TESTVM","sequence":10048640,"domain":"heartbeat","lastEpochMicrosec":1518616063564476,"eventName":"Heartbeat_vVnf","sourceName":"TESTVM","nfNamingCode":"vVNF"}}}) - print("Policy URL : "+pol_url) - httpretty.register_uri(httpretty.POST, pol_url, body=pol_body, status=200, content_type='text/json') + pol_url = mr_url + "/events/" + outopic + "/DefaultGroup/1?timeout=15000" + pol_body = json.dumps( + { + "event": { + "commonEventHeader": { + "startEpochMicrosec": 1518616063564475, + "sourceId": "587c14b3-72c0-4581-b5cb-6567310b9bb7", + "eventId": "10048640", + "reportingEntityId": "587c14b3-72c0-4581-b5cb-6567310b9bb7", + "priority": "Normal", + "version": 3, + "reportingEntityName": "TESTVM", + "sequence": 10048640, + "domain": "heartbeat", + "lastEpochMicrosec": 1518616063564476, + "eventName": "Heartbeat_vVnf", + "sourceName": "TESTVM", + "nfNamingCode": "vVNF", + } + } + } + ) + print("Policy URL : " + pol_url) + httpretty.register_uri(httpretty.POST, pol_url, body=pol_body, status=200, content_type="text/json") + def test_full(): - p = subprocess.Popen(['./miss_htbt_service/misshtbtd.py'], stdout=subprocess.PIPE,shell=True) + p = subprocess.Popen(["./miss_htbt_service/misshtbtd.py"], stdout=subprocess.PIPE, shell=True) diff --git a/tests/test_check_health.py b/tests/test_check_health.py index d3088f9..d77beb1 100644 --- a/tests/test_check_health.py +++ b/tests/test_check_health.py @@ -1,7 +1,7 @@ # ============LICENSE_START======================================================= -# Copyright (c) 2017-2020 AT&T Intellectual Property. All rights reserved. +# Copyright (c) 2017-2021 AT&T Intellectual Property. All rights reserved. # Copyright (c) 2019 Pantheon.tech. All rights reserved. -# Copyright 2021 Fujitsu Ltd. +# Copyright (c) 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. @@ -22,9 +22,11 @@ from miss_htbt_service import check_health import io + class MockSocket(object): def getsockname(self): - return 'sockname', + return ("sockname",) + class MockRequest(object): _sock = MockSocket() @@ -35,32 +37,40 @@ class MockRequest(object): self._body = body def makefile(self, *args, **kwargs): - if args[0] == 'rb': - if self._rqtype == 'GET': - return io.BytesIO(bytes("%s %s HTTP/1.0" % (self._rqtype, self._path), 'utf-8')) + if args[0] == "rb": + if self._rqtype == "GET": + return io.BytesIO(bytes("%s %s HTTP/1.0" % (self._rqtype, self._path), "utf-8")) else: - return io.BytesIO(bytes("%s %s HTTP/1.0\r\nContent-Length: %s\r\n\r\n%s" % (self._rqtype, self._path, len(self._body), self._body), 'utf-8')) - elif args[0] == 'wb': - return io.BytesIO(b'') + return io.BytesIO( + bytes( + "%s %s HTTP/1.0\r\nContent-Length: %s\r\n\r\n%s" + % (self._rqtype, self._path, len(self._body), self._body), + "utf-8", + ) + ) + elif args[0] == "wb": + return io.BytesIO(b"") else: raise ValueError("Unknown file type to make", args, kwargs) def sendall(self, bstr): pass + class MockServer(object): def __init__(self, rqtype, path, ip_port, Handler, body=None): handler = Handler(MockRequest(rqtype, path, body), ip_port, self) + def test_check_health_get(): """ test the check_health GET and POST handlers using a mock server """ - server = MockServer('GET', '/', ('0.0.0.0', 8888), check_health.GetHandler) + server = MockServer("GET", "/", ("0.0.0.0", 8888), check_health.GetHandler) + def test_check_health_post(): """ test the check_health GET and POST handlers using a mock server """ - server = MockServer('POST', '/', ('0.0.0.0', 8888), check_health.GetHandler, - '{ "health": "" }') + server = MockServer("POST", "/", ("0.0.0.0", 8888), check_health.GetHandler, '{ "health": "" }') diff --git a/tests/test_get_logger.py b/tests/test_get_logger.py index cbef9c6..a4ceea5 100644 --- a/tests/test_get_logger.py +++ b/tests/test_get_logger.py @@ -1,6 +1,6 @@ # ============LICENSE_START======================================================= -# Copyright (c) 2020 AT&T Intellectual Property. All rights reserved. -# Copyright 2021 Fujitsu Ltd. +# Copyright (c) 2020-2021 AT&T Intellectual Property. All rights reserved. +# Copyright (c) 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. @@ -24,20 +24,20 @@ log = logging.getLogger(__name__) def test_configure_logger(): - expected_log_path = Path('./hb_logs.txt') + expected_log_path = Path("./hb_logs.txt") if expected_log_path.exists(): os.remove(expected_log_path) - get_logger.configure_logger('') + get_logger.configure_logger("") log.info("hi there") assert expected_log_path.exists() os.remove(expected_log_path) def test_configure_logger_with_name(): - expected_log_path = Path('./hb_htbtworker_logs.txt') + expected_log_path = Path("./hb_htbtworker_logs.txt") if expected_log_path.exists(): os.remove(expected_log_path) - get_logger.configure_logger('htbtworker') + get_logger.configure_logger("htbtworker") log.info("hi there") assert expected_log_path.exists() os.remove(expected_log_path) diff --git a/tests/test_htbtworker.py b/tests/test_htbtworker.py index 97e61b4..11cdc4a 100644 --- a/tests/test_htbtworker.py +++ b/tests/test_htbtworker.py @@ -1,5 +1,5 @@ # ============LICENSE_START======================================================= -# Copyright (c) 2020 AT&T Intellectual Property. All rights reserved. +# Copyright (c) 2020-2021 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. @@ -18,6 +18,7 @@ from miss_htbt_service import htbtworker import os, tempfile, json + def run_test(i): """ read_json_file() opens the file CWD/prefix/test{j}.json and returns the json value found there @@ -28,18 +29,20 @@ def run_test(i): pdir = f"{prefix}{tdir.name}" fname = f"{tdir.name}/test{j}.json" with open(fname, "w") as fp: - json.dump({ "test": i }, fp) - assert(os.path.isfile(f"{tdir.name}/test{j}.json")) - assert(os.path.isfile(f"{pdir}/test{j}.json")) + json.dump({"test": i}, fp) + assert os.path.isfile(f"{tdir.name}/test{j}.json") + assert os.path.isfile(f"{pdir}/test{j}.json") cfg = htbtworker.read_json_file(i, prefix=pdir) - assert(cfg["test"] == i) + assert cfg["test"] == i + def test_read_json_file_0(): run_test(0) + def test_read_json_file_1(): run_test(1) + def test_read_json_file_2(): run_test(2) - diff --git a/tests/test_trapd_exit.py b/tests/test_trapd_exit.py index 35b7111..8803b29 100644 --- a/tests/test_trapd_exit.py +++ b/tests/test_trapd_exit.py @@ -1,7 +1,7 @@ # ============LICENSE_START======================================================= -# Copyright (c) 2017-2020 AT&T Intellectual Property. All rights reserved. +# Copyright (c) 2017-2021 AT&T Intellectual Property. All rights reserved. # Copyright (c) 2019 Pantheon.tech. All rights reserved. -# Copyright 2021 Fujitsu Ltd. +# Copyright (c) 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. @@ -20,8 +20,9 @@ import pytest import unittest from miss_htbt_service.mod import trapd_exit -pid_file="/tmp/test_pid_file" -pid_file_dne="/tmp/test_pid_file_NOT" +pid_file = "/tmp/test_pid_file" +pid_file_dne = "/tmp/test_pid_file_NOT" + class test_cleanup_and_exit(unittest.TestCase): """ @@ -32,10 +33,10 @@ class test_cleanup_and_exit(unittest.TestCase): """ Test normal exit works as expected """ - open(pid_file, 'w') + open(pid_file, "w") with pytest.raises(SystemExit) as pytest_wrapped_sys_exit: - result = trapd_exit.cleanup_and_exit(0,pid_file) + result = trapd_exit.cleanup_and_exit(0, pid_file) assert pytest_wrapped_sys_exit.type == SystemExit assert pytest_wrapped_sys_exit.value.code == 0 @@ -44,6 +45,6 @@ class test_cleanup_and_exit(unittest.TestCase): Test exit with missing PID file exits non-zero """ with pytest.raises(SystemExit) as pytest_wrapped_sys_exit: - result = trapd_exit.cleanup_and_exit(0,pid_file_dne) + result = trapd_exit.cleanup_and_exit(0, pid_file_dne) assert pytest_wrapped_sys_exit.type == SystemExit assert pytest_wrapped_sys_exit.value.code == 1 diff --git a/tests/test_trapd_get_cbs_config.py b/tests/test_trapd_get_cbs_config.py index 6cebef2..ffb9bfb 100644 --- a/tests/test_trapd_get_cbs_config.py +++ b/tests/test_trapd_get_cbs_config.py @@ -1,7 +1,7 @@ # ============LICENSE_START======================================================= -# Copyright (c) 2017-2020 AT&T Intellectual Property. All rights reserved. +# Copyright (c) 2017-2021 AT&T Intellectual Property. All rights reserved. # Copyright (c) 2019 Pantheon.tech. All rights reserved. -# Copyright 2021 Fujitsu Ltd. +# Copyright (c) 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. @@ -22,19 +22,19 @@ import os from miss_htbt_service.mod import trapd_get_cbs_config + class test_get_cbs_config(unittest.TestCase): """ Test the trapd_get_cbs_config mod """ - pytest_json_data ="{ \"heartbeat_config\": { \"vnfs\": [{ \"eventName\": \"Heartbeat_vDNS\", \"heartbeatcountmissed\": 3, \"heartbeatinterval\": 60, \"closedLoopControlName\": \"ControlLoopEvent1\", \"policyVersion\": \"1.0.0.5\", \"policyName\": \"vFireWall\", \"policyScope\": \"resource=sampleResource,type=sampletype,CLName=sampleCLName\", \"target_type\": \"VNF\", \"target\": \"genVnfName\", \"version\": \"1.0\" }, { \"eventName\": \"Heartbeat_vFW\", \"heartbeatcountmissed\": 3, \"heartbeatinterval\": 60, \"closedLoopControlName\": \"ControlLoopEvent1\", \"policyVersion\": \"1.0.0.5\", \"policyName\": \"vFireWall\", \"policyScope\": \"resource=sampleResource,type=sampletype,CLName=sampleCLName\", \"target_type\": \"VNF\", \"target\": \"genVnfName\", \"version\": \"1.0\" }, { \"eventName\": \"Heartbeat_xx\", \"heartbeatcountmissed\": 3, \"heartbeatinterval\": 60, \"closedLoopControlName\": \"ControlLoopEvent1\", \"policyVersion\": \"1.0.0.5\", \"policyName\": \"vFireWall\", \"policyScope\": \"resource=sampleResource,type=sampletype,CLName=sampleCLName\", \"target_type\": \"VNF\", \"target\": \"genVnfName\", \"version\": \"1.0\" } ] }, \"streams_publishes\": { \"ves_heartbeat\": { \"dmaap_info\": { \"topic_url\": \"http://message-router:3904/events/unauthenticated.DCAE_CL_OUTPUT/\" }, \"type\": \"message_router\" } }, \"streams_subscribes\": { \"ves_heartbeat\": { \"dmaap_info\": { \"topic_url\": \"http://message-router:3904/events/unauthenticated.SEC_HEARTBEAT_INPUT/\" }, \"type\": \"message_router\" } } }" + pytest_json_data = '{ "heartbeat_config": { "vnfs": [{ "eventName": "Heartbeat_vDNS", "heartbeatcountmissed": 3, "heartbeatinterval": 60, "closedLoopControlName": "ControlLoopEvent1", "policyVersion": "1.0.0.5", "policyName": "vFireWall", "policyScope": "resource=sampleResource,type=sampletype,CLName=sampleCLName", "target_type": "VNF", "target": "genVnfName", "version": "1.0" }, { "eventName": "Heartbeat_vFW", "heartbeatcountmissed": 3, "heartbeatinterval": 60, "closedLoopControlName": "ControlLoopEvent1", "policyVersion": "1.0.0.5", "policyName": "vFireWall", "policyScope": "resource=sampleResource,type=sampletype,CLName=sampleCLName", "target_type": "VNF", "target": "genVnfName", "version": "1.0" }, { "eventName": "Heartbeat_xx", "heartbeatcountmissed": 3, "heartbeatinterval": 60, "closedLoopControlName": "ControlLoopEvent1", "policyVersion": "1.0.0.5", "policyName": "vFireWall", "policyScope": "resource=sampleResource,type=sampletype,CLName=sampleCLName", "target_type": "VNF", "target": "genVnfName", "version": "1.0" } ] }, "streams_publishes": { "ves_heartbeat": { "dmaap_info": { "topic_url": "http://message-router:3904/events/unauthenticated.DCAE_CL_OUTPUT/" }, "type": "message_router" } }, "streams_subscribes": { "ves_heartbeat": { "dmaap_info": { "topic_url": "http://message-router:3904/events/unauthenticated.SEC_HEARTBEAT_INPUT/" }, "type": "message_router" } } }' # create copy of snmptrapd.json for pytest pytest_json_config = "/tmp/opt/app/miss_htbt_service/etc/config.json" - with open(pytest_json_config, 'w') as outfile: + with open(pytest_json_config, "w") as outfile: outfile.write(pytest_json_data) - def test_cbs_env_present(self): """ Test that CONSUL_HOST env variable exists but fails to @@ -45,13 +45,12 @@ class test_get_cbs_config(unittest.TestCase): result = trapd_get_cbs_config.get_cbs_config() assert pytest_wrapped_sys_exit.type == SystemExit - def test_cbs_fallback_env_present(self): """ Test that CBS fallback env variable exists and we can get config from fallback env var """ - os.environ.update(CBS_HTBT_JSON='/tmp/opt/app/miss_htbt_service/etc/config.json') + os.environ.update(CBS_HTBT_JSON="/tmp/opt/app/miss_htbt_service/etc/config.json") result = True print("result: %s" % result) self.assertEqual(result, True) diff --git a/tests/test_trapd_http_session.py b/tests/test_trapd_http_session.py index 47380fc..070fc93 100644 --- a/tests/test_trapd_http_session.py +++ b/tests/test_trapd_http_session.py @@ -1,7 +1,7 @@ # ============LICENSE_START======================================================= -# Copyright (c) 2017-2020 AT&T Intellectual Property. All rights reserved. +# Copyright (c) 2017-2021 AT&T Intellectual Property. All rights reserved. # Copyright (c) 2019 Pantheon.tech. All rights reserved. -# Copyright 2021 Fujitsu Ltd. +# Copyright (c) 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. @@ -19,6 +19,7 @@ import unittest from miss_htbt_service.mod import trapd_http_session + class test_init_session_obj(unittest.TestCase): """ Test the init_session_obj mod diff --git a/tests/test_trapd_runtime_pid.py b/tests/test_trapd_runtime_pid.py index f31c4db..47bc642 100644 --- a/tests/test_trapd_runtime_pid.py +++ b/tests/test_trapd_runtime_pid.py @@ -1,7 +1,7 @@ # ============LICENSE_START======================================================= -# Copyright (c) 2017-2020 AT&T Intellectual Property. All rights reserved. +# Copyright (c) 2017-2021 AT&T Intellectual Property. All rights reserved. # Copyright (c) 2019 Pantheon.tech. All rights reserved. -# Copyright 2021 Fujitsu Ltd. +# Copyright (c) 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. @@ -19,6 +19,7 @@ import unittest from miss_htbt_service.mod import trapd_runtime_pid + class test_save_pid(unittest.TestCase): """ Test the save_pid mod @@ -28,16 +29,17 @@ class test_save_pid(unittest.TestCase): """ Test that attempt to create pid file in standard location works """ - result = trapd_runtime_pid.save_pid('/tmp/snmptrap_test_pid_file') + result = trapd_runtime_pid.save_pid("/tmp/snmptrap_test_pid_file") self.assertEqual(result, True) def test_missing_directory(self): """ Test that attempt to create pid file in missing dir fails """ - result = trapd_runtime_pid.save_pid('/bogus/directory/for/snmptrap_test_pid_file') + result = trapd_runtime_pid.save_pid("/bogus/directory/for/snmptrap_test_pid_file") self.assertEqual(result, False) + class test_rm_pid(unittest.TestCase): """ Test the rm_pid mod @@ -48,14 +50,14 @@ class test_rm_pid(unittest.TestCase): Test that attempt to remove pid file in standard location works """ # must create it before removing it - result = trapd_runtime_pid.save_pid('/tmp/snmptrap_test_pid_file') + result = trapd_runtime_pid.save_pid("/tmp/snmptrap_test_pid_file") self.assertEqual(result, True) - result = trapd_runtime_pid.rm_pid('/tmp/snmptrap_test_pid_file') + result = trapd_runtime_pid.rm_pid("/tmp/snmptrap_test_pid_file") self.assertEqual(result, True) def test_missing_file(self): """ Test that attempt to rm non-existent pid file fails """ - result = trapd_runtime_pid.rm_pid('/tmp/snmptrap_test_pid_file_9999') + result = trapd_runtime_pid.rm_pid("/tmp/snmptrap_test_pid_file_9999") self.assertEqual(result, False) diff --git a/tests/test_trapd_settings.py b/tests/test_trapd_settings.py index d800119..f3aebdd 100644 --- a/tests/test_trapd_settings.py +++ b/tests/test_trapd_settings.py @@ -1,7 +1,7 @@ # ============LICENSE_START======================================================= -# Copyright (c) 2017-2020 AT&T Intellectual Property. All rights reserved. +# Copyright (c) 2017-2021 AT&T Intellectual Property. All rights reserved. # Copyright (c) 2019 Pantheon.tech. All rights reserved. -# Copyright 2021 Fujitsu Ltd. +# Copyright (c) 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. @@ -20,8 +20,8 @@ import unittest from miss_htbt_service.mod import trapd_settings as tds -pid_file="/tmp/test_pid_file" -pid_file_dne="/tmp/test_pid_file_NOT" +pid_file = "/tmp/test_pid_file" +pid_file_dne = "/tmp/test_pid_file_NOT" class test_cleanup_and_exit(unittest.TestCase): @@ -29,7 +29,6 @@ class test_cleanup_and_exit(unittest.TestCase): Test for presense of required vars """ - def test_nonexistent_dict(self): """ Test nosuch var diff --git a/tests/test_trapd_vnf_table.py b/tests/test_trapd_vnf_table.py index e8ef5d0..b88b4d5 100644 --- a/tests/test_trapd_vnf_table.py +++ b/tests/test_trapd_vnf_table.py @@ -1,10 +1,8 @@ # ============LICENSE_START======================================================= -# org.onap.dcae -# ================================================================================ -# Copyright (c) 2017-2020 AT&T Intellectual Property. All rights reserved. +# Copyright (c) 2017-2021 AT&T Intellectual Property. All rights reserved. # Copyright (c) 2019 Pantheon.tech. All rights reserved. -# Copyright 2020 Deutsche Telekom. All rights reserved. -# Copyright 2021 Fujitsu Ltd. +# Copyright (c) 2020 Deutsche Telekom. All rights reserved. +# Copyright (c) 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. @@ -26,12 +24,20 @@ test_trapd_vnf_table contains test cases related to DB Tables and cbs polling. import logging import unittest from mod.trapd_vnf_table import ( - verify_DB_creation_1, verify_DB_creation_2, verify_DB_creation_hb_common, - hb_properties, verify_cbspolling, - verify_sendControlLoop_VNF_ONSET, verify_sendControlLoop_VM_ONSET, - verify_sendControlLoop_VNF_ABATED, verify_sendControlLoop_VM_ABATED, - verify_fetch_json_file, verify_misshtbtdmain, verify_dbmonitoring, - verify_dbmon_startup) + verify_DB_creation_1, + verify_DB_creation_2, + verify_DB_creation_hb_common, + hb_properties, + verify_cbspolling, + verify_sendControlLoop_VNF_ONSET, + verify_sendControlLoop_VM_ONSET, + verify_sendControlLoop_VNF_ABATED, + verify_sendControlLoop_VM_ABATED, + verify_fetch_json_file, + verify_misshtbtdmain, + verify_dbmonitoring, + verify_dbmon_startup, +) _logger = logging.getLogger(__name__) @@ -40,19 +46,20 @@ class test_vnf_tables(unittest.TestCase): """ Test the DB Creation """ + 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() def test_validate_vnf_table_1(self): - result =verify_DB_creation_1(user_name,password,ip_address,port_num,db_name) + result = verify_DB_creation_1(user_name, password, ip_address, port_num, db_name) self.assertEqual(result, True) def test_validate_vnf_table_2(self): - result =verify_DB_creation_2(user_name,password,ip_address,port_num,db_name) + result = verify_DB_creation_2(user_name, password, ip_address, port_num, db_name) self.assertEqual(result, True) def test_validate_hb_common(self): - result =verify_DB_creation_hb_common(user_name,password,ip_address,port_num,db_name) + result = verify_DB_creation_hb_common(user_name, password, ip_address, port_num, db_name) self.assertEqual(result, True) def test_cbspolling(self): @@ -60,41 +67,41 @@ class test_vnf_tables(unittest.TestCase): verify_cbspolling() def test_fetch_json_file(self): - result= verify_fetch_json_file() + result = verify_fetch_json_file() _logger.info(result) self.assertEqual(result, True) def test_misshtbtdmain(self): - result= verify_misshtbtdmain() + result = verify_misshtbtdmain() _logger.info(result) self.assertEqual(result, True) def test_dbmon_startup(self): - result= verify_dbmon_startup() + result = verify_dbmon_startup() _logger.info(result) self.assertEqual(result, True) def test_dbmonitoring(self): - result= verify_dbmonitoring() + result = verify_dbmonitoring() _logger.info(result) self.assertEqual(result, True) def test_sendControlLoop_VNF_ONSET(self): - result= verify_sendControlLoop_VNF_ONSET() + result = verify_sendControlLoop_VNF_ONSET() _logger.info(result) self.assertEqual(result, True) def test_sendControlLoop_VM_ONSET(self): - result= verify_sendControlLoop_VM_ONSET() + result = verify_sendControlLoop_VM_ONSET() _logger.info(result) self.assertEqual(result, True) def test_sendControlLoop_VNF_ABATED(self): - result= verify_sendControlLoop_VNF_ABATED() + result = verify_sendControlLoop_VNF_ABATED() _logger.info(result) self.assertEqual(result, True) def test_sendControlLoop_VM_ABATED(self): - result= verify_sendControlLoop_VM_ABATED() + result = verify_sendControlLoop_VM_ABATED() _logger.info(result) self.assertEqual(result, True) |