Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test():
"""Test for FileTailer Python Class"""
# Grab a test file
data_path = file_utils.relative_dir(__file__, '../../data')
test_path = os.path.join(data_path, 'http.log')
print('Opening Data File: {:s}'.format(test_path))
# Create the Class
tailer = FileTailer(test_path, tail=False) # First with no tailing
for line in tailer.readlines():
print(line)
print('Read with NoTail Test successful!')
# Now include tailing (note: as an automated test this needs to timeout quickly)
try:
from interruptingcow import timeout
# Spin up the class
tailer = FileTailer(test_path) # Tail = True
# Tail the file for 2 seconds and then quit
try:
with timeout(2, exception=RuntimeError):
for line in tailer.readlines():
print(line)
data_path = file_utils.relative_dir(__file__, '../../data')
test_path = os.path.join(data_path, 'http.log')
print('Opening Data File: {:s}'.format(test_path))
# Create the Class
tailer = FileTailer(test_path, tail=False) # First with no tailing
for line in tailer.readlines():
print(line)
print('Read with NoTail Test successful!')
# Now include tailing (note: as an automated test this needs to timeout quickly)
try:
from interruptingcow import timeout
# Spin up the class
tailer = FileTailer(test_path) # Tail = True
# Tail the file for 2 seconds and then quit
try:
with timeout(2, exception=RuntimeError):
for line in tailer.readlines():
print(line)
except RuntimeError: # InterruptingCow raises a RuntimeError on timeout
print('Tailing Test successful!')
except ImportError:
print('Tailing Test not run, need interruptcow module...')
- Read contents + 'tail -f' Zeek log file (tail=True)
Args:
filepath (str): The full path the file (/full/path/to/the/file.txt)
delimiter (str): The delimiter in the Zeek logs (default='\t')
tail (bool): Do a dynamic tail on the file (i.e. tail -f) (default=False)
"""
from __future__ import print_function
import os
import time
import datetime
# Local Imports
from zat.utils import file_tailer, file_utils
class BroLogReader(file_tailer.FileTailer):
"""BroLogReader: This class reads in various Zeek logs. The class inherits from
the FileTailer class so it supports the following use cases:
- Read contents of a Zeek log file (tail=False)
- Read contents + 'tail -f' Zeek log file (tail=True)
Args:
filepath (str): The full path the file (/full/path/to/the/file.txt)
delimiter (str): The delimiter in the Zeek logs (default='\t')
tail (bool): Do a dynamic tail on the file (i.e. tail -f) (default=False)
strict (bool): Raise an exception on conversions errors (default=False)
"""
def __init__(self, filepath, delimiter='\t', tail=False, strict=False):
"""Initialization for the BroLogReader Class"""
# First check if the file exists and is readable
if not os.access(filepath, os.R_OK):