else:
perf_counter = time.perf_counter
+DEBUG = False
+
class LogFileWriter(Thread):
""" thread that creates and writes to given file """
if self.do_encode:
text = text.encode(self.do_encode)
write_func(text)
- #print('wrote {0}'.format(counter))
time.sleep(self.pause_time)
def setUp(self):
""" called before each test """
- print('setup test')
+ if DEBUG:
+ print('setup test')
temp_handle, temp_name = mkstemp()
os.close(temp_handle)
self.temp_file = temp_name
- print('created temp file ' + self.temp_file)
+ if DEBUG:
+ print('created temp file ' + self.temp_file)
def tearDown(self):
""" called after each test """
- print('tear down test')
+ if DEBUG:
+ print('tear down test')
if os.path.isfile(self.temp_file):
- print('delete temp file' + 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):
LogFileWriter(self.temp_file, text_pattern, n_writes=n_texts,
pause_time=pause_time, do_encode=encoding,
use_logging=use_logging).start()
- print('testing with log file {0}'.format(self.temp_file))
- print('encoding is {0}, use logging = {1}'.format(encoding,
- use_logging))
+ 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:
for counter, (desc, text) in enumerate(reader):
receive_time = perf_counter()
text = text.strip()
- print('{1}: received text "{0}" at {2}'
- .format(text, counter, receive_time))
+ if DEBUG:
+ print('{1}: received text "{0}" at {2}'
+ .format(text, counter, receive_time))
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:
- print('stop since have {0} reads'.format(counter))
+ if DEBUG:
+ print('stop since have {0} reads'.format(counter))
break
- print('time diffs in ms: {0}'.format(time_diffs))
+ 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)))
if 'end' in line_read:
break
else:
- print('expect "{0}", read "{1}"'.format(line_expected,
- line_read))
+ if DEBUG:
+ print('expect "{0}", read "{1}"'.format(line_expected,
+ line_read))
self.assertEqual(line_expected, line_read)