From 341b5bb2347c30344662675936b90b325efe5520 Mon Sep 17 00:00:00 2001
From: Vijay Venkatesh Kumar <vv770d@att.com>
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 <vv770d@att.com>
Change-Id: I99229d966c13ad666ac994ab5a582aeeaa306639
Signed-off-by: Vijay Venkatesh Kumar <vv770d@att.com>
---
 tests/hbproperties-test.yaml       |  11 ++
 tests/test-config.json             |   1 +
 tests/test4.json                   |   1 +
 tests/test_db_monitoring.py        |  75 +++++++++++++
 tests/test_get_logger.py           |   6 +-
 tests/test_htbt_exit.py            |  50 +++++++++
 tests/test_htbt_get_cbs_config.py  | 148 +++++++++++++++++++++++++
 tests/test_htbt_http_session.py    |  34 ++++++
 tests/test_htbt_runtime_pid.py     |  63 +++++++++++
 tests/test_htbt_settings.py        |  81 ++++++++++++++
 tests/test_htbt_vnf_table.py       | 128 ++++++++++++++++++++++
 tests/test_htbtworker.py           | 163 +++++++++++++++++++++++-----
 tests/test_misshtbtd.py            | 215 +++++++++++++++++++++++++++++++++++++
 tests/test_trapd_exit.py           |  50 ---------
 tests/test_trapd_get_cbs_config.py | 121 ---------------------
 tests/test_trapd_http_session.py   |  34 ------
 tests/test_trapd_runtime_pid.py    |  63 -----------
 tests/test_trapd_settings.py       |  81 --------------
 tests/test_trapd_vnf_table.py      | 107 ------------------
 19 files changed, 951 insertions(+), 481 deletions(-)
 create mode 100644 tests/hbproperties-test.yaml
 create mode 100644 tests/test-config.json
 create mode 100644 tests/test4.json
 create mode 100644 tests/test_db_monitoring.py
 create mode 100644 tests/test_htbt_exit.py
 create mode 100644 tests/test_htbt_get_cbs_config.py
 create mode 100644 tests/test_htbt_http_session.py
 create mode 100644 tests/test_htbt_runtime_pid.py
 create mode 100644 tests/test_htbt_settings.py
 create mode 100644 tests/test_htbt_vnf_table.py
 create mode 100644 tests/test_misshtbtd.py
 delete mode 100644 tests/test_trapd_exit.py
 delete mode 100644 tests/test_trapd_get_cbs_config.py
 delete mode 100644 tests/test_trapd_http_session.py
 delete mode 100644 tests/test_trapd_runtime_pid.py
 delete mode 100644 tests/test_trapd_settings.py
 delete mode 100644 tests/test_trapd_vnf_table.py

(limited to 'tests')

diff --git a/tests/hbproperties-test.yaml b/tests/hbproperties-test.yaml
new file mode 100644
index 0000000..1dc649c
--- /dev/null
+++ b/tests/hbproperties-test.yaml
@@ -0,0 +1,11 @@
+#Postgres database input
+#pg_ipAddress: 127.0.0.1
+pg_ipAddress: 10.0.0.0
+pg_portNum: "1234"
+pg_userName: postgres-test
+pg_passwd: postgres-test
+pg_dbName: postgres
+
+#Periodic polling of CBS config download
+CBS_polling_allowed: "True"
+CBS_polling_interval: "300"
diff --git a/tests/test-config.json b/tests/test-config.json
new file mode 100644
index 0000000..e853f6a
--- /dev/null
+++ b/tests/test-config.json
@@ -0,0 +1 @@
+{"pg_ipAddress": "10.0.4.1", "pg_userName": "postgres", "pg_dbName": "postgres", "streams_subscribes": {"ves-heartbeat": {"type": "message_router", "dmaap_info": {"topic_url": "http://10.12.5.252:3904/events/unauthenticated.SEC_HEARTBEAT_INPUT"}}}, "consumerID": "1", "CBS_polling_interval": "300", "pg_passwd": "postgres", "streams_publishes": {"dcae_cl_out": {"type": "message_router", "dmaap_info": {"topic_url": "http://10.12.5.252:3904/events/unauthenticated.DCAE_CL_OUTPUT"}}}, "pg_portNum": "5432", "CBS_polling_allowed": "True", "groupID": "groupID", "heartbeat_config": "{\"vnfs\": [{\"eventName\": \"Heartbeat_vDNS\",\"heartbeatcountmissed\": 3,\"heartbeatinterval\": 60,\"closedLoopControlName\": \"ControlLoopEvent1\",\t\"policyVersion\": \"1.0.0.5\",\t\"policyName\":\"vFireWall\",\"policyScope\": \"resource=sampleResource,type=sampletype,CLName=sampleCLName\",\"target_type\": \"VNF\",\t\"target\": \"genVnfName\",\t\"version\": \"1.0\"}, {\"eventName\": \"Heartbeat_vFW\",\"heartbeatcountmissed\": 3,\t\"heartbeatinterval\": 60,\"closedLoopControlName\": \"ControlLoopEvent1\",\"policyVersion\": \"1.0.0.5\",\"policyName\": \"vFireWall\",\"policyScope\": \"resource=sampleResource,type=sampletype,CLName=sampleCLName\",\t\"target_type\":\"VNF\",\t\"target\": \"genVnfName\",\t\"version\": \"1.0\"}, {\"eventName\": \"Heartbeat_xx\",\"heartbeatcountmissed\": 3,\t\"heartbeatinterval\": 60,\"closedLoopControlName\": \"ControlLoopEvent1\",\"policyVersion\": \"1.0.0.5\",\"policyName\": \"vFireWall\",\t\"policyScope\": \"resource=sampleResource,type=sampletype,CLName=sampleCLName\",\"target_type\": \"VNF\",\"target\": \"genVnfName\",\"version\": \"1.0\"}]}"}
diff --git a/tests/test4.json b/tests/test4.json
new file mode 100644
index 0000000..b223497
--- /dev/null
+++ b/tests/test4.json
@@ -0,0 +1 @@
+{"event":{"commonEventHeader":{"startEpochMicrosec":1621969018000,"sourceId":"79e90d76-513a-4f79-886d-470a0037c5cf","eventId":"Heartbeat_vDNS_10.0.0.1","nfcNamingCode":"DNS","reportingEntityId":"79e90d76-513a-4f79-886d-470a0037c5cf","internalHeaderFields":{"collectorTimeStamp":"Wed, 01 04 2023 01:36:35 GMT"},"eventType":"applicationVnf","priority":"Normal","version":3,"reportingEntityName":"VVVVVVcmd010","sequence":36312,"domain":"heartbeat","lastEpochMicrosec":1621969018000,"eventName":"Heartbeat_vDNS","sourceName":"zalp1bmdns01cmd010","nfNamingCode":"MDNS"}}}
diff --git a/tests/test_db_monitoring.py b/tests/test_db_monitoring.py
new file mode 100644
index 0000000..b9a644e
--- /dev/null
+++ b/tests/test_db_monitoring.py
@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+# ============LICENSE_START=======================================================
+# Copyright (c) 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 db_monitoring
+import htbtworker
+import logging
+import requests
+import tempfile
+import os
+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')
+    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")
+        self.assertEqual(status, True)
+        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")
+        self.assertEqual(status, True)
+        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')
+    def test_db_monitoring(self, mock1, mock2):
+        status = True
+        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")
+        self.assertEqual(status, True)
+        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")
+        self.assertEqual(status, True)
+
+    def test_db_monitoring_wrapper(self):
+        status = True
+        db_monitoring.db_monitoring_wrapper("111", htbtworker.configjsonfile, number_of_iterations=0)
+        self.assertEqual(status, True)
+
+if __name__ == "__main__": # pragma: no cover
+    unittest.main()
diff --git a/tests/test_get_logger.py b/tests/test_get_logger.py
index a4ceea5..1096271 100644
--- a/tests/test_get_logger.py
+++ b/tests/test_get_logger.py
@@ -1,5 +1,5 @@
 # ============LICENSE_START=======================================================
-# Copyright (c) 2020-2021 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2020-2023 AT&T Intellectual Property. All rights reserved.
 # Copyright (c) 2021 Fujitsu Ltd.
 # ================================================================================
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -24,6 +24,8 @@ log = logging.getLogger(__name__)
 
 
 def test_configure_logger():
+    # logpath = (os.path.dirname(__file__))+"hb_logs.txt"
+    # expected_log_path = Path(logpath)
     expected_log_path = Path("./hb_logs.txt")
     if expected_log_path.exists():
         os.remove(expected_log_path)
@@ -34,6 +36,8 @@ def test_configure_logger():
 
 
 def test_configure_logger_with_name():
+    # logpath = (os.path.dirname(__file__))+"hb_htbtworker_logs.txt"
+    # expected_log_path = Path(logpath)
     expected_log_path = Path("./hb_htbtworker_logs.txt")
     if expected_log_path.exists():
         os.remove(expected_log_path)
diff --git a/tests/test_htbt_exit.py b/tests/test_htbt_exit.py
new file mode 100644
index 0000000..f3698d5
--- /dev/null
+++ b/tests/test_htbt_exit.py
@@ -0,0 +1,50 @@
+# ============LICENSE_START=======================================================
+# Copyright (c) 2017-2023 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019 Pantheon.tech. 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.
+# 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 pytest
+import unittest
+from miss_htbt_service.mod import htbt_exit
+
+pid_file = "/tmp/test_pid_file"
+pid_file_dne = "/tmp/test_pid_file_NOT"
+
+
+class test_cleanup_and_exit(unittest.TestCase):
+    """
+    Test the cleanup_and_exit mod
+    """
+
+    def test_normal_exit(self):
+        """
+        Test normal exit works as expected
+        """
+        open(pid_file, "w")
+
+        with pytest.raises(SystemExit) as pytest_wrapped_sys_exit:
+            result = htbt_exit.cleanup_and_exit(0, pid_file)
+            assert pytest_wrapped_sys_exit.type == SystemExit
+            assert pytest_wrapped_sys_exit.value.code == 0
+
+    def test_abnormal_exit(self):
+        """
+        Test exit with missing PID file exits non-zero
+        """
+        with pytest.raises(SystemExit) as pytest_wrapped_sys_exit:
+            result = htbt_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_htbt_get_cbs_config.py b/tests/test_htbt_get_cbs_config.py
new file mode 100644
index 0000000..99e347c
--- /dev/null
+++ b/tests/test_htbt_get_cbs_config.py
@@ -0,0 +1,148 @@
+# ============LICENSE_START=======================================================
+# Copyright (c) 2017-2023 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019 Pantheon.tech. 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.
+# 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 pytest
+import unittest
+import os
+import cbs_polling as cp
+import time
+import misshtbtd
+from unittest.mock import *
+
+from miss_htbt_service.mod import htbt_get_cbs_config
+
+
+class test_get_cbs_config(unittest.TestCase):
+    """
+    Test the htbt_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": {'
+        '        "dcae_cl_out": {'
+        '            "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"'
+        "        }"
+        "    },"
+        '    "pg_ipAddress": "10.0.4.1",'
+        '    "pg_userName": "postgres",'
+        '    "pg_dbName": "postgres",'
+        '    "pg_passwd": "postgres",'
+        '    "pg_portNum": "5432"'
+        "}"
+    )
+
+    # create copy of snmphtbt.json for pytest
+    #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)
+
+    def test_cbs_env_present(self):
+        """
+        Test that CONSUL_HOST env variable exists but fails to
+        respond
+        """
+
+        with pytest.raises(Exception) as pytest_wrapped_sys_exit:
+            result = htbt_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")
+        result = True
+        print("result: %s" % result)
+        self.assertEqual(result, True)
+
+    @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)
+        # 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_http_session.py b/tests/test_htbt_http_session.py
new file mode 100644
index 0000000..ed3a33a
--- /dev/null
+++ b/tests/test_htbt_http_session.py
@@ -0,0 +1,34 @@
+# ============LICENSE_START=======================================================
+# Copyright (c) 2017-2023 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019 Pantheon.tech. 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.
+# 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 unittest
+from miss_htbt_service.mod import htbt_http_session
+
+
+class test_init_session_obj(unittest.TestCase):
+    """
+    Test the init_session_obj mod
+    """
+
+    def test_correct_usage(self):
+        """
+        Test that attempt to create http session object works
+        """
+        result = htbt_http_session.init_session_obj()
+        compare = str(result).startswith("<requests.sessions.Session object at")
+        self.assertEqual(compare, True)
diff --git a/tests/test_htbt_runtime_pid.py b/tests/test_htbt_runtime_pid.py
new file mode 100644
index 0000000..96379d9
--- /dev/null
+++ b/tests/test_htbt_runtime_pid.py
@@ -0,0 +1,63 @@
+# ============LICENSE_START=======================================================
+# Copyright (c) 2017-2023 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019 Pantheon.tech. 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.
+# 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 unittest
+from miss_htbt_service.mod import htbt_runtime_pid
+
+
+class test_save_pid(unittest.TestCase):
+    """
+    Test the save_pid mod
+    """
+
+    def test_correct_usage(self):
+        """
+        Test that attempt to create pid file in standard location works
+        """
+        result = htbt_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 = htbt_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
+    """
+
+    def test_correct_usage(self):
+        """
+        Test that attempt to remove pid file in standard location works
+        """
+        # must create it before removing it
+        result = htbt_runtime_pid.save_pid("/tmp/snmptrap_test_pid_file")
+        self.assertEqual(result, True)
+        result = htbt_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 = htbt_runtime_pid.rm_pid("/tmp/snmptrap_test_pid_file_9999")
+        self.assertEqual(result, False)
diff --git a/tests/test_htbt_settings.py b/tests/test_htbt_settings.py
new file mode 100644
index 0000000..3db86a1
--- /dev/null
+++ b/tests/test_htbt_settings.py
@@ -0,0 +1,81 @@
+# ============LICENSE_START=======================================================
+# Copyright (c) 2017-2023 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019 Pantheon.tech. 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.
+# 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 unittest
+
+from miss_htbt_service.mod import htbt_settings as tds
+
+pid_file = "/tmp/test_pid_file"
+pid_file_dne = "/tmp/test_pid_file_NOT"
+
+
+class test_cleanup_and_exit(unittest.TestCase):
+    """
+    Test for presense of required vars
+    """
+
+    def test_nonexistent_dict(self):
+        """
+        Test nosuch var
+        """
+        tds.init()
+        try:
+            tds.no_such_var
+            result = True
+        except Exception:
+            result = False
+
+        self.assertEqual(result, False)
+
+    def test_config_dict(self):
+        """
+        Test config dict
+        """
+        tds.init()
+        try:
+            tds.c_config
+            result = True
+        except Exception:
+            result = False
+        self.assertEqual(result, True)
+
+    def test_dns_cache_ip_to_name(self):
+        """
+        Test dns cache name dict
+        """
+
+        tds.init()
+        try:
+            tds.dns_cache_ip_to_name
+            result = True
+        except Exception:
+            result = False
+        self.assertEqual(result, True)
+
+    def test_dns_cache_ip_expires(self):
+        """
+        Test dns cache ip expires dict
+        """
+
+        tds.init()
+        try:
+            tds.dns_cache_ip_expires
+            result = True
+        except Exception:
+            result = False
+        self.assertEqual(result, True)
diff --git a/tests/test_htbt_vnf_table.py b/tests/test_htbt_vnf_table.py
new file mode 100644
index 0000000..4b6c3f1
--- /dev/null
+++ b/tests/test_htbt_vnf_table.py
@@ -0,0 +1,128 @@
+# ============LICENSE_START=======================================================
+# Copyright (c) 2017-2023 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019 Pantheon.tech. All rights reserved.
+# 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.
+# 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=========================================================
+#
+#  Author Prakask H (ph553f)
+"""
+test_htbt_vnf_table contains test cases related to DB Tables and cbs polling.
+"""
+import logging
+import unittest
+import os
+from unittest.mock import *
+from mod.htbt_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,
+)
+
+_logger = logging.getLogger(__name__)
+
+
+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()
+
+    @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)
+    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)
+    def test_validate_hb_common(self, mock, mock1):
+        result = verify_DB_creation_hb_common()
+        self.assertEqual(result, True)
+    
+    @patch('cbs_polling.poll_cbs')
+    def test_cbspolling(self, mock):
+        # Check if no exception thrown
+        verify_cbspolling()
+
+    @patch('misshtbtd.fetch_json_file')
+    def test_fetch_json_file(self, mock1):
+        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')
+    def test_misshtbtdmain(self, mock):
+        result = verify_misshtbtdmain()
+        _logger.info(result)
+        self.assertEqual(result, True)
+
+    def test_dbmon_startup(self):
+        result = verify_dbmon_startup()
+        _logger.info(result)
+        self.assertEqual(result, True)
+
+    @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"
+        mock1.return_value = configjsonfile
+        mock2.return_value = ("1234","RUNNING", "XYZ", 1234)
+        
+        result = verify_dbmonitoring()
+        _logger.info(result)
+        self.assertEqual(result, True)
+
+    def test_sendControlLoop_VNF_ONSET(self):
+        result = verify_sendControlLoop_VNF_ONSET()
+        _logger.info(result)
+        self.assertEqual(result, True)
+
+    def test_sendControlLoop_VM_ONSET(self):
+        result = verify_sendControlLoop_VM_ONSET()
+        _logger.info(result)
+        self.assertEqual(result, True)
+
+    def test_sendControlLoop_VNF_ABATED(self):
+        result = verify_sendControlLoop_VNF_ABATED()
+        _logger.info(result)
+        self.assertEqual(result, True)
+
+    def test_sendControlLoop_VM_ABATED(self):
+        result = verify_sendControlLoop_VM_ABATED()
+        _logger.info(result)
+        self.assertEqual(result, True)
diff --git a/tests/test_htbtworker.py b/tests/test_htbtworker.py
index 78c9087..ee03ddb 100644
--- a/tests/test_htbtworker.py
+++ b/tests/test_htbtworker.py
@@ -1,5 +1,5 @@
 # ============LICENSE_START=======================================================
-# Copyright (c) 2020-2022 AT&T Intellectual Property. All rights reserved.
+# 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.
@@ -14,37 +14,152 @@
 # limitations under the License.
 # ============LICENSE_END=========================================================
 
-from miss_htbt_service import htbtworker
 
+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):
 
-def run_test(i):
-    """
-    read_json_file() opens the file CWD/prefix/test{j}.json and returns the json value found there
-    """
-    j = i + 1
-    tdir = tempfile.TemporaryDirectory()
-    prefix = "../../../../../../../../../../../../.."
-    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")
-    cfg = htbtworker.read_json_file(i, prefix=pdir)
-    assert cfg["test"] == i
+    def setUp(self):
+        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):  
+        """
+        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)
+            }  
+         )             
+        mock3.return_value = [("Heartbeat_vDNS", "Heartbeat_vFW")]
+        mock1.return_value = mock_resp  
 
-def test_read_json_file_0():
-    run_test(0)
+        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)
+        self.assertEqual(srcname, "SOURCE_NAME1")
+        self.assertEqual(event_name, "Heartbeat_vDNS")
 
-def test_read_json_file_1():
-    run_test(1)
+        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")
 
 
-def test_read_json_file_2():
-    run_test(2)
+    @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_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):
+        """
+        Check to verify if new node entry is made for tracking HB
+        """
+        status = True
+        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):
+        """
+        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
+            }
+        )
+        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):
+        """
+        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,")
+        connection_db = db_mock
+        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):
+        """
+        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):
+        """
+        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")
+        self.assertEqual(False, flag)
+        
+    @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())):
+            flag = htbtworker.commit_and_close_db(connection_db)
+            self.assertEqual(False, flag)
+            
+     
+if __name__ == "__main__": # pragma: no cover
+    unittest.main()
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()
diff --git a/tests/test_trapd_exit.py b/tests/test_trapd_exit.py
deleted file mode 100644
index 8803b29..0000000
--- a/tests/test_trapd_exit.py
+++ /dev/null
@@ -1,50 +0,0 @@
-# ============LICENSE_START=======================================================
-# Copyright (c) 2017-2021 AT&T Intellectual Property. All rights reserved.
-# Copyright (c) 2019 Pantheon.tech. 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.
-# 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 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"
-
-
-class test_cleanup_and_exit(unittest.TestCase):
-    """
-    Test the cleanup_and_exit mod
-    """
-
-    def test_normal_exit(self):
-        """
-        Test normal exit works as expected
-        """
-        open(pid_file, "w")
-
-        with pytest.raises(SystemExit) as pytest_wrapped_sys_exit:
-            result = trapd_exit.cleanup_and_exit(0, pid_file)
-            assert pytest_wrapped_sys_exit.type == SystemExit
-            assert pytest_wrapped_sys_exit.value.code == 0
-
-    def test_abnormal_exit(self):
-        """
-        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)
-            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
deleted file mode 100644
index 75e2055..0000000
--- a/tests/test_trapd_get_cbs_config.py
+++ /dev/null
@@ -1,121 +0,0 @@
-# ============LICENSE_START=======================================================
-# Copyright (c) 2017-2022 AT&T Intellectual Property. All rights reserved.
-# Copyright (c) 2019 Pantheon.tech. 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.
-# 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 pytest
-import unittest
-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"'
-        "        }"
-        "    }"
-        "}"
-    )
-
-    # 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:
-        outfile.write(pytest_json_data)
-
-    def test_cbs_env_present(self):
-        """
-        Test that CONSUL_HOST env variable exists but fails to
-        respond
-        """
-
-        with pytest.raises(Exception) as pytest_wrapped_sys_exit:
-            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")
-        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
deleted file mode 100644
index 070fc93..0000000
--- a/tests/test_trapd_http_session.py
+++ /dev/null
@@ -1,34 +0,0 @@
-# ============LICENSE_START=======================================================
-# Copyright (c) 2017-2021 AT&T Intellectual Property. All rights reserved.
-# Copyright (c) 2019 Pantheon.tech. 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.
-# 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 unittest
-from miss_htbt_service.mod import trapd_http_session
-
-
-class test_init_session_obj(unittest.TestCase):
-    """
-    Test the init_session_obj mod
-    """
-
-    def test_correct_usage(self):
-        """
-        Test that attempt to create http session object works
-        """
-        result = trapd_http_session.init_session_obj()
-        compare = str(result).startswith("<requests.sessions.Session object at")
-        self.assertEqual(compare, True)
diff --git a/tests/test_trapd_runtime_pid.py b/tests/test_trapd_runtime_pid.py
deleted file mode 100644
index 47bc642..0000000
--- a/tests/test_trapd_runtime_pid.py
+++ /dev/null
@@ -1,63 +0,0 @@
-# ============LICENSE_START=======================================================
-# Copyright (c) 2017-2021 AT&T Intellectual Property. All rights reserved.
-# Copyright (c) 2019 Pantheon.tech. 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.
-# 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 unittest
-from miss_htbt_service.mod import trapd_runtime_pid
-
-
-class test_save_pid(unittest.TestCase):
-    """
-    Test the save_pid mod
-    """
-
-    def test_correct_usage(self):
-        """
-        Test that attempt to create pid file in standard location works
-        """
-        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")
-        self.assertEqual(result, False)
-
-
-class test_rm_pid(unittest.TestCase):
-    """
-    Test the rm_pid mod
-    """
-
-    def test_correct_usage(self):
-        """
-        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")
-        self.assertEqual(result, True)
-        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")
-        self.assertEqual(result, False)
diff --git a/tests/test_trapd_settings.py b/tests/test_trapd_settings.py
deleted file mode 100644
index 743fa68..0000000
--- a/tests/test_trapd_settings.py
+++ /dev/null
@@ -1,81 +0,0 @@
-# ============LICENSE_START=======================================================
-# Copyright (c) 2017-2022 AT&T Intellectual Property. All rights reserved.
-# Copyright (c) 2019 Pantheon.tech. 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.
-# 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 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"
-
-
-class test_cleanup_and_exit(unittest.TestCase):
-    """
-    Test for presense of required vars
-    """
-
-    def test_nonexistent_dict(self):
-        """
-        Test nosuch var
-        """
-        tds.init()
-        try:
-            tds.no_such_var
-            result = True
-        except Exception:
-            result = False
-
-        self.assertEqual(result, False)
-
-    def test_config_dict(self):
-        """
-        Test config dict
-        """
-        tds.init()
-        try:
-            tds.c_config
-            result = True
-        except Exception:
-            result = False
-        self.assertEqual(result, True)
-
-    def test_dns_cache_ip_to_name(self):
-        """
-        Test dns cache name dict
-        """
-
-        tds.init()
-        try:
-            tds.dns_cache_ip_to_name
-            result = True
-        except Exception:
-            result = False
-        self.assertEqual(result, True)
-
-    def test_dns_cache_ip_expires(self):
-        """
-        Test dns cache ip expires dict
-        """
-
-        tds.init()
-        try:
-            tds.dns_cache_ip_expires
-            result = True
-        except Exception:
-            result = False
-        self.assertEqual(result, True)
diff --git a/tests/test_trapd_vnf_table.py b/tests/test_trapd_vnf_table.py
deleted file mode 100644
index 2e0f22a..0000000
--- a/tests/test_trapd_vnf_table.py
+++ /dev/null
@@ -1,107 +0,0 @@
-# ============LICENSE_START=======================================================
-# Copyright (c) 2017-2022 AT&T Intellectual Property. All rights reserved.
-# Copyright (c) 2019 Pantheon.tech. All rights reserved.
-# 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.
-# 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=========================================================
-#
-#  Author Prakask H (ph553f)
-"""
-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,
-)
-
-_logger = logging.getLogger(__name__)
-
-
-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)
-        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)
-        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)
-        self.assertEqual(result, True)
-
-    def test_cbspolling(self):
-        # Check if no exception thrown
-        verify_cbspolling()
-
-    def test_fetch_json_file(self):
-        result = verify_fetch_json_file()
-        _logger.info(result)
-        self.assertEqual(result, True)
-
-    def test_misshtbtdmain(self):
-        result = verify_misshtbtdmain()
-        _logger.info(result)
-        self.assertEqual(result, True)
-
-    def test_dbmon_startup(self):
-        result = verify_dbmon_startup()
-        _logger.info(result)
-        self.assertEqual(result, True)
-
-    def test_dbmonitoring(self):
-        result = verify_dbmonitoring()
-        _logger.info(result)
-        self.assertEqual(result, True)
-
-    def test_sendControlLoop_VNF_ONSET(self):
-        result = verify_sendControlLoop_VNF_ONSET()
-        _logger.info(result)
-        self.assertEqual(result, True)
-
-    def test_sendControlLoop_VM_ONSET(self):
-        result = verify_sendControlLoop_VM_ONSET()
-        _logger.info(result)
-        self.assertEqual(result, True)
-
-    def test_sendControlLoop_VNF_ABATED(self):
-        result = verify_sendControlLoop_VNF_ABATED()
-        _logger.info(result)
-        self.assertEqual(result, True)
-
-    def test_sendControlLoop_VM_ABATED(self):
-        result = verify_sendControlLoop_VM_ABATED()
-        _logger.info(result)
-        self.assertEqual(result, True)
-- 
cgit 1.2.3-korg