Encapsulate NFTestAssert class, rewrite tests

CHANGELOG.md

@@ -16,6 +16,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Use `shell=False` for subprocess
 - Capture Nextflow logs via syslog rather than console
 - Format all code with ruff
+- Encapsulate NFTestAssert interface, rewrite tests
 
 ### Fixed
 - Make `nftest` with no arguments print usage and exit

nftest/NFTestAssert.py

@@ -1,21 +1,31 @@
 """NF Test assert"""
 
 import datetime
-import errno
-import os
 import subprocess
-from typing import Callable
+from typing import Callable, Optional
 from logging import getLogger, DEBUG
 
 from nftest.common import calculate_checksum, resolve_single_path, popen_with_logger
 from nftest.NFTestENV import NFTestENV
 
 
+class NotUpdatedError(Exception):
+    "An exception indicating that file was not updated."
+
+
+class MismatchedContentsError(Exception):
+    "An exception that the contents are mismatched."
+
+
 class NFTestAssert:
     """Defines how nextflow test results are asserted."""
 
     def __init__(
-        self, actual: str, expect: str, method: str = "md5", script: str = None
+        self,
+        actual: str,
+        expect: str,
+        method: str = "md5",
+        script: Optional[str] = None,
     ):
         """Constructor"""
         self._env = NFTestENV()
@@ -27,55 +37,45 @@ def __init__(
 
         self.startup_time = datetime.datetime.now(tz=datetime.timezone.utc)
 
-    def identify_assertion_files(self) -> None:
-        """Resolve actual and expected paths"""
-        self.actual = resolve_single_path(self.actual)
-        self.expect = resolve_single_path(self.expect)
-
-    def assert_exists(self) -> None:
-        "Assert that the expected and actual files exist."
-        if not self.actual.exists():
-            self._logger.error("Actual file not found: %s", self.actual)
-            raise FileNotFoundError(
-                errno.ENOENT, os.strerror(errno.ENOENT), self.actual
-            )
-
-        if not self.expect.exists():
-            self._logger.error("Expect file not found: %s", self.expect)
-            raise FileNotFoundError(
-                errno.ENOENT, os.strerror(errno.ENOENT), self.expect
-            )
+    def perform_assertions(self):
+        "Perform the appropriate assertions on the named files."
+        # Ensure that there is exactly one file for each input glob pattern
+        actual_path = resolve_single_path(self.actual)
+        self._logger.debug(
+            "Actual path `%s` resolved to `%s`", self.actual, actual_path
+        )
+        expect_path = resolve_single_path(self.expect)
+        self._logger.debug(
+            "Expected path `%s` resolved to `%s`", self.expect, expect_path
+        )
 
-    def assert_updated(self) -> None:
-        "Assert that the actual file was updated during this test run."
+        # Assert that the actual file was updated during this test run
         file_mod_time = datetime.datetime.fromtimestamp(
-            self.actual.stat().st_mtime, tz=datetime.timezone.utc
+            actual_path.stat().st_mtime, tz=datetime.timezone.utc
         )
 
         self._logger.debug("Test creation time: %s", self.startup_time)
         self._logger.debug("Actual mod time:    %s", file_mod_time)
-        assert (
-            file_mod_time > self.startup_time
-        ), f"{str(self.actual)} was not modified by this pipeline"
-
-    def assert_expected(self) -> None:
-        "Assert the results match with the expected values."
-        assert_method = self.get_assert_method()
-        try:
-            assert assert_method(self.actual, self.expect)
-            self._logger.debug("Assertion passed")
-        except AssertionError as error:
+
+        if self.startup_time >= file_mod_time:
+            raise NotUpdatedError(
+                f"{str(self.actual)} was not modified by this pipeline"
+            )
+
+        # Assert that the files match
+        if not self.get_assert_method()(actual_path, expect_path):
             self._logger.error("Assertion failed")
             self._logger.error("Actual: %s", self.actual)
             self._logger.error("Expect: %s", self.expect)
-            raise error
+            raise MismatchedContentsError("File comparison failed")
+
+        self._logger.debug("Assertion passed")
 
     def get_assert_method(self) -> Callable:
         """Get the assert method"""
-        # pylint: disable=E0102
         if self.script is not None:
 
-            def func(actual, expect):
+            def script_function(actual, expect):
                 cmd = [self.script, actual, expect]
                 self._logger.debug(subprocess.list2cmdline(cmd))
 
@@ -84,15 +84,17 @@ def func(actual, expect):
                 )
                 return process.returncode == 0
 
-            return func
+            return script_function
+
         if self.method == "md5":
 
-            def func(actual, expect):
+            def md5_function(actual, expect):
                 self._logger.debug("md5 %s %s", actual, expect)
                 actual_value = calculate_checksum(actual)
                 expect_value = calculate_checksum(expect)
                 return actual_value == expect_value
 
-            return func
+            return md5_function
+
         self._logger.error("assert method %s unknown.", self.method)
         raise ValueError(f"assert method {self.method} unknown.")

nftest/NFTestCase.py

@@ -27,8 +27,7 @@
 class NFTestCase:
     """Defines the NF test case"""
 
-    # pylint: disable=R0902
-    # pylint: disable=R0913
+    # pylint: disable=too-many-instance-attributes
     def __init__(
         self,
         name: str = None,
@@ -109,12 +108,9 @@ def test(self) -> bool:
             self._logger.error(" [ failed ]")
             return False
 
-        for ass in self.asserts:
+        for assertion in self.asserts:
             try:
-                ass.identify_assertion_files()
-                ass.assert_exists()
-                ass.assert_updated()
-                ass.assert_expected()
+                assertion.perform_assertions()
             except Exception as error:
                 self._logger.error(error.args)
                 self._logger.error(" [ failed ]")

nftest/NFTestENV.py

@@ -7,11 +7,11 @@
 from nftest.Singleton import Singleton
 
 
-# pylint: disable=C0103
 @dataclass
 class NFTestENV(metaclass=Singleton):
     """Class for initializng and holding environment variables."""
 
+    # pylint: disable=invalid-name
     NFT_OUTPUT: str = field(init=False)
     NFT_TEMP: str = field(init=False)
     NFT_INIT: str = field(init=False)

nftest/NFTestGlobal.py

@@ -7,7 +7,8 @@
 class NFTestGlobal:
     """Test global settings"""
 
-    # pylint: disable=R0903
+    # pylint: disable=too-few-public-methods
+
     def __init__(
         self,
         temp_dir: str,

nftest/__main__.py

@@ -35,7 +35,6 @@ def parse_args() -> argparse.Namespace:
     return args
 
 
-# pylint: disable=W0212
 def add_subparser_init(subparsers: argparse._SubParsersAction):
     """Add subparser for init"""
     parser: argparse.ArgumentParser = subparsers.add_parser(

nftest/common.py

@@ -20,8 +20,7 @@
 from nftest.syslog import syslog_filter
 
 
-# pylint: disable=W0613
-def validate_yaml(path: Path):
+def validate_yaml(path: Path):  # pylint: disable=unused-argument
     """Validate the yaml. Potentially use yaml schema
     https://rx.codesimply.com/
     """
@@ -41,10 +40,13 @@ def remove_nextflow_logs() -> None:
 def resolve_single_path(path: str) -> Path:
     """Resolve wildcards in path and ensure only a single path is identified"""
     expanded_paths = glob.glob(path)
-    if 1 != len(expanded_paths):
+
+    if not expanded_paths:
+        raise ValueError(f"Expression `{path}` did not resolve to any files")
+
+    if len(expanded_paths) > 1:
         raise ValueError(
-            f"Path `{path}` resolved to 0 or more than 1 file: {expanded_paths}."
-            " Assertion failed."
+            f"Expression `{path}` resolved to multiple files: {expanded_paths}"
         )
 
     return Path(expanded_paths[0])
@@ -79,6 +81,9 @@ def validate_reference(
 
     _logger = logging.getLogger("NFTest")
 
+    if reference_checksum_type.lower() not in {"md5"}:
+        _logger.warning("reference_checksum_type must be `md5`")
+
     actual_checksum = calculate_checksum(Path(reference_parameter_path))
 
     if actual_checksum != reference_checksum:
@@ -126,7 +131,7 @@ def setup_loggers():
     # Make a stream handler with the requested verbosity
     stream_handler = logging.StreamHandler(sys.stdout)
     try:
-        stream_handler.setLevel(logging._checkLevel(_env.NFT_LOG_LEVEL))  # pylint: disable=W0212
+        stream_handler.setLevel(_env.NFT_LOG_LEVEL)
     except ValueError:
         stream_handler.setLevel(logging.INFO)