import re
from .rndc import RNDCBinaryExecutor, RNDCException, RNDCExecutor
-from .watchlog import WatchLogFromStart, WatchLogFromHere
+from .log import LogFile, WatchLogFromStart, WatchLogFromHere
class NamedPorts(NamedTuple):
"""
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()
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:
"""
# 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
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):
"""