Increase version to 1.7.4

[pyi2ncommon] / test / test_log_read.py

# 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 <info@intra2net.com>

""" 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: <cmu-lmtpd-3962-1667891464-0@system-name.m.i2n> 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: <cmu-lmtpd-2715-1666331574-0@system-name.m.i2n> 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: <cmu-lmtpd-3997-1666331695-0@system-name.m.i2n> 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=<test-sender@system-name.net.lan>, 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: <test-sender@system-name.net.lan> -> <test-recipient@system-name.net.lan> 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 <test-recipient@system-name.net.lan>; Mon,  7 Nov 2022 10:08:11 +0100 (CET)
                Nov  7 10:08:12 system-name postfix/qmgr[3868]: E59146D: from=<test-sender@system-name.net.lan>, 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] <test-sender@system-name.net.lan> -> <test-recipient@system-name.net.lan>
                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=<test-sender@system-name.net.lan>, 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()