class TestSyslog(VppTestCase):
- """ Syslog Protocol Test Cases """
+ """Syslog Protocol Test Cases"""
@property
def SYSLOG_SEVERITY(self):
super(TestSyslog, cls).setUpClass()
try:
- cls.pg0, = cls.create_pg_interfaces(range(1))
+ (cls.pg0,) = cls.create_pg_interfaces(range(1))
cls.pg0.admin_up()
cls.pg0.config_ip4()
cls.pg0.resolve_arp()
def tearDownClass(cls):
super(TestSyslog, cls).tearDownClass()
- def syslog_generate(self, facility, severity, appname, msgid, sd=None,
- msg=None):
+ def syslog_generate(self, facility, severity, appname, msgid, sd=None, msg=None):
"""
Generate syslog message
:param sd: structured data (optional)
:param msg: free-form message (optional)
"""
- facility_str = ['kernel', 'user-level', 'mail-system',
- 'system-daemons', 'security-authorization', 'syslogd',
- 'line-printer', 'network-news', 'uucp', 'clock-daemon',
- '', 'ftp-daemon', 'ntp-subsystem', 'log-audit',
- 'log-alert', '', 'local0', 'local1', 'local2',
- 'local3', 'local4', 'local5', 'local6', 'local7']
-
- severity_str = ['emergency', 'alert', 'critical', 'error', 'warning',
- 'notice', 'informational', 'debug']
-
- cli_str = "test syslog %s %s %s %s" % (facility_str[facility],
- severity_str[severity],
- appname,
- msgid)
+ facility_str = [
+ "kernel",
+ "user-level",
+ "mail-system",
+ "system-daemons",
+ "security-authorization",
+ "syslogd",
+ "line-printer",
+ "network-news",
+ "uucp",
+ "clock-daemon",
+ "",
+ "ftp-daemon",
+ "ntp-subsystem",
+ "log-audit",
+ "log-alert",
+ "",
+ "local0",
+ "local1",
+ "local2",
+ "local3",
+ "local4",
+ "local5",
+ "local6",
+ "local7",
+ ]
+
+ severity_str = [
+ "emergency",
+ "alert",
+ "critical",
+ "error",
+ "warning",
+ "notice",
+ "informational",
+ "debug",
+ ]
+
+ cli_str = "test syslog %s %s %s %s" % (
+ facility_str[facility],
+ severity_str[severity],
+ appname,
+ msgid,
+ )
if sd is not None:
for sd_id, sd_params in sd.items():
cli_str += " sd-id %s" % (sd_id)
cli_str += " %s" % (msg)
self.vapi.cli(cli_str)
- def syslog_verify(self, data, facility, severity, appname, msgid, sd=None,
- msg=None):
+ def syslog_verify(
+ self, data, facility, severity, appname, msgid, sd=None, msg=None
+ ):
"""
Verify syslog message
:param sd: structured data (optional)
:param msg: free-form message (optional)
"""
- message = data.decode('utf-8')
+ message = data.decode("utf-8")
if sd is None:
sd = {}
try:
self.assertEqual(message.hostname, self.pg0.local_ip4)
def test_syslog(self):
- """ Syslog Protocol test """
- self.vapi.syslog_set_sender(src_address=self.pg0.local_ip4,
- collector_address=self.pg0.remote_ip4)
+ """Syslog Protocol test"""
+ self.vapi.syslog_set_sender(
+ src_address=self.pg0.local_ip4, collector_address=self.pg0.remote_ip4
+ )
config = self.vapi.syslog_get_sender()
- self.assertEqual(str(config.collector_address),
- self.pg0.remote_ip4)
+ self.assertEqual(str(config.collector_address), self.pg0.remote_ip4)
self.assertEqual(config.collector_port, 514)
self.assertEqual(str(config.src_address), self.pg0.local_ip4)
self.assertEqual(config.vrf_id, 0)
self.assertEqual(config.max_msg_size, 480)
- appname = 'test'
- msgid = 'testMsg'
- msg = 'this is message'
- sd1 = {'exampleSDID@32473': {'iut': '3',
- 'eventSource': 'App',
- 'eventID': '1011'}}
- sd2 = {'exampleSDID@32473': {'iut': '3',
- 'eventSource': 'App',
- 'eventID': '1011'},
- 'examplePriority@32473': {'class': 'high'}}
+ appname = "test"
+ msgid = "testMsg"
+ msg = "this is message"
+ sd1 = {
+ "exampleSDID@32473": {"iut": "3", "eventSource": "App", "eventID": "1011"}
+ }
+ sd2 = {
+ "exampleSDID@32473": {"iut": "3", "eventSource": "App", "eventID": "1011"},
+ "examplePriority@32473": {"class": "high"},
+ }
self.pg_enable_capture(self.pg_interfaces)
- self.syslog_generate(SyslogFacility.local7,
- SyslogSeverity.info,
- appname,
- msgid,
- None,
- msg)
+ self.syslog_generate(
+ SyslogFacility.local7, SyslogSeverity.info, appname, msgid, None, msg
+ )
capture = self.pg0.get_capture(1)
try:
self.assertEqual(capture[0][IP].src, self.pg0.local_ip4)
except:
self.logger.error(ppp("invalid packet:", capture[0]))
raise
- self.syslog_verify(capture[0][Raw].load,
- SyslogFacility.local7,
- SyslogSeverity.info,
- appname,
- msgid,
- None,
- msg)
+ self.syslog_verify(
+ capture[0][Raw].load,
+ SyslogFacility.local7,
+ SyslogSeverity.info,
+ appname,
+ msgid,
+ None,
+ msg,
+ )
self.pg_enable_capture(self.pg_interfaces)
- self.vapi.syslog_set_filter(
- self.SYSLOG_SEVERITY.SYSLOG_API_SEVERITY_WARN)
+ self.vapi.syslog_set_filter(self.SYSLOG_SEVERITY.SYSLOG_API_SEVERITY_WARN)
filter = self.vapi.syslog_get_filter()
- self.assertEqual(filter.severity,
- self.SYSLOG_SEVERITY.SYSLOG_API_SEVERITY_WARN)
- self.syslog_generate(SyslogFacility.local7,
- SyslogSeverity.info,
- appname,
- msgid,
- None,
- msg)
+ self.assertEqual(filter.severity, self.SYSLOG_SEVERITY.SYSLOG_API_SEVERITY_WARN)
+ self.syslog_generate(
+ SyslogFacility.local7, SyslogSeverity.info, appname, msgid, None, msg
+ )
self.pg0.assert_nothing_captured()
self.pg_enable_capture(self.pg_interfaces)
- self.syslog_generate(SyslogFacility.local6,
- SyslogSeverity.warning,
- appname,
- msgid,
- sd1,
- msg)
+ self.syslog_generate(
+ SyslogFacility.local6, SyslogSeverity.warning, appname, msgid, sd1, msg
+ )
capture = self.pg0.get_capture(1)
- self.syslog_verify(capture[0][Raw].load,
- SyslogFacility.local6,
- SyslogSeverity.warning,
- appname,
- msgid,
- sd1,
- msg)
-
- self.vapi.syslog_set_sender(self.pg0.local_ip4,
- self.pg0.remote_ip4,
- collector_port=12345)
+ self.syslog_verify(
+ capture[0][Raw].load,
+ SyslogFacility.local6,
+ SyslogSeverity.warning,
+ appname,
+ msgid,
+ sd1,
+ msg,
+ )
+
+ self.vapi.syslog_set_sender(
+ self.pg0.local_ip4, self.pg0.remote_ip4, collector_port=12345
+ )
config = self.vapi.syslog_get_sender()
self.assertEqual(config.collector_port, 12345)
self.pg_enable_capture(self.pg_interfaces)
- self.syslog_generate(SyslogFacility.local5,
- SyslogSeverity.err,
- appname,
- msgid,
- sd2,
- None)
+ self.syslog_generate(
+ SyslogFacility.local5, SyslogSeverity.err, appname, msgid, sd2, None
+ )
capture = self.pg0.get_capture(1)
try:
self.assertEqual(capture[0][UDP].dport, 12345)
except:
self.logger.error(ppp("invalid packet:", capture[0]))
raise
- self.syslog_verify(capture[0][Raw].load,
- SyslogFacility.local5,
- SyslogSeverity.err,
- appname,
- msgid,
- sd2,
- None)
-
-
-if __name__ == '__main__':
+ self.syslog_verify(
+ capture[0][Raw].load,
+ SyslogFacility.local5,
+ SyslogSeverity.err,
+ appname,
+ msgid,
+ sd2,
+ None,
+ )
+
+
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff