[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 IterativeReader, LineReader, LogReadWarning

# 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, deamon 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 idx 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
        lines_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)


if __name__ == '__main__':
    unittest.main()
Commit	Line	Data
3237d2a6 CH	1	# The software in this package is distributed under the GNU General
	2	# Public License version 2 (with a special exception described below).
	3	#
	4	# A copy of GNU General Public License (GPL) is included in this distribution,
	5	# in the file COPYING.GPL.
	6	#
	7	# As a special exception, if other files instantiate templates or use macros
	8	# or inline functions from this file, or you compile this file and link it
	9	# with other works to produce a work based on this file, this file
	10	# does not by itself cause the resulting work to be covered
	11	# by the GNU General Public License.
	12	#
	13	# However the source code for this file must still be made available
	14	# in accordance with section (3) of the GNU General Public License.
	15	#
	16	# This exception does not invalidate any other reasons why a work based
	17	# on this file might be covered by the GNU General Public License.
f365f614 CH	18	#
f365f614 CH	19	# Copyright (c) 2016-2018 Intra2net AG <info@intra2net.com>
3237d2a6	20
e7d49180 CH	21	""" Unittests for log_read
	22
	23	Creates own thread to write data to a log file
e7d49180 CH	24	"""
	25
	26	import unittest
	27	from threading import Thread
	28	from tempfile import mkstemp
	29	import os
	30	import time
	31	import logging
ea8b01a3	32	from warnings import warn
e7d49180	33
ea8b01a3	34	from src.log_read import IterativeReader, LineReader, LogReadWarning
e7d49180	35
59281ef9	36	# get best clock
7628bc48	37	perf_counter = time.perf_counter
59281ef9	38
7c00b0af CH	39	DEBUG = False
7c00b0af CH	40
e7d49180 CH	41
	42	class LogFileWriter(Thread):
	43	""" thread that creates and writes to given file """
	44
	45	def __init__(self, file_name, text_pattern, n_writes=None,
	46	pause_time=0.1, do_encode=None, use_logging=True):
	47	""" creates thread, deamon is True
	48
	49	if n_writes is None, will write indefinitely; else writes text_pattern
01fe1580	50	n_writes times, formatted with (counter, perf_counter)
e7d49180 CH	51	If do_encode is True, will encode text to bytes and open file handle
	52	in 'wb' mode; otherwise opens in 'wt' mode and writes unicode text.
	53	If use_logging is False, will open file and run file_handle.write;
	54	If use_logging is True, will create logger that logs to file and use
	55	logging.info (no file_handle.write)
	56	"""
01fe1580 CH	57	super(LogFileWriter, self).__init__()
01fe1580 CH	58	self.daemon = True
e7d49180 CH	59	self.file_name = file_name
	60	self.text_pattern = text_pattern
	61	self.n_writes = n_writes
	62	self.pause_time = pause_time
	63	self.do_encode = do_encode
	64	self.use_logging = use_logging
	65
	66	def run(self):
	67	counter = 0
	68	if self.do_encode:
	69	mode = 'wb'
	70	buffering = 0 # no buffering -- only allowed for byte mode
	71	else:
	72	mode = 'wt'
	73	buffering = 1 # line buffering -- only allowed for text mode
	74
	75	if self.use_logging:
	76	logging.basicConfig(filename=self.file_name, level=logging.INFO,
	77	format='%(msg)s')
	78	while True:
	79	if self.n_writes is not None and counter >= self.n_writes:
	80	break
	81	self.write_and_sleep(logging.info, counter)
	82	counter += 1
	83	else:
	84	with open(self.file_name, mode=mode, buffering=buffering) \
	85	as file_handle:
	86	while True:
	87	if self.n_writes is not None and counter >= self.n_writes:
	88	break
	89	self.write_and_sleep(file_handle.write, counter)
	90	counter += 1
	91
	92	def write_and_sleep(self, write_func, counter):
d910eba5 CH	93	""" format text, write it using given function and sleep """
	94	if isinstance(self.text_pattern, (list, tuple)):
	95	text = self.text_pattern[counter]
	96	else:
	97	text = self.text_pattern
01fe1580	98	text = text.format(counter, perf_counter())
d910eba5	99
e7d49180 CH	100	if self.do_encode:
	101	text = text.encode(self.do_encode)
	102	write_func(text)
e7d49180 CH	103	time.sleep(self.pause_time)
	104
	105
	106	class LogReadTester(unittest.TestCase):
	107	""" class with all the tests """
	108
	109	def setUp(self):
	110	""" called before each test """
7c00b0af CH	111	if DEBUG:
7c00b0af CH	112	print('setup test')
e7d49180 CH	113	temp_handle, temp_name = mkstemp()
	114	os.close(temp_handle)
	115	self.temp_file = temp_name
7c00b0af CH	116	if DEBUG:
7c00b0af CH	117	print('created temp file ' + self.temp_file)
e7d49180 CH	118
	119	def tearDown(self):
	120	""" called after each test """
7c00b0af CH	121	if DEBUG:
7c00b0af CH	122	print('tear down test')
e7d49180	123	if os.path.isfile(self.temp_file):
7c00b0af CH	124	if DEBUG:
7c00b0af CH	125	print('delete temp file' + self.temp_file)
e7d49180 CH	126	os.unlink(self.temp_file)
e7d49180 CH	127
d910eba5 CH	128	def helper_test_len(self, reader, n_expected):
	129	""" helper function that tests length of vars in reader """
	130	self.assertEqual(reader.n_sources(), n_expected)
	131	self.assertEqual(len(reader.file_objs), n_expected)
3e0b3965	132	self.assertEqual(len(reader.file_handles), n_expected)
d910eba5 CH	133	self.assertEqual(len(reader.descriptions), n_expected)
	134	self.assertEqual(len(reader.ignore), n_expected)
	135	self.assertEqual(len(reader.last_sizes), n_expected)
e7d49180 CH	136
	137	def test_args(self):
	138	self.assertRaises(TypeError, IterativeReader) # no args
	139	self.assertRaises(ValueError, IterativeReader, [], 'test')
	140	self.assertRaises(ValueError, IterativeReader, [], ['test', ])
	141	self.assertRaises(ValueError, IterativeReader, self.temp_file)
	142	self.assertRaises(ValueError, IterativeReader, [self.temp_file, ])
	143	with open(self.temp_file, 'rt') as file_handle:
	144	reader = IterativeReader(file_handle)
d910eba5	145	self.helper_test_len(reader, 1)
e7d49180	146	reader = IterativeReader([file_handle, ])
d910eba5	147	self.helper_test_len(reader, 1)
e7d49180	148	reader = IterativeReader(file_handle, 'desc')
d910eba5	149	self.helper_test_len(reader, 1)
e7d49180	150	reader = IterativeReader([file_handle, ], ['desc', ])
d910eba5	151	self.helper_test_len(reader, 1)
e7d49180	152	reader = IterativeReader(file_handle, ['desc', ])
d910eba5	153	self.helper_test_len(reader, 1)
e7d49180 CH	154	self.assertRaises(ValueError, IterativeReader,
	155	[file_handle, ], 'desc', )
	156	reader = IterativeReader([file_handle, file_handle],
	157	['desc1', 'desc2'])
d910eba5	158	self.helper_test_len(reader, 2)
e7d49180	159	reader = IterativeReader((file_handle for idx in range(5)))
d910eba5	160	self.helper_test_len(reader, 5)
e7d49180 CH	161	self.assertRaises(ValueError, IterativeReader,
	162	(file_handle for idx in range(5)),
	163	tuple('desc' for idx in range(4)))
	164	self.assertRaises(ValueError, IterativeReader,
	165	(file_handle for idx in range(5)),
	166	('desc' for idx in range(6)))
	167
	168	def test_simple_read(self):
	169	""" write fixed number of lines, see how fast they are retrieved """
	170
	171	# need newline only when writing text (because of write buffering)
	172	param_combinations = ('{0}:{1}\n', None, False), \
	173	('{0}:{1}\n', 'ascii', False), \
	174	('{0}:{1} ' , 'ascii', False)
	175	#('{0}:{1}\n', None , True), \ logging seems
	176	#('{0}:{1}\n', 'ascii', True), \ to buffer writes
	177	#('{0}:{1} ' , None , True), \ to files
	178	#('{0}:{1} ' , 'ascii', True)
	179
	180	n_texts = 10
	181	pause_time = 0.01 # 100 tps (texts per second)
	182
	183	for text_pattern, encoding, use_logging in param_combinations:
	184	LogFileWriter(self.temp_file, text_pattern, n_writes=n_texts,
	185	pause_time=pause_time, do_encode=encoding,
	186	use_logging=use_logging).start()
7c00b0af CH	187	if DEBUG:
	188	print('testing with log file {0}'.format(self.temp_file))
	189	print('encoding is {0}, use logging = {1}'.format(encoding,
	190	use_logging))
e7d49180 CH	191	time_diffs = []
	192
	193	with open(self.temp_file, 'rt') as file_handle:
545f3458	194	reader = IterativeReader(file_handle, keep_watching=True)
d910eba5	195	self.helper_test_len(reader, 1)
ea8b01a3	196	counter = -1 # we may have to adapt this manually
aef38def	197	for desc, text, source_idx in reader:
01fe1580	198	receive_time = perf_counter()
f75cc662	199	self.assertEqual(desc, self.temp_file)
aef38def	200	self.assertEqual(source_idx, 0)
ea8b01a3	201	counter += 1
e7d49180	202	text = text.strip()
7c00b0af CH	203	if DEBUG:
	204	print('{1}: received text "{0}" at {2}'
	205	.format(text, counter, receive_time))
ea8b01a3 CH	206	if counter == 0 and not text:
	207	# if reader runs stat() before we write, we might get
	208	# a warning and one empty read here
	209	counter -= 1
	210	warn('Got an empty read, you should have seen another '
	211	'warning about file shrinking',
	212	category=LogReadWarning)
	213	continue
e7d49180 CH	214	index = text.index(':')
	215	count_text = int(text[:index].strip())
	216	self.assertEqual(count_text, counter)
	217	write_time = float(text[index+1:].strip())
58347979	218	time_diffs.append((receive_time - write_time)*1000.)
e7d49180	219	if counter == n_texts-1:
7c00b0af CH	220	if DEBUG:
7c00b0af CH	221	print('stop since have {0} reads'.format(counter))
e7d49180	222	break
7c00b0af CH	223	if DEBUG:
7c00b0af CH	224	print('time diffs in ms: {0}'.format(time_diffs))
58347979 CH	225	self.assertTrue(max(time_diffs) < 100.,
58347979 CH	226	'read took more than 100ms (max was {0:.3f}ms)!'
6ab9df9b	227	.format(max(time_diffs)))
e7d49180	228
d910eba5 CH	229	def test_line_read(self):
	230	""" write partial lines, full lines and multiple lines """
	231
	232	pause_time = 0.01 # 100 tps (texts per second)
	233	encoding = None
	234	use_logging = False
	235	texts = ['line0\n', 'line1\n', 'li', 'ne2\n', 'line3\n',
	236	'line4\nline5\n', 'li', 'ne6\nli', 'ne7\nl', 'i', 'n', 'e',
	237	'8', '\n', 'l', 'ine9\nline10\nline1', '1', '\n', '\n', '\n',
	238	'end\n', '\nend\n', '\n\nend\n\n']
	239	lines_expected = ['line{0}'.format(idx) for idx in range(12)] \
	240	+ ['', '', '']
	241
	242	# create writer
	243	LogFileWriter(self.temp_file, texts, n_writes=len(texts),
	244	pause_time=pause_time, do_encode=encoding,
	245	use_logging=use_logging).start()
	246
	247	# read
	248	lines_read = []
	249	with open(self.temp_file, 'rt') as file_handle:
545f3458	250	reader = LineReader(file_handle, keep_watching=True)
d910eba5 CH	251	self.helper_test_len(reader, 1)
d910eba5 CH	252
aef38def	253	for line_expected, (_, line_read, _) in zip(lines_expected, reader):
d910eba5 CH	254	if 'end' in line_read:
	255	break
	256	else:
7c00b0af CH	257	if DEBUG:
	258	print('expect "{0}", read "{1}"'.format(line_expected,
	259	line_read))
d910eba5 CH	260	self.assertEqual(line_expected, line_read)
	261
	262
e7d49180 CH	263	if __name__ == '__main__':
e7d49180 CH	264	unittest.main()