8 files changed, 317 insertions, 0 deletions
diff --git a/tests/_test_trapd_get_cbs_config.py b/tests/_test_trapd_get_cbs_config.py
new file mode 100644
index 0000000..5fcfc2a
--- /dev/null
+++ b/tests/_test_trapd_get_cbs_config.py
@@ -0,0 +1,44 @@
+import pytest
+import unittest
+import os
+from onap_dcae_cbs_docker_client.client import get_config
+from trapd_exit import cleanup_and_exit
+from trapd_logging import stdout_logger
+import trapd_get_cbs_config
+ 
+class test_get_cbs_config(unittest.TestCase):
+    """
+    Test the trapd_get_cbs_config mod
+    """
+ 
+    def test_cbs_env_present(self):
+        """
+        Test that CBS env variable exists and we can get config even
+        if CONSUL_HOST doesn't provide
+        """
+        os.environ.update(CONSUL_HOST='nosuchhost')
+        result = trapd_get_cbs_config.trapd_get_cbs_config()
+        compare = str(result).startswith("{'snmptrap': ")
+        self.assertEqual(compare, False)
+ 
+    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_SIM_JSON='../etc/snmptrapd.json')
+        result = trapd_get_cbs_config.trapd_get_cbs_config()
+        compare = str(result).startswith("{'snmptrap': ")
+        self.assertEqual(compare, False)
+ 
+    def test_cbs_fallback_env_not_present(self):
+        """
+        Test that CBS fallback env variable does not exists fails
+        """
+        os.environ.update(CBS_SIM_JSON='../etc/no_such_file.json')
+        result = trapd_get_cbs_config.trapd_get_cbs_config()
+        compare = str(result).startswith("{'snmptrap': ")
+        self.assertEqual(compare, False)
+ 
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/setup.py b/tests/setup.py
new file mode 100644
index 0000000..7ff184c
--- /dev/null
+++ b/tests/setup.py
@@ -0,0 +1,68 @@
+# org.onap.dcae
+# ================================================================================
+# Copyright (c) 2017 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=========================================================
+#
+# ECOMP is a trademark and service mark of AT&T Intellectual Property.
+
+import argparse
+import array
+import asyncio
+import collections
+import datetime
+import errno
+from pysnmp.carrier.asyncio.dgram import udp, udp6
+from pysnmp.entity import engine, config
+from pysnmp.entity.rfc3413 import ntfrcv
+from pysnmp.proto.api import v2c
+import json
+import logging
+import logging.handlers
+import os
+import pprint
+import re
+import requests
+import signal
+import socket
+import string
+import sys
+import time
+import traceback
+from trapd_dmaap_config import read_dmaap_config
+from trapd_exit import cleanup_and_exit
+from trapd_http_session import init_session_obj
+from trapd_perm_status import log_to_perm_status
+from trapd_runtime_pid import save_pid, rm_pid
+from trapd_trap_config import read_trap_config
+from trapd_yaml_config import read_yaml_config
+import unicodedata
+import uuid as uuid_mod
+import yaml
+
+install_reqs = parse_requirements("requirements.txt", session=PipSession())
+reqs = [str(ir.req) for ir in install_reqs]
+
+setup(
+    name = "onap_dcae_cbs_docker_client",
+    description = "snmp trap receiver for a DCAE docker image",
+    version = "1.2",
+    packages=find_packages(),
+    author = "Dave L",
+    author_email = "dl3158@att.com",
+    license='Apache 2',
+    keywords = "",
+    url = "",
+    install_requires=reqs
+)
diff --git a/tests/test_snmptrapd.py b/tests/test_snmptrapd.py
new file mode 100644
index 0000000..2f1783c
--- /dev/null
+++ b/tests/test_snmptrapd.py
@@ -0,0 +1,46 @@
+import pytest
+import unittest
+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')
+        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)
+ 
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/test_snmptrapd_send_test_trap.py b/tests/test_snmptrapd_send_test_trap.py
new file mode 100755
index 0000000..54d522e
--- /dev/null
+++ b/tests/test_snmptrapd_send_test_trap.py
@@ -0,0 +1,40 @@
+from pysnmp.hlapi import *
+from pysnmp import debug
+
+# debug.setLogger(debug.Debug('msgproc'))
+
+iters = range(0, 10, 1)
+for i in iters:
+    errorIndication, errorStatus, errorIndex, varbinds = next(sendNotification(SnmpEngine(),
+         CommunityData('not_public'),
+         UdpTransportTarget(('localhost', 6164)),
+         ContextData(),
+         'trap',
+         [ObjectType(ObjectIdentity('.1.3.6.1.4.1.999.1'), OctetString('test trap - ignore')),
+          ObjectType(ObjectIdentity('.1.3.6.1.4.1.999.2'), OctetString('ONAP pytest trap'))])
+    )
+    
+    if errorIndication:
+        print(errorIndication)
+    else:
+        print("successfully sent first trap example, number %d" % i)
+
+for i in iters:
+    errorIndication, errorStatus, errorIndex, varbinds = next(sendNotification(SnmpEngine(),
+         CommunityData('public'),
+         UdpTransportTarget(('localhost', 6164)),
+         ContextData(),
+         'trap',
+            NotificationType(
+                ObjectIdentity('.1.3.6.1.4.1.74.2.46.12.1.1')
+            ).addVarBinds(
+                ('.1.3.6.1.4.1.999.1', OctetString('ONAP pytest trap - ignore (varbind 1)')),
+                ('.1.3.6.1.4.1.999.2', OctetString('ONAP pytest trap - ignore (varbind 2)'))
+            )
+        )
+    )
+
+    if errorIndication:
+        print(errorIndication)
+    else:
+        print("successfully sent second trap example, number %d" % i)
diff --git a/tests/test_trapd_exit.py b/tests/test_trapd_exit.py
new file mode 100644
index 0000000..594624f
--- /dev/null
+++ b/tests/test_trapd_exit.py
@@ -0,0 +1,37 @@
+import pytest
+import unittest
+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
+
+        # compare = str(result).startswith("SystemExit: 0")
+        # self.assertEqual(compare, True)
+ 
+    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
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/test_trapd_http_session.py b/tests/test_trapd_http_session.py
new file mode 100644
index 0000000..8f61d08
--- /dev/null
+++ b/tests/test_trapd_http_session.py
@@ -0,0 +1,20 @@
+import pytest
+import unittest
+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)
+ 
+ 
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/test_trapd_runtime_pid.py b/tests/test_trapd_runtime_pid.py
new file mode 100644
index 0000000..c48ef34
--- /dev/null
+++ b/tests/test_trapd_runtime_pid.py
@@ -0,0 +1,47 @@
+import pytest
+import unittest
+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)
+ 
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/tox.ini b/tests/tox.ini
new file mode 100644
index 0000000..3d8e842
--- /dev/null
+++ b/tests/tox.ini
@@ -0,0 +1,15 @@
+[tox]
+envlist = py36
+ 
+[testenv]
+deps = coverage
+commands = coverage erase
+ 
+[testenv:py36]
+deps = coverage pytest
+commands = coverage run ../bin/snmptrapd.py &
+           pytest test_trapd_exit.py  
+           pytest test_trapd_http_session.py  
+           pytest test_trapd_runtime_pid.py
+           ./test_snmptrapd_send_test_trap.py &
+           coverage report -m