Add LogFile helper

LogFile class contains a log path and has means to find a string in the
log file.
This commit is contained in:
Tom Krizek
2023-12-22 10:55:17 +01:00
committed by Michał Kępień
parent 46b7cc2ed2
commit 05b0ebac0f
3 changed files with 40 additions and 6 deletions

View File

@@ -13,4 +13,4 @@ from . import check
from . import instance
from . import query
from . import rndc
from . import watchlog
from . import log

View File

@@ -18,7 +18,7 @@ import os
import re
from .rndc import RNDCBinaryExecutor, RNDCException, RNDCExecutor
from .watchlog import WatchLogFromStart, WatchLogFromHere
from .log import LogFile, WatchLogFromStart, WatchLogFromHere
class NamedPorts(NamedTuple):
@@ -63,7 +63,7 @@ class NamedInstance:
"""
self.ip = self._identifier_to_ip(identifier)
self.ports = ports
self._log_file = os.path.join(identifier, "named.run")
self.log = LogFile(os.path.join(identifier, "named.run"))
self._rndc_executor = rndc_executor or RNDCBinaryExecutor()
self._rndc_logger = rndc_logger or logging.getLogger()
@@ -133,14 +133,14 @@ class NamedInstance:
Return an instance of the `WatchLogFromStart` context manager for this
`named` instance's log file.
"""
return WatchLogFromStart(self._log_file)
return WatchLogFromStart(self.log.path)
def watch_log_from_here(self) -> WatchLogFromHere:
"""
Return an instance of the `WatchLogFromHere` context manager for this
`named` instance's log file.
"""
return WatchLogFromHere(self._log_file)
return WatchLogFromHere(self.log.path)
def reconfigure(self) -> None:
"""

View File

@@ -9,7 +9,7 @@
# See the COPYRIGHT file distributed with this work for additional
# information regarding copyright ownership.
from typing import Optional, TextIO, Dict, Any, overload, List, Union
from typing import Iterator, Optional, TextIO, Dict, Any, overload, List, Union
import abc
import os
@@ -22,6 +22,40 @@ class WatchLogException(Exception):
pass
class LogFile:
"""
Log file wrapper with a path and means to find a string in its contents.
"""
def __init__(self, path: str):
self.path = path
@property
def _lines(self) -> Iterator[str]:
with open(self.path, encoding="utf-8") as f:
yield from f
def __contains__(self, substring: str) -> bool:
"""
Return whether any of the lines in the log contains a given string.
"""
for line in self._lines:
if substring in line:
return True
return False
def expect(self, msg: str):
"""Check the string is present anywhere in the log file."""
if msg in self:
return
assert False, f"log message not found in log {self.path}: {msg}"
def prohibit(self, msg: str):
"""Check the string is not present in the entire log file."""
if msg in self:
assert False, f"forbidden message appeared in log {self.path}: {msg}"
class WatchLog(abc.ABC):
"""