Stream NextFlow logs in realtime

CHANGELOG.md

@@ -8,6 +8,9 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ---
 
 ## [Unreleased]
+### Changed
+- Wrap all NextFlow output with real-time line-by-line log statements
+- Add DEBUG lines about specific asserts
 
 ---
 

nftest/NFTestAssert.py

@@ -35,6 +35,7 @@ def assert_expected(self):
         assert_method = self.get_assert_method()
         try:
             assert assert_method(self.actual, self.expect)
+            self._logger.debug('Assertion passed')
         except AssertionError as error:
             self._logger.error('Assertion failed')
             self._logger.error('Actual: %s', self.actual)
@@ -47,11 +48,13 @@ def get_assert_method(self) -> Callable:
         if self.script is not None:
             def func(actual, expect):
                 cmd = f"{self.script} {actual} {expect}"
+                self._logger.debug(cmd)
                 process_out = sp.run(cmd, shell=True, check=False)
                 return process_out.returncode == 0
             return func
         if self.method == 'md5':
             def func(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

nftest/NFTestCase.py

@@ -1,13 +1,17 @@
 """ NF Test case """
 from __future__ import annotations
-import shutil
+
 import re
+import selectors
+import shutil
+import subprocess as sp
+
+from logging import getLogger, INFO, ERROR
 from pathlib import Path
 from shlex import quote
-import subprocess as sp
 from subprocess import PIPE
-from logging import getLogger
 from typing import Callable, List, TYPE_CHECKING
+
 from nftest.common import remove_nextflow_logs
 from nftest.NFTestENV import NFTestENV
 
@@ -29,6 +33,7 @@ def __init__(self, name:str=None, message:str=None, nf_script:str=None,
         """ Constructor """
         self._env = NFTestENV()
         self._logger = getLogger('NFTest')
+        self._nflogger = getLogger("NextFlow")
         self.name = name
         self.name_for_output = re.sub(r'[^a-zA-Z0-9_\-.]', '', self.name.replace(' ', '-'))
         self.message = message
@@ -107,22 +112,40 @@ def submit(self) -> sp.CompletedProcess:
         """
         self._logger.info(' '.join(cmd.split()))
 
-        res = sp.run(cmd,
-            shell=True,
-            stdout=PIPE,
-            stderr=PIPE,
-            check=False,
-            universal_newlines=True,
-            capture_output=(not self.verbose))
-
-        self._logger.info(res.stdout)
-        if res.stderr.strip():
-            self._logger.error(res.stderr)
-
-        return res
-
-
-    def combine_global(self, _global:NFTestGlobal) -> None:
+        with sp.Popen(cmd,
+                      shell=True,
+                      stdout=PIPE,
+                      stderr=PIPE,
+                      universal_newlines=True) as process:
+
+            # Route stdout to INFO and stderr to ERROR in real-time
+            with selectors.DefaultSelector() as selector:
+                selector.register(
+                    fileobj=process.stdout,
+                    events=selectors.EVENT_READ,
+                    data=INFO
+                )
+                selector.register(
+                    fileobj=process.stderr,
+                    events=selectors.EVENT_READ,
+                    data=ERROR
+                )
+
+                while process.poll() is None:
+                    events = selector.select()
+                    for key, _ in events:
+                        line = key.fileobj.readline()
+                        if line:
+                            # The only case in which this won't be true is when
+                            # the pipe is closed
+                            self._nflogger.log(
+                                level=key.data,
+                                msg=line.rstrip()
+                            )
+
+        return process
+
+    def combine_global(self, _global: NFTestGlobal) -> None:
         """ Combine test case configs with the global configs. """
         if _global.nf_config:
             self.nf_configs.insert(0, _global.nf_config)

nftest/common.py

@@ -53,39 +53,30 @@ def print_version_and_exist():
     print(__version__, file=sys.stdout)
     sys.exit()
 
+
 def setup_loggers():
     """ Initialize loggers for both init and run """
-    _ = generate_logger('NFTest')
-    _ = generate_logger('NFTestInit')
+    # Always log times in UTC
+    logging.Formatter.converter = time.gmtime
 
-# pylint: disable=W0212
-def generate_logger(logger_name:str):
-    """ Generate program-specific logger """
     _env = NFTestENV()
-    try:
-        log_level = logging._checkLevel(_env.NFT_LOG_LEVEL)
-    except ValueError:
-        log_level = logging._checkLevel('INFO')
-
-    logger = logging.getLogger(logger_name)
-    logger.setLevel(log_level)
-
-    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
-    logging.Formatter.converter = time.gmtime
 
+    # Make a file handler that accepts all logs
     try:
         file_handler = logging.FileHandler(_env.NFT_LOG)
+        file_handler.setLevel(logging.DEBUG)
     except (FileNotFoundError, PermissionError) as file_error:
         raise Exception(f'Unable to create log file: {_env.NFT_LOG}') from file_error
-    file_handler.setLevel(log_level)
 
+    # Make a stream handler with the requested verbosity
     stream_handler = logging.StreamHandler(sys.stdout)
-    stream_handler.setLevel(log_level)
-
-    file_handler.setFormatter(formatter)
-    stream_handler.setFormatter(formatter)
-
-    logger.addHandler(file_handler)
-    logger.addHandler(stream_handler)
+    try:
+        stream_handler.setLevel(logging._checkLevel(_env.NFT_LOG_LEVEL)) # pylint: disable=W0212
+    except ValueError:
+        stream_handler.setLevel(logging.INFO)
 
-    return logger
+    logging.basicConfig(
+        level=logging.DEBUG,
+        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+        handlers=(file_handler, stream_handler)
+    )