test(project): keep the #1851 and #1878 tests green on the KTLO legs (#1898, #1899)

tests/project/test_ast_unparse_is_gated_1899_unit.py

@@ -0,0 +1,128 @@
+# -*- coding: utf-8 -*-
+"""No test calls :func:`ast.unparse` outside a version gate (#1899).
+
+:func:`ast.unparse` is 3.9+, and the unit tier also runs on the
+``Python 3.8 (KTLO)`` leg, where the call raises :class:`AttributeError`. That
+leg reads green even when a test fails (#1679), so the failure shows only in its
+annotations and log, which is how the #1878 engine-stub tests broke there
+unnoticed.
+
+A test reads a node back as written with :func:`ast.get_source_segment`,
+compares it with :func:`ast.dump`, or matches its fields, all of which 3.8 has.
+One that has to render a node puts the call under a gate on
+:data:`sys.version_info`: an enclosing ``if``, or a decorator such as
+:func:`unittest.skipIf` on the test.
+
+This module is unit-tier: it parses source and imports nothing it scans.
+
+"""
+from __future__ import annotations
+
+import ast
+import unittest
+from typing import TYPE_CHECKING
+
+from tests._tiers import ROOT
+
+if TYPE_CHECKING:
+    from typing import Iterator
+
+#: Directory that is scanned: every module of it runs on the 3.8 leg.
+TESTS = ROOT / 'tests'
+
+
+def _reads_version_info(node: 'ast.AST') -> bool:
+    """Whether ``node`` reads ``version_info``, as ``sys.version_info`` or imported bare."""
+    return any((isinstance(sub, ast.Attribute) and sub.attr == 'version_info')
+               or (isinstance(sub, ast.Name) and sub.id == 'version_info') for sub in ast.walk(node))
+
+
+def _is_gate(node: 'ast.AST') -> bool:
+    """Whether ``node`` gates what it holds on the interpreter's version."""
+    if isinstance(node, (ast.If, ast.IfExp)):
+        return _reads_version_info(node.test)
+    if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
+        return any(_reads_version_info(decorator) for decorator in node.decorator_list)
+    return False
+
+
+def ungated(source: str) -> 'Iterator[int]':
+    """Yield the line of each use of :func:`ast.unparse` that no version gate encloses.
+
+    A use is ``<name>.unparse`` for a name :mod:`ast` is imported as, an
+    ``from ast import unparse``, or a load of the name that import binds. Imports
+    are collected from the whole file, function bodies included.
+
+    Args:
+        source: Module source text.
+
+    Yields:
+        The line number of each ungated use.
+
+    """
+    tree = ast.parse(source)
+    modules = set()  # type: set[str]
+    functions = set()  # type: set[str]
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            modules.update(alias.asname or alias.name for alias in node.names if alias.name == 'ast')
+        elif isinstance(node, ast.ImportFrom) and node.module == 'ast' and not node.level:
+            functions.update(alias.asname or alias.name for alias in node.names if alias.name == 'unparse')
+
+    def use(node: 'ast.AST') -> bool:
+        if isinstance(node, ast.Attribute):
+            return node.attr == 'unparse' and isinstance(node.value, ast.Name) and node.value.id in modules
+        if isinstance(node, ast.ImportFrom):
+            return node.module == 'ast' and not node.level and any(alias.name == 'unparse' for alias in node.names)
+        return isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load) and node.id in functions
+
+    def walk(node: 'ast.AST', gated: bool) -> 'Iterator[int]':
+        gated = gated or _is_gate(node)
+        if not gated and isinstance(node, (ast.expr, ast.stmt)) and use(node):
+            yield node.lineno
+        for child in ast.iter_child_nodes(node):
+            yield from walk(child, gated)
+
+    yield from walk(tree, False)
+
+
+class UnparseGateTests(unittest.TestCase):
+    """Every use of :func:`ast.unparse` under :file:`tests/` is gated to 3.9+."""
+
+    def test_no_test_calls_unparse_outside_a_version_gate(self) -> None:
+        found = []  # type: list[str]
+        for path in sorted(TESTS.rglob('*.py')):
+            relative = path.relative_to(ROOT).as_posix()
+            found.extend(f'{relative}:{lineno}' for lineno in ungated(path.read_text(encoding='utf-8')))
+        self.assertEqual(found, [], 'ast.unparse is 3.9+, so these fail on the 3.8 leg: read the node with '
+                                    'ast.get_source_segment or match its fields, or gate on sys.version_info')
+
+    def test_each_use_and_gate_is_matched(self) -> None:
+        """The scan sees each spelling of a use, and each gate, so a pass is not vacuous."""
+        source = (
+            'import ast\n'
+            'import ast as tree\n'
+            'import sys\n'
+            'import unittest\n'
+            'ast.unparse(node)\n'
+            'tree.unparse(node)\n'
+            'from ast import unparse\n'
+            'from ast import unparse as render\n'
+            'render(node)\n'
+            'if sys.version_info >= (3, 9):\n'
+            '    ast.unparse(node)\n'
+            'else:\n'
+            '    ast.dump(node)\n'
+            'text = ast.unparse(node) if sys.version_info >= (3, 9) else ast.dump(node)\n'
+            "@unittest.skipIf(sys.version_info < (3, 9), 'ast.unparse is 3.9+')\n"
+            'def test():\n'
+            '    return ast.unparse(node)\n'
+            'def helper(node):\n'
+            '    """ast.unparse is 3.9+"""\n'
+            '    return ast.get_source_segment(source, node), node.unparse\n'
+        )
+        self.assertEqual(list(ungated(source)), [5, 6, 7, 8, 9])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/project/test_conda_platform_pin_refresh_1851.py

@@ -21,6 +21,8 @@
   ``conda-update`` in :file:`.github/workflows/cron-conda.yml` installs
   requests and runs ``--refresh-pins`` between installing ``.[all]`` and
   writing the requirements, read with :func:`yaml.safe_load`.
+* Every test reaches the script through :func:`load_vendor`, which skips without
+  :mod:`tomllib`, so none fails on the 3.8 to 3.10 legs (#1898).
 
 """
 
@@ -67,16 +69,26 @@
 _MODULE = []  # type: list[Any]
 
 
-def vendor() -> 'Any':
-    """:file:`util/conda_vendor.py`, loaded once; skipped without :mod:`tomllib`, which it imports."""
+def load_vendor(name: 'str' = 'conda_vendor') -> 'Any':
+    """:file:`util/conda_vendor.py`, loaded afresh as ``name``; skipped without :mod:`tomllib`, which it imports.
+
+    Every test reaches the script through here, so none fails on the 3.8 to 3.10
+    legs, which lack :mod:`tomllib` (#1898): :class:`TomllibGuardTests`.
+
+    """
     if tomllib is None:  # pragma: no cover
         raise unittest.SkipTest('tomllib, which util/conda_vendor.py imports, needs Python 3.11+')
+    spec = importlib.util.spec_from_file_location(name, ROOT / 'util' / 'conda_vendor.py')
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)  # type: ignore[union-attr]
+    return module
+
+
+def vendor() -> 'Any':
+    """:file:`util/conda_vendor.py`, loaded once through :func:`load_vendor`."""
     if not _MODULE:
-        spec = importlib.util.spec_from_file_location('conda_vendor', ROOT / 'util' / 'conda_vendor.py')
-        assert spec is not None and spec.loader is not None
-        module = importlib.util.module_from_spec(spec)
-        spec.loader.exec_module(module)  # type: ignore[union-attr]
-        _MODULE.append(module)
+        _MODULE.append(load_vendor())
     return _MODULE[0]
 
 
@@ -398,11 +410,7 @@ def test_without_requests_only_the_request_fails(self) -> None:
 
         """
         with no_urllib(), mock.patch.dict(sys.modules, {'requests': None}):
-            spec = importlib.util.spec_from_file_location('conda_vendor_without_requests',
-                                                          ROOT / 'util' / 'conda_vendor.py')
-            assert spec is not None and spec.loader is not None
-            fresh = importlib.util.module_from_spec(spec)
-            spec.loader.exec_module(fresh)  # type: ignore[union-attr]
+            fresh = load_vendor('conda_vendor_without_requests')
             with self.assertRaises(ImportError):
                 fresh._fetch('cryptography')
             with tempfile.TemporaryDirectory() as temp:
@@ -635,5 +643,32 @@ def test_no_other_job_reads_pypi_for_the_pins(self) -> None:
                     self.assertNotIn('--refresh-pins', str(step.get('run', '')))
 
 
+class TomllibGuardTests(unittest.TestCase):
+    """Without :mod:`tomllib`, as on the 3.8 to 3.10 legs, every other test here passes or skips (#1898).
+
+    Those legs read green even when a test fails (#1679), so a test reaching the
+    script's module-level ``import tomllib`` other than through :func:`load_vendor`
+    shows only in their annotations. This runs the rest of the module with
+    :mod:`tomllib` gone instead, on any interpreter.
+
+    """
+
+    def test_every_other_test_passes_or_skips_without_tomllib(self) -> None:
+        module = sys.modules[__name__]
+        loader = unittest.defaultTestLoader
+        suite = unittest.TestSuite(
+            loader.loadTestsFromTestCase(case) for case in vars(module).values()
+            if isinstance(case, type) and issubclass(case, unittest.TestCase)
+            and case.__module__ == __name__ and case is not type(self))
+        result = unittest.TestResult()
+        with mock.patch.dict(sys.modules, {'tomllib': None}), mock.patch.object(module, 'tomllib', None), \
+                mock.patch.object(module, '_MODULE', []):
+            suite.run(result)
+        self.assertEqual([f'{test.id()}: {trace.strip().splitlines()[-1]}'
+                          for test, trace in result.errors + result.failures], [])
+        skipped = {test.id().rpartition('.')[2] for test, reason in result.skipped if 'tomllib' in reason}
+        self.assertIn('test_without_requests_only_the_request_fails', skipped)
+
+
 if __name__ == '__main__':
     unittest.main()

tests/project/test_engine_stubs_1878_unit.py

@@ -15,7 +15,9 @@
 annotations, the protocols covering what each engine reads, and the stubs
 covering what :mod:`pcapkit` imports.
 
-This module is unit-tier: it parses source and imports no engine package.
+This module is unit-tier: it parses source and imports no engine package. The
+unit tier runs on Python 3.8 too, so it reads nodes without :func:`ast.unparse`,
+which is 3.9+ (#1899).
 
 """
 from __future__ import annotations
@@ -76,36 +78,56 @@ def _class(tree: 'ast.Module', name: str) -> 'ast.ClassDef':
     raise AssertionError(f'no top-level class {name}')
 
 
+def _named(annotation: 'ast.expr') -> str:
+    """An annotation as the name it gives: a quoted one unquoted, a dotted one joined.
+
+    Only a protocol's name is looked up in what this returns, so any other form,
+    which names no protocol, is given as :func:`ast.dump` renders it.
+
+    """
+    if isinstance(annotation, ast.Constant) and isinstance(annotation.value, str):
+        return annotation.value
+    if isinstance(annotation, ast.Name):
+        return annotation.id
+    if isinstance(annotation, ast.Attribute):
+        return f'{_named(annotation.value)}.{annotation.attr}'
+    return ast.dump(annotation)
+
+
+def _is_protocol(base: 'ast.expr') -> bool:
+    """Whether a class base is written ``Protocol`` or ``typing.Protocol``."""
+    if isinstance(base, ast.Name):
+        return base.id == 'Protocol'
+    return (isinstance(base, ast.Attribute) and base.attr == 'Protocol'
+            and isinstance(base.value, ast.Name) and base.value.id == 'typing')
+
+
 def _expkg_annotation(cls: 'ast.ClassDef') -> 'Optional[str]':
     """The annotation of ``_expkg`` in the class body, a quoted one unquoted."""
     for statement in [*cls.body, *_guarded(cls.body)]:
         if (isinstance(statement, ast.AnnAssign) and isinstance(statement.target, ast.Name)
                 and statement.target.id == '_expkg'):
-            annotation = statement.annotation
-            if isinstance(annotation, ast.Constant) and isinstance(annotation.value, str):
-                return annotation.value
-            return ast.unparse(annotation)
+            return _named(statement.annotation)
     return None
 
 
 def _guarded_protocols(tree: 'ast.Module') -> 'dict[str, ast.ClassDef]':
     """Every class the module-level guard defines as a ``Protocol``, by name."""
     return {
         statement.name: statement for statement in _guarded(tree.body)
-        if isinstance(statement, ast.ClassDef)
-        and any(ast.unparse(base) in ('Protocol', 'typing.Protocol') for base in statement.bases)
+        if isinstance(statement, ast.ClassDef) and any(_is_protocol(base) for base in statement.bases)
     }
 
 
 def _members(cls: 'ast.ClassDef') -> 'dict[str, str]':
-    """A protocol's members, each mapped to the annotation of its value."""
+    """A protocol's members, each mapped to the annotation of its value, a quoted one unquoted."""
     members = {}
     for statement in cls.body:
         if isinstance(statement, ast.FunctionDef):
-            members[statement.name] = ast.unparse(statement.returns) if statement.returns else ''
+            members[statement.name] = _named(statement.returns) if statement.returns else ''
         elif isinstance(statement, ast.AnnAssign) and isinstance(statement.target, ast.Name):
-            members[statement.target.id] = ast.unparse(statement.annotation)
-    return {name: value.strip('\'"') for name, value in members.items()}
+            members[statement.target.id] = _named(statement.annotation)
+    return members
 
 
 def _is_expkg(node: 'ast.AST') -> bool: