Tidy logger

janus_core/log.py

@@ -5,42 +5,60 @@
 
 from janus_core.janus_types import LogLevel
 
-FORMAT = """
-- timestamp: %(asctime)s
-  level: %(levelname)s
-  message: %(message)s
-  trace: %(module)s
-  line: %(lineno)d
-"""
 
-
-class CustomFormatter(logging.Formatter):  # numpydoc ignore=PR02
+class YamlFormatter(logging.Formatter):  # numpydoc ignore=PR02
     """
     Custom formatter to convert multiline messages into yaml list.
 
     Parameters
     ----------
-    fmt : str
+    fmt : Optional[str]
         A format string in the given style for the logged output as a whole. Default is
-        '%(message)s'.
-    datefmt : str
+        defined by `FORMAT`.
+    datefmt : Optional[str]
         A format string in the given style for the date/time portion of the logged
         output. Default is taken from logging.Formatter.formatTime().
-    style : str
+    style : Literal['%'] | Literal['{'] | Literal['$']
         Determines how the format string will be merged with its data. Can be one of
         '%', '{' or '$'. Default is '%'.
     validate : bool
         If True, incorrect or mismatched fmt and style will raise a ValueError. Default
         is True.
-    defaults : dict[str, Any]
-         A dictionary with default values to use in custom fields.
+    defaults : Optional[dict[str, logging.Any]]
+         A dictionary with default values to use in custom fields. Default is None.
+    *args
+        Arguments to pass to logging.Formatter.__init__.
+    **kwargs
+        Keyword arguments to pass to logging.Formatter.__init__.
 
     Methods
     -------
     format(record)
         Format log message to convert new lines into a yaml list.
     """
 
+    FORMAT = (
+        "\n- timestamp: %(asctime)s\n"
+        "  level: %(levelname)s\n"
+        "  message: %(message)s\n"
+        "  trace: %(module)s\n"
+        "  line: %(lineno)d\n"
+    )
+
+    def __init__(self, *args, **kwargs) -> None:
+        """
+        Set default string format to yaml style.
+
+        Parameters
+        ----------
+        *args
+            Arguments to pass to logging.Formatter.__init__.
+        **kwargs
+            Keyword arguments to pass to logging.Formatter.__init__.
+        """
+        kwargs.setdefault("fmt", self.FORMAT)
+        super().__init__(*args, **kwargs)
+
     def format(self, record: logging.LogRecord) -> str:
         """
         Format log message to convert new lines into a yaml list.
@@ -59,19 +77,10 @@ def format(self, record: logging.LogRecord) -> str:
         record.msg = record.msg.replace('"', "'")
 
         # Convert new lines into yaml list
-        if len(record.msg.split("\n")) > 1:
-            msg = record.msg
-            record.msg = "\n"
-            for line in msg.split("\n"):
-                # Exclude empty lines
-                if line.strip():
-                    record.msg += f'    - "{line.strip()}"\n'
-            # Remove final newline (single character)
-            record.msg = record.msg[:-1]
-        else:
-            # Wrap line in quotes here rather than in FORMAT due to list
-            record.msg = f'"{record.msg}"'
-
+        msg = record.msg
+        record.msg = "\n" + "\n".join(
+            [f'    - "{line.strip()}"' for line in msg.split("\n") if line.strip()],
+        )
         return super().format(record)
 
 
@@ -116,7 +125,7 @@ def config_logger(
             encoding="utf-8",
         )
         handler.setLevel(level)
-        formatter = CustomFormatter(FORMAT)
+        formatter = YamlFormatter()
         handler.setFormatter(formatter)
 
         logging.basicConfig(

tests/test_geomopt_cli.py

@@ -76,7 +76,7 @@ def test_log(tmp_path):
     assert result.exit_code == 0
 
     check_log_contents(
-        log_path, contains="Starting geometry optimization", excludes="Using filter"
+        log_path, includes="Starting geometry optimization", excludes="Using filter"
     )
 
 
@@ -131,7 +131,7 @@ def test_fully_opt(tmp_path):
     )
     assert result.exit_code == 0
 
-    check_log_contents(log_path, contains=["Using filter", "hydrostatic_strain: False"])
+    check_log_contents(log_path, includes=["Using filter", "hydrostatic_strain: False"])
 
     atoms = read(results_path)
     expected = [5.68834069, 5.68893345, 5.68932555, 89.75938298, 90.0, 90.0]
@@ -162,7 +162,7 @@ def test_fully_opt_and_vectors(tmp_path):
     )
     assert result.exit_code == 0
 
-    check_log_contents(log_path, contains=["Using filter", "hydrostatic_strain: True"])
+    check_log_contents(log_path, includes=["Using filter", "hydrostatic_strain: True"])
 
     atoms = read(results_path)
     expected = [5.69139709, 5.69139709, 5.69139709, 89.0, 90.0, 90.0]
@@ -192,7 +192,7 @@ def test_vectors_not_fully_opt(tmp_path):
     )
     assert result.exit_code == 0
 
-    check_log_contents(log_path, contains=["Using filter", "hydrostatic_strain: True"])
+    check_log_contents(log_path, includes=["Using filter", "hydrostatic_strain: True"])
 
 
 def test_duplicate_traj(tmp_path):

tests/test_log.py

@@ -3,23 +3,28 @@
 import yaml
 
 from janus_core.log import config_logger
+from tests.utils import check_log_contents
 
 
 def test_multiline_log(tmp_path):
     """Test multiline log is written correctly."""
     log_path = tmp_path / "test.log"
 
-    logger = config_logger(name=__name__, filename=log_path, force=True)
+    logger = config_logger(name=__name__, filename=log_path)
     logger.info(
         """
         Line 1
         Line 2
         Line 3
         """
     )
+    logger.info("Line 4")
+
     assert log_path.exists()
     with open(log_path, encoding="utf8") as log:
         log_dicts = yaml.safe_load(log)
 
     assert log_dicts[0]["level"] == "INFO"
     assert len(log_dicts[0]["message"]) == 3
+
+    check_log_contents(log_path, includes=["Line 1", "Line 4"])

tests/test_md_cli.py

@@ -105,7 +105,7 @@ def test_log(tmp_path):
     )
     assert result.exit_code == 0
 
-    check_log_contents(log_path, contains=["Starting molecular dynamics simulation"])
+    check_log_contents(log_path, includes=["Starting molecular dynamics simulation"])
 
     with open(tmp_path / "nvt-T300-stats.dat", encoding="utf8") as stats_file:
         lines = stats_file.readlines()

tests/test_singlepoint_cli.py

@@ -192,7 +192,7 @@ def test_log(tmp_path):
     )
     assert result.exit_code == 0
 
-    check_log_contents(log_path, contains=["Starting single point calculation"])
+    check_log_contents(log_path, includes=["Starting single point calculation"])
 
 
 def test_summary(tmp_path):

tests/utils.py

@@ -37,7 +37,7 @@ def read_atoms(path: Path) -> Union[Atoms, None]:
 
 def check_log_contents(
     log_path: PathLike,
-    contains: Optional[MaybeSequence[str]] = None,
+    includes: Optional[MaybeSequence[str]] = None,
     excludes: Optional[MaybeSequence[str]] = None,
 ) -> None:
     """
@@ -47,22 +47,23 @@ def check_log_contents(
     ----------
     log_path : PathLike
         Path to log file to check messsages of.
-    contains : MaybeSequence[str]
+    includes : MaybeSequence[str]
         Messages that must appear in the log file. Default is None.
     excludes : MaybeSequence[str]
         Messages that must not appear in the log file. Default is None.
     """
     # Convert single strings to iterable
-    contains = [contains] if isinstance(contains, str) else contains
+    includes = [includes] if isinstance(includes, str) else includes
     excludes = [excludes] if isinstance(excludes, str) else excludes
 
     # Read log file
     with open(log_path, encoding="utf8") as log_file:
         logs = yaml.safe_load(log_file)
-    messages = "".join(log["message"] for log in logs)
+    # Nested join as log["message"] may be a list
+    messages = "".join("".join(log["message"]) for log in logs)
 
-    if contains:
-        for msg in contains:
+    if includes:
+        for msg in includes:
             assert msg in messages
     if excludes:
         for msg in excludes: