aboutsummaryrefslogtreecommitdiffstats
path: root/tests/test_snmptrapd.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_snmptrapd.py')
-rw-r--r--tests/test_snmptrapd.py73
1 files changed, 71 insertions, 2 deletions
diff --git a/tests/test_snmptrapd.py b/tests/test_snmptrapd.py
index a361bfc..d02b4f7 100644
--- a/tests/test_snmptrapd.py
+++ b/tests/test_snmptrapd.py
@@ -8,7 +8,6 @@ import trapd_settings as tds
import trapd_http_session
import trapd_runtime_pid
import trapd_io
-import trapd_logging
import trapd_get_cbs_config
class test_snmptrapd(unittest.TestCase):
@@ -52,6 +51,23 @@ class test_snmptrapd(unittest.TestCase):
result = snmptrapd.load_all_configs(0, 1)
self.assertEqual(result, True)
+ def test_load_all_configs_signal(self):
+ """
+ Test load of all configs via runtime signal
+ """
+
+ # init vars
+ tds.init()
+
+ # request load of CBS data
+ os.environ.update(CBS_SIM_JSON='/tmp/opt/app/snmptrap/etc/snmptrapd.json')
+ result = trapd_get_cbs_config.get_cbs_config()
+ self.assertEqual(result, True)
+
+ # request load of CBS data
+ result = snmptrapd.load_all_configs(1, 1)
+ self.assertEqual(result, True)
+
def test_log_all_arriving_traps(self):
"""
Test logging of traps
@@ -85,12 +101,65 @@ class test_snmptrapd(unittest.TestCase):
tds.json_traps_fd = trapd_io.open_file(tds.json_traps_filename)
msg = ("published traps logged to: %s" % tds.json_traps_filename)
trapd_io.stdout_logger(msg)
- trapd_logging.ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
+ trapd_io.ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
# don't open files, but try to log - should raise exception
with pytest.raises(Exception) as pytest_wrapped_exception:
result = snmptrapd.log_all_arriving_traps()
assert pytest_wrapped_exception.type == AttributeError
+ def test_log_all_incorrect_log_type(self):
+ """
+ Test logging of traps
+ """
+
+ # init vars
+ tds.init()
+
+ # request load of CBS data
+ os.environ.update(CBS_SIM_JSON='/tmp/opt/app/snmptrap/etc/snmptrapd.json')
+ trapd_get_cbs_config.get_cbs_config()
+
+ # open eelf logs
+ trapd_io.open_eelf_logs()
+
+ def test_post_dmaap(self):
+ """
+ Test publish of traps
+ """
+
+ # init vars
+ tds.init()
+
+ # request load of CBS data
+ os.environ.update(CBS_SIM_JSON='/tmp/opt/app/snmptrap/etc/snmptrapd.json')
+ result = trapd_get_cbs_config.get_cbs_config()
+
+ # set last day to current
+ tds.last_day = datetime.datetime.now().day
+
+ # trap dict for publish
+ tds.trap_dict = {'uuid': '06f6e91c-3236-11e8-9953-005056865aac', 'agent address': '1.2.3.4', 'agent name': 'test-agent.nodomain.com', 'cambria.partition': 'test-agent.nodomain.com', 'community': '', 'community len': 0, 'epoch_serno': 15222068260000, 'protocol version': 'v2c', 'time received': 1522206826.2938566, 'trap category': 'ONAP-COLLECTOR-SNMPTRAP', 'sysUptime': '218567736', 'notify OID': '1.3.6.1.4.1.9999.9.9.999', 'notify OID len': 10}
+
+ # open eelf logs
+ trapd_io.open_eelf_logs()
+
+ # open trap logs
+ tds.arriving_traps_filename = tds.c_config['files.runtime_base_dir'] + "/" + \
+ tds.c_config['files.log_dir'] + "/" + \
+ (tds.c_config['files.arriving_traps_log'])
+ tds.arriving_traps_fd = trapd_io.open_file(tds.arriving_traps_filename)
+
+ # name and open json trap log
+ tds.json_traps_filename = tds.c_config['files.runtime_base_dir'] + "/" + tds.c_config['files.log_dir'] + "/" + "DMAAP_" + (
+ tds.c_config['streams_publishes']['sec_fault_unsecure']['dmaap_info']['topic_url'].split('/')[-1]) + ".json"
+ tds.json_traps_fd = trapd_io.open_file(tds.json_traps_filename)
+ msg = ("published traps logged to: %s" % tds.json_traps_filename)
+ trapd_io.stdout_logger(msg)
+ trapd_io.ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
+
+ result = snmptrapd.post_dmaap()
+ print("result from post_dmaap -> %s" % result)
+
if __name__ == '__main__':
unittest.main()