# TESTING
# #############################################################################
-from datetime import datetime as dt, timedelta as td
+from datetime import date, datetime as dt, timedelta as td
from subprocess import Popen, PIPE
syslog_file = '/var/log/messages'
-time_diff_seconds = 300
+time_diff_seconds = 60
syslog_time_format = '%b %d %H:%M:%S'
-def test_syslog_line(n_lines, source, desc, line, start_time):
+def test_syslog_line(n_lines, source, desc, line,
+ today_str, start_time):
""" called by test functions for each line of syslog """
if n_lines % 1000 == 0:
- print(('{0} old lines, showing lines after {1}; '
+ print(('{0:6d} old lines, showing lines after {1}; '
'abort using Ctrl-C').format(n_lines, start_time))
+ if line[:6] != today_str:
+ return
try:
log_time = dt.strptime(line[:15], syslog_time_format)
log_time = log_time.replace(year=start_time.year)
except ValueError:
log_time = None
- if False: # log_time is None or log_time > start_time:
+ if log_time is None or log_time > start_time:
print('line {0} from "{1}", (orig from {2}): {3}'.format(
n_lines, desc, log_time, line))
"""
start_time = dt.now() - td(seconds=time_diff_seconds)
+ today_str = date.today().strftime(syslog_time_format)[:6]
+ if today_str[4] == '0':
+ today_str = today_str[:4] + ' ' + today_str[5:] # 'Oct 01' --> 'Oct 1'
# create process for 'tail -f syslog' --> 2 pipes (stdout + stderr)
proc = Popen(['tail', '-f', syslog_file], stdout=PIPE, stderr=PIPE)
def test_follower_context():
""" test FollowContextManager and follow() function """
+ today_str = date.today().strftime(syslog_time_format)[:6]
+ if today_str[4] == '0':
+ today_str = today_str[:4] + ' ' + today_str[5:] # 'Oct 01' --> 'Oct 1'
start_time = dt.now() - td(seconds=time_diff_seconds)
-
with follow(syslog_file) as flwr:
for n_lines, (source, desc, line, flags) in enumerate(flwr):
test_syslog_line(n_lines, source, desc, line,
today_str, start_time)
+def test_context_proc():
+ """ test FollowContextManager's ability to wrap proc args """
+
+ today_str = date.today().strftime(syslog_time_format)[:6]
+ if today_str[4] == '0':
+ today_str = today_str[:4] + ' ' + today_str[5:] # 'Oct 01' --> 'Oct 1'
+ start_time = dt.now() - td(seconds=time_diff_seconds)
+ with follow(procs=(['tail', '-f', syslog_file], )) as flwr:
+ for n_lines, (source, desc, line, flags) in enumerate(flwr):
+ test_syslog_line(n_lines, source, desc, line,
+ today_str, start_time)
+
def test():
""" Main function, tests some of class's functionality """
- if False:
- test_follower_syslog()
- if True:
- test_follower_context()
+ #test_follower_syslog()
+ #test_follower_context()
+ test_context_proc()
# end: function main