run the black formatting tool on python code
Change-Id: I41badbf7ea2b2cd243e9f2acd5c9f2d3ec907964
Issue-ID: DCAEGEN2-2995
Signed-off-by: Hansen, Tony (th1395) <th1395@att.com>
diff --git a/Changelog.md b/Changelog.md
index a6126b5..78b3a8a 100644
--- a/Changelog.md
+++ b/Changelog.md
@@ -5,7 +5,9 @@
and this project adheres to [Semantic Versioning](http://semver.org/).
## [2.0.6] - 2021/10/26
-* Changed [DCAEGEN2-2957] SNMP Trap collector - STDOUT complaince
+### Changed
+* [DCAEGEN2-2957] SNMP Trap collector - STDOUT complaince
+* [DCAEGEN2-2995] run the black formatting tool on python code
## [2.0.5] - 2021/07/19
* Changed to use version 2.2.1 of pypi onap_dcae_cbs_docker_client
diff --git a/setup.py b/setup.py
index b1fb070..8716437 100644
--- a/setup.py
+++ b/setup.py
@@ -1,7 +1,6 @@
-# org.onap.dcae
-# ================================================================================
+# ============LICENSE_START=======================================================
# Copyright (c) 2017-2021 AT&T Intellectual Property. All rights reserved.
-# Copyright 2021 Deutsche Telekom. All rights reserved.
+# Copyright (c) 2021 Deutsche Telekom. 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.
@@ -15,8 +14,6 @@
# 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 os
import string
@@ -26,19 +23,14 @@
setup(
- name = "snmptrap",
- description = "snmp trap receiver for ONAP docker image",
- version = "2.0.6",
+ name="snmptrap",
+ description="snmp trap receiver for ONAP docker image",
+ version="2.0.6",
packages=find_packages(),
- install_requires=[
- "pysnmp==4.4.12",
- "requests==2.18.3",
- "onap_dcae_cbs_docker_client==2.2.1",
- "pyyaml"
- ],
- author = "Dave L",
- author_email = "dl3158@att.com",
- license='Apache 2',
- keywords = "",
- url = ""
+ install_requires=["pysnmp==4.4.12", "requests==2.18.3", "onap_dcae_cbs_docker_client==2.2.1", "pyyaml"],
+ author="Dave L",
+ author_email="dl3158@att.com",
+ license="Apache 2",
+ keywords="",
+ url="",
)
diff --git a/snmptrap/mod/trapd_exit.py b/snmptrap/mod/trapd_exit.py
index 6d1ea45..1d1ea16 100644
--- a/snmptrap/mod/trapd_exit.py
+++ b/snmptrap/mod/trapd_exit.py
@@ -1,5 +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.
@@ -18,7 +18,7 @@
file, and exiting with the provided (param 1) exit code
"""
-__docformat__ = 'restructuredtext'
+__docformat__ = "restructuredtext"
import sys
import os
diff --git a/snmptrap/mod/trapd_get_cbs_config.py b/snmptrap/mod/trapd_get_cbs_config.py
index 272aabe..cfae994 100644
--- a/snmptrap/mod/trapd_get_cbs_config.py
+++ b/snmptrap/mod/trapd_get_cbs_config.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2018-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2018-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.
@@ -19,7 +19,7 @@
testing purposes)
"""
-__docformat__ = 'restructuredtext'
+__docformat__ = "restructuredtext"
import json
import os
@@ -76,17 +76,14 @@
stdout_logger(msg)
cleanup_and_exit(1, None)
else:
- msg = ("ONAP controller override specified via CBS_SIM_JSON: %s" %
- _cbs_sim_json_file)
+ msg = "ONAP controller override specified via CBS_SIM_JSON: %s" % _cbs_sim_json_file
stdout_logger(msg)
try:
tds.c_config = json.load(open(_cbs_sim_json_file))
- msg = ("%s loaded and parsed successfully" %
- _cbs_sim_json_file)
+ msg = "%s loaded and parsed successfully" % _cbs_sim_json_file
stdout_logger(msg)
except Exception as e:
- msg = "Unable to load CBS_SIM_JSON " + _cbs_sim_json_file + \
- " (invalid json?) - FATAL ERROR, exiting"
+ msg = "Unable to load CBS_SIM_JSON " + _cbs_sim_json_file + " (invalid json?) - FATAL ERROR, exiting"
stdout_logger(msg)
cleanup_and_exit(1, None)
@@ -96,24 +93,24 @@
# recalc timeout, set default if not present
try:
- tds.timeout_seconds = float(tds.c_config['publisher']['http_milliseconds_timeout'] / 1000.0)
+ tds.timeout_seconds = float(tds.c_config["publisher"]["http_milliseconds_timeout"] / 1000.0)
except Exception as e:
tds.timeout_seconds = float(1.5)
# recalc seconds_between_retries, set default if not present
try:
- tds.seconds_between_retries = float(tds.c_config['publisher']['http_milliseconds_between_retries'] / 1000.0)
+ tds.seconds_between_retries = float(tds.c_config["publisher"]["http_milliseconds_between_retries"] / 1000.0)
except Exception as e:
- tds.seconds_between_retries = float(.750)
+ tds.seconds_between_retries = float(0.750)
# recalc min_severity_to_log, set default if not present
try:
- tds.minimum_severity_to_log = int(tds.c_config['files']['minimum_severity_to_log'])
+ tds.minimum_severity_to_log = int(tds.c_config["files"]["minimum_severity_to_log"])
except Exception as e:
tds.minimum_severity_to_log = int(3)
try:
- tds.publisher_retries = int(tds.c_config['publisher']['http_retries'])
+ tds.publisher_retries = int(tds.c_config["publisher"]["http_retries"])
except Exception as e:
tds.publisher_retries = int(2)
diff --git a/snmptrap/mod/trapd_http_session.py b/snmptrap/mod/trapd_http_session.py
index 14abb21..7118d00 100644
--- a/snmptrap/mod/trapd_http_session.py
+++ b/snmptrap/mod/trapd_http_session.py
@@ -1,5 +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.
@@ -18,7 +18,7 @@
messages to the dmaap cluster.
"""
-__docformat__ = 'restructuredtext'
+__docformat__ = "restructuredtext"
import os
import requests
@@ -81,8 +81,7 @@
return True
except Exception as e:
msg = "Unable to close current http session - FATAL ERROR, exiting"
- ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_FATAL,
- tds.CODE_GENERAL, msg)
+ ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_FATAL, tds.CODE_GENERAL, msg)
stdout_logger(msg)
cleanup_and_exit(1, tds.pid_file_name)
diff --git a/snmptrap/mod/trapd_io.py b/snmptrap/mod/trapd_io.py
index 20b99f9..f7c48b7 100644
--- a/snmptrap/mod/trapd_io.py
+++ b/snmptrap/mod/trapd_io.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2018-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2018-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.
@@ -16,7 +16,7 @@
"""
"""
-__docformat__ = 'restructuredtext'
+__docformat__ = "restructuredtext"
# basics
import datetime
@@ -54,8 +54,14 @@
# NOTE: this will go away when onap logging is standardized/available
try:
# open various ecomp logs - if any fails, exit
- for fd in [tds.eelf_error_fd, tds.eelf_debug_fd, tds.eelf_audit_fd,
- tds.eelf_metrics_fd, tds.arriving_traps_fd, tds.json_traps_fd]:
+ for fd in [
+ tds.eelf_error_fd,
+ tds.eelf_debug_fd,
+ tds.eelf_audit_fd,
+ tds.eelf_metrics_fd,
+ tds.arriving_traps_fd,
+ tds.json_traps_fd,
+ ]:
fd.close()
roll_file(tds.eelf_error_file_name)
@@ -80,8 +86,7 @@
try:
tds.json_traps_fd = open_file(tds.json_traps_filename)
except Exception as e:
- msg = ("Error opening json_log %s : %s" %
- (json_traps_filename, str(e)))
+ msg = "Error opening json_log %s : %s" % (json_traps_filename, str(e))
stdout_logger(msg)
cleanup_and_exit(1, tds.pid_file_name)
@@ -91,8 +96,7 @@
try:
tds.arriving_traps_fd = open_file(tds.arriving_traps_filename)
except Exception as e:
- msg = ("Error opening arriving traps %s : %s" %
- (arriving_traps_filename, str(e)))
+ msg = "Error opening arriving traps %s : %s" % (arriving_traps_filename, str(e))
stdout_logger(msg)
cleanup_and_exit(1, tds.pid_file_name)
@@ -111,8 +115,7 @@
try:
# open various ecomp logs - if any fails, exit
- tds.eelf_error_file_name = (
- tds.c_config['files']['eelf_base_dir'] + "/" + tds.c_config['files']['eelf_error'])
+ tds.eelf_error_file_name = tds.c_config["files"]["eelf_base_dir"] + "/" + tds.c_config["files"]["eelf_error"]
tds.eelf_error_fd = open_file(tds.eelf_error_file_name)
except Exception as e:
@@ -121,8 +124,7 @@
cleanup_and_exit(1, tds.pid_file_name)
try:
- tds.eelf_debug_file_name = (
- tds.c_config['files']['eelf_base_dir'] + "/" + tds.c_config['files']['eelf_debug'])
+ tds.eelf_debug_file_name = tds.c_config["files"]["eelf_base_dir"] + "/" + tds.c_config["files"]["eelf_debug"]
tds.eelf_debug_fd = open_file(tds.eelf_debug_file_name)
except Exception as e:
@@ -131,8 +133,7 @@
cleanup_and_exit(1, tds.pid_file_name)
try:
- tds.eelf_audit_file_name = (
- tds.c_config['files']['eelf_base_dir'] + "/" + tds.c_config['files']['eelf_audit'])
+ tds.eelf_audit_file_name = tds.c_config["files"]["eelf_base_dir"] + "/" + tds.c_config["files"]["eelf_audit"]
tds.eelf_audit_fd = open_file(tds.eelf_audit_file_name)
except Exception as e:
msg = "Error opening eelf audit log : " + str(e)
@@ -141,7 +142,8 @@
try:
tds.eelf_metrics_file_name = (
- tds.c_config['files']['eelf_base_dir'] + "/" + tds.c_config['files']['eelf_metrics'])
+ tds.c_config["files"]["eelf_base_dir"] + "/" + tds.c_config["files"]["eelf_metrics"]
+ )
tds.eelf_metrics_fd = open_file(tds.eelf_metrics_file_name)
except Exception as e:
msg = "Error opening eelf metric log : " + str(e)
@@ -150,6 +152,7 @@
return True
+
# # # # # # # # # # # # # # # # # # #
# fx: roll_log_file -> move provided filename to timestamped version
# # # # # # # # # # ## # # # # # # #
@@ -160,11 +163,11 @@
move active file to timestamped archive
"""
- _file_name_suffix = "%s" % (datetime.datetime.fromtimestamp(time.time()).
- fromtimestamp(time.time()).
- strftime('%Y-%m-%dT%H:%M:%S'))
+ _file_name_suffix = "%s" % (
+ datetime.datetime.fromtimestamp(time.time()).fromtimestamp(time.time()).strftime("%Y-%m-%dT%H:%M:%S")
+ )
- _loc_file_name_bak = _loc_file_name + '.' + _file_name_suffix
+ _loc_file_name_bak = _loc_file_name + "." + _file_name_suffix
# roll existing file if present
if os.path.isfile(_loc_file_name):
@@ -172,15 +175,13 @@
os.rename(_loc_file_name, _loc_file_name_bak)
return True
except Exception as e:
- _msg = ("ERROR: Unable to rename %s to %s"
- % (_loc_file_name,
- _loc_file_name_bak))
- ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_ERROR,
- tds.CODE_GENERAL, _msg)
+ _msg = "ERROR: Unable to rename %s to %s" % (_loc_file_name, _loc_file_name_bak)
+ ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_ERROR, tds.CODE_GENERAL, _msg)
return False
return False
+
# # # # # # # # # # # # #
# fx: open_log_file
# # # # # # # # # # # # #
@@ -194,17 +195,16 @@
try:
# open append mode just in case so nothing is lost, but should be
# non-existent file
- _loc_fd = open(_loc_file_name, 'a', 1)
+ _loc_fd = open(_loc_file_name, "a", 1)
return _loc_fd
except Exception as e:
msg = "Error opening " + _loc_file_name + " append mode - " + str(e)
stdout_logger(msg)
cleanup_and_exit(1, tds.pid_file_name)
-
-# # # # # # # # # # # # #
-# fx: close_file
-# # # # # # # # # # # # #
+ # # # # # # # # # # # # #
+ # fx: close_file
+ # # # # # # # # # # # # #
"""
close _loc_file_name, return True with success, False otherwise
"""
@@ -216,11 +216,11 @@
_loc_fd.close()
return True
except Exception as e:
- msg = "Error closing %s : %s - results indeterminate" % (
- _loc_filename, str(e))
+ msg = "Error closing %s : %s - results indeterminate" % (_loc_filename, str(e))
ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_FATAL, tds.CODE_GENERAL, msg)
return False
+
# # # # # # # # # # # # # # # # # # #
# fx: ecomp_logger -> log in eelf format until standard
# is released for python via LOG-161
@@ -317,13 +317,22 @@
# catch invalid log type
if _log_type < 1 or _log_type > 5:
- msg = ("INVALID log type: %s " % _log_type)
- _out_rec = ("%s|%s|%s|%s|%s|%s|%s|%s|%s"
- % (calling_fx, "snmptrapd", unused, unused, unused, tds.SEV_TYPES[_sev], _error_code, unused, (msg + _msg)))
+ msg = "INVALID log type: %s " % _log_type
+ _out_rec = "%s|%s|%s|%s|%s|%s|%s|%s|%s" % (
+ calling_fx,
+ "snmptrapd",
+ unused,
+ unused,
+ unused,
+ tds.SEV_TYPES[_sev],
+ _error_code,
+ unused,
+ (msg + _msg),
+ )
try:
- tds.eelf_error_fd.write('%s|%s\n' % (t_out, str(_out_rec)))
+ tds.eelf_error_fd.write("%s|%s\n" % (t_out, str(_out_rec)))
if log_to_stdout:
- print('%s|%s' % (t_out, str(_out_rec)))
+ print("%s|%s" % (t_out, str(_out_rec)))
except Exception as e:
stdout_logger(str(_out_rec))
@@ -335,34 +344,79 @@
# _out_rec = ('%s|%s|%s|%s|%s|%s|%s|%s|%s'
# _out_rec = ('%s|%s|%s|%s|%s|%s|%s|%s|%s'
# % (calling_fx, "snmptrapd", unused, unused, unused, tds.SEV_TYPES[_sev], _error_code, unused, _msg))
- _out_rec = ('%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s'
- % (unused, unused, calling_fx, unused, "snmptrapd", unused, unused, unused, unused, unused, unused, unused, tds.SEV_TYPES[_sev], unused, unused, unused, unused, unused, unused, unused, unused, unused, unused, unused, unused, unused, _msg))
+ _out_rec = "%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s" % (
+ unused,
+ unused,
+ calling_fx,
+ unused,
+ "snmptrapd",
+ unused,
+ unused,
+ unused,
+ unused,
+ unused,
+ unused,
+ unused,
+ tds.SEV_TYPES[_sev],
+ unused,
+ unused,
+ unused,
+ unused,
+ unused,
+ unused,
+ unused,
+ unused,
+ unused,
+ unused,
+ unused,
+ unused,
+ unused,
+ _msg,
+ )
try:
- tds.eelf_error_fd.write('%s|%s|%s\n' % (t_out, t_out, str(_out_rec)))
+ tds.eelf_error_fd.write("%s|%s|%s\n" % (t_out, t_out, str(_out_rec)))
if log_to_stdout:
- print('%s|%s|%s' % (t_out, t_out, str(_out_rec)))
+ print("%s|%s|%s" % (t_out, t_out, str(_out_rec)))
except Exception as e:
stdout_logger(str(_out_rec))
elif _log_type == tds.LOG_TYPE_AUDIT:
# log message in AUDIT format
# _out_rec = ('%s|%s|%s|%s|%s|%s|%s|%s|%s'
# % (calling_fx, "snmptrapd", unused, unused, unused, tds.SEV_TYPES[_sev], _error_code, unused, _msg))
- _out_rec = ('%s|%s|%s|%s|%s|%s|%s|%s|%s'
- % (calling_fx, "snmptrapd", unused, unused, unused, tds.SEV_TYPES[_sev], _error_code, unused, _msg))
+ _out_rec = "%s|%s|%s|%s|%s|%s|%s|%s|%s" % (
+ calling_fx,
+ "snmptrapd",
+ unused,
+ unused,
+ unused,
+ tds.SEV_TYPES[_sev],
+ _error_code,
+ unused,
+ _msg,
+ )
try:
- tds.eelf_audit_fd.write('%s|%s\n' % (t_out, str(_out_rec)))
+ tds.eelf_audit_fd.write("%s|%s\n" % (t_out, str(_out_rec)))
if log_to_stdout:
- print('%s|%s' % (t_out, str(_out_rec)))
+ print("%s|%s" % (t_out, str(_out_rec)))
except Exception as e:
stdout_logger(str(_out_rec))
elif _log_type == tds.LOG_TYPE_METRICS:
# log message in METRICS format
- _out_rec = ('%s|%s|%s|%s|%s|%s|%s|%s|%s'
- % (calling_fx, "snmptrapd", unused, unused, unused, tds.SEV_TYPES[_sev], _error_code, unused, _msg))
+ _out_rec = "%s|%s|%s|%s|%s|%s|%s|%s|%s" % (
+ calling_fx,
+ "snmptrapd",
+ unused,
+ unused,
+ unused,
+ tds.SEV_TYPES[_sev],
+ _error_code,
+ unused,
+ _msg,
+ )
try:
- tds.eelf_metrics_fd.write('%s|%s\n' % (t_out, str(_out_rec)))
+ tds.eelf_metrics_fd.write("%s|%s\n" % (t_out, str(_out_rec)))
if log_to_stdout:
- print('%s|%s' % (t_out, str(_out_rec)))
+ print("%s|%s" % (t_out, str(_out_rec)))
except Exception as e:
stdout_logger(str(_out_rec))
@@ -370,17 +424,28 @@
# DLFM: too much I/O !!!
# always write to debug; we need ONE logfile that has time-sequence full view !!!
# log message in DEBUG format
- _out_rec = ("%s|%s|%s|%s|%s|%s|%s|%s|%s|%s"
- % (unused, calling_fx, "snmptrapd", unused, unused, unused, tds.SEV_TYPES[_sev], _error_code, unused, _msg))
+ _out_rec = "%s|%s|%s|%s|%s|%s|%s|%s|%s|%s" % (
+ unused,
+ calling_fx,
+ "snmptrapd",
+ unused,
+ unused,
+ unused,
+ tds.SEV_TYPES[_sev],
+ _error_code,
+ unused,
+ _msg,
+ )
try:
- tds.eelf_debug_fd.write('%s|%s\n' % (t_out, str(_out_rec)))
+ tds.eelf_debug_fd.write("%s|%s\n" % (t_out, str(_out_rec)))
if log_to_stdout:
- print('%s|%s' % (t_out, str(_out_rec)))
+ print("%s|%s" % (t_out, str(_out_rec)))
except Exception as e:
stdout_logger(str(_out_rec))
return True
+
# # # # # # # # # # # # #
# fx: stdout_logger
# # # # # # # # # # # # #
@@ -403,4 +468,4 @@
t_out = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S,%f")[:-3]
- print('%s %s' % (t_out, _msg))
+ print("%s %s" % (t_out, _msg))
diff --git a/snmptrap/mod/trapd_runtime_pid.py b/snmptrap/mod/trapd_runtime_pid.py
index 74668f5..c647885 100644
--- a/snmptrap/mod/trapd_runtime_pid.py
+++ b/snmptrap/mod/trapd_runtime_pid.py
@@ -1,5 +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.
@@ -18,7 +18,7 @@
PID of currently running trap receiver)
"""
-__docformat__ = 'restructuredtext'
+__docformat__ = "restructuredtext"
import logging
import os
@@ -47,8 +47,8 @@
"""
try:
- pid_fd = open(_pid_file_name, 'w')
- pid_fd.write('%d' % os.getpid())
+ pid_fd = open(_pid_file_name, "w")
+ pid_fd.write("%d" % os.getpid())
pid_fd.close()
except IOError:
print("IOError saving PID file %s :" % _pid_file_name)
diff --git a/snmptrap/mod/trapd_settings.py b/snmptrap/mod/trapd_settings.py
index 41dc18f..26e7c9b 100644
--- a/snmptrap/mod/trapd_settings.py
+++ b/snmptrap/mod/trapd_settings.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2018-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2018-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.
@@ -16,7 +16,7 @@
"""
"""
-__docformat__ = 'restructuredtext'
+__docformat__ = "restructuredtext"
def init():
diff --git a/snmptrap/mod/trapd_snmpv3.py b/snmptrap/mod/trapd_snmpv3.py
index b421ae1..56ecb5c 100644
--- a/snmptrap/mod/trapd_snmpv3.py
+++ b/snmptrap/mod/trapd_snmpv3.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2018-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2018-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.
@@ -20,7 +20,7 @@
"""
-__docformat__ = 'restructuredtext'
+__docformat__ = "restructuredtext"
import json
import os
@@ -60,7 +60,7 @@
try:
v3_users = _cbs_config["snmpv3_config"]["usm_users"]
except Exception as e:
- msg = ("No V3 users defined")
+ msg = "No V3 users defined"
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
return _py_config, _snmp_engine
@@ -68,13 +68,13 @@
# engineId
try:
- ctx_engine_id = v3_user['engineId']
+ ctx_engine_id = v3_user["engineId"]
except Exception as e:
ctx_engine_id = None
# user
try:
- userName = v3_user['user']
+ userName = v3_user["user"]
except Exception as e:
userName = None
@@ -86,40 +86,38 @@
# usmHMACMD5AuthProtocol
try:
- authKey = v3_user['usmHMACMD5AuthProtocol']
+ authKey = v3_user["usmHMACMD5AuthProtocol"]
authProtocol = config.usmHMACMD5AuthProtocol
except Exception as e:
try:
- authKey = v3_user['usmHMACSHAAuthProtocol']
+ authKey = v3_user["usmHMACSHAAuthProtocol"]
authProtocol = config.usmHMACSHAAuthProtocol
except Exception as e:
try:
- authKey = v3_user['usmHMAC128SHA224AuthProtocol']
+ authKey = v3_user["usmHMAC128SHA224AuthProtocol"]
authProtocol = config.usmHMAC128SHA224AuthProtocol
except Exception as e:
try:
- authKey = v3_user['usmHMAC192SHA256AuthProtocol']
+ authKey = v3_user["usmHMAC192SHA256AuthProtocol"]
authProtocol = config.usmHMAC192SHA256AuthProtocol
except Exception as e:
try:
- authKey = v3_user['usmHMAC256SHA384AuthProtocol']
+ authKey = v3_user["usmHMAC256SHA384AuthProtocol"]
authProtocol = config.usmHMAC256SHA384AuthProtocol
except Exception as e:
try:
- authKey = v3_user['usmHMAC384SHA512AuthProtocol']
+ authKey = v3_user["usmHMAC384SHA512AuthProtocol"]
authProtocol = config.usmHMAC384SHA512AuthProtocol
except Exception as e:
try:
- authKey = v3_user['usmNoAuthProtocol']
+ authKey = v3_user["usmNoAuthProtocol"]
authProtocol = config.usmNoAuthProtocol
except Exception as e:
# FMDL: default to NoAuth, or error/skip entry?
- msg = (
- "No auth specified for user %s ?" % (userName))
+ msg = "No auth specified for user %s ?" % (userName)
authKey = None
authProtocol = config.usmNoAuthProtocol
- ecomp_logger(
- tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
# privacy
# find options at -> site-packages/pysnmp/entity/config.py
@@ -129,56 +127,57 @@
# usm3DESEDEPriv
try:
- privKey = v3_user['usm3DESEDEPrivProtocol']
+ privKey = v3_user["usm3DESEDEPrivProtocol"]
privProtocol = config.usm3DESEDEPrivProtocol
except Exception as e:
# usmAesCfb128Protocol
try:
- privKey = v3_user['usmAesCfb128Protocol']
+ privKey = v3_user["usmAesCfb128Protocol"]
privProtocol = config.usmAesCfb128Protocol
except Exception as e:
# usmAesCfb192Protocol
try:
- privKey = v3_user['usmAesCfb192Protocol']
+ privKey = v3_user["usmAesCfb192Protocol"]
privProtocol = config.usmAesCfb192Protocol
except Exception as e:
# usmAesBlumenthalCfb192Protocol
try:
- privKey = v3_user['usmAesBlumenthalCfb192Protocol']
+ privKey = v3_user["usmAesBlumenthalCfb192Protocol"]
privProtocol = config.usmAesBlumenthalCfb192Protocol
except Exception as e:
# usmAesCfb256Protocol
try:
- privKey = v3_user['usmAesCfb256Protocol']
+ privKey = v3_user["usmAesCfb256Protocol"]
privProtocol = config.usmAesCfb256Protocol
except Exception as e:
# usmAesBlumenthalCfb256Protocol
try:
- privKey = v3_user['usmAesBlumenthalCfb256Protocol']
+ privKey = v3_user["usmAesBlumenthalCfb256Protocol"]
privProtocol = config.usmAesBlumenthalCfb256Protocol
except Exception as e:
# usmDESPrivProtocol
try:
- privKey = v3_user['usmDESPrivProtocol']
+ privKey = v3_user["usmDESPrivProtocol"]
privProtocol = config.usmDESPrivProtocol
except Exception as e:
# usmNoPrivProtocol
try:
- privKey = v3_user['usmNoPrivProtocol']
+ privKey = v3_user["usmNoPrivProtocol"]
privProtocol = config.usmNoPrivProtocol
except Exception as e:
# FMDL: default to NoPriv, or error/skip entry?
- msg = (
- "No priv specified for user %s" % (userName))
- ecomp_logger(
- tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
+ msg = "No priv specified for user %s" % (userName)
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
privKey = None
privProtocol = config.usmNoPrivProtocol
# break
# msg = ("userName: %s authKey: %s authProtocol: %s privKey: %s privProtocol: %s engineId: %s % (userName, authKey, authProtocol, privKey, privProtocol, ctx_engine_id))
- msg = ("userName: %s authKey: **** authProtocol: %s privKey: **** privProtocol: %s engineId: ****" %
- (userName, authProtocol, privProtocol))
+ msg = "userName: %s authKey: **** authProtocol: %s privKey: **** privProtocol: %s engineId: ****" % (
+ userName,
+ authProtocol,
+ privProtocol,
+ )
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
# user: usr-md5-des, auth: MD5, priv DES, contextEngineId: 8000000001020304
@@ -189,16 +188,15 @@
if ctx_engine_id is not None:
config.addV3User(
- _snmp_engine, userName,
- authProtocol, authKey,
- privProtocol, privKey,
- contextEngineId=v2c.OctetString(hexValue=ctx_engine_id)
+ _snmp_engine,
+ userName,
+ authProtocol,
+ authKey,
+ privProtocol,
+ privKey,
+ contextEngineId=v2c.OctetString(hexValue=ctx_engine_id),
)
else:
- config.addV3User(
- _snmp_engine, userName,
- authProtocol, authKey,
- privProtocol, privKey
- )
+ config.addV3User(_snmp_engine, userName, authProtocol, authKey, privProtocol, privKey)
return _py_config, _snmp_engine
diff --git a/snmptrap/mod/trapd_stats_settings.py b/snmptrap/mod/trapd_stats_settings.py
index df4da0e..2aacb2e 100644
--- a/snmptrap/mod/trapd_stats_settings.py
+++ b/snmptrap/mod/trapd_stats_settings.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.
@@ -16,7 +16,7 @@
"""
"""
-__docformat__ = 'restructuredtext'
+__docformat__ = "restructuredtext"
def init():
diff --git a/snmptrap/mod/trapd_stormwatch.py b/snmptrap/mod/trapd_stormwatch.py
index 9e41bd0..4c374bb 100644
--- a/snmptrap/mod/trapd_stormwatch.py
+++ b/snmptrap/mod/trapd_stormwatch.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.
@@ -20,7 +20,7 @@
so the trap can be logged and immediately discarded
"""
-__docformat__ = 'restructuredtext'
+__docformat__ = "restructuredtext"
import sys
import os
@@ -98,11 +98,11 @@
sws.sw_config_category.clear()
return True
except Exception as e:
- msg = "unable to reset stormwatch dictionaries - results will be indeterminate: %s" % (
- e)
+ msg = "unable to reset stormwatch dictionaries - results will be indeterminate: %s" % (e)
ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_WARN, tds.CODE_GENERAL, msg)
return False
+
# # # # # # # # # # # # #
# fx: sw_load_trap_config
# - load trap configurations from CBS response
@@ -121,10 +121,10 @@
try:
sws.sw_storm_active_dict
ret = sw_clear_dicts()
- msg = ("reset existing sws dictionaries to empty")
+ msg = "reset existing sws dictionaries to empty"
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
except NameError:
- msg = ("sws dictionaries not present - initializing")
+ msg = "sws dictionaries not present - initializing"
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
ret = sw_init()
@@ -134,24 +134,22 @@
# get metric % threshold for logging trap count by-agent to metric log
try:
stats.metric_log_notification_threshold_pct = int(
- _config["trap_config"]["metric_log_notification_threshold_pct"])
- msg = ("metric_log_notification_threshold_pct value: %d" %
- stats.metric_log_notification_threshold_pct)
+ _config["trap_config"]["metric_log_notification_threshold_pct"]
+ )
+ msg = "metric_log_notification_threshold_pct value: %d" % stats.metric_log_notification_threshold_pct
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
except Exception as e:
- msg = (
- "metric_log_notification_threshold_pct not present in config - default to 25")
+ msg = "metric_log_notification_threshold_pct not present in config - default to 25"
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_WARN, tds.CODE_GENERAL, msg)
stats.metric_log_notification_threshold_pct = 25
# get stormwatch interval; default to 60 seconds
try:
- sws.sw_interval_in_seconds = int(
- _config["trap_config"]["sw_interval_in_seconds"])
- msg = ("sw_interval_in_seconds value: %d" % sws.sw_interval_in_seconds)
+ sws.sw_interval_in_seconds = int(_config["trap_config"]["sw_interval_in_seconds"])
+ msg = "sw_interval_in_seconds value: %d" % sws.sw_interval_in_seconds
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
except Exception as e:
- msg = ("sw_interval_in_seconds not present in config - default to 60 seconds")
+ msg = "sw_interval_in_seconds not present in config - default to 60 seconds"
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_WARN, tds.CODE_GENERAL, msg)
sws.sw_interval_in_seconds = 60
@@ -159,7 +157,7 @@
try:
notify_oids = _config["trap_config"]["notify_oids"]
except Exception as e:
- msg = ("no trap_config or notify_oids defined")
+ msg = "no trap_config or notify_oids defined"
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_WARN, tds.CODE_GENERAL, msg)
return False
@@ -167,65 +165,66 @@
for trap_block in notify_oids:
# oid
try:
- _oid = trap_block['oid']
+ _oid = trap_block["oid"]
except Exception as e:
- msg = (
- "missing oid value in notify_oids - oid section of CBS config - using empty value, disregard entry")
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_WARN,
- tds.CODE_GENERAL, msg)
+ msg = "missing oid value in notify_oids - oid section of CBS config - using empty value, disregard entry"
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_WARN, tds.CODE_GENERAL, msg)
_oid = None
# sw_high_water_in_interval
try:
- _sw_high_water_in_interval = int(
- trap_block['sw_high_water_in_interval'])
+ _sw_high_water_in_interval = int(trap_block["sw_high_water_in_interval"])
except Exception as e:
msg = (
- "missing sw_high_water_in_interval value in notify_oids - oid section of CBS config - using empty value")
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_WARN,
- tds.CODE_GENERAL, msg)
+ "missing sw_high_water_in_interval value in notify_oids - oid section of CBS config - using empty value"
+ )
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_WARN, tds.CODE_GENERAL, msg)
_sw_high_water_in_interval = None
# sw_low_water_in_interval
try:
- _sw_low_water_in_interval = int(
- trap_block['sw_low_water_in_interval'])
+ _sw_low_water_in_interval = int(trap_block["sw_low_water_in_interval"])
except Exception as e:
msg = (
- "missing sw_low_water_in_interval value in notify_oids - oid section of CBS config - using empty value")
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_WARN,
- tds.CODE_GENERAL, msg)
+ "missing sw_low_water_in_interval value in notify_oids - oid section of CBS config - using empty value"
+ )
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_WARN, tds.CODE_GENERAL, msg)
_sw_low_water_in_interval = None
# category
try:
- _category = trap_block['category']
+ _category = trap_block["category"]
except Exception as e:
- msg = (
- "missing category value in notify_oids - oid section of CBS config - using empty value")
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_WARN,
- tds.CODE_GENERAL, msg)
+ msg = "missing category value in notify_oids - oid section of CBS config - using empty value"
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_WARN, tds.CODE_GENERAL, msg)
_category = None
- if (_oid is not None and _category is not None and
- _sw_low_water_in_interval is not None and _sw_high_water_in_interval is not None and
- _sw_low_water_in_interval < _sw_high_water_in_interval):
+ if (
+ _oid is not None
+ and _category is not None
+ and _sw_low_water_in_interval is not None
+ and _sw_high_water_in_interval is not None
+ and _sw_low_water_in_interval < _sw_high_water_in_interval
+ ):
# FMDL: Do we actually need sw_config_oid_dict?
- msg = ("oid: %s sw_high_water_in_interval: %d sw_low_water_in_interval: %d category: %s" % (
- _oid, _sw_high_water_in_interval, _sw_low_water_in_interval, _category))
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ msg = "oid: %s sw_high_water_in_interval: %d sw_low_water_in_interval: %d category: %s" % (
+ _oid,
+ _sw_high_water_in_interval,
+ _sw_low_water_in_interval,
+ _category,
+ )
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
sws.sw_config_oid_dict[_oid] = True
sws.sw_config_low_water_in_interval_dict[_oid] = _sw_low_water_in_interval
sws.sw_config_high_water_in_interval_dict[_oid] = _sw_high_water_in_interval
sws.sw_config_category[_oid] = _category
trap_block_counter += 1
else:
- msg = ("Missing or incorrect value for stormwatch config entry %d: skipping: %s" % (
- trap_block_counter, trap_block))
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
-
+ msg = "Missing or incorrect value for stormwatch config entry %d: skipping: %s" % (
+ trap_block_counter,
+ trap_block,
+ )
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
return trap_block_counter
@@ -246,17 +245,18 @@
:Variables:
"""
- msg = "total notifications: %d, interval in seconds: %d" % (
- stats.total_notifications, sws.sw_interval_in_seconds)
+ msg = "total notifications: %d, interval in seconds: %d" % (stats.total_notifications, sws.sw_interval_in_seconds)
ecomp_logger(tds.LOG_TYPE_METRICS, tds.SEV_INFO, tds.CODE_GENERAL, msg)
# print metrics for total traps and traps-per-second avg
# during sample interval
avg_traps_per_second = stats.total_notifications / 60
msg = "total traps: %d, interval in seconds: %d, average traps-per-second: %d" % (
- stats.total_notifications, sws.sw_interval_in_seconds, avg_traps_per_second)
- ecomp_logger(tds.LOG_TYPE_METRICS, tds.SEV_WARN,
- tds.CODE_GENERAL, msg)
+ stats.total_notifications,
+ sws.sw_interval_in_seconds,
+ avg_traps_per_second,
+ )
+ ecomp_logger(tds.LOG_TYPE_METRICS, tds.SEV_WARN, tds.CODE_GENERAL, msg)
# print metrics for any agent that represents more than stats.metric_log_notification_threshold_pct
# during sample interval
@@ -265,9 +265,13 @@
p = c / stats.total_notifications * 100
if p > stats.metric_log_notification_threshold_pct:
msg = "agent: %s, notifications: %d, interval in seconds: %d, percent of total traps: %d" % (
- k, c, sws.sw_interval_in_seconds, p)
- ecomp_logger(tds.LOG_TYPE_METRICS, tds.SEV_WARN,
- tds.CODE_GENERAL, msg)
+ k,
+ c,
+ sws.sw_interval_in_seconds,
+ p,
+ )
+ ecomp_logger(tds.LOG_TYPE_METRICS, tds.SEV_WARN, tds.CODE_GENERAL, msg)
+
# # # # # # # # # # # # #
# fx: stats_increment_counters
@@ -339,8 +343,10 @@
# if we are at or above stormwatch interval, re-eval and re-set
elapsed_time = int(time.time()) - sws.sw_last_stormwatch_dict_analysis
if elapsed_time >= sws.sw_interval_in_seconds:
- msg = "%d seconds has elapsed since stormwatch dictionary eval (%d second threshold) - check and reset counters " % (
- elapsed_time, sws.sw_interval_in_seconds)
+ msg = (
+ "%d seconds has elapsed since stormwatch dictionary eval (%d second threshold) - check and reset counters "
+ % (elapsed_time, sws.sw_interval_in_seconds)
+ )
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
sw_log_metrics()
sw_reset_counter_dict()
@@ -349,8 +355,7 @@
# that means it's participating in stormwatch, otherwise bail out
try:
_high_water_val = sws.sw_config_high_water_in_interval_dict[_loc_oid]
- msg = "%s present in stormwatch config - high water value: %d" % (
- _loc_oid, _high_water_val)
+ msg = "%s present in stormwatch config - high water value: %d" % (_loc_oid, _high_water_val)
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
except Exception as e:
return False
@@ -375,25 +380,32 @@
# if we got this far, trap is in stormwatch configs, we've incremented
# counter in sw_storm_counter_dict - figure out if we are over limit
if sws.sw_storm_counter_dict[_dict_key] > _high_water_val:
- print(f"sws.sw_storm_counter_dict[{_dict_key}]({sws.sw_storm_counter_dict[_dict_key]}) > _high_water_val ({_high_water_val})")
+ print(
+ f"sws.sw_storm_counter_dict[{_dict_key}]({sws.sw_storm_counter_dict[_dict_key]}) > _high_water_val ({_high_water_val})"
+ )
_loc_agent = _dict_key.split()[0]
_loc_oid = _dict_key.split()[1]
msg = "STORM ACTIVE: received %d events (%s) from %s (greater than high water threshold: %d)" % (
- sws.sw_storm_counter_dict[_dict_key], _loc_oid, _loc_agent, _high_water_val)
+ sws.sw_storm_counter_dict[_dict_key],
+ _loc_oid,
+ _loc_agent,
+ _high_water_val,
+ )
ecomp_logger(tds.LOG_TYPE_AUDIT, tds.SEV_WARN, tds.CODE_GENERAL, msg)
try:
sws.sw_storm_active_dict[_dict_key] = True
except Exception as e:
- msg = "ERROR setting %s in storm active state: %s " % (
- _dict_key, e)
- ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_ERROR,
- tds.CODE_GENERAL, msg)
+ msg = "ERROR setting %s in storm active state: %s " % (_dict_key, e)
+ ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_ERROR, tds.CODE_GENERAL, msg)
return True
else:
- print(f"NOT sws.sw_storm_counter_dict[{_dict_key}]({sws.sw_storm_counter_dict[_dict_key]}) > _high_water_val ({_high_water_val})")
+ print(
+ f"NOT sws.sw_storm_counter_dict[{_dict_key}]({sws.sw_storm_counter_dict[_dict_key]}) > _high_water_val ({_high_water_val})"
+ )
return False
+
# # # # # # # # # # # # #
# fx: sw_reset_counter_dict
# - reset counter dictionary on <interval> boundaries
@@ -416,11 +428,9 @@
# publish stats to MR...
try:
msg = "publish counts-by-oid from stats.oid_counter_dict"
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
msg = "publish count-by-agent from stats.agent_counter_dict"
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
except Exception as e:
msg = "unable to publish counts by oid and agent to MR: " % (e)
ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_WARN, tds.CODE_GENERAL, msg)
@@ -431,8 +441,7 @@
stats.agent_counter_dict.clear()
stats.total_notifications = 0
except Exception as e:
- msg = "unable to reset counts by oid and agent dictionaries - stats will be INNACURATE: " % (
- e)
+ msg = "unable to reset counts by oid and agent dictionaries - stats will be INNACURATE: " % (e)
ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_WARN, tds.CODE_GENERAL, msg)
# </stats>
@@ -457,36 +466,45 @@
if sws.sw_storm_counter_dict[k] >= _high_water_val:
msg = "%s remaining in storm state, received %d events (GE to upper threshold: %d)" % (
- k, sws.sw_storm_counter_dict[k], _high_water_val)
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ k,
+ sws.sw_storm_counter_dict[k],
+ _high_water_val,
+ )
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
sws.sw_storm_counter_dict[k] = 0
else:
_low_water_val = sws.sw_config_low_water_in_interval_dict[_loc_oid]
if sws.sw_storm_counter_dict[k] < _low_water_val:
try:
msg = "STORM OVER: received %d events (%s) from %s (less than low water threshold: %d)" % (
- sws.sw_storm_counter_dict[k], _loc_oid, _loc_agent, _low_water_val)
- ecomp_logger(tds.LOG_TYPE_AUDIT, tds.SEV_WARN,
- tds.CODE_GENERAL, msg)
+ sws.sw_storm_counter_dict[k],
+ _loc_oid,
+ _loc_agent,
+ _low_water_val,
+ )
+ ecomp_logger(tds.LOG_TYPE_AUDIT, tds.SEV_WARN, tds.CODE_GENERAL, msg)
del sws.sw_storm_active_dict[k]
sws.sw_storm_counter_dict[k] = 0
except Exception as e:
- msg = "unable to remove %s from storm active dictionary - TRAPS MAY BE DISCARDED UNINTENTIONALLY! Reason: %s " % (
- k, e)
- ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_ERROR,
- tds.CODE_GENERAL, msg)
+ msg = (
+ "unable to remove %s from storm active dictionary - TRAPS MAY BE DISCARDED UNINTENTIONALLY! Reason: %s "
+ % (k, e)
+ )
+ ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_ERROR, tds.CODE_GENERAL, msg)
else:
msg = "%s remaining in storm state, received %d events (GE to lower threshold: %d)" % (
- k, sws.sw_storm_counter_dict[k], _low_water_val)
- ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ k,
+ sws.sw_storm_counter_dict[k],
+ _low_water_val,
+ )
+ ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_INFO, tds.CODE_GENERAL, msg)
sws.sw_storm_counter_dict[k] = 0
sws.sw_last_stormwatch_dict_analysis = int(time.time())
return True
+
# # # # # # # # # # # # #
# fx: sw_increment_counter
# - increment OID and agent trap counters
@@ -511,14 +529,11 @@
try:
sws.sw_storm_counter_dict[_dict_key] += 1
- msg = "stormwatch counter for %s now: %d" % (
- _dict_key, sws.sw_storm_counter_dict[_dict_key])
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ msg = "stormwatch counter for %s now: %d" % (_dict_key, sws.sw_storm_counter_dict[_dict_key])
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
return True
except Exception as E:
msg = "first trap for %s - init stormwatch counter to 1" % (_dict_key)
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
sws.sw_storm_counter_dict[_dict_key] = 1
return True
diff --git a/snmptrap/mod/trapd_stormwatch_settings.py b/snmptrap/mod/trapd_stormwatch_settings.py
index f2586f9..a4bc410 100644
--- a/snmptrap/mod/trapd_stormwatch_settings.py
+++ b/snmptrap/mod/trapd_stormwatch_settings.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.
@@ -16,7 +16,7 @@
"""
"""
-__docformat__ = 'restructuredtext'
+__docformat__ = "restructuredtext"
def init():
diff --git a/snmptrap/mod/trapd_vb_types.py b/snmptrap/mod/trapd_vb_types.py
index 98d5d2c..0b73e68 100644
--- a/snmptrap/mod/trapd_vb_types.py
+++ b/snmptrap/mod/trapd_vb_types.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2018-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2018-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.
@@ -20,7 +20,7 @@
"""
-__docformat__ = 'restructuredtext'
+__docformat__ = "restructuredtext"
import json
import os
@@ -42,18 +42,18 @@
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
_pysnmp_to_netsnmp_vb_type = {
- 'Integer32': 'integer',
- 'Integer': 'integer',
- 'Gauge32': 'unsigned',
- 'Counter32': 'counter32',
- 'OctetString': 'octet',
- 'py_type_5': 'hex',
- 'py_type_6': 'decimal',
- 'Null': 'null',
- 'ObjectIdentifier': 'oid',
- 'TimeTicks': 'timeticks',
- 'IpAddress': 'ipaddress',
- 'Bits': 'bits'
+ "Integer32": "integer",
+ "Integer": "integer",
+ "Gauge32": "unsigned",
+ "Counter32": "counter32",
+ "OctetString": "octet",
+ "py_type_5": "hex",
+ "py_type_6": "decimal",
+ "Null": "null",
+ "ObjectIdentifier": "oid",
+ "TimeTicks": "timeticks",
+ "IpAddress": "ipaddress",
+ "Bits": "bits",
}
default_vb_type = "octet"
@@ -77,7 +77,6 @@
return _netsnmp_vb_type
except Exception as e:
# if not found, return original pysnmp type
- msg = ("%s not configured as pysnmp varbind type - returning %s"
- % (_pysnmp_vb_type,default_vb_type))
+ msg = "%s not configured as pysnmp varbind type - returning %s" % (_pysnmp_vb_type, default_vb_type)
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
return default_vb_type
diff --git a/snmptrap/snmptrapd.py b/snmptrap/snmptrapd.py
index 5f46f34..ddb6b32 100644
--- a/snmptrap/snmptrapd.py
+++ b/snmptrap/snmptrapd.py
@@ -1,5 +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.
@@ -28,7 +28,7 @@
onap dcae snmp trap publish dmaap
"""
-__docformat__ = 'restructuredtext'
+__docformat__ = "restructuredtext"
# basics
import argparse
@@ -57,6 +57,7 @@
# pysnmp
from pysnmp.entity import engine, config
from pysnmp.carrier.asyncio.dgram import udp, udp6
+
# from pysnmp.carrier.asyncore.dgram import udp
from pysnmp.entity.rfc3413 import ntfrcv
from pysnmp.proto.api import v2c
@@ -68,12 +69,10 @@
from trapd_runtime_pid import save_pid, rm_pid
from trapd_get_cbs_config import get_cbs_config
from trapd_exit import cleanup_and_exit
-from trapd_http_session import init_session_obj, close_session_obj,\
- reset_session_obj
+from trapd_http_session import init_session_obj, close_session_obj, reset_session_obj
from trapd_snmpv3 import load_snmpv3_credentials
from trapd_vb_types import pysnmp_to_netsnmp_varbind_convert
-from trapd_io import roll_all_logs, open_eelf_logs, roll_file, open_file,\
- close_file, ecomp_logger, stdout_logger
+from trapd_io import roll_all_logs, open_eelf_logs, roll_file, open_file, close_file, ecomp_logger, stdout_logger
import trapd_stormwatch_settings as sws
import trapd_stormwatch as stormwatch
@@ -99,8 +98,8 @@
usage args
"""
- print('Incorrect usage invoked. Correct usage:')
- print(' %s [-v]' % prog_name)
+ print("Incorrect usage invoked. Correct usage:")
+ print(" %s [-v]" % prog_name)
cleanup_and_exit(1, "undefined")
@@ -127,10 +126,8 @@
"""
if int(_signum) != 0:
- msg = ("received signal %s at frame %s; re-reading configs"
- % (_signum, _frame))
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ msg = "received signal %s at frame %s; re-reading configs" % (_signum, _frame)
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
# re-request config from broker:
if not get_cbs_config():
@@ -138,19 +135,14 @@
stdout_logger(msg)
cleanup_and_exit(1, tds.pid_file_name)
else:
- current_runtime_config_file_name = (
- tds.c_config['files']['runtime_base_dir'] +
- "/tmp/current_config.json")
+ current_runtime_config_file_name = tds.c_config["files"]["runtime_base_dir"] + "/tmp/current_config.json"
if int(_signum) != 0:
- msg = "updated config logged to : %s" % \
- current_runtime_config_file_name
+ msg = "updated config logged to : %s" % current_runtime_config_file_name
else:
- msg = "current config logged to : %s" % \
- current_runtime_config_file_name
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL,
- msg)
+ msg = "current config logged to : %s" % current_runtime_config_file_name
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
- with open(current_runtime_config_file_name, 'w') as outfile:
+ with open(current_runtime_config_file_name, "w") as outfile:
json.dump(tds.c_config, outfile)
# reset http session based on latest config
@@ -158,10 +150,8 @@
# reload sw participating entries, reset counter dictionary
traps_configured = stormwatch.sw_load_trap_config(tds.c_config)
- msg = "encountered %d trap configurations in CBS/json config" % \
- traps_configured
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL,
- msg)
+ msg = "encountered %d trap configurations in CBS/json config" % traps_configured
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
tds.last_minute = datetime.datetime.now().minute
tds.last_hour = datetime.datetime.now().hour
@@ -178,34 +168,33 @@
try:
if int(tds.dns_cache_ip_expires[_loc_ip_addr_str] < int(time.time())):
- raise Exception('cache expired for %s at %d - updating value' %
- (_loc_ip_addr_str,
- (tds.dns_cache_ip_expires[_loc_ip_addr_str])))
+ raise Exception(
+ "cache expired for %s at %d - updating value"
+ % (_loc_ip_addr_str, (tds.dns_cache_ip_expires[_loc_ip_addr_str]))
+ )
else:
agent_fqdn = tds.dns_cache_ip_to_name[_loc_ip_addr_str]
except Exception as e:
- msg = "dns cache expired or missing for %s - refreshing" % \
- _loc_ip_addr_str
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ msg = "dns cache expired or missing for %s - refreshing" % _loc_ip_addr_str
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
try:
- agent_fqdn, alias, addresslist = socket.gethostbyaddr(
- _loc_ip_addr_str)
+ agent_fqdn, alias, addresslist = socket.gethostbyaddr(_loc_ip_addr_str)
except Exception as e:
agent_fqdn = _loc_ip_addr_str
tds.dns_cache_ip_to_name[_loc_ip_addr_str] = agent_fqdn
- tds.dns_cache_ip_expires[_loc_ip_addr_str] = (
- time.time() + int(tds.c_config['cache']['dns_cache_ttl_seconds']))
- msg = "cache for %s (%s) updated - set to expire at %d" % \
- (agent_fqdn, _loc_ip_addr_str,
- tds.dns_cache_ip_expires[_loc_ip_addr_str])
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ tds.dns_cache_ip_expires[_loc_ip_addr_str] = time.time() + int(tds.c_config["cache"]["dns_cache_ttl_seconds"])
+ msg = "cache for %s (%s) updated - set to expire at %d" % (
+ agent_fqdn,
+ _loc_ip_addr_str,
+ tds.dns_cache_ip_expires[_loc_ip_addr_str],
+ )
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
return agent_fqdn
+
# # # # # # # # # # # # #
# fx: log_all_arriving_traps
# # # # # # # # # # # # #
@@ -214,12 +203,12 @@
def log_all_arriving_traps():
# roll logs as needed/defined in files.roll_frequency
- if tds.c_config['files']['roll_frequency'] == "minute":
+ if tds.c_config["files"]["roll_frequency"] == "minute":
curr_minute = datetime.datetime.now().minute
if curr_minute != tds.last_minute:
roll_all_logs()
tds.last_minute = curr_minute
- elif tds.c_config['files']['roll_frequency'] == "hour":
+ elif tds.c_config["files"]["roll_frequency"] == "hour":
curr_hour = datetime.datetime.now().hour
if curr_hour != tds.last_hour:
roll_all_logs()
@@ -233,28 +222,35 @@
# always log arriving trap
try:
- time_now=int(round(time.time(),0))
- arrived_epoch_time_int=int(tds.trap_dict["time received"])
- tds.arriving_traps_fd.write('%d %s; %s %s %s %s %s %s %s %s %s %s %s %s %s\n' %
- (time_now,
- datetime.datetime.fromtimestamp(time_now).strftime("%a %b %d %H:%M:%S %Y"),
- tds.trap_dict["time received"],
- datetime.datetime.fromtimestamp(arrived_epoch_time_int).strftime("%a %b %d %H:%M:%S %Y"),
- tds.trap_dict["trap category"],
- tds.trap_dict["epoch_serno"],
- tds.trap_dict["notify OID"],
- tds.trap_dict["agent name"],
- tds.trap_dict["agent address"],
- tds.trap_dict["pdu agent name"],
- tds.trap_dict["pdu agent address"],
- tds.trap_dict["cambria.partition"],
- tds.trap_dict["protocol version"],
- tds.trap_dict["uuid"],
- tds.all_vb_str))
+ time_now = int(round(time.time(), 0))
+ arrived_epoch_time_int = int(tds.trap_dict["time received"])
+ tds.arriving_traps_fd.write(
+ "%d %s; %s %s %s %s %s %s %s %s %s %s %s %s %s\n"
+ % (
+ time_now,
+ datetime.datetime.fromtimestamp(time_now).strftime("%a %b %d %H:%M:%S %Y"),
+ tds.trap_dict["time received"],
+ datetime.datetime.fromtimestamp(arrived_epoch_time_int).strftime("%a %b %d %H:%M:%S %Y"),
+ tds.trap_dict["trap category"],
+ tds.trap_dict["epoch_serno"],
+ tds.trap_dict["notify OID"],
+ tds.trap_dict["agent name"],
+ tds.trap_dict["agent address"],
+ tds.trap_dict["pdu agent name"],
+ tds.trap_dict["pdu agent address"],
+ tds.trap_dict["cambria.partition"],
+ tds.trap_dict["protocol version"],
+ tds.trap_dict["uuid"],
+ tds.all_vb_str,
+ )
+ )
except Exception as e:
msg = "Error writing to %s : %s - arriving trap %s NOT LOGGED" % (
- tds.arriving_traps_filename, str(e), tds.trap_dict["uuid"])
+ tds.arriving_traps_filename,
+ str(e),
+ tds.trap_dict["uuid"],
+ )
ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_ERROR, tds.CODE_GENERAL, msg)
@@ -272,14 +268,11 @@
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
try:
- tds.json_traps_fd.write('%s\n' % _post_data_enclosed)
- msg = "successfully logged json for %s to %s" % (
- tds.trap_dict["uuid"], tds.json_traps_filename)
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ tds.json_traps_fd.write("%s\n" % _post_data_enclosed)
+ msg = "successfully logged json for %s to %s" % (tds.trap_dict["uuid"], tds.json_traps_filename)
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
except Exception as e:
- msg = "Error writing to %s : %s - trap %s NOT LOGGED" % (
- tds.json_traps_filename, str(e), tds.trap_dict["uuid"])
+ msg = "Error writing to %s : %s - trap %s NOT LOGGED" % (tds.json_traps_filename, str(e), tds.trap_dict["uuid"])
ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_ERROR, tds.CODE_GENERAL, msg)
@@ -303,8 +296,7 @@
if tds.http_requ_session is None:
msg = "tds.http_requ_session is None - getting new (%s)" % tds.http_requ_session
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
tds.http_requ_session = init_session_obj()
# if only 1 trap, ship as-is
@@ -312,7 +304,7 @@
post_data_enclosed = tds.all_traps_json_str
else:
# otherwise, add brackets around package
- post_data_enclosed = '[' + tds.all_traps_json_str + ']'
+ post_data_enclosed = "[" + tds.all_traps_json_str + "]"
msg = "post_data_enclosed: %s" % (post_data_enclosed)
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
@@ -320,68 +312,91 @@
k = 0
dmaap_pub_success = False
- while not dmaap_pub_success and k < (int(tds.c_config['publisher']['http_retries'])):
+ while not dmaap_pub_success and k < (int(tds.c_config["publisher"]["http_retries"])):
try:
- if tds.c_config['streams_publishes']['sec_fault_unsecure']['aaf_username'] == "" or tds.c_config['streams_publishes']['sec_fault_unsecure']['aaf_username'] is None:
+ if (
+ tds.c_config["streams_publishes"]["sec_fault_unsecure"]["aaf_username"] == ""
+ or tds.c_config["streams_publishes"]["sec_fault_unsecure"]["aaf_username"] is None
+ ):
msg = "%d trap(s) : %s - attempt %d (unsecure)" % (
- tds.traps_since_last_publish, tds.trap_uuids_in_buffer, k)
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
- http_resp = tds.http_requ_session.post(tds.c_config['streams_publishes']['sec_fault_unsecure']['dmaap_info']['topic_url'], post_data_enclosed,
- headers=http_headers,
- timeout=tds.timeout_seconds)
+ tds.traps_since_last_publish,
+ tds.trap_uuids_in_buffer,
+ k,
+ )
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
+ http_resp = tds.http_requ_session.post(
+ tds.c_config["streams_publishes"]["sec_fault_unsecure"]["dmaap_info"]["topic_url"],
+ post_data_enclosed,
+ headers=http_headers,
+ timeout=tds.timeout_seconds,
+ )
else:
msg = "%d trap(s) : %s - attempt %d (secure)" % (
- tds.traps_since_last_publish, tds.trap_uuids_in_buffer, k)
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
- http_resp = tds.http_requ_session.post(tds.c_config['streams_publishes']['sec_fault_unsecure']['dmaap_info']['topic_url'], post_data_enclosed,
- auth=(tds.c_config['streams_publishes']['sec_fault_unsecure']['aaf_username'],
- tds.c_config['streams_publishes']['sec_fault_unsecure']['aaf_password']),
- headers=http_headers,
- timeout=tds.timeout_seconds)
+ tds.traps_since_last_publish,
+ tds.trap_uuids_in_buffer,
+ k,
+ )
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
+ http_resp = tds.http_requ_session.post(
+ tds.c_config["streams_publishes"]["sec_fault_unsecure"]["dmaap_info"]["topic_url"],
+ post_data_enclosed,
+ auth=(
+ tds.c_config["streams_publishes"]["sec_fault_unsecure"]["aaf_username"],
+ tds.c_config["streams_publishes"]["sec_fault_unsecure"]["aaf_password"],
+ ),
+ headers=http_headers,
+ timeout=tds.timeout_seconds,
+ )
if http_resp.status_code == requests.codes.ok:
- msg = "%d trap(s) successfully published: %s" % (
- tds.traps_since_last_publish, tds.trap_uuids_in_buffer)
- ecomp_logger(tds.LOG_TYPE_METRICS, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ msg = "%d trap(s) successfully published: %s" % (tds.traps_since_last_publish, tds.trap_uuids_in_buffer)
+ ecomp_logger(tds.LOG_TYPE_METRICS, tds.SEV_INFO, tds.CODE_GENERAL, msg)
log_published_messages(post_data_enclosed)
tds.last_pub_time = time.time()
dmaap_pub_success = True
break
else:
msg = "Trap(s) %s publish attempt %d returned non-normal: %d %s" % (
- tds.trap_uuids_in_buffer, k, http_resp.status_code, http_resp.text)
- ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_WARN,
- tds.CODE_GENERAL, msg)
+ tds.trap_uuids_in_buffer,
+ k,
+ http_resp.status_code,
+ http_resp.text,
+ )
+ ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_WARN, tds.CODE_GENERAL, msg)
except OSError as e:
msg = "OS exception while attempting to post %s attempt %d: (%s) %s %s" % (
- tds.trap_uuids_in_buffer, k, e.errno, e.strerror, str(e))
- ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_WARN,
- tds.CODE_GENERAL, msg)
+ tds.trap_uuids_in_buffer,
+ k,
+ e.errno,
+ e.strerror,
+ str(e),
+ )
+ ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_WARN, tds.CODE_GENERAL, msg)
except requests.exceptions.RequestException as e:
msg = "Requests exception while attempting to post %s attempt %d: (%d) %s" % (
- tds.trap_uuids_in_buffer, int(k), int(e.errno), str(e.strerror))
- ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_WARN,
- tds.CODE_GENERAL, msg)
+ tds.trap_uuids_in_buffer,
+ int(k),
+ int(e.errno),
+ str(e.strerror),
+ )
+ ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_WARN, tds.CODE_GENERAL, msg)
k += 1
- if k < int(tds.c_config['publisher']['http_retries']):
- msg = "sleeping %.4f seconds and retrying" % (
- tds.seconds_between_retries)
- ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_WARN,
- tds.CODE_GENERAL, msg)
+ if k < int(tds.c_config["publisher"]["http_retries"]):
+ msg = "sleeping %.4f seconds and retrying" % (tds.seconds_between_retries)
+ ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_WARN, tds.CODE_GENERAL, msg)
time.sleep(tds.seconds_between_retries)
else:
break
if not dmaap_pub_success:
- msg = "ALL publish attempts failed for traps %s to URL %s "\
- % (tds.trap_uuids_in_buffer, tds.c_config['streams_publishes']['sec_fault_unsecure']['dmaap_info']['topic_url'])
+ msg = "ALL publish attempts failed for traps %s to URL %s " % (
+ tds.trap_uuids_in_buffer,
+ tds.c_config["streams_publishes"]["sec_fault_unsecure"]["dmaap_info"]["topic_url"],
+ )
ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_ERROR, tds.CODE_GENERAL, msg)
# FMDL: This currently tries, then logs error and trashes buffer if all dmaap attempts fail. Better way?
@@ -390,6 +405,7 @@
tds.all_traps_json_str = ""
tds.first_trap = True
+
# # # # # # # # # # # # # # # # # # #
# fx: request_observer for community string rewrite
# # # # # # # # # # # # # # # # # # #
@@ -398,8 +414,9 @@
def comm_string_rewrite_observer(snmpEngine, execpoint, variables, cbCtx):
# match ALL community strings
- if re.match('.*', str(variables['communityName'])):
- variables['communityName'] = variables['communityName'].clone('public')
+ if re.match(".*", str(variables["communityName"])):
+ variables["communityName"] = variables["communityName"].clone("public")
+
# # # # # # # # # # # # # # # # # # #
# fx: snmp_engine_observer_cb
@@ -457,21 +474,20 @@
else:
tds.traps_in_epoch = 0
tds.last_epoch_second = epoch_second
- traps_in_epoch_04d = format(tds.traps_in_epoch, '04d')
- tds.trap_dict['epoch_arrived'] = epoch_second
- tds.trap_dict['epoch_serno'] = int(
- (str(epoch_second) + str(traps_in_epoch_04d)))
+ traps_in_epoch_04d = format(tds.traps_in_epoch, "04d")
+ tds.trap_dict["epoch_arrived"] = epoch_second
+ tds.trap_dict["epoch_serno"] = int((str(epoch_second) + str(traps_in_epoch_04d)))
# assign uuid to trap
tds.trap_dict["uuid"] = str(uuid_mod.uuid1())
# ip and hostname
- ip_addr_str = str(variables['transportAddress'][0])
- # set agent address and name to source of packet, OVERWRITE if
+ ip_addr_str = str(variables["transportAddress"][0])
+ # set agent address and name to source of packet, OVERWRITE if
# .1.3.6.1.6.3.18.1.3.0 varbind encountered later in trap processing
tds.trap_dict["agent address"] = ip_addr_str
tds.trap_dict["agent name"] = resolve_ip(ip_addr_str)
- # set overridden/logical address and name to source of packet so we know
+ # set overridden/logical address and name to source of packet so we know
# original value if .1.3.6.1.6.3.18.1.3.0 shows up
# NOTE: This does NOT change ever, label may change to
# "overridden agent..." in the future for truth in nameing
@@ -479,17 +495,15 @@
tds.trap_dict["pdu agent name"] = tds.trap_dict["agent name"]
# log arrival now that we have agent addr
- msg = 'trap from %s %s, assigned uuid: %s' % \
- (ip_addr_str, tds.trap_dict["agent name"], tds.trap_dict["uuid"])
+ msg = "trap from %s %s, assigned uuid: %s" % (ip_addr_str, tds.trap_dict["agent name"], tds.trap_dict["uuid"])
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
-
tds.trap_dict["cambria.partition"] = str(tds.trap_dict["agent name"])
# do not include cleartext community in pub
tds.trap_dict["community"] = ""
tds.trap_dict["community len"] = 0
- snmp_version = variables['securityModel']
+ snmp_version = variables["securityModel"]
if snmp_version == 1:
tds.trap_dict["protocol version"] = "v1"
# enterprise = variables['pdu']['enterprise'].prettyPrint()
@@ -517,16 +531,19 @@
tds.trap_dict["protocol version"] = "unknown"
# tds.trap_dict['time received'] = epoch_msecond
- tds.trap_dict['time received'] = epoch_second
- tds.trap_dict['trap category'] = (
- tds.c_config['streams_publishes']['sec_fault_unsecure']['dmaap_info']['topic_url']).split('/')[-1]
+ tds.trap_dict["time received"] = epoch_second
+ tds.trap_dict["trap category"] = (
+ tds.c_config["streams_publishes"]["sec_fault_unsecure"]["dmaap_info"]["topic_url"]
+ ).split("/")[-1]
return True
+
# # # # # # # # # # # # # # # # # # #
# fx: request_observer for community string rewrite
# # # # # # # # # # # # # # # # # # #
+
def add_varbind_to_log_string(vb_idx, vb_oid, vb_type, vb_val):
"""
Called for each varbind, adds individual attributes of varbind instance to
@@ -548,10 +565,11 @@
"""
if vb_idx == 0:
- tds.all_vb_str = 'varbinds:'
+ tds.all_vb_str = "varbinds:"
- tds.all_vb_str = tds.all_vb_str + " [" + str(vb_idx) + "] " \
- + str(vb_oid) + " {" + vb_type + "} " + str(vb_val.prettyPrint())
+ tds.all_vb_str = (
+ tds.all_vb_str + " [" + str(vb_idx) + "] " + str(vb_oid) + " {" + vb_type + "} " + str(vb_val.prettyPrint())
+ )
# try:
# tds.all_vb_str = tds.all_vb_str + " [" + str(vb_idx) + "] " + vb_oid + " {" + vb_type + "} " + vb_val
@@ -562,6 +580,7 @@
# ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_WARN, tds.CODE_GENERAL, msg)
# return 1
+
def add_varbind_to_json(vb_idx, vb_oid, vb_type, vb_val):
"""
Called for each varbind, adds individual attributes of varbind instance to
@@ -595,8 +614,7 @@
# if second varbind, use as notifyOID for all snmp versions
if vb_idx == 1:
tds.trap_dict["notify OID"] = "." + _vb_value
- tds.trap_dict["notify OID len"] = tds.trap_dict["notify OID"].count(
- '.')
+ tds.trap_dict["notify OID len"] = tds.trap_dict["notify OID"].count(".")
return 0
# if override varbind OID, use value as agent address
@@ -614,15 +632,15 @@
return 0
if tds.first_varbind:
- tds.all_vb_json_str = ', \"varbinds\": ['
+ tds.all_vb_json_str = ', "varbinds": ['
tds.first_varbind = False
else:
tds.all_vb_json_str = tds.all_vb_json_str + " ,"
_individual_vb_dict.clear()
- _individual_vb_dict['varbind_oid'] = _vb_oid
- _individual_vb_dict['varbind_type'] = _vb_type
- _individual_vb_dict['varbind_value'] = _vb_value
+ _individual_vb_dict["varbind_oid"] = _vb_oid
+ _individual_vb_dict["varbind_type"] = _vb_type
+ _individual_vb_dict["varbind_value"] = _vb_value
_individual_vb_json_str = json.dumps(_individual_vb_dict)
@@ -632,8 +650,7 @@
# Callback function for receiving notifications
# noinspection PyUnusedLocal,PyUnusedLocal,PyUnusedLocal
-def notif_receiver_cb(snmp_engine, stateReference, contextEngineId, contextName,
- varBinds, cbCtx):
+def notif_receiver_cb(snmp_engine, stateReference, contextEngineId, contextName, varBinds, cbCtx):
"""
Callback executed when trap arrives
:Parameters:
@@ -673,10 +690,8 @@
# iterate over varbinds, add to json struct
for vb_oid, vb_val in varBinds:
- log_ret = add_varbind_to_log_string(
- pdu_varbind_count, vb_oid, vb_val.__class__.__name__, vb_val)
- varbinds_added = add_varbind_to_json(
- pdu_varbind_count, vb_oid, vb_val.__class__.__name__, vb_val)
+ log_ret = add_varbind_to_log_string(pdu_varbind_count, vb_oid, vb_val.__class__.__name__, vb_val)
+ varbinds_added = add_varbind_to_json(pdu_varbind_count, vb_oid, vb_val.__class__.__name__, vb_val)
payload_varbinds += varbinds_added
pdu_varbind_count += 1
@@ -686,7 +701,7 @@
# if varbinds present - which will almost always be the case - add all_vb_json_str to trap_json_message
if payload_varbinds != 0:
# close out vb array
- tds.all_vb_json_str = tds.all_vb_json_str + ']'
+ tds.all_vb_json_str = tds.all_vb_json_str + "]"
# remove last close bracket from curr_trap_json_str
curr_trap_json_str = curr_trap_json_str[:-1]
@@ -695,7 +710,7 @@
curr_trap_json_str = curr_trap_json_str + tds.all_vb_json_str
# add last close brace back in
- curr_trap_json_str = curr_trap_json_str + '}'
+ curr_trap_json_str = curr_trap_json_str + "}"
msg = "trap %s : %s" % (tds.trap_dict["uuid"], curr_trap_json_str)
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
@@ -710,38 +725,41 @@
# only add to publish buffer if stormwatch is NOT active
if stormwatch.sw_storm_active(tds.trap_dict["agent address"], tds.trap_dict["notify OID"]):
msg = "stormwatch active - deflecting notification %s from %s" % (
- tds.trap_dict["notify OID"], tds.trap_dict["agent address"])
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ tds.trap_dict["notify OID"],
+ tds.trap_dict["agent address"],
+ )
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
else:
msg = "adding %s to buffer" % (tds.trap_dict["uuid"])
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
if tds.first_trap:
tds.all_traps_json_str = curr_trap_json_str
tds.trap_uuids_in_buffer = tds.trap_dict["uuid"]
tds.first_trap = False
else:
- tds.trap_uuids_in_buffer = tds.trap_uuids_in_buffer + \
- ', ' + tds.trap_dict["uuid"]
- tds.all_traps_json_str = tds.all_traps_json_str + ', ' + curr_trap_json_str
+ tds.trap_uuids_in_buffer = tds.trap_uuids_in_buffer + ", " + tds.trap_dict["uuid"]
+ tds.all_traps_json_str = tds.all_traps_json_str + ", " + curr_trap_json_str
- if tds.traps_since_last_publish >= int(tds.c_config['publisher']['max_traps_between_publishes']):
+ if tds.traps_since_last_publish >= int(tds.c_config["publisher"]["max_traps_between_publishes"]):
msg = "num traps since last publish (%d) exceeds threshold (%d) - publish traps" % (
- tds.traps_since_last_publish, int(tds.c_config['publisher']['max_traps_between_publishes']))
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ tds.traps_since_last_publish,
+ int(tds.c_config["publisher"]["max_traps_between_publishes"]),
+ )
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
post_dmaap()
- elif milliseconds_since_last_publish >= int(tds.c_config['publisher']['max_milliseconds_between_publishes']):
- msg = "num milliseconds since last publish (%.0f) exceeds threshold - publish traps" % milliseconds_since_last_publish
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ elif milliseconds_since_last_publish >= int(tds.c_config["publisher"]["max_milliseconds_between_publishes"]):
+ msg = (
+ "num milliseconds since last publish (%.0f) exceeds threshold - publish traps"
+ % milliseconds_since_last_publish
+ )
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
post_dmaap()
else:
- msg = "neither milliseconds_since_last_publish (%.0f) or traps_since_last_publish (%d) exceed threshold - continue" % (
- milliseconds_since_last_publish, tds.traps_since_last_publish)
- ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO,
- tds.CODE_GENERAL, msg)
+ msg = (
+ "neither milliseconds_since_last_publish (%.0f) or traps_since_last_publish (%d) exceed threshold - continue"
+ % (milliseconds_since_last_publish, tds.traps_since_last_publish)
+ )
+ ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
# # # # # # # # # # # # #
@@ -751,12 +769,9 @@
if __name__ == "__main__":
- parser = argparse.ArgumentParser(description='Post SNMP traps '
- 'to message bus')
- parser.add_argument('-v', action="store_true", dest="verbose",
- help="verbose logging")
- parser.add_argument('-?', action="store_true", dest="usage_requested",
- help="show command line use")
+ parser = argparse.ArgumentParser(description="Post SNMP traps " "to message bus")
+ parser.add_argument("-v", action="store_true", dest="verbose", help="verbose logging")
+ parser.add_argument("-?", action="store_true", dest="usage_requested", help="show command line use")
# parse args
args = parser.parse_args()
@@ -784,7 +799,10 @@
# get config binding service (CBS) values (either broker, or json file override)
load_all_configs(0, 0)
msg = "%s : %s version %s starting" % (
- prog_name, tds.c_config['snmptrapd']['title'], tds.c_config['snmptrapd']['version'])
+ prog_name,
+ tds.c_config["snmptrapd"]["title"],
+ tds.c_config["snmptrapd"]["version"],
+ )
stdout_logger(msg)
# open various ecomp logs
@@ -797,22 +815,33 @@
stdout_logger(msg)
# use specific flags or 'all' for full debugging
# debug.setLogger(debug.Debug('dsp', 'msgproc'))
- debug.setLogger(debug.Debug('all'))
+ debug.setLogger(debug.Debug("all"))
# name and open arriving trap log
- 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_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 = open_file(tds.arriving_traps_filename)
- msg = ("arriving traps logged to: %s" % tds.arriving_traps_filename)
+ msg = "arriving traps logged to: %s" % tds.arriving_traps_filename
stdout_logger(msg)
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
# 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_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 = open_file(tds.json_traps_filename)
- msg = ("published traps logged to: %s" % tds.json_traps_filename)
+ msg = "published traps logged to: %s" % tds.json_traps_filename
stdout_logger(msg)
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
@@ -820,8 +849,9 @@
signal.signal(signal.SIGUSR1, load_all_configs)
# save current PID for future/external reference
- tds.pid_file_name = tds.c_config['files']['runtime_base_dir'] + \
- '/' + tds.c_config['files']['pid_dir'] + '/' + prog_name + ".pid"
+ tds.pid_file_name = (
+ tds.c_config["files"]["runtime_base_dir"] + "/" + tds.c_config["files"]["pid_dir"] + "/" + prog_name + ".pid"
+ )
msg = "Runtime PID file: %s" % tds.pid_file_name
ecomp_logger(tds.LOG_TYPE_DEBUG, tds.SEV_INFO, tds.CODE_GENERAL, msg)
rc = save_pid(tds.pid_file_name)
@@ -839,8 +869,8 @@
# UDP over IPv4
try:
- ipv4_interface = tds.c_config['protocols']['ipv4_interface']
- ipv4_port = int(tds.c_config['protocols']['ipv4_port'])
+ ipv4_interface = tds.c_config["protocols"]["ipv4_interface"]
+ ipv4_port = int(tds.c_config["protocols"]["ipv4_port"])
try:
# FIXME: this doesn't appear to throw an exception even if
@@ -850,16 +880,11 @@
# means to confirm proper privileges (then
# close it and reopen w/ pysnmp api)
config.addTransport(
- snmp_engine,
- udp.domainName + (1,),
- udp.UdpTransport().openServerMode(
- (ipv4_interface, ipv4_port))
+ snmp_engine, udp.domainName + (1,), udp.UdpTransport().openServerMode((ipv4_interface, ipv4_port))
)
except Exception as e:
- msg = "Unable to bind to %s:%d - %s" % (
- ipv4_interface, ipv4_port, str(e))
- ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_FATAL,
- tds.CODE_GENERAL, msg)
+ msg = "Unable to bind to %s:%d - %s" % (ipv4_interface, ipv4_port, str(e))
+ ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_FATAL, tds.CODE_GENERAL, msg)
stdout_logger(msg)
cleanup_and_exit(1, tds.pid_file_name)
@@ -870,22 +895,17 @@
# UDP over IPv6
try:
- ipv6_interface = tds.c_config['protocols']['ipv6_interface']
- ipv6_port = int(tds.c_config['protocols']['ipv6_port'])
+ ipv6_interface = tds.c_config["protocols"]["ipv6_interface"]
+ ipv6_port = int(tds.c_config["protocols"]["ipv6_port"])
try:
config.addTransport(
- snmp_engine,
- udp6.domainName,
- udp6.Udp6Transport().openServerMode(
- (ipv6_interface, ipv6_port))
+ snmp_engine, udp6.domainName, udp6.Udp6Transport().openServerMode((ipv6_interface, ipv6_port))
)
except Exception as e:
- msg = "Unable to bind to %s:%d - %s" % (
- ipv6_interface, ipv6_port, str(e))
+ msg = "Unable to bind to %s:%d - %s" % (ipv6_interface, ipv6_port, str(e))
stdout_logger(msg)
- ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_FATAL,
- tds.CODE_GENERAL, msg)
+ ecomp_logger(tds.LOG_TYPE_ERROR, tds.SEV_FATAL, tds.CODE_GENERAL, msg)
cleanup_and_exit(1, tds.pid_file_name)
except Exception as e:
@@ -900,25 +920,21 @@
# SecurityName <-> CommunityName mapping
# to restrict trap reception to only those with specific community
# strings
- config.addV1System(snmp_engine, 'my-area', 'public')
+ config.addV1System(snmp_engine, "my-area", "public")
# register comm_string_rewrite_observer for message arrival
- snmp_engine.observer.registerObserver(
- comm_string_rewrite_observer,
- 'rfc2576.processIncomingMsg:writable'
- )
+ snmp_engine.observer.registerObserver(comm_string_rewrite_observer, "rfc2576.processIncomingMsg:writable")
# # # # # # # # # # # #
# SNMPv3 setup
# # # # # # # # # # # #
- config, snmp_engine = load_snmpv3_credentials(
- config, snmp_engine, tds.c_config)
+ config, snmp_engine = load_snmpv3_credentials(config, snmp_engine, tds.c_config)
# register snmp_engine_observer_cb for message arrival
snmp_engine.observer.registerObserver(
snmp_engine_observer_cb,
- 'rfc3412.receiveMessage:request',
- 'rfc3412.returnResponsePdu',
+ "rfc3412.receiveMessage:request",
+ "rfc3412.returnResponsePdu",
)
# Register SNMP Application at the SNMP engine
diff --git a/tests/__init__.py b/tests/__init__.py
index af0fbb2..1d97a6a 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -1,5 +1,5 @@
-# ================================================================================
-# Copyright (c) 2018-2020 AT&T Intellectual Property. All rights reserved.
+# ============LICENSE_START=======================================================
+# Copyright (c) 2018-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.
@@ -13,8 +13,6 @@
# 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.
# empty __init__.py so that pytest can add correct path to coverage report, -- per pytest
diff --git a/tests/snmp.setup.py b/tests/snmp.setup.py
index ed565f2..c5a533c 100644
--- a/tests/snmp.setup.py
+++ b/tests/snmp.setup.py
@@ -1,6 +1,5 @@
-# org.onap.dcae
-# ================================================================================
-# Copyright (c) 2017-2018 AT&T Intellectual Property. All rights reserved.
+# ============LICENSE_START=======================================================
+# 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.
@@ -14,8 +13,6 @@
# 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
@@ -44,6 +41,7 @@
from collections import Counter
from onap_dcae_cbs_docker_client.client import get_config
from pysnmp.carrier.asyncio.dgram import udp, udp6
+
# from pysnmp.carrier.asyncore.dgram import udp
from pysnmp.entity import engine, config
from pysnmp.entity.rfc3413 import ntfrcv
@@ -60,14 +58,14 @@
reqs = [str(ir.req) for ir in install_reqs]
setup(
- name = "dcaegen2-collectors-snmptrap",
- description = "snmp trap receiver for ONAP docker image",
- version = "1.4.0",
+ name="dcaegen2-collectors-snmptrap",
+ description="snmp trap receiver for ONAP docker image",
+ version="1.4.0",
packages=find_packages(),
- author = "Dave L",
- author_email = "dl3158@att.com",
- license='Apache 2',
- keywords = "",
- url = "",
- install_requires=reqs
+ 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
index cdd19a4..dee2aa0 100644
--- a/tests/test_snmptrapd.py
+++ b/tests/test_snmptrapd.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2018-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2018-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.
@@ -31,19 +31,19 @@
from pysnmp.hlapi import *
from pysnmp import debug
+
class test_snmptrapd(unittest.TestCase):
"""
Test the save_pid mod
"""
- pytest_json_data = "{ \"snmptrapd\": { \"version\": \"2.0.3\", \"title\": \"ONAP SNMP Trap Receiver\" }, \"protocols\": { \"transport\": \"udp\", \"ipv4_interface\": \"0.0.0.0\", \"ipv4_port\": 6162, \"ipv6_interface\": \"::1\", \"ipv6_port\": 6162 }, \"cache\": { \"dns_cache_ttl_seconds\": 60 }, \"publisher\": { \"http_timeout_milliseconds\": 1500, \"http_retries\": 3, \"http_milliseconds_between_retries\": 750, \"http_primary_publisher\": \"true\", \"http_peer_publisher\": \"unavailable\", \"max_traps_between_publishes\": 10, \"max_milliseconds_between_publishes\": 10000 }, \"streams_publishes\": { \"sec_fault_unsecure\": { \"type\": \"message_router\", \"aaf_password\": null, \"dmaap_info\": { \"location\": \"mtl5\", \"client_id\": null, \"client_role\": null, \"topic_url\": \"http://uebsb91kcdc.it.att.com:3904/events/ONAP-COLLECTOR-SNMPTRAP\" }, \"aaf_username\": null } }, \"files\": { \"runtime_base_dir\": \"/tmp/opt/app/snmptrap\", \"log_dir\": \"logs\", \"data_dir\": \"data\", \"pid_dir\": \"tmp\", \"arriving_traps_log\": \"snmptrapd_arriving_traps.log\", \"snmptrapd_diag\": \"snmptrapd_prog_diag.log\", \"traps_stats_log\": \"snmptrapd_stats.csv\", \"perm_status_file\": \"snmptrapd_status.log\", \"eelf_base_dir\": \"/tmp/opt/app/snmptrap/logs\", \"eelf_error\": \"error.log\", \"eelf_debug\": \"debug.log\", \"eelf_audit\": \"audit.log\", \"eelf_metrics\": \"metrics.log\", \"roll_frequency\": \"day\", \"minimum_severity_to_log\": 2 }, \"trap_config\": { \"sw_interval_in_seconds\": 60, \"notify_oids\": { \".1.3.6.1.4.1.9.0.1\": { \"sw_high_water_in_interval\": 102, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.2\": { \"sw_high_water_in_interval\": 101, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.3\": { \"sw_high_water_in_interval\": 102, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.4\": { \"sw_high_water_in_interval\": 10, \"sw_low_water_in_interval\": 3, \"category\": \"logonly\" } } }, \"snmpv3_config\": { \"usm_users\": [ { \"user\": \"usr-sha-aes256\", \"engineId\": \"8000000001020304\", \"usmHMACSHAAuth\": \"authkey1\", \"usmAesCfb256\": \"privkey1\" }, { \"user\": \"user1\", \"engineId\": \"8000000000000001\", \"usmHMACMD5Auth\": \"authkey1\", \"usmDESPriv\": \"privkey1\" }, { \"user\": \"user2\", \"engineId\": \"8000000000000002\", \"usmHMACSHAAuth\": \"authkey2\", \"usmAesCfb128\": \"privkey2\" }, { \"user\": \"user3\", \"engineId\": \"8000000000000003\", \"usmHMACSHAAuth\": \"authkey3\", \"usmAesCfb256\": \"privkey3\" } ] } }"
+ pytest_json_data = '{ "snmptrapd": { "version": "2.0.3", "title": "ONAP SNMP Trap Receiver" }, "protocols": { "transport": "udp", "ipv4_interface": "0.0.0.0", "ipv4_port": 6162, "ipv6_interface": "::1", "ipv6_port": 6162 }, "cache": { "dns_cache_ttl_seconds": 60 }, "publisher": { "http_timeout_milliseconds": 1500, "http_retries": 3, "http_milliseconds_between_retries": 750, "http_primary_publisher": "true", "http_peer_publisher": "unavailable", "max_traps_between_publishes": 10, "max_milliseconds_between_publishes": 10000 }, "streams_publishes": { "sec_fault_unsecure": { "type": "message_router", "aaf_password": null, "dmaap_info": { "location": "mtl5", "client_id": null, "client_role": null, "topic_url": "http://uebsb91kcdc.it.att.com:3904/events/ONAP-COLLECTOR-SNMPTRAP" }, "aaf_username": null } }, "files": { "runtime_base_dir": "/tmp/opt/app/snmptrap", "log_dir": "logs", "data_dir": "data", "pid_dir": "tmp", "arriving_traps_log": "snmptrapd_arriving_traps.log", "snmptrapd_diag": "snmptrapd_prog_diag.log", "traps_stats_log": "snmptrapd_stats.csv", "perm_status_file": "snmptrapd_status.log", "eelf_base_dir": "/tmp/opt/app/snmptrap/logs", "eelf_error": "error.log", "eelf_debug": "debug.log", "eelf_audit": "audit.log", "eelf_metrics": "metrics.log", "roll_frequency": "day", "minimum_severity_to_log": 2 }, "trap_config": { "sw_interval_in_seconds": 60, "notify_oids": { ".1.3.6.1.4.1.9.0.1": { "sw_high_water_in_interval": 102, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.2": { "sw_high_water_in_interval": 101, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.3": { "sw_high_water_in_interval": 102, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.4": { "sw_high_water_in_interval": 10, "sw_low_water_in_interval": 3, "category": "logonly" } } }, "snmpv3_config": { "usm_users": [ { "user": "usr-sha-aes256", "engineId": "8000000001020304", "usmHMACSHAAuth": "authkey1", "usmAesCfb256": "privkey1" }, { "user": "user1", "engineId": "8000000000000001", "usmHMACMD5Auth": "authkey1", "usmDESPriv": "privkey1" }, { "user": "user2", "engineId": "8000000000000002", "usmHMACSHAAuth": "authkey2", "usmAesCfb128": "privkey2" }, { "user": "user3", "engineId": "8000000000000003", "usmHMACSHAAuth": "authkey3", "usmAesCfb256": "privkey3" } ] } }'
# create copy of snmptrapd.json for pytest
pytest_json_config = "/tmp/opt/app/snmptrap/etc/snmptrapd.json"
- with open(pytest_json_config, 'w') as outfile:
+ with open(pytest_json_config, "w") as outfile:
outfile.write(pytest_json_data)
-
def test_usage_err(self):
"""
Test usage error
@@ -54,7 +54,6 @@
assert pytest_wrapped_sys_exit.type == SystemExit
assert pytest_wrapped_sys_exit.value.code == 1
-
def test_load_all_configs(self):
"""
Test load of all configs
@@ -65,7 +64,7 @@
sw.sw_init()
# request load of CBS data
- os.environ.update(CBS_SIM_JSON='/tmp/opt/app/snmptrap/etc/snmptrapd.json')
+ 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)
@@ -82,7 +81,7 @@
tds.init()
# request load of CBS data
- os.environ.update(CBS_SIM_JSON='/tmp/opt/app/snmptrap/etc/snmptrapd.json')
+ 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)
@@ -99,29 +98,54 @@
tds.init()
# request load of CBS data
- os.environ.update(CBS_SIM_JSON='/tmp/opt/app/snmptrap/etc/snmptrapd.json')
+ 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 logging
- 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}
+ 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_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_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)
+ 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)
@@ -139,7 +163,7 @@
tds.init()
# request load of CBS data
- os.environ.update(CBS_SIM_JSON='/tmp/opt/app/snmptrap/etc/snmptrapd.json')
+ os.environ.update(CBS_SIM_JSON="/tmp/opt/app/snmptrap/etc/snmptrapd.json")
trapd_get_cbs_config.get_cbs_config()
# open eelf logs
@@ -154,20 +178,26 @@
tds.init()
# request load of CBS data
- os.environ.update(CBS_SIM_JSON='/tmp/opt/app/snmptrap/etc/snmptrapd.json')
+ os.environ.update(CBS_SIM_JSON="/tmp/opt/app/snmptrap/etc/snmptrapd.json")
trapd_get_cbs_config.get_cbs_config()
- errorIndication, errorStatus, errorIndex, varbinds = next(sendNotification(SnmpEngine(),
- CommunityData('not_public'),
- UdpTransportTarget(('localhost', 6162)),
- 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'))])
+ errorIndication, errorStatus, errorIndex, varbinds = next(
+ sendNotification(
+ SnmpEngine(),
+ CommunityData("not_public"),
+ UdpTransportTarget(("localhost", 6162)),
+ 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")),
+ ],
+ )
)
result = errorIndication
self.assertEqual(result, None)
-if __name__ == '__main__':
+
+if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_snmptrapd_send_test_trap.py b/tests/test_snmptrapd_send_test_trap.py
index 0e23def..8158629 100755
--- a/tests/test_snmptrapd_send_test_trap.py
+++ b/tests/test_snmptrapd_send_test_trap.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2018-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2018-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.
@@ -21,32 +21,37 @@
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'))])
+ 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)'))
- )
+ 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)")),
+ ),
)
)
diff --git a/tests/test_trapd_exit.py b/tests/test_trapd_exit.py
index 3bbc386..0e38461 100644
--- a/tests/test_trapd_exit.py
+++ b/tests/test_trapd_exit.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2018-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2018-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,36 +18,38 @@
import unittest
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):
"""
Test the cleanup_and_exit mod
"""
-
+
def test_normal_exit(self):
"""
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
# 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)
+ 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__':
+
+if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_trapd_get_cbs_config.py b/tests/test_trapd_get_cbs_config.py
index 67bccfc..b7c5d7b 100644
--- a/tests/test_trapd_get_cbs_config.py
+++ b/tests/test_trapd_get_cbs_config.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019-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.
@@ -41,16 +41,16 @@
# env var for CBS_SIM_JSON
try:
- os.environ['CBS_SIM_JSON'] = "/tmp/opt/app/snmptrap/etc/snmptrapd.json"
+ os.environ["CBS_SIM_JSON"] = "/tmp/opt/app/snmptrap/etc/snmptrapd.json"
except Exception as e:
print("Error while running %s : %s" % (os.path.basename(__file__), str(e.strerror)))
sys.exit(1)
-pytest_json_data = "{ \"snmptrapd\": { \"version\": \"1.4.0\", \"title\": \"ONAP SNMP Trap Receiver\" }, \"protocols\": { \"transport\": \"udp\", \"ipv4_interface\": \"0.0.0.0\", \"ipv4_port\": 6162, \"ipv6_interface\": \"::1\", \"ipv6_port\": 6162 }, \"cache\": { \"dns_cache_ttl_seconds\": 60 }, \"publisher\": { \"http_timeout_milliseconds\": 1500, \"http_retries\": 3, \"http_milliseconds_between_retries\": 750, \"http_primary_publisher\": \"true\", \"http_peer_publisher\": \"unavailable\", \"max_traps_between_publishes\": 10, \"max_milliseconds_between_publishes\": 10000 }, \"streams_publishes\": { \"sec_fault_unsecure\": { \"type\": \"message_router\", \"aaf_password\": null, \"dmaap_info\": { \"location\": \"mtl5\", \"client_id\": null, \"client_role\": null, \"topic_url\": \"http://localhost:3904/events/ONAP-COLLECTOR-SNMPTRAP\" }, \"aaf_username\": null } }, \"files\": { \"runtime_base_dir\": \"/tmp/opt/app/snmptrap\", \"log_dir\": \"logs\", \"data_dir\": \"data\", \"pid_dir\": \"tmp\", \"arriving_traps_log\": \"snmptrapd_arriving_traps.log\", \"snmptrapd_diag\": \"snmptrapd_prog_diag.log\", \"traps_stats_log\": \"snmptrapd_stats.csv\", \"perm_status_file\": \"snmptrapd_status.log\", \"eelf_base_dir\": \"/tmp/opt/app/snmptrap/logs\", \"eelf_error\": \"error.log\", \"eelf_debug\": \"debug.log\", \"eelf_audit\": \"audit.log\", \"eelf_metrics\": \"metrics.log\", \"roll_frequency\": \"day\", \"minimum_severity_to_log\": 2 }, \"trap_config\": { \"sw_interval_in_seconds\": 60, \"notify_oids\": { \".1.3.6.1.4.1.9.0.1\": { \"sw_high_water_in_interval\": 102, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.2\": { \"sw_high_water_in_interval\": 101, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.3\": { \"sw_high_water_in_interval\": 102, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.4\": { \"sw_high_water_in_interval\": 10, \"sw_low_water_in_interval\": 3, \"category\": \"logonly\" } } }, \"snmpv3_config\": { \"usm_users\": [ { \"user\": \"usr-sha-aes256\", \"engineId\": \"8000000001020304\", \"usmHMACSHAAuth\": \"authkey1\", \"usmAesCfb256\": \"privkey1\" }, { \"user\": \"user1\", \"engineId\": \"8000000000000001\", \"usmHMACMD5Auth\": \"authkey1\", \"usmDESPriv\": \"privkey1\" }, { \"user\": \"user2\", \"engineId\": \"8000000000000002\", \"usmHMACSHAAuth\": \"authkey2\", \"usmAesCfb128\": \"privkey2\" }, { \"user\": \"user3\", \"engineId\": \"8000000000000003\", \"usmHMACSHAAuth\": \"authkey3\", \"usmAesCfb256\": \"privkey3\" } ] } }"
+pytest_json_data = '{ "snmptrapd": { "version": "1.4.0", "title": "ONAP SNMP Trap Receiver" }, "protocols": { "transport": "udp", "ipv4_interface": "0.0.0.0", "ipv4_port": 6162, "ipv6_interface": "::1", "ipv6_port": 6162 }, "cache": { "dns_cache_ttl_seconds": 60 }, "publisher": { "http_timeout_milliseconds": 1500, "http_retries": 3, "http_milliseconds_between_retries": 750, "http_primary_publisher": "true", "http_peer_publisher": "unavailable", "max_traps_between_publishes": 10, "max_milliseconds_between_publishes": 10000 }, "streams_publishes": { "sec_fault_unsecure": { "type": "message_router", "aaf_password": null, "dmaap_info": { "location": "mtl5", "client_id": null, "client_role": null, "topic_url": "http://localhost:3904/events/ONAP-COLLECTOR-SNMPTRAP" }, "aaf_username": null } }, "files": { "runtime_base_dir": "/tmp/opt/app/snmptrap", "log_dir": "logs", "data_dir": "data", "pid_dir": "tmp", "arriving_traps_log": "snmptrapd_arriving_traps.log", "snmptrapd_diag": "snmptrapd_prog_diag.log", "traps_stats_log": "snmptrapd_stats.csv", "perm_status_file": "snmptrapd_status.log", "eelf_base_dir": "/tmp/opt/app/snmptrap/logs", "eelf_error": "error.log", "eelf_debug": "debug.log", "eelf_audit": "audit.log", "eelf_metrics": "metrics.log", "roll_frequency": "day", "minimum_severity_to_log": 2 }, "trap_config": { "sw_interval_in_seconds": 60, "notify_oids": { ".1.3.6.1.4.1.9.0.1": { "sw_high_water_in_interval": 102, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.2": { "sw_high_water_in_interval": 101, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.3": { "sw_high_water_in_interval": 102, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.4": { "sw_high_water_in_interval": 10, "sw_low_water_in_interval": 3, "category": "logonly" } } }, "snmpv3_config": { "usm_users": [ { "user": "usr-sha-aes256", "engineId": "8000000001020304", "usmHMACSHAAuth": "authkey1", "usmAesCfb256": "privkey1" }, { "user": "user1", "engineId": "8000000000000001", "usmHMACMD5Auth": "authkey1", "usmDESPriv": "privkey1" }, { "user": "user2", "engineId": "8000000000000002", "usmHMACSHAAuth": "authkey2", "usmAesCfb128": "privkey2" }, { "user": "user3", "engineId": "8000000000000003", "usmHMACSHAAuth": "authkey3", "usmAesCfb256": "privkey3" } ] } }'
# create snmptrapd.json for pytest
-pytest_json_config = os.getenv('CBS_SIM_JSON')
-with open(pytest_json_config, 'w') as outfile:
+pytest_json_config = os.getenv("CBS_SIM_JSON")
+with open(pytest_json_config, "w") as outfile:
outfile.write(pytest_json_data)
outfile.close()
@@ -65,7 +65,7 @@
Test that CBS fallback env variable exists and we can get config
from fallback env var
"""
- os.environ.update(CBS_SIM_JSON='/tmp/opt/app/snmptrap/etc/snmptrapd.json')
+ os.environ.update(CBS_SIM_JSON="/tmp/opt/app/snmptrap/etc/snmptrapd.json")
result = trapd_get_cbs_config.get_cbs_config()
print("result: %s" % result)
# compare = str(result).startswith("{'snmptrap': ")
@@ -73,9 +73,8 @@
self.assertEqual(result, True)
def test_cbs_override_env_invalid(self):
- """
- """
- os.environ.update(CBS_SIM_JSON='/tmp/opt/app/snmptrap/etc/nosuchfile.json')
+ """ """
+ os.environ.update(CBS_SIM_JSON="/tmp/opt/app/snmptrap/etc/nosuchfile.json")
# result = trapd_get_cbs_config.get_cbs_config()
# print("result: %s" % result)
# compare = str(result).startswith("{'snmptrap': ")
@@ -85,14 +84,14 @@
result = trapd_get_cbs_config.get_cbs_config()
assert pytest_wrapped_sys_exit.type == SystemExit
# assert pytest_wrapped_sys_exit.value.code == 1
-
+
def test_cbs_env_present(self):
"""
Test that CONSUL_HOST env variable exists but fails to
respond
"""
- os.environ.update(CONSUL_HOST='localhost')
- del os.environ['CBS_SIM_JSON']
+ os.environ.update(CONSUL_HOST="localhost")
+ del os.environ["CBS_SIM_JSON"]
# result = trapd_get_cbs_config.get_cbs_config()
# print("result: %s" % result)
# compare = str(result).startswith("{'snmptrap': ")
@@ -100,16 +99,16 @@
with pytest.raises(SystemExit) as sys_exit:
trapd_get_cbs_config.get_cbs_config()
- assert sys_exit.value.errno == errno.ECONNREFUSED
-
+ assert sys_exit.value.errno == errno.ECONNREFUSED
+
def test_cbs_override_env_undefined(self):
- """
- """
+ """ """
print("------>>> RUNNING test_no_cbs_override_env_var:")
- del os.environ['CBS_SIM_JSON']
-
+ del os.environ["CBS_SIM_JSON"]
+
with pytest.raises(SystemExit) as pytest_wrapped_sys_exit:
assert trapd_get_cbs_config.get_cbs_config() == SystemExit
-
-if __name__ == '__main__':
+
+
+if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_trapd_http_session.py b/tests/test_trapd_http_session.py
index 478b29d..7ec89a6 100644
--- a/tests/test_trapd_http_session.py
+++ b/tests/test_trapd_http_session.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019-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.
@@ -17,20 +17,21 @@
import pytest
import unittest
import trapd_http_session
-
+
+
class test_init_session_obj(unittest.TestCase):
"""
Test the init_session_obj mod
"""
-
+
def close_nonexisting_session(self):
"""
test close of existing http session
"""
- sess="no session"
+ sess = "no session"
result = trapd_http_session.close_session_obj(sess)
self.assertEqual(result, True)
-
+
def init_session(self):
"""
test creation of http session
@@ -38,7 +39,7 @@
result = trapd_http_session.init_session_obj()
compare = str(result).startswith("<requests.sessions.Session object at")
self.assertEqual(compare, True)
-
+
def test_reset(self):
"""
test reset of existing http session
@@ -47,7 +48,7 @@
result = trapd_http_session.reset_session_obj(sess)
compare = str(result).startswith("<requests.sessions.Session object at")
self.assertEqual(compare, True)
-
+
def close_existing_session(self):
"""
test close of existing http session
@@ -55,6 +56,7 @@
sess = trapd_http_session.init_session_obj()
result = trapd_http_session.close_session_obj(sess)
self.assertEqual(result, True)
-
-if __name__ == '__main__':
+
+
+if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_trapd_io.py b/tests/test_trapd_io.py
index 3e32493..c1702aa 100644
--- a/tests/test_trapd_io.py
+++ b/tests/test_trapd_io.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019-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.
@@ -31,8 +31,10 @@
"""
Test the save_pid mod
"""
- tds.c_config = json.loads("{ \"snmptrapd\": { \"version\": \"1.4.0\", \"title\": \"ONAP SNMP Trap Receiver\" }, \"protocols\": { \"transport\": \"udp\", \"ipv4_interface\": \"0.0.0.0\", \"ipv4_port\": 6162, \"ipv6_interface\": \"::1\", \"ipv6_port\": 6162 }, \"cache\": { \"dns_cache_ttl_seconds\": 60 }, \"publisher\": { \"http_timeout_milliseconds\": 1500, \"http_retries\": 3, \"http_milliseconds_between_retries\": 750, \"http_primary_publisher\": \"true\", \"http_peer_publisher\": \"unavailable\", \"max_traps_between_publishes\": 10, \"max_milliseconds_between_publishes\": 10000 }, \"streams_publishes\": { \"sec_fault_unsecure\": { \"type\": \"message_router\", \"aaf_password\": null, \"dmaap_info\": { \"location\": \"mtl5\", \"client_id\": null, \"client_role\": null, \"topic_url\": \"http://localhost:3904/events/ONAP-COLLECTOR-SNMPTRAP\" }, \"aaf_username\": null } }, \"files\": { \"runtime_base_dir\": \"/tmp/opt/app/snmptrap\", \"log_dir\": \"logs\", \"data_dir\": \"data\", \"pid_dir\": \"tmp\", \"arriving_traps_log\": \"snmptrapd_arriving_traps.log\", \"snmptrapd_diag\": \"snmptrapd_prog_diag.log\", \"traps_stats_log\": \"snmptrapd_stats.csv\", \"perm_status_file\": \"snmptrapd_status.log\", \"eelf_base_dir\": \"/tmp/opt/app/snmptrap/logs\", \"eelf_error\": \"error.log\", \"eelf_debug\": \"debug.log\", \"eelf_audit\": \"audit.log\", \"eelf_metrics\": \"metrics.log\", \"roll_frequency\": \"day\", \"minimum_severity_to_log\": 2 }, \"trap_config\": { \"sw_interval_in_seconds\": 60, \"notify_oids\": { \".1.3.6.1.4.1.9.0.1\": { \"sw_high_water_in_interval\": 102, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.2\": { \"sw_high_water_in_interval\": 101, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.3\": { \"sw_high_water_in_interval\": 102, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.4\": { \"sw_high_water_in_interval\": 10, \"sw_low_water_in_interval\": 3, \"category\": \"logonly\" } } }, \"snmpv3_config\": { \"usm_users\": [ { \"user\": \"usr-sha-aes256\", \"engineId\": \"8000000001020304\", \"usmHMACSHAAuth\": \"authkey1\", \"usmAesCfb256\": \"privkey1\" }, { \"user\": \"user1\", \"engineId\": \"8000000000000001\", \"usmHMACMD5Auth\": \"authkey1\", \"usmDESPriv\": \"privkey1\" }, { \"user\": \"user2\", \"engineId\": \"8000000000000002\", \"usmHMACSHAAuth\": \"authkey2\", \"usmAesCfb128\": \"privkey2\" }, { \"user\": \"user3\", \"engineId\": \"8000000000000003\", \"usmHMACSHAAuth\": \"authkey3\", \"usmAesCfb256\": \"privkey3\" } ] } }")
+ tds.c_config = json.loads(
+ '{ "snmptrapd": { "version": "1.4.0", "title": "ONAP SNMP Trap Receiver" }, "protocols": { "transport": "udp", "ipv4_interface": "0.0.0.0", "ipv4_port": 6162, "ipv6_interface": "::1", "ipv6_port": 6162 }, "cache": { "dns_cache_ttl_seconds": 60 }, "publisher": { "http_timeout_milliseconds": 1500, "http_retries": 3, "http_milliseconds_between_retries": 750, "http_primary_publisher": "true", "http_peer_publisher": "unavailable", "max_traps_between_publishes": 10, "max_milliseconds_between_publishes": 10000 }, "streams_publishes": { "sec_fault_unsecure": { "type": "message_router", "aaf_password": null, "dmaap_info": { "location": "mtl5", "client_id": null, "client_role": null, "topic_url": "http://localhost:3904/events/ONAP-COLLECTOR-SNMPTRAP" }, "aaf_username": null } }, "files": { "runtime_base_dir": "/tmp/opt/app/snmptrap", "log_dir": "logs", "data_dir": "data", "pid_dir": "tmp", "arriving_traps_log": "snmptrapd_arriving_traps.log", "snmptrapd_diag": "snmptrapd_prog_diag.log", "traps_stats_log": "snmptrapd_stats.csv", "perm_status_file": "snmptrapd_status.log", "eelf_base_dir": "/tmp/opt/app/snmptrap/logs", "eelf_error": "error.log", "eelf_debug": "debug.log", "eelf_audit": "audit.log", "eelf_metrics": "metrics.log", "roll_frequency": "day", "minimum_severity_to_log": 2 }, "trap_config": { "sw_interval_in_seconds": 60, "notify_oids": { ".1.3.6.1.4.1.9.0.1": { "sw_high_water_in_interval": 102, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.2": { "sw_high_water_in_interval": 101, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.3": { "sw_high_water_in_interval": 102, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.4": { "sw_high_water_in_interval": 10, "sw_low_water_in_interval": 3, "category": "logonly" } } }, "snmpv3_config": { "usm_users": [ { "user": "usr-sha-aes256", "engineId": "8000000001020304", "usmHMACSHAAuth": "authkey1", "usmAesCfb256": "privkey1" }, { "user": "user1", "engineId": "8000000000000001", "usmHMACMD5Auth": "authkey1", "usmDESPriv": "privkey1" }, { "user": "user2", "engineId": "8000000000000002", "usmHMACSHAAuth": "authkey2", "usmAesCfb128": "privkey2" }, { "user": "user3", "engineId": "8000000000000003", "usmHMACSHAAuth": "authkey3", "usmAesCfb256": "privkey3" } ] } }'
+ )
def test_roll_all_files_notopen(self):
"""
@@ -48,9 +50,9 @@
"""
Test bad error file location
"""
-
+
# open eelf error logs
- tds.c_config['files.eelf_error']="/bad_dir/error.log"
+ tds.c_config["files.eelf_error"] = "/bad_dir/error.log"
# try to open file in non-existent dir
with pytest.raises(SystemExit) as pytest_wrapped_exception:
@@ -61,9 +63,9 @@
"""
Test bad debug file location
"""
-
+
# open eelf debug logs
- tds.c_config['files.eelf_debug']="/bad_dir/debug.log"
+ tds.c_config["files.eelf_debug"] = "/bad_dir/debug.log"
# try to open file in non-existent dir
with pytest.raises(SystemExit) as pytest_wrapped_exception:
@@ -74,9 +76,9 @@
"""
Test bad audit file location
"""
-
+
# open eelf debug logs
- tds.c_config['files.eelf_audit']="/bad_dir/audit.log"
+ tds.c_config["files.eelf_audit"] = "/bad_dir/audit.log"
# try to open file in non-existent dir
with pytest.raises(SystemExit) as pytest_wrapped_exception:
@@ -87,9 +89,9 @@
"""
Test bad metrics file location
"""
-
+
# open eelf debug logs
- tds.c_config['files.eelf_metrics']="/bad_dir/metrics.log"
+ tds.c_config["files.eelf_metrics"] = "/bad_dir/metrics.log"
# try to open file in non-existent dir
with pytest.raises(SystemExit) as pytest_wrapped_exception:
@@ -100,7 +102,7 @@
"""
Test roll of logs when not open
"""
-
+
# try to roll logs when not open
with pytest.raises(SystemExit) as pytest_wrapped_exception:
result = trapd_io.roll_all_logs()
@@ -110,7 +112,7 @@
"""
Test roll of individual file when not present
"""
-
+
# try to roll logs when not open
result = trapd_io.roll_file("/file/not/present")
self.assertEqual(result, False)
@@ -119,22 +121,22 @@
"""
try to roll logs when not enough perms
"""
-
- no_perms_dir="/tmp/opt/app/snmptrap/no_perms"
- no_perms_file="test.dat"
- no_perms_fp= no_perms_dir + "/" + no_perms_file
+
+ no_perms_dir = "/tmp/opt/app/snmptrap/no_perms"
+ no_perms_file = "test.dat"
+ no_perms_fp = no_perms_dir + "/" + no_perms_file
# required directory tree
try:
Path(no_perms_dir).mkdir(parents=True, exist_ok=True)
- os.chmod(no_perms_dir,0o777)
+ os.chmod(no_perms_dir, 0o777)
except Exception as e:
print("Error while running %s : %s" % (os.path.basename(__file__), str(e.strerror)))
sys.exit(1)
# create empty file
- open(no_perms_fp,'a').close()
- os.chmod(no_perms_dir,0o444)
+ open(no_perms_fp, "a").close()
+ os.chmod(no_perms_dir, 0o444)
result = trapd_io.roll_file(no_perms_fp)
self.assertEqual(result, False)
@@ -144,7 +146,6 @@
result = trapd_io.open_file(no_perms_fp)
assert pytest_wrapped_exception.type == SystemExit
-
def test_open_file_exists(self):
"""
Test file open in directory present
@@ -152,7 +153,7 @@
# create copy of snmptrapd.json for pytest
test_file = "/tmp/snmptrap_pytest"
-
+
# try to roll logs when not open
result = trapd_io.open_file(test_file)
compare = str(result).startswith("<_io.TextIOWrapper name=")
@@ -165,7 +166,7 @@
# create copy of snmptrapd.json for pytest
test_file = "/tmp/no_such_dir/snmptrap_pytest"
-
+
# try to open file when dir not present
with pytest.raises(SystemExit) as pytest_wrapped_exception:
result = trapd_io.open_file(test_file)
@@ -179,14 +180,14 @@
# create copy of snmptrapd.json for pytest
test_file_name = "/tmp/snmptrap_pytest"
test_file = trapd_io.open_file(test_file_name)
-
+
# close active file
result = trapd_io.close_file(test_file, test_file_name)
self.assertEqual(result, True)
def test_close_file_does_not_exists(self):
"""
- Test closing non-existent file
+ Test closing non-existent file
"""
# try to roll logs when not open
@@ -194,5 +195,5 @@
self.assertEqual(result, False)
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_trapd_runtime_pid.py b/tests/test_trapd_runtime_pid.py
index e22b6cb..c6b8601 100644
--- a/tests/test_trapd_runtime_pid.py
+++ b/tests/test_trapd_runtime_pid.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019-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,48 +18,50 @@
import unittest
import trapd_runtime_pid
import trapd_io
-
+
+
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')
+ 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
"""
-
+
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.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)
-
-
-if __name__ == '__main__':
+
+
+if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_trapd_settings.py b/tests/test_trapd_settings.py
index 9d5cee2..92a3144 100644
--- a/tests/test_trapd_settings.py
+++ b/tests/test_trapd_settings.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019-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,16 +18,16 @@
import unittest
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"
import trapd_settings as tds
+
class test_cleanup_and_exit(unittest.TestCase):
"""
Test for presense of required vars
"""
-
def test_nonexistent_dict(self):
"""
@@ -41,7 +41,7 @@
result = False
self.assertEqual(result, False)
-
+
def test_config_dict(self):
"""
Test config dict
@@ -54,10 +54,10 @@
result = False
self.assertEqual(result, True)
-
+
def test_dns_cache_ip_to_name(self):
"""
- Test dns cache name dict
+ Test dns cache name dict
"""
tds.init()
@@ -71,7 +71,7 @@
def test_dns_cache_ip_expires(self):
"""
- Test dns cache ip expires dict
+ Test dns cache ip expires dict
"""
tds.init()
@@ -83,6 +83,7 @@
self.assertEqual(result, True)
-if __name__ == '__main__':
+
+if __name__ == "__main__":
# tds.init()
unittest.main()
diff --git a/tests/test_trapd_snmpv3.py b/tests/test_trapd_snmpv3.py
index 478f479..eac6082 100644
--- a/tests/test_trapd_snmpv3.py
+++ b/tests/test_trapd_snmpv3.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019-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.
@@ -26,7 +26,8 @@
import trapd_snmpv3
from pysnmp.entity import engine, config
-
+
+
class test_snmpv3_config(unittest.TestCase):
"""
Test snmpv3 module
@@ -36,7 +37,9 @@
"""
Test that snmpv3 config is present
"""
- tds.c_config = json.loads("{ \"snmptrapd\": { \"version\": \"1.4.0\", \"title\": \"ONAP SNMP Trap Receiver\" }, \"protocols\": { \"transport\": \"udp\", \"ipv4_interface\": \"0.0.0.0\", \"ipv4_port\": 6162, \"ipv6_interface\": \"::1\", \"ipv6_port\": 6162 }, \"cache\": { \"dns_cache_ttl_seconds\": 60 }, \"publisher\": { \"http_timeout_milliseconds\": 1500, \"http_retries\": 3, \"http_milliseconds_between_retries\": 750, \"http_primary_publisher\": \"true\", \"http_peer_publisher\": \"unavailable\", \"max_traps_between_publishes\": 10, \"max_milliseconds_between_publishes\": 10000 }, \"streams_publishes\": { \"sec_fault_unsecure\": { \"type\": \"message_router\", \"aaf_password\": null, \"dmaap_info\": { \"location\": \"mtl5\", \"client_id\": null, \"client_role\": null, \"topic_url\": \"http://localhost:3904/events/ONAP-COLLECTOR-SNMPTRAP\" }, \"aaf_username\": null } }, \"files\": { \"runtime_base_dir\": \"/tmp/opt/app/snmptrap\", \"log_dir\": \"logs\", \"data_dir\": \"data\", \"pid_dir\": \"tmp\", \"arriving_traps_log\": \"snmptrapd_arriving_traps.log\", \"snmptrapd_diag\": \"snmptrapd_prog_diag.log\", \"traps_stats_log\": \"snmptrapd_stats.csv\", \"perm_status_file\": \"snmptrapd_status.log\", \"eelf_base_dir\": \"/tmp/opt/app/snmptrap/logs\", \"eelf_error\": \"error.log\", \"eelf_debug\": \"debug.log\", \"eelf_audit\": \"audit.log\", \"eelf_metrics\": \"metrics.log\", \"roll_frequency\": \"day\", \"minimum_severity_to_log\": 2 }, \"trap_config\": { \"sw_interval_in_seconds\": 60, \"notify_oids\": { \".1.3.6.1.4.1.9.0.1\": { \"sw_high_water_in_interval\": 102, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.2\": { \"sw_high_water_in_interval\": 101, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.3\": { \"sw_high_water_in_interval\": 102, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.4\": { \"sw_high_water_in_interval\": 10, \"sw_low_water_in_interval\": 3, \"category\": \"logonly\" } } }, \"snmpv3_config\": { \"usm_users\": [ { \"user\": \"usr-sha-aes256\", \"engineId\": \"8000000001020304\", \"usmHMACSHAAuth\": \"authkey1\", \"usmAesCfb256\": \"privkey1\" }, { \"user\": \"user1\", \"engineId\": \"8000000000000001\", \"usmHMACMD5Auth\": \"authkey1\", \"usmDESPriv\": \"privkey1\" }, { \"user\": \"user2\", \"engineId\": \"8000000000000002\", \"usmHMACSHAAuth\": \"authkey2\", \"usmAesCfb128\": \"privkey2\" }, { \"user\": \"user3\", \"engineId\": \"8000000000000003\", \"usmHMACSHAAuth\": \"authkey3\", \"usmAesCfb256\": \"privkey3\" }, { \"user\": \"user4\"} , { \"engineId\": \"1\"}, { \"user\": \"user6\", \"engineId\": \"8000000000000006\", \"usmNoAuth\": \"authkey3\", \"usmAesCfb192\": \"privkey6\" }, { \"user\": \"user7\", \"engineId\": \"8000000000000007\", \"usmNoAuth\": \"\", \"usmNoPriv\": \"\" }] } }")
+ tds.c_config = json.loads(
+ '{ "snmptrapd": { "version": "1.4.0", "title": "ONAP SNMP Trap Receiver" }, "protocols": { "transport": "udp", "ipv4_interface": "0.0.0.0", "ipv4_port": 6162, "ipv6_interface": "::1", "ipv6_port": 6162 }, "cache": { "dns_cache_ttl_seconds": 60 }, "publisher": { "http_timeout_milliseconds": 1500, "http_retries": 3, "http_milliseconds_between_retries": 750, "http_primary_publisher": "true", "http_peer_publisher": "unavailable", "max_traps_between_publishes": 10, "max_milliseconds_between_publishes": 10000 }, "streams_publishes": { "sec_fault_unsecure": { "type": "message_router", "aaf_password": null, "dmaap_info": { "location": "mtl5", "client_id": null, "client_role": null, "topic_url": "http://localhost:3904/events/ONAP-COLLECTOR-SNMPTRAP" }, "aaf_username": null } }, "files": { "runtime_base_dir": "/tmp/opt/app/snmptrap", "log_dir": "logs", "data_dir": "data", "pid_dir": "tmp", "arriving_traps_log": "snmptrapd_arriving_traps.log", "snmptrapd_diag": "snmptrapd_prog_diag.log", "traps_stats_log": "snmptrapd_stats.csv", "perm_status_file": "snmptrapd_status.log", "eelf_base_dir": "/tmp/opt/app/snmptrap/logs", "eelf_error": "error.log", "eelf_debug": "debug.log", "eelf_audit": "audit.log", "eelf_metrics": "metrics.log", "roll_frequency": "day", "minimum_severity_to_log": 2 }, "trap_config": { "sw_interval_in_seconds": 60, "notify_oids": { ".1.3.6.1.4.1.9.0.1": { "sw_high_water_in_interval": 102, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.2": { "sw_high_water_in_interval": 101, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.3": { "sw_high_water_in_interval": 102, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.4": { "sw_high_water_in_interval": 10, "sw_low_water_in_interval": 3, "category": "logonly" } } }, "snmpv3_config": { "usm_users": [ { "user": "usr-sha-aes256", "engineId": "8000000001020304", "usmHMACSHAAuth": "authkey1", "usmAesCfb256": "privkey1" }, { "user": "user1", "engineId": "8000000000000001", "usmHMACMD5Auth": "authkey1", "usmDESPriv": "privkey1" }, { "user": "user2", "engineId": "8000000000000002", "usmHMACSHAAuth": "authkey2", "usmAesCfb128": "privkey2" }, { "user": "user3", "engineId": "8000000000000003", "usmHMACSHAAuth": "authkey3", "usmAesCfb256": "privkey3" }, { "user": "user4"} , { "engineId": "1"}, { "user": "user6", "engineId": "8000000000000006", "usmNoAuth": "authkey3", "usmAesCfb192": "privkey6" }, { "user": "user7", "engineId": "8000000000000007", "usmNoAuth": "", "usmNoPriv": "" }] } }'
+ )
# del os.environ['CBS_SIM_JSON']
# result = trapd_get_cbs_config.get_cbs_config()
@@ -51,12 +54,13 @@
assert pytest_wrapped_sys_exit.type == SystemExit
# assert pytest_wrapped_sys_exit.value.code == 1
-
def test_v3_config_not_present(self):
"""
Test that app is ok if v3 config not present
"""
- tds.c_config = json.loads("{ \"snmptrapd\": { \"version\": \"1.4.0\", \"title\": \"ONAP SNMP Trap Receiver\" }, \"protocols\": { \"transport\": \"udp\", \"ipv4_interface\": \"0.0.0.0\", \"ipv4_port\": 6162, \"ipv6_interface\": \"::1\", \"ipv6_port\": 6162 }, \"cache\": { \"dns_cache_ttl_seconds\": 60 }, \"publisher\": { \"http_timeout_milliseconds\": 1500, \"http_retries\": 3, \"http_milliseconds_between_retries\": 750, \"http_primary_publisher\": \"true\", \"http_peer_publisher\": \"unavailable\", \"max_traps_between_publishes\": 10, \"max_milliseconds_between_publishes\": 10000 }, \"streams_publishes\": { \"sec_fault_unsecure\": { \"type\": \"message_router\", \"aaf_password\": null, \"dmaap_info\": { \"location\": \"mtl5\", \"client_id\": null, \"client_role\": null, \"topic_url\": \"http://localhost:3904/events/ONAP-COLLECTOR-SNMPTRAP\" }, \"aaf_username\": null } }, \"files\": { \"runtime_base_dir\": \"/tmp/opt/app/snmptrap\", \"log_dir\": \"logs\", \"data_dir\": \"data\", \"pid_dir\": \"tmp\", \"arriving_traps_log\": \"snmptrapd_arriving_traps.log\", \"snmptrapd_diag\": \"snmptrapd_prog_diag.log\", \"traps_stats_log\": \"snmptrapd_stats.csv\", \"perm_status_file\": \"snmptrapd_status.log\", \"eelf_base_dir\": \"/tmp/opt/app/snmptrap/logs\", \"eelf_error\": \"error.log\", \"eelf_debug\": \"debug.log\", \"eelf_audit\": \"audit.log\", \"eelf_metrics\": \"metrics.log\", \"roll_frequency\": \"day\", \"minimum_severity_to_log\": 2 }, \"trap_config\": { \"sw_interval_in_seconds\": 60, \"notify_oids\": { \".1.3.6.1.4.1.9.0.1\": { \"sw_high_water_in_interval\": 102, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.2\": { \"sw_high_water_in_interval\": 101, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.3\": { \"sw_high_water_in_interval\": 102, \"sw_low_water_in_interval\": 7, \"category\": \"logonly\" }, \".1.3.6.1.4.1.9.0.4\": { \"sw_high_water_in_interval\": 10, \"sw_low_water_in_interval\": 3, \"category\": \"logonly\" } } } }")
+ tds.c_config = json.loads(
+ '{ "snmptrapd": { "version": "1.4.0", "title": "ONAP SNMP Trap Receiver" }, "protocols": { "transport": "udp", "ipv4_interface": "0.0.0.0", "ipv4_port": 6162, "ipv6_interface": "::1", "ipv6_port": 6162 }, "cache": { "dns_cache_ttl_seconds": 60 }, "publisher": { "http_timeout_milliseconds": 1500, "http_retries": 3, "http_milliseconds_between_retries": 750, "http_primary_publisher": "true", "http_peer_publisher": "unavailable", "max_traps_between_publishes": 10, "max_milliseconds_between_publishes": 10000 }, "streams_publishes": { "sec_fault_unsecure": { "type": "message_router", "aaf_password": null, "dmaap_info": { "location": "mtl5", "client_id": null, "client_role": null, "topic_url": "http://localhost:3904/events/ONAP-COLLECTOR-SNMPTRAP" }, "aaf_username": null } }, "files": { "runtime_base_dir": "/tmp/opt/app/snmptrap", "log_dir": "logs", "data_dir": "data", "pid_dir": "tmp", "arriving_traps_log": "snmptrapd_arriving_traps.log", "snmptrapd_diag": "snmptrapd_prog_diag.log", "traps_stats_log": "snmptrapd_stats.csv", "perm_status_file": "snmptrapd_status.log", "eelf_base_dir": "/tmp/opt/app/snmptrap/logs", "eelf_error": "error.log", "eelf_debug": "debug.log", "eelf_audit": "audit.log", "eelf_metrics": "metrics.log", "roll_frequency": "day", "minimum_severity_to_log": 2 }, "trap_config": { "sw_interval_in_seconds": 60, "notify_oids": { ".1.3.6.1.4.1.9.0.1": { "sw_high_water_in_interval": 102, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.2": { "sw_high_water_in_interval": 101, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.3": { "sw_high_water_in_interval": 102, "sw_low_water_in_interval": 7, "category": "logonly" }, ".1.3.6.1.4.1.9.0.4": { "sw_high_water_in_interval": 10, "sw_low_water_in_interval": 3, "category": "logonly" } } } }'
+ )
# del os.environ['CBS_SIM_JSON']
# result = trapd_get_cbs_config.get_cbs_config()
@@ -71,6 +75,6 @@
assert pytest_wrapped_sys_exit.type == SystemExit
# assert pytest_wrapped_sys_exit.value.code == 1
-
-if __name__ == '__main__':
+
+if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_trapd_stormwatch.py b/tests/test_trapd_stormwatch.py
index 6463a0c..16b0a17 100644
--- a/tests/test_trapd_stormwatch.py
+++ b/tests/test_trapd_stormwatch.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.
@@ -19,15 +19,15 @@
import trapd_exit
import time
-import trapd_stormwatch as sw
-import trapd_stormwatch_settings as sws
+import trapd_stormwatch as sw
+import trapd_stormwatch_settings as sws
import trapd_stats_settings as stats
+
class test_cleanup_and_exit(unittest.TestCase):
"""
Test for presense of required vars
"""
-
def test_increment_existing_counter(self):
"""
@@ -36,7 +36,7 @@
sw.sw_init()
stats.init()
- oid=".1.2.3.4.5.6"
+ oid = ".1.2.3.4.5.6"
sws.sw_config_oid_dict[oid] = True
sws.sw_config_low_water_in_interval_dict[oid] = 1
sws.sw_config_high_water_in_interval_dict[oid] = 10
@@ -79,7 +79,7 @@
# now make sure we get an exception
sws.sw_config_category = 3
self.assertFalse(sw.sw_clear_dicts())
-
+
# clean up the attributes we added above
delattr(sws, "sw_storm_counter_dict")
delattr(stats, "agent_counter_dict")
@@ -95,12 +95,12 @@
stats.total_notifications = 3
stats.total_notifications = 50
sws.sw_interval_in_seconds = 30
- stats.agent_counter_dict = { "a": 3, "b": 40 }
+ stats.agent_counter_dict = {"a": 3, "b": 40}
stats.metric_log_notification_threshold_pct = 30
sw.sw_log_metrics()
# make sure we got this far
- assert(True)
+ assert True
def test_sw_storm_active(self):
"""
@@ -160,9 +160,10 @@
stats.oid_counter_dict = {}
stats.oid_counter_dict["abc"] = 5
self.assertTrue(sw.sw_storm_active(loc_agent, loc_oid))
- self.assertTrue(not hasattr(sws,"oid_counter_dict"))
+ self.assertTrue(not hasattr(sws, "oid_counter_dict"))
# .get("abc") != None)
-if __name__ == '__main__':
+
+if __name__ == "__main__":
# sws.init()
unittest.main()
diff --git a/tests/test_trapd_stormwatch_settings.py b/tests/test_trapd_stormwatch_settings.py
index ba8ddb0..bcb04a7 100644
--- a/tests/test_trapd_stormwatch_settings.py
+++ b/tests/test_trapd_stormwatch_settings.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,16 +18,16 @@
import unittest
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"
-import trapd_stormwatch_settings as sws
+import trapd_stormwatch_settings as sws
+
class test_cleanup_and_exit(unittest.TestCase):
"""
Test for presense of required vars
"""
-
def test_nonexistent_dict(self):
"""
@@ -41,7 +41,7 @@
result = False
self.assertEqual(result, False)
-
+
def test_storm_counter_dict(self):
"""
Test storm_counter_dict
@@ -54,7 +54,7 @@
result = False
self.assertEqual(result, True)
-
+
def test_storm_active_dict(self):
"""
Test storm_active_dict
@@ -153,6 +153,7 @@
self.assertEqual(result, True)
-if __name__ == '__main__':
+
+if __name__ == "__main__":
# sws.init()
unittest.main()
diff --git a/tests/test_trapd_vb_types.py b/tests/test_trapd_vb_types.py
index 1cf75ab..5792b8c 100644
--- a/tests/test_trapd_vb_types.py
+++ b/tests/test_trapd_vb_types.py
@@ -1,5 +1,5 @@
# ============LICENSE_START=======================================================
-# Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019-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.
@@ -26,13 +26,22 @@
import trapd_vb_types
from pysnmp.entity import engine, config
-
+
+
class test_trapd_vb_types(unittest.TestCase):
"""
Test snmpv3 module
"""
- good_varbind_types = ["Integer", "Unsigned32", "Counter32", "OctetString", "ObjectIdentifier", "TimeTicks", "IpAddress"]
+ good_varbind_types = [
+ "Integer",
+ "Unsigned32",
+ "Counter32",
+ "OctetString",
+ "ObjectIdentifier",
+ "TimeTicks",
+ "IpAddress",
+ ]
def trapd_vb_type_conversion_integer(self):
"""
@@ -139,6 +148,6 @@
# should return default of octet if not defined
self.assertEqual(result, "octet")
-
-if __name__ == '__main__':
+
+if __name__ == "__main__":
unittest.main()