4 from framework import VppTestCase, VppTestRunner
6 from scapy.packet import Raw
7 from scapy.layers.inet import IP, UDP
8 from vpp_papi_provider import SYSLOG_SEVERITY
9 from syslog_rfc5424_parser import SyslogMessage, ParseError
10 from syslog_rfc5424_parser.constants import SyslogFacility, SyslogSeverity
13 class TestSyslog(VppTestCase):
14 """ Syslog Protocol Test Cases """
18 super(TestSyslog, cls).setUpClass()
21 cls.pg0, = cls.create_pg_interfaces(range(1))
27 super(TestSyslog, cls).tearDownClass()
30 def syslog_generate(self, facility, severity, appname, msgid, sd=None,
33 Generate syslog message
35 :param facility: facility value
36 :param severity: severity level
37 :param appname: application name that originate message
38 :param msgid: message identifier
39 :param sd: structured data (optional)
40 :param msg: free-form message (optional)
42 facility_str = ['kernel', 'user-level', 'mail-system',
43 'system-daemons', 'security-authorization', 'syslogd',
44 'line-printer', 'network-news', 'uucp', 'clock-daemon',
45 '', 'ftp-daemon', 'ntp-subsystem', 'log-audit',
46 'log-alert', '', 'local0', 'local1', 'local2',
47 'local3', 'local4', 'local5', 'local6', 'local7']
49 severity_str = ['emergency', 'alert', 'critical', 'error', 'warning',
50 'notice', 'informational', 'debug']
52 cli_str = "test syslog %s %s %s %s" % (facility_str[facility],
53 severity_str[severity],
57 for sd_id, sd_params in sd.items():
58 cli_str += " sd-id %s" % (sd_id)
59 for name, value in sd_params.items():
60 cli_str += " sd-param %s %s" % (name, value)
62 cli_str += " %s" % (msg)
63 self.vapi.cli(cli_str)
65 def syslog_verify(self, data, facility, severity, appname, msgid, sd=None,
70 :param data: syslog message
71 :param facility: facility value
72 :param severity: severity level
73 :param appname: application name that originate message
74 :param msgid: message identifier
75 :param sd: structured data (optional)
76 :param msg: free-form message (optional)
78 message = data.decode('utf-8')
82 message = SyslogMessage.parse(message)
83 except ParseError as e:
87 self.assertEqual(message.facility, facility)
88 self.assertEqual(message.severity, severity)
89 self.assertEqual(message.appname, appname)
90 self.assertEqual(message.msgid, msgid)
91 self.assertEqual(message.msg, msg)
92 self.assertEqual(message.sd, sd)
93 self.assertEqual(message.version, 1)
94 self.assertEqual(message.hostname, self.pg0.local_ip4)
96 def test_syslog(self):
97 """ Syslog Protocol test """
98 self.vapi.syslog_set_sender(self.pg0.remote_ip4n, self.pg0.local_ip4n)
99 config = self.vapi.syslog_get_sender()
100 self.assertEqual(str(config.collector_address),
102 self.assertEqual(config.collector_port, 514)
103 self.assertEqual(str(config.src_address), self.pg0.local_ip4)
104 self.assertEqual(config.vrf_id, 0)
105 self.assertEqual(config.max_msg_size, 480)
109 msg = 'this is message'
110 sd1 = {'exampleSDID@32473': {'iut': '3',
111 'eventSource': 'App',
113 sd2 = {'exampleSDID@32473': {'iut': '3',
114 'eventSource': 'App',
116 'examplePriority@32473': {'class': 'high'}}
118 self.pg_enable_capture(self.pg_interfaces)
119 self.syslog_generate(SyslogFacility.local7,
125 capture = self.pg0.get_capture(1)
127 self.assertEqual(capture[0][IP].src, self.pg0.local_ip4)
128 self.assertEqual(capture[0][IP].dst, self.pg0.remote_ip4)
129 self.assertEqual(capture[0][UDP].dport, 514)
130 self.assert_packet_checksums_valid(capture[0], False)
132 self.logger.error(ppp("invalid packet:", capture[0]))
134 self.syslog_verify(capture[0][Raw].load,
135 SyslogFacility.local7,
142 self.pg_enable_capture(self.pg_interfaces)
143 self.vapi.syslog_set_filter(SYSLOG_SEVERITY.WARN)
144 filter = self.vapi.syslog_get_filter()
145 self.assertEqual(filter.severity, SYSLOG_SEVERITY.WARN)
146 self.syslog_generate(SyslogFacility.local7,
152 self.pg0.assert_nothing_captured()
154 self.pg_enable_capture(self.pg_interfaces)
155 self.syslog_generate(SyslogFacility.local6,
156 SyslogSeverity.warning,
161 capture = self.pg0.get_capture(1)
162 self.syslog_verify(capture[0][Raw].load,
163 SyslogFacility.local6,
164 SyslogSeverity.warning,
170 self.vapi.syslog_set_sender(self.pg0.remote_ip4n,
172 collector_port=12345)
173 config = self.vapi.syslog_get_sender()
174 self.assertEqual(config.collector_port, 12345)
176 self.pg_enable_capture(self.pg_interfaces)
177 self.syslog_generate(SyslogFacility.local5,
183 capture = self.pg0.get_capture(1)
185 self.assertEqual(capture[0][UDP].dport, 12345)
187 self.logger.error(ppp("invalid packet:", capture[0]))
189 self.syslog_verify(capture[0][Raw].load,
190 SyslogFacility.local5,
198 if __name__ == '__main__':
199 unittest.main(testRunner=VppTestRunner)