diff options
Diffstat (limited to 'test/test_syslog.py')
| -rw-r--r-- | test/test_syslog.py | 196 |
1 files changed, 196 insertions, 0 deletions
diff --git a/test/test_syslog.py b/test/test_syslog.py new file mode 100644 index 00000000000..db7d7bef2c7 --- /dev/null +++ b/test/test_syslog.py @@ -0,0 +1,196 @@ +#!/usr/bin/env python + +from framework import VppTestCase, VppTestRunner +from util import ppp +from scapy.packet import Raw +from scapy.layers.inet import IP, UDP +from vpp_papi_provider import SYSLOG_SEVERITY +from syslog_rfc5424_parser import SyslogMessage, ParseError +from syslog_rfc5424_parser.constants import SyslogFacility, SyslogSeverity + + +class TestSyslog(VppTestCase): + """ Syslog Protocol Test Cases """ + + @classmethod + def setUpClass(cls): + super(TestSyslog, cls).setUpClass() + + try: + cls.create_pg_interfaces(range(1)) + cls.pg0.admin_up() + cls.pg0.config_ip4() + cls.pg0.resolve_arp() + + except Exception: + super(TestSyslog, cls).tearDownClass() + raise + + def syslog_generate(self, facility, severity, appname, msgid, sd=None, + msg=None): + """ + Generate syslog message + + :param facility: facility value + :param severity: severity level + :param appname: application name that originate message + :param msgid: message indetifier + :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) + if sd is not None: + for sd_id, sd_params in sd.items(): + cli_str += " sd-id %s" % (sd_id) + for name, value in sd_params.items(): + cli_str += " sd-param %s %s" % (name, value) + if msg is not None: + cli_str += " %s" % (msg) + self.vapi.cli(cli_str) + + def syslog_verify(self, data, facility, severity, appname, msgid, sd=None, + msg=None): + """ + Verify syslog message + + :param data: syslog message + :param facility: facility value + :param severity: severity level + :param appname: application name that originate message + :param msgid: message indetifier + :param sd: structured data (optional) + :param msg: free-form message (optional) + """ + message = data.decode('utf-8') + if sd is None: + sd = {} + try: + message = SyslogMessage.parse(message) + self.assertEqual(message.facility, facility) + self.assertEqual(message.severity, severity) + self.assertEqual(message.appname, appname) + self.assertEqual(message.msgid, msgid) + self.assertEqual(message.msg, msg) + self.assertEqual(message.sd, sd) + self.assertEqual(message.version, 1) + self.assertEqual(message.hostname, self.pg0.local_ip4) + except ParseError as e: + self.logger.error(e) + + def test_syslog(self): + """ Syslog Protocol test """ + self.vapi.syslog_set_sender(self.pg0.remote_ip4n, self.pg0.local_ip4n) + config = self.vapi.syslog_get_sender() + self.assertEqual(config.collector_address.address, + self.pg0.remote_ip4n) + self.assertEqual(config.collector_port, 514) + self.assertEqual(config.src_address.address, self.pg0.local_ip4n) + 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'}} + + self.pg_enable_capture(self.pg_interfaces) + 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) + self.assertEqual(capture[0][IP].dst, self.pg0.remote_ip4) + self.assertEqual(capture[0][UDP].dport, 514) + self.assert_packet_checksums_valid(capture[0], False) + 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.pg_enable_capture(self.pg_interfaces) + self.vapi.syslog_set_filter(SYSLOG_SEVERITY.WARN) + filter = self.vapi.syslog_get_filter() + self.assertEqual(filter.severity, SYSLOG_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) + 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.remote_ip4n, + self.pg0.local_ip4n, + 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) + 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__': + unittest.main(testRunner=VppTestRunner) |