popen options for subprocessing

phlop/proc.py

@@ -2,13 +2,15 @@
 #
 #
 #
+
 import subprocess
 import sys
 
 from phlop.procs.runtimer import RunTimer
 
 
-class ProcessNonZeroExitCode(RuntimeError): ...
+class ProcessNonZeroExitCode(RuntimeError):
+    ...
 
 
 def run(cmd, shell=True, capture_output=True, check=False, print_cmd=True, **kwargs):

phlop/procs/runtimer.py

@@ -2,13 +2,11 @@
 #
 #
 #
-#
 
 
 import os
 import subprocess
 import time
-from copy import deepcopy
 
 from phlop.os import pushd, write_to_file
 from phlop.string import decode_bytes
@@ -26,63 +24,99 @@ def __init__(
         working_dir=None,
         log_file_path=None,
         logging=2,
+        popen=True,
         **kwargs,
     ):
         self.cmd = cmd
-        start = time.time()
-
-        self.run_time = None
         self.stdout = ""
         self.stderr = ""
+        self.run_time = None
+        self.logging = logging
+        self.log_file_path = log_file_path
+        self.capture_output = capture_output
         benv = os.environ.copy()
         benv.update(env)
-        self.logging = logging
-
-        ekwargs = {}
+        ekwargs = dict(
+            shell=shell,
+            env=benv,
+            close_fds=True,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
         if not capture_output and log_file_path:
             ekwargs.update(
                 dict(
                     stdout=open(f"{log_file_path}.stdout", "w"),
                     stderr=open(f"{log_file_path}.stderr", "w"),
                 ),
             )
+        else:
+            ekwargs.update(
+                dict(stdout=subprocess.PIPE, stderr=subprocess.PIPE),
+            )
 
-        def run():
-            logging = deepcopy(self.logging)
-            try:
-                self.run = subprocess.run(
-                    self.cmd,
-                    shell=shell,
-                    check=check,
-                    env=benv,
-                    capture_output=capture_output,
-                    close_fds=True,
-                    **kwargs,
-                    **ekwargs,
+        def go():
+            if popen:
+                self._popen(*kwargs, **ekwargs)
+            else:
+                self._run(
+                    check=check, capture_output=capture_output, *kwargs, **ekwargs
                 )
-                self.run_time = time.time() - start
-                self.exitcode = self.run.returncode
-                if logging == 2 and capture_output:
-                    self.stdout = decode_bytes(self.run.stdout)
-                    self.stderr = decode_bytes(self.run.stderr)
-            except (
-                subprocess.CalledProcessError
-            ) as e:  # only triggers on failure if check=True
-                self.exitcode = e.returncode
-                self.run_time = time.time() - start
-                if logging >= 1 and capture_output:
-                    self.stdout = decode_bytes(e.stdout)
-                    self.stderr = decode_bytes(e.stderr)
-                    logging = 2  # force logging as exception occurred
-            if logging == 2 and capture_output and log_file_path:
-                write_to_file(f"{log_file_path}.stdout", self.stdout)
-                write_to_file(f"{log_file_path}.stderr", self.stderr)
 
         if working_dir:
             with pushd(working_dir):
-                run()
+                go()
         else:
-            run()
+            go()
+
+    def _locals(self):
+        return self.capture_output, self.log_file_path, self.logging
+
+    def _run(self, **kwargs):
+        capture_output, log_file_path, logging = self._locals()
+        try:
+            start = time.time()
+            self.run = subprocess.run(self.cmd, **kwargs)
+            self.run_time = time.time() - start
+            self.exitcode = self.run.returncode
+            if logging == 2 and capture_output:
+                self.stdout = decode_bytes(self.run.stdout)
+                self.stderr = decode_bytes(self.run.stderr)
+        except subprocess.CalledProcessError as e:
+            # only triggers on failure if check=True
+            self.run_time = time.time() - start
+            self.exitcode = e.returncode
+            if logging >= 1 and capture_output:
+                self.stdout = decode_bytes(e.stdout)
+                self.stderr = decode_bytes(e.stderr)
+                logging = 2  # force logging as exception occurred
+        if logging == 2 and capture_output and log_file_path:
+            write_to_file(f"{log_file_path}.stdout", self.stdout)
+            write_to_file(f"{log_file_path}.stderr", self.stderr)
+
+    def _popen(self, **kwargs):
+        capture_output, log_file_path, logging = self._locals()
+        start = time.time()
+        p = subprocess.Popen(self.cmd, **kwargs)
+        self.stdout, self.stderr = p.communicate()
+        self.run_time = time.time() - start
+        self.exitcode = p.returncode
+        if not capture_output and log_file_path:
+            kwargs["stdout"].close()
+            kwargs["stderr"].close()
+        elif capture_output:
+            p.stdout.close()
+            p.stderr.close()
+        p = None
+
+        if self.exitcode > 0 and capture_output:
+            logging = 2  # force logging as exception occurred
+        if logging == 2 and capture_output:
+            self.stdout = decode_bytes(self.stdout)
+            self.stderr = decode_bytes(self.stderr)
+        if logging == 2 and capture_output and log_file_path:
+            write_to_file(f"{log_file_path}.stdout", self.stdout)
+            write_to_file(f"{log_file_path}.stderr", self.stderr)
 
     def out(self, ignore_exit_code=False):
         if not ignore_exit_code and self.exitcode > 0:

phlop/reflection.py

@@ -2,7 +2,6 @@
 #
 #
 #
-#
 
 import importlib
 import inspect
@@ -17,7 +16,7 @@
 ).lower() in ("true", "1", "t")
 
 
-def classes_in_file(file_path, subclasses_only=None, fail_on_import_error=False):
+def classes_in_file(file_path, subclasses_only=None, fail_on_import_error=True):
     file = Path(file_path)
     module = str(file).replace(os.path.sep, ".")[:-3]
     assert module

phlop/testing/parallel_processor.py

@@ -11,8 +11,11 @@
 
 from phlop.proc import run
 
+timeout = 60  # seconds - give chance to interrupt
 
-class TestCaseFailure(Exception): ...
+
+class TestCaseFailure(Exception):
+    ...
 
 
 class LoggingMode(Enum):
@@ -32,7 +35,7 @@ def __call__(self, **kwargs):
         self.run = run(
             self.test_case.cmd.split(),
             shell=False,
-            capture_output=True,
+            capture_output=False,
             check=True,
             print_cmd=False,
             env=self.test_case.env,
@@ -93,10 +96,7 @@ def launch_tests():
                     )
                     cc.cores_avail -= batch.cores
                     cc.procs[batch_index] += [
-                        Process(
-                            target=runner,
-                            args=(test, (pqueue)),
-                        )
+                        Process(target=runner, args=(test, (pqueue)))
                     ]
                     cc.procs[batch_index][-1].daemon = True
                     cc.procs[batch_index][-1].start()
@@ -110,7 +110,13 @@ def finished():
     def waiter(queue):
         fail = 0
         while True:
-            proc = queue.get()
+            proc = None
+            try:
+                proc = queue.get(timeout=timeout)
+            except Exception:
+                print("Queue Exception! - no jobs finished - polling")
+                continue
+
             time.sleep(0.01)  # don't throttle!
             if isinstance(proc, CallableTest):
                 status = "finished" if proc.run.exitcode == 0 else "FAILED"

phlop/testing/test_cases.py

@@ -18,7 +18,6 @@
 from phlop.sys import extend_sys_path
 
 _LOG_DIR = Path(os.environ.get("PHLOP_LOG_DIR", os.getcwd()))
-
 CMD_PREFIX = ""
 CMD_POSTFIX = ""
 
@@ -143,8 +142,10 @@ def determine_cores_for_test_case(test_case):
 
 
 def binless(test_case):
-    if test_case.cmd.startswith("/usr/bin/"):
-        test_case.cmd = test_case.cmd[9:]
+    if test_case.cmd.startswith("/usr/"):
+        bits = test_case.cmd.split(" ")
+        test_case.cmd = " ".join([bits[0].split("/")[-1]] + bits[1:])
+    # print("test_case.cmd ", test_case.cmd)
     return test_case
 
 

phlop/timing/scope_timer.py

@@ -2,9 +2,10 @@
 # parsing PHARE scope funtion timers
 #
 
-import numpy as np
-from pathlib import Path
 from dataclasses import dataclass, field
+from pathlib import Path
+
+import numpy as np
 
 
 @dataclass

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "phlop"
-version = "0.0.27"
+version = "0.0.28"
 
 dependencies = [
 

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name="phlop",
-    version="0.0.27",
+    version="0.0.28",
     cmdclass={},
     classifiers=[],
     include_package_data=True,

sh/lint.sh

@@ -7,7 +7,7 @@ set -e
 
 PY_FILES=$(find . -name "*.py")
 
-python3 -m black phlop tests
+# python3 -m black phlop tests
 pylint --errors-only phlop tests
 isort phlop tests
 python3 -m ruff check phlop tests