# The software in this package is distributed under the GNU General # Public License version 2 (with a special exception described below). # # A copy of GNU General Public License (GPL) is included in this distribution, # in the file COPYING.GPL. # # As a special exception, if other files instantiate templates or use macros # or inline functions from this file, or you compile this file and link it # with other works to produce a work based on this file, this file # does not by itself cause the resulting work to be covered # by the GNU General Public License. # # However the source code for this file must still be made available # in accordance with section (3) of the GNU General Public License. # # This exception does not invalidate any other reasons why a work based # on this file might be covered by the GNU General Public License. # # Copyright (c) 2016-2018 Intra2net AG """ Unittests for log_read Creates own thread to write data to a log file """ import unittest from threading import Thread from tempfile import mkstemp import os import time import logging from warnings import warn from src.log_read import * # get best clock perf_counter = time.perf_counter DEBUG = False class LogFileWriter(Thread): """ thread that creates and writes to given file """ def __init__(self, file_name, text_pattern, n_writes=None, pause_time=0.1, do_encode=None, use_logging=True): """ creates thread, daemon is True if n_writes is None, will write indefinitely; else writes text_pattern n_writes times, formatted with (counter, perf_counter) If do_encode is True, will encode text to bytes and open file handle in 'wb' mode; otherwise opens in 'wt' mode and writes unicode text. If use_logging is False, will open file and run file_handle.write; If use_logging is True, will create logger that logs to file and use logging.info (no file_handle.write) """ super(LogFileWriter, self).__init__() self.daemon = True self.file_name = file_name self.text_pattern = text_pattern self.n_writes = n_writes self.pause_time = pause_time self.do_encode = do_encode self.use_logging = use_logging def run(self): counter = 0 if self.do_encode: mode = 'wb' buffering = 0 # no buffering -- only allowed for byte mode else: mode = 'wt' buffering = 1 # line buffering -- only allowed for text mode if self.use_logging: logging.basicConfig(filename=self.file_name, level=logging.INFO, format='%(msg)s') while True: if self.n_writes is not None and counter >= self.n_writes: break self.write_and_sleep(logging.info, counter) counter += 1 else: with open(self.file_name, mode=mode, buffering=buffering) \ as file_handle: while True: if self.n_writes is not None and counter >= self.n_writes: break self.write_and_sleep(file_handle.write, counter) counter += 1 def write_and_sleep(self, write_func, counter): """ format text, write it using given function and sleep """ if isinstance(self.text_pattern, (list, tuple)): text = self.text_pattern[counter] else: text = self.text_pattern text = text.format(counter, perf_counter()) if self.do_encode: text = text.encode(self.do_encode) write_func(text) time.sleep(self.pause_time) class LogReadTester(unittest.TestCase): """ class with all the tests """ def setUp(self): """ called before each test """ if DEBUG: print('setup test') temp_handle, temp_name = mkstemp() os.close(temp_handle) self.temp_file = temp_name if DEBUG: print('created temp file ' + self.temp_file) def tearDown(self): """ called after each test """ if DEBUG: print('tear down test') if os.path.isfile(self.temp_file): if DEBUG: print('delete temp file' + self.temp_file) os.unlink(self.temp_file) def helper_test_len(self, reader, n_expected): """ helper function that tests length of vars in reader """ self.assertEqual(reader.n_sources(), n_expected) self.assertEqual(len(reader.file_objs), n_expected) self.assertEqual(len(reader.file_handles), n_expected) self.assertEqual(len(reader.descriptions), n_expected) self.assertEqual(len(reader.ignore), n_expected) self.assertEqual(len(reader.last_sizes), n_expected) def test_args(self): self.assertRaises(TypeError, IterativeReader) # no args self.assertRaises(ValueError, IterativeReader, [], 'test') self.assertRaises(ValueError, IterativeReader, [], ['test', ]) self.assertRaises(ValueError, IterativeReader, self.temp_file) self.assertRaises(ValueError, IterativeReader, [self.temp_file, ]) with open(self.temp_file, 'rt') as file_handle: reader = IterativeReader(file_handle) self.helper_test_len(reader, 1) reader = IterativeReader([file_handle, ]) self.helper_test_len(reader, 1) reader = IterativeReader(file_handle, 'desc') self.helper_test_len(reader, 1) reader = IterativeReader([file_handle, ], ['desc', ]) self.helper_test_len(reader, 1) reader = IterativeReader(file_handle, ['desc', ]) self.helper_test_len(reader, 1) self.assertRaises(ValueError, IterativeReader, [file_handle, ], 'desc', ) reader = IterativeReader([file_handle, file_handle], ['desc1', 'desc2']) self.helper_test_len(reader, 2) reader = IterativeReader((file_handle for _ in range(5))) self.helper_test_len(reader, 5) self.assertRaises(ValueError, IterativeReader, (file_handle for idx in range(5)), tuple('desc' for idx in range(4))) self.assertRaises(ValueError, IterativeReader, (file_handle for idx in range(5)), ('desc' for idx in range(6))) def test_simple_read(self): """ write fixed number of lines, see how fast they are retrieved """ # need newline only when writing text (because of write buffering) param_combinations = ('{0}:{1}\n', None, False), \ ('{0}:{1}\n', 'ascii', False), \ ('{0}:{1} ' , 'ascii', False) #('{0}:{1}\n', None , True), \ logging seems #('{0}:{1}\n', 'ascii', True), \ to buffer writes #('{0}:{1} ' , None , True), \ to files #('{0}:{1} ' , 'ascii', True) n_texts = 10 pause_time = 0.01 # 100 tps (texts per second) for text_pattern, encoding, use_logging in param_combinations: LogFileWriter(self.temp_file, text_pattern, n_writes=n_texts, pause_time=pause_time, do_encode=encoding, use_logging=use_logging).start() if DEBUG: print('testing with log file {0}'.format(self.temp_file)) print('encoding is {0}, use logging = {1}'.format(encoding, use_logging)) time_diffs = [] with open(self.temp_file, 'rt') as file_handle: reader = IterativeReader(file_handle, keep_watching=True) self.helper_test_len(reader, 1) counter = -1 # we may have to adapt this manually for desc, text, source_idx in reader: receive_time = perf_counter() self.assertEqual(desc, self.temp_file) self.assertEqual(source_idx, 0) counter += 1 text = text.strip() if DEBUG: print('{1}: received text "{0}" at {2}' .format(text, counter, receive_time)) if counter == 0 and not text: # if reader runs stat() before we write, we might get # a warning and one empty read here counter -= 1 warn('Got an empty read, you should have seen another ' 'warning about file shrinking', category=LogReadWarning) continue index = text.index(':') count_text = int(text[:index].strip()) self.assertEqual(count_text, counter) write_time = float(text[index+1:].strip()) time_diffs.append((receive_time - write_time)*1000.) if counter == n_texts-1: if DEBUG: print('stop since have {0} reads'.format(counter)) break if DEBUG: print('time diffs in ms: {0}'.format(time_diffs)) self.assertTrue(max(time_diffs) < 100., 'read took more than 100ms (max was {0:.3f}ms)!' .format(max(time_diffs))) def test_line_read(self): """ write partial lines, full lines and multiple lines """ pause_time = 0.01 # 100 tps (texts per second) encoding = None use_logging = False texts = ['line0\n', 'line1\n', 'li', 'ne2\n', 'line3\n', 'line4\nline5\n', 'li', 'ne6\nli', 'ne7\nl', 'i', 'n', 'e', '8', '\n', 'l', 'ine9\nline10\nline1', '1', '\n', '\n', '\n', 'end\n', '\nend\n', '\n\nend\n\n'] lines_expected = ['line{0}'.format(idx) for idx in range(12)] + ['', '', ''] # create writer LogFileWriter(self.temp_file, texts, n_writes=len(texts), pause_time=pause_time, do_encode=encoding, use_logging=use_logging).start() # read with open(self.temp_file, 'rt') as file_handle: reader = LineReader(file_handle, keep_watching=True) self.helper_test_len(reader, 1) for line_expected, (_, line_read, _) in zip(lines_expected, reader): if 'end' in line_read: break else: if DEBUG: print('expect "{0}", read "{1}"'.format(line_expected, line_read)) self.assertEqual(line_expected, line_read) def test_parse_syslog_lines(self): """Test parsing a few fixed lines of syslog""" syslog_lines = """ Nov 8 08:10:01 system-name rsyslogd: [origin software="rsyslogd" swVersion="8.39.0" x-pid="1968" x-info="http://www.rsyslog.com"] rsyslogd was HUPed Nov 8 08:10:12 system-name squid[2762]: Starting new redirector helpers...#012 current master transaction: master58 Nov 8 08:10:28 system-name pop30[3922]: counts: retr=<0> top=<0> dele=<0> Nov 8 08:10:43 system-name pop3loc[3923]: counts: retr=<0> top=<0> dele=<0> Nov 8 08:10:44 system-name pingcheck[3004]: Status (0 down, limit=4, notify up now): report link up? Nov 8 08:10:44 system-name connd[2917]: [subsys_pingcheck] received notification that connection for id 1 is online Nov 8 08:10:44 system-name connd[2917]: [subsys_pingcheck] Connection [ConnInfo P1 (connection name): online] was confirmed online Nov 8 08:10:44 system-name connd[2917]: [connection_manager] Confirm that provider "P1" is online Nov 8 08:11:01 system-name mon[3563]: failure for intranator dirwatchd 1667891461 (NO SUMMARY) Nov 8 08:11:04 system-name lmtpunix[3962]: Delivered: to mailbox: user.cyrus as uid: 49792 Nov 8 08:11:04 system-name lmtpunix[3962]: USAGE cyrus user: 0.000000 sys: 0.006803 Nov 8 08:11:20 system-name pop30[3982]: counts: retr=<0> top=<0> dele=<0> Oct 21 07:52:39 system-name avupdate[3671]: successfully installed new antivirus database Oct 21 07:52:39 system-name savapi_version_info[3973]: AVE=8.3.64.214, VDF=8.19.26.110, Released=21.10.2022 05:39 Oct 21 07:52:39 system-name connd[3180]: [connection_manager] online mode set to always online Oct 21 07:52:47 system-name kernel: REJECT local IN=eth0 OUT= MACSRC=68:4f:64:75:7b:82 MACDST=01:00:5e:00:00:01 MACPROTO=0800 SRC=123.45.67.89 DST=224.0.0.1 LEN=36 TOS=0x00 PREC=0x00 TTL=1 ID=34431 PROTO=ICMP TYPE=9 CODE=0 Oct 21 07:52:54 system-name lmtpunix[2715]: Delivered: to mailbox: user.cyrus as uid: 46558 Oct 21 07:52:54 system-name lmtpunix[2715]: USAGE cyrus user: 0.010564 sys: 0.000000 Oct 21 07:53:08 system-name pop30[2703]: counts: retr=<0> top=<0> dele=<0> Oct 21 07:53:08 system-name pop3loc[2701]: counts: retr=<0> top=<0> dele=<0> Oct 21 07:53:30 system-name ntpd[3701]: adjusting local clock by -0.943974s Oct 21 07:54:15 system-name keymakerd[3206]: [acmeautorenew] Starting auto renew check. @acmeautorenew#285 Oct 21 07:54:15 system-name keymakerd[3206]: [acmeautorenew] Finishing auto renew thread. @acmeautorenew#206 Oct 21 07:54:23 system-name pop3loc[2701]: counts: retr=<0> top=<0> dele=<0> Oct 21 07:54:41 system-name pop30[2703]: counts: retr=<0> top=<0> dele=<0> Oct 21 07:54:51 system-name squid[3025]: Starting new redirector helpers...#012 current master transaction: master54 Oct 21 07:54:55 system-name lmtpunix[3997]: Delivered: to mailbox: user.cyrus as uid: 46559 Oct 21 07:54:55 system-name lmtpunix[3997]: USAGE cyrus user: 0.002022 sys: 0.005926 """ lines = [line.lstrip() for line in syslog_lines.splitlines() if line.lstrip()] LogFileWriter(self.temp_file, lines, n_writes=len(lines)).start() with open(self.temp_file, 'rt') as file_handle: parser = LogParser(file_handle, SYS_LOG_PATTERN) for _, data, _ in parser: self.assertIsNotNone(data, f"Failed to parse {parser.last_unparsed_line}") def test_parse_proxy_lines(self): proxy_lines = """ 1667901672.645 0 127.0.0.1 NONE_NONE/000 0 - error:transaction-end-before-headers - HIER_NONE/- - - 1667901792.812 0 127.0.0.1 NONE_NONE/000 0 - error:transaction-end-before-headers - HIER_NONE/- - - 1667901912.997 0 127.0.0.1 NONE_NONE/000 0 - error:transaction-end-before-headers - HIER_NONE/- - - 1667902032.168 0 127.0.0.1 NONE_NONE/000 0 - error:transaction-end-before-headers - HIER_NONE/- - - 1667902055.130 571 192.168.1.1 TCP_TUNNEL/200 10403 CONNECT apod.nasa.gov:443 - HIER_DIRECT/apod.nasa.gov - - 1667902063.333 969 192.168.1.1 TCP_TUNNEL/200 174425 CONNECT www.startpage.com:443 - HIER_DIRECT/www.startpage.com - - 1667902063.352 17 192.168.1.1 TCP_TUNNEL/200 39 CONNECT www.startpage.com:443 - HIER_DIRECT/www.startpage.com - - 1667902063.539 24 192.168.1.1 TCP_TUNNEL/200 39 CONNECT www.startpage.com:443 - HIER_DIRECT/www.startpage.com - - 1667902063.545 207 192.168.1.1 TCP_TUNNEL/200 1720 CONNECT www.startpage.com:443 - HIER_DIRECT/www.startpage.com - - 1667902064.141 611 192.168.1.1 TCP_TUNNEL/200 10100 CONNECT www.startpage.com:443 - HIER_DIRECT/www.startpage.com - - 1667902072.816 974 192.168.1.1 TCP_MISS/200 981 POST http://r3.o.lencr.org/ - FIRSTUP_PARENT/127.0.0.1 application/ocsp-response - 1667902089.338 16331 192.168.1.1 TCP_TUNNEL/200 103739 CONNECT www.roma-pizza-tuebingen.de:443 - HIER_DIRECT/www.roma-pizza-tuebingen.de - - 1667902089.339 16332 192.168.1.1 TCP_TUNNEL/200 49795 CONNECT www.roma-pizza-tuebingen.de:443 - HIER_DIRECT/www.roma-pizza-tuebingen.de - - 1667902089.339 16330 192.168.1.1 TCP_TUNNEL/200 24641 CONNECT www.roma-pizza-tuebingen.de:443 - HIER_DIRECT/www.roma-pizza-tuebingen.de - - 1667902105.718 0 192.168.1.1 NONE_NONE/000 0 - error:transaction-end-before-headers - HIER_NONE/- - - 1667902152.347 0 127.0.0.1 NONE_NONE/000 0 - error:transaction-end-before-headers - HIER_NONE/- - - """ lines = [line.lstrip() for line in proxy_lines.splitlines() if line.lstrip()] LogFileWriter(self.temp_file, lines, n_writes=len(lines)).start() with open(self.temp_file, 'rt') as file_handle: parser = LogParser(file_handle, PROXY_LOG_PATTERN) for _, data, _ in parser: self.assertIsNotNone(data, f"Failed to parse {parser.last_unparsed_line}") def test_parse_maillog_lines(self): mail_lines = """ Nov 7 10:07:33 system-name amavis[1]: SA info: rules: meta test I2N_RAZOR_ADJUST_4 has dependency 'RAZOR2_CF_RANGE_E8_51_100' with a zero score Nov 7 10:07:33 system-name amavis[1]: SpamControl: init_pre_fork on SpamAssassin done Nov 7 10:07:35 system-name quarantined[3574]: quarantined version 0.15 (build Jun 28 2022 17:26:39) up and running Nov 7 10:07:52 system-name mailreconfig[3713]: called by sh[3712]|generate[3663], mode update, changed files: aliases,recipients Nov 7 10:07:53 system-name postfix/postfix-script[3724]: refreshing the Postfix mail system Nov 7 10:07:53 system-name postfix/master[2548]: reload -- version 3.3.22, configuration /etc/postfix Nov 7 10:08:07 system-name mailreconfig[3851]: called by sh[3850]|generate[3804], mode update, changed files: aliases,recipients Nov 7 10:08:07 system-name postfix/postfix-script[3862]: refreshing the Postfix mail system Nov 7 10:08:07 system-name postfix/master[2548]: reload -- version 3.3.22, configuration /etc/postfix Nov 7 10:08:11 system-name postfix/smtpd[3933]: connect from system-name.net.lan[127.0.0.1] Nov 7 10:08:11 system-name postfix/smtpd[3933]: D9C626B: client=system-name.net.lan[127.0.0.1] Nov 7 10:08:11 system-name postfix/cleanup[3936]: D9C626B: message-id=<20221107090811.D9C626B@system-name.net.lan> Nov 7 10:08:11 system-name postfix/qmgr[3868]: D9C626B: from=, size=10888, nrcpt=1 (queue active) Nov 7 10:08:11 system-name postfix/smtpd[3933]: disconnect from system-name.net.lan[127.0.0.1] ehlo=1 mail=1 rcpt=1 data=1 quit=1 commands=5 Nov 7 10:08:11 system-name postfix/smtpd[3933]: connect from system-name.net.lan[127.0.0.1] Nov 7 10:08:11 system-name postfix/smtpd[3933]: E59146D: client=system-name.net.lan[127.0.0.1] Nov 7 10:08:11 system-name postfix/cleanup[3936]: E59146D: message-id=<20221107090811.E59146D@system-name.net.lan> Nov 7 10:08:11 system-name amavis[11]: (00011-01) LMTP::10024 /var/spool/vscan/amavis/tmp/amavis-20221107T100811-00011: -> SIZE=10888 Received: from system-name.net.lan ([127.0.0.1]) by localhost (system-name.net.lan [127.0.0.1]) (amavisd-new, port 10024) with LMTP for ; Mon, 7 Nov 2022 10:08:11 +0100 (CET) Nov 7 10:08:12 system-name postfix/qmgr[3868]: E59146D: from=, size=9472, nrcpt=1 (queue active) Nov 7 10:08:12 system-name postfix/smtpd[3933]: disconnect from system-name.net.lan[127.0.0.1] ehlo=1 mail=1 rcpt=1 data=1 quit=1 commands=5 Nov 7 10:08:12 system-name postfix/smtpd[3933]: connect from system-name.net.lan[127.0.0.1] Nov 7 10:08:12 system-name amavis[11]: (00011-01) Checking: q4eVqFn169Kq [127.0.0.1] -> Nov 7 10:08:12 system-name amavis[11]: (00011-01) p.path BANNED:1 test-recipient@system-name.net.lan: "P=p003,L=1,M=multipart/mixed | P=p002,L=1/2,M=application/x-msdownload,T=exe,T=exe-ms,N=cloudcar.exe", matching_key="(?^i:\\.exe$)" Nov 7 10:08:12 system-name postfix/qmgr[3868]: 4B0CF70: from=, size=164391, nrcpt=1 (queue active) """ lines = [line.lstrip() for line in mail_lines.splitlines() if line.lstrip()] LogFileWriter(self.temp_file, lines, n_writes=len(lines)).start() with open(self.temp_file, 'rt') as file_handle: parser = LogParser(file_handle, SYS_LOG_PATTERN) for _, data, _ in parser: self.assertIsNotNone(data, f"Failed to parse {parser.last_unparsed_line}") @unittest.skipIf(not os.access('/var/log/messages', os.R_OK), "messages not accessible") def test_parse_messages(self): """Try parsing first 100 lines of messages if running on linux""" with LogParser.create_for('/var/log/messages', SYS_LOG_PATTERN) as parser: for line_count, (_, data, _) in enumerate(parser): if line_count > 100: break self.assertIsNotNone(data, f"Failed to parse {parser.last_unparsed_line}") @unittest.skipIf(not os.access('/var/log/syslog', os.R_OK), "syslog not accessible") def test_parse_syslog(self): """Try parsing first 100 lines of syslog if running on linux""" with LogParser.create_for('/var/log/syslog', SYS_LOG_PATTERN) as parser: for line_count, (_, data, _) in enumerate(parser): if line_count > 100: break self.assertIsNotNone(data, f"Failed to parse {parser.last_unparsed_line}") @unittest.skipIf(not os.access('/var/log/maillog', os.R_OK), "maillog not accessible") def test_parse_maillog(self): """Try parsing first 100 lines of maillog if running on linux""" with LogParser.create_for('/var/log/maillog', SYS_LOG_PATTERN) as parser: for line_count, (_, data, _) in enumerate(parser): if line_count > 100: break self.assertIsNotNone(data, f"Failed to parse {parser.last_unparsed_line}") @unittest.skipIf(not os.access('/var/log/squid/access.log', os.R_OK), "proxy log not accessible") def test_parse_proxy_log(self): """Try parsing first 100 lines of proxy log if running on linux""" with LogParser.create_for('/var/log/squid/access.log', PROXY_LOG_PATTERN) as parser: for line_count, (_, data, _) in enumerate(parser): if line_count > 100: break self.assertIsNotNone(data, f"Failed to parse {parser.last_unparsed_line}") if __name__ == '__main__': unittest.main()