From abb283c5a3b72eb2d91e6adb4eda68184f040f35 Mon Sep 17 00:00:00 2001 From: "Ladue, David (dl3158)" Date: Wed, 14 Mar 2018 14:01:36 -0400 Subject: copyright and controller ENV changes Change-Id: Ic12aa439a03f19c7a06a536368a8d2a1f26855aa Issue-ID: DCAEGEN2-271 Signed-off-by: Ladue, David (dl3158) --- tests/_test_trapd_get_cbs_config.py | 44 ++++++++++++++++++++++ tests/setup.py | 68 ++++++++++++++++++++++++++++++++++ tests/test_snmptrapd.py | 46 +++++++++++++++++++++++ tests/test_snmptrapd_send_test_trap.py | 40 ++++++++++++++++++++ tests/test_trapd_exit.py | 37 ++++++++++++++++++ tests/test_trapd_http_session.py | 20 ++++++++++ tests/test_trapd_runtime_pid.py | 47 +++++++++++++++++++++++ tests/tox.ini | 15 ++++++++ 8 files changed, 317 insertions(+) create mode 100644 tests/_test_trapd_get_cbs_config.py create mode 100644 tests/setup.py create mode 100644 tests/test_snmptrapd.py create mode 100755 tests/test_snmptrapd_send_test_trap.py create mode 100644 tests/test_trapd_exit.py create mode 100644 tests/test_trapd_http_session.py create mode 100644 tests/test_trapd_runtime_pid.py create mode 100644 tests/tox.ini (limited to 'tests') 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("