Fix #646 pkcli.test support file::case (pytest node ID syntax)

pykern/pkcli/test.py

@@ -70,7 +70,9 @@ def default_command(*args):
     files.
 
     An argument which is ``case=<pattern>``, is passed to pytest
-    as ``-k <pattern>``.
+    as ``-k <pattern>``. An argument of the form ``<file>::<case>``
+    runs that specific test by node ID. These two forms are mutually
+    exclusive.
 
     ``skip_past=<last_to_ignore>`` causes collection to ignore all
     files up to and including ``<last_to_ignore>``` (may be partial
@@ -91,11 +93,12 @@ def default_command(*args):
 
 
 class _Case:
-    def __init__(self, rel_path, runner):
+    def __init__(self, entry, runner):
         self.timed_out_secs = 0
         self.runner = runner
-        self.rel_path = rel_path
-        self.abs_path = pkio.py_path(rel_path)
+        self.rel_path = entry.path
+        self.case_func = entry.case_func
+        self.abs_path = pkio.py_path(self.rel_path)
         self.tries = _MAX_RESTARTS if _cfg.restartable else 1
         self.run()
 
@@ -180,9 +183,7 @@ def _ignore_warnings():
             return rv
 
         def _remove_work_dir():
-            w = _TEST_PY.sub(pkunit.WORK_DIR_SUFFIX, self.rel_path)
-            if w != self.rel_path:
-                pkio.unchecked_remove(w)
+            pkio.unchecked_remove(pkunit.test_path_to_work_dir(self.rel_path))
 
         def _process():
             c = (
@@ -193,7 +194,7 @@ def _process():
                     "-v",
                     "-s",
                     "-rs",
-                    self.rel_path,
+                    self.case_func or self.rel_path,
                 ]
                 + self.runner.pytest_flags
             )
@@ -242,9 +243,9 @@ def _too_many_failures():
         self.failures = []
         self.cases = set()
         with _SignalCascade() as self.signal_cascade:
-            for p in self.rel_paths:
+            for v in self.rel_paths:
                 c += 1
-                self._run(p)
+                self._run(v)
                 if a := _too_many_failures():
                     break
             while self._wait_for_one(aborting=a):
@@ -253,12 +254,18 @@ def _too_many_failures():
         self.result = f"passed={c}"
 
     def _args(self, tests):
-        def _file(path):
+        def _case_funcs(case_funcs, cwd):
+            for p, c in case_funcs:
+                if not (t := pkio.py_path(p)).exists():
+                    pykern.pkcli.command_error("not found test={}", t)
+                _file(str(cwd.bestrelpath(t)), case_func=c)
+
+        def _file(path, case_func=None):
             if self.skip_past:
                 if self.skip_past in path:
                     self.skip_past = None
                 return
-            self.rel_paths.append(path)
+            self.rel_paths.append(PKDict(path=path, case_func=case_func))
 
         def _find(paths):
             i = re.compile(r"(?:_work|_data)/")
@@ -276,9 +283,11 @@ def _find(paths):
                         _file(p)
 
         def _flag(name, value):
+            rv = False
             if len(value) <= 0:
                 pykern.pkcli.command_error(f"empty value for option={name}")
             elif name == "case":
+                rv = True
                 self.pytest_flags.extend(("-k", value))
             elif name == "max_procs":
                 try:
@@ -297,6 +306,7 @@ def _flag(name, value):
                 self.skip_past = value
             else:
                 pykern.pkcli.command_error(f"unsupported option={name}")
+            return rv
 
         def _resolve_test_paths(paths, current_dir):
             if not paths:
@@ -306,17 +316,27 @@ def _resolve_test_paths(paths, current_dir):
                 paths = (p,)
             return paths
 
-        p = []
+        case_flag = False
+        paths = []
+        case_funcs = []
         self.pytest_flags = []
         self.max_procs = _cfg.max_procs
         self.skip_past = None
         for t in tests:
             if "=" in t:
-                _flag(*(t.split("=")))
+                case_flag = _flag(*(t.split("=")))
+            elif "::" in t:
+                v = t.split("::", 1)
+                case_funcs.append((v[0], t))
             else:
-                p.append(t)
+                paths.append(t)
         self.rel_paths = []
-        _find(p)
+        if case_funcs:
+            if case_flag:
+                pykern.pkcli.command_error("use case= or test::case, not both")
+            _case_funcs(case_funcs, pkio.py_path())
+        if paths or not case_funcs:
+            _find(paths)
 
     def _assert_failures(self, failures, count):
         if len(failures) <= 0:
@@ -343,15 +363,16 @@ def _info(self, case, lines):
                 # other output on its own line, ensure newline at end
                 lines[-1] += "\n"
         else:
+            v = case.case_func or case.rel_path
             if lines[0] == _FAIL_MSG:
                 # add the failure context
                 lines[0] += f" {case.output_path}"
             if self.max_procs > 1:
                 # line by line when multiprocess
-                lines[0] = case.rel_path + " " + lines[0]
+                lines[0] = v + " " + lines[0]
             elif lines[0] == _START_MSG:
                 # starting a case
-                lines[0] = case.rel_path
+                lines[0] = v
             else:
                 # completing a case
                 lines[0] = " " + lines[0]
@@ -363,8 +384,8 @@ def _info(self, case, lines):
         # TODO(robnagler) is this necessary?
         sys.stdout.flush()
 
-    def _run(self, rel_path):
-        c = _Case(rel_path, self)
+    def _run(self, entry):
+        c = _Case(entry, self)
         self.cases.add(c)
         self._info(c, [_START_MSG])
         if len(self.cases) >= self.max_procs:

pykern/pkio.py

@@ -226,7 +226,7 @@ def read_text(filename):
         filename (str or py.path.Local): File to open
 
     Returns:
-        Str: contents of `filename`
+        str: contents of `filename`
     """
     try:
         with open_text(filename) as f:

pykern/pkunit.py

@@ -522,6 +522,25 @@ def __exit__(self, *args):
         return False
 
 
+def test_path_to_work_dir(path):
+    """Convert a test file path to its work directory path.
+
+    Strips ``_test`` suffix or ``test_`` prefix from the basename and
+    appends ``_work``.
+
+    Args:
+        path (str or py.path.local): test file path ending in ``_test`` or starting with ``test_``
+
+    Returns:
+        py.path.local: work directory path
+    """
+    p = pkio.py_path(path)
+    b = _strip_test_affix(p.purebasename)
+    if b is None:
+        pkfail("{}: path must be a test file (_test suffix or test_ prefix)", p)
+    return p.new(basename=b + WORK_DIR_SUFFIX)
+
+
 def work_dir():
     """Returns ephemeral work directory, created if necessary.
 
@@ -538,7 +557,10 @@ def work_dir():
     Returns:
         py.path: directory name
     """
-    return _base_dir(WORK_DIR_SUFFIX).ensure(dir=True)
+    f = _test_file()
+    if not f:
+        raise PKFail("unable to find test file path; not running in pykern.pkcli.test?")
+    return test_path_to_work_dir(f).realpath().ensure(dir=True)
 
 
 class _FileEq:
@@ -747,8 +769,8 @@ def _base_dir(postfix):
     f = _test_file()
     if not f:
         raise PKFail("unable to find test file path; not running in pykern.pkcli.test?")
-    b = re.sub(r"_test$|^test_", "", f.purebasename)
-    assert b != f.purebasename, "{}: module name must end in _test".format(f)
+    b = _strip_test_affix(f.purebasename)
+    assert b is not None, "{}: module name must end in _test".format(f)
     return f.new(basename=b + postfix).realpath()
 
 
@@ -774,6 +796,13 @@ def _pkdlog(*args, **kwargs):
     pkdlog(*args, **kwargs)
 
 
+def _strip_test_affix(purebasename):
+    b = re.sub(r"_test$", "", purebasename)
+    if b == purebasename:
+        b = re.sub(r"^test_", "", purebasename)
+    return None if b == purebasename else b
+
+
 def _test_file():
     """Various ways to initialize _test_file"""
     global _init_test_file, __test_file

tests/pkcli/test_case_data/1.in/1_test.py

@@ -0,0 +1,13 @@
+"""fixture for test_case_test: one passing and one failing test
+
+:copyright: Copyright (c) 2026 RadiaSoft LLC.  All Rights Reserved.
+:license: http://www.apache.org/licenses/LICENSE-2.0.html
+"""
+
+
+def test_fail():
+    x = 1 / 0
+
+
+def test_pass():
+    pass

tests/pkcli/test_case_data/1.in/args

@@ -0,0 +1 @@
+1_test.py::test_pass

tests/pkcli/test_case_data/1.in/pkre

@@ -0,0 +1 @@
+passed=1

tests/pkcli/test_case_data/2.in/1_test.py

@@ -0,0 +1,13 @@
+"""fixture for test_case_test: one passing and one failing test
+
+:copyright: Copyright (c) 2026 RadiaSoft LLC.  All Rights Reserved.
+:license: http://www.apache.org/licenses/LICENSE-2.0.html
+"""
+
+
+def test_fail():
+    x = 1 / 0
+
+
+def test_pass():
+    pass

tests/pkcli/test_case_data/2.in/args

@@ -0,0 +1 @@
+1_test.py::test_fail 1_test.py::test_pass

tests/pkcli/test_case_data/2.in/pkre

@@ -0,0 +1 @@
+FAILED=1

tests/pkcli/test_case_data/2.out/pkexcept

@@ -0,0 +1 @@
+FAILED=1 passed=1

tests/pkcli/test_case_test.py

@@ -0,0 +1,17 @@
+"""tests for file::case node ID syntax in pkcli.test
+
+:copyright: Copyright (c) 2026 RadiaSoft LLC.  All Rights Reserved.
+:license: http://www.apache.org/licenses/LICENSE-2.0.html
+"""
+
+
+def test_file_case(capsys):
+    from pykern import pkunit, pkio
+    from pykern.pkcli import test
+
+    for d in pkunit.case_dirs():
+        with pkunit.ExceptToFile():
+            pkunit.pkre(
+                pkio.read_text(d.join("pkre")).strip(),
+                test.default_command(*pkio.read_text(d.join("args")).split()),
+            )