diff --git a/tests/conftest.py b/tests/conftest.py index 27de232..44e1b4f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -17,6 +17,9 @@ from __future__ import annotations +import contextlib +import functools +import importlib import sys from pathlib import Path @@ -557,22 +560,218 @@ class StandInHost: SUBCOMMAND_EXTENSIONS: tuple = () -@_pytest.fixture(autouse=True) -def _a_hosts_plane_for_the_token_reading_suites(request): - if request.node.path.name not in HOST_PLANE_SUITES: - yield - return +@contextlib.contextmanager +def a_hosts_plane(host_factory): + """Register the host `host_factory()` builds with openDox's registry for + the length of the block, then put back what was registered before. + + The registration is dropped first, so the host's is made on a clean slate, + and the host is built only then. Whatever the block does, the registry is + emptied again afterwards and the earlier registration restored.""" from opendox import default_profile as _default_profile from opendox import domain_profile as _registry previous = _registry.current() if _registry.is_registered() else None _registry.unregister() - _registry.register(StandInHost()) try: - yield + yield _registry.register(host_factory()) finally: _registry.unregister() # The entry point's default is not put back as a registration: the # next entry point registers it again, as it does in any process. if previous is not None and previous is not _default_profile: _registry.register(previous) + + +@_pytest.fixture(autouse=True) +def _a_hosts_plane_for_the_token_reading_suites(request): + if plane_host_for(request.node.path.name) != "stand-in": + yield + return + with a_hosts_plane(StandInHost): + yield + + +# --------------------------------------------------------------------------- +# THE GOVERNED HOST FOR THE GOVERNED SUITES THAT DRIVE THE GATE VERBS (plan 038 +# T020, U-1; R2Q8 (a), openxFactory#656 comment 6003486656; P4F-5, ruled with +# plan 038 on comment 6013547504). +# +# #1144's 12.5 runs openxFactory's governed suites, each alone, "with the +# host's implementation registered". Eight of them drive openXdox's `gate` +# verbs, and with them its gate and projection routes. Those columns reach +# openDox only through a registered host: openDox's default profile +# contributes none, and `StandInHost` above contributes none either. So their +# cases read `invalid choice: 'gate'` or `unknown_action` (R2-INV-P4F's +# host-reg bucket and its one-off groups H and T). +# +# For the length of each test of the modules GOVERNED_HOST_SUITES names, this +# registers THE GOVERNED HOST, and puts the registry back afterwards, as the +# stand-in's fixture above does. Which host that is depends on the run: +# * Where the run is composed, with openxFactory's `scripts/` on the path +# (R1Q23 (a)), it is openxFactory's own composite, `opendox_host.profile()`: +# the object openxFactory's process-start call registers. It carries +# openXdox's gate and projection columns and openxFactory's own lane +# column, whose `POST /actions/refresh` two cases request. It is found as +# the home corpus's registration above finds its adapter: only a missing +# `opendox_host` itself means "absent". +# * Elsewhere it is `GovernedStandInHost`: openXdox's three columns and +# nothing else, as `tests/test_hermeticity_gate_verbs.py`'s `_GateHost` +# carries the gate verbs. Today these suites reach `doc_health`, so they +# run composed or not at all; this is the host they get wherever else they +# come to run. +# +# THE COMPOSITE ALONE, NOT openxFactory'S WHOLE START. `register_openxfactory()` +# also fills openxFactory's process-wide seams, and openxFactory's host has no +# call that takes them back, so they would outlive the test into every later +# module of the run. And it refuses while openXdox's own registry holds the +# profile the root `conftest.py` registers. This harness makes a host's other registrations +# itself (the profile, the home corpus), so only the composite is registered, +# with openDox's registry, as the stand-in is. +# +# AND openxFactory'S REACH STAYS OUT OF THIS PROCESS. Importing openxFactory's +# lane column runs `carved_reach.install()`, which puts openxFactory's pinned +# `openXdox/code/src` at the head of `sys.path`. `openxdox` is a namespace +# package, so every later `openxdox.*` import would come from that pinned copy +# and not from this checkout, and the run would measure another commit. So +# openXdox's columns are imported from this checkout first, the composite's +# facets are read once with `sys.path` saved, `sys.path` is put back, and an +# `openxdox` module imported from anywhere else refuses the case, by name. +# +# THE LIST IS THE SCAN'S, as HOST_PLANE_SUITES is. `tests/test_host_plane.py` +# holds it equal to the governed suites, computed as 12.5 computes them +# (`governed_suites()` below), that name a `gate` verb of openXdox's command +# line. A module on both lists runs under the governed host, which is a host's +# plane too (`plane_host_for`). +# --------------------------------------------------------------------------- + +GOVERNED_HOST_SUITES = frozenset({ + "test_branch_session.py", + "test_gateway_provenance.py", + "test_session_confinement.py", + "test_session_gates.py", + "test_session_lifecycle.py", + "test_session_notebook.py", + "test_session_transaction.py", + "test_session_verbs.py", +}) + +#: 12.5's governed set is every `tests/test_*.py` that names one of these +#: (its falsifier's `git grep -l -e … -- 'tests/test_*.py'`). They are spelled +#: here, in a file that pathspec does not match, so the check that holds the +#: list to its scan does not join the set it scans. +GOVERNED_SUITE_MARKERS: tuple[str, ...] = ("open-pr", "open_pr", "FakePullRequests") + + +def governed_suites(tests: Path = HERE) -> frozenset[str]: + """12.5's governed set: the `test_*.py` modules directly in `tests` that + name one of GOVERNED_SUITE_MARKERS.""" + return frozenset( + path.name for path in tests.glob("test_*.py") + if any(marker in path.read_text(encoding="utf-8") + for marker in GOVERNED_SUITE_MARKERS)) + + +def plane_host_for(suite: str) -> str | None: + """The host a test module's cases run under: "governed" for + GOVERNED_HOST_SUITES, "stand-in" for the rest of HOST_PLANE_SUITES, and + None for every other module. The governed host wins where a module is on + both lists, because it is a host's plane as well.""" + if suite in GOVERNED_HOST_SUITES: + return "governed" + if suite in HOST_PLANE_SUITES: + return "stand-in" + return None + + +#: openXdox's governed columns, as openxFactory's profile contributes them: +#: (the facet, the module, the class). Named rather than imported, because +#: each of the three modules reaches `doc_health`. +GOVERNED_COLUMNS: tuple[tuple[str, str, str], ...] = ( + ("SUBCOMMAND_EXTENSIONS", "openxdox.cli_gate", "GateSubcommands"), + ("ROUTE_EXTENSIONS", "openxdox.serve_gate", "GateRoutesExtension"), + ("ROUTE_EXTENSIONS", "openxdox.serve_projection", "ProjectionRoutesExtension"), +) + + +def _governed_columns(facet: str) -> tuple: + return tuple(getattr(importlib.import_module(module), name)() + for each, module, name in GOVERNED_COLUMNS if each == facet) + + +class GovernedStandInHost: + """A host's profile that carries openXdox's governed columns and nothing + else. Each facet is built on first read, so the host itself can be made + where `doc_health` cannot be imported.""" + + @functools.cached_property + def SUBCOMMAND_EXTENSIONS(self) -> tuple: # noqa: N802 (the facet's name) + return _governed_columns("SUBCOMMAND_EXTENSIONS") + + @functools.cached_property + def ROUTE_EXTENSIONS(self) -> tuple: # noqa: N802 (the facet's name) + return _governed_columns("ROUTE_EXTENSIONS") + + +def openxfactory_host(): + """openxFactory's host module (`scripts/opendox_host.py`) where the run is + composed, and None where it is not. Only a missing `opendox_host` itself + means "absent": a present one that cannot be imported fails here, loudly.""" + try: + return importlib.import_module("opendox_host") + except ModuleNotFoundError as absent: + if absent.name != "opendox_host": + raise + return None + + +#: The facets openDox's parser and server builders read off a host's profile. +#: Reading them is what imports a composite's columns. +BUILT_FACETS: tuple[str, ...] = ( + "SUBCOMMAND_EXTENSIONS", "ROUTE_EXTENSIONS", "HANDLER_CONTRIBUTIONS") + + +def _refuse_an_openxdox_from_elsewhere() -> None: + """Every `openxdox` module this process holds is this checkout's, or the + case is refused, naming each module that is not.""" + home = Path(_domain_profile.__file__).resolve().parent + elsewhere = sorted( + name for name, module in list(sys.modules.items()) + if name.startswith("openxdox.") + and getattr(module, "__file__", None) + and not Path(module.__file__).resolve().is_relative_to(home)) + if elsewhere: + raise RuntimeError( + f"{elsewhere} were imported from outside this checkout's openxdox " + f"({home}), so the run would measure another copy of them. " + "openxFactory's reach puts its pinned openXdox leg ahead of this " + "checkout when its lane column is imported; import the module " + "from this checkout before the composite's facets are read.") + + +def governed_host(): + """THE GOVERNED HOST for one case: openxFactory's composite where the run + is composed, and `GovernedStandInHost` where it is not.""" + host_module = openxfactory_host() + if host_module is None: + return GovernedStandInHost() + for _, module, _ in GOVERNED_COLUMNS: + importlib.import_module(module) + saved = list(sys.path) + try: + composite = host_module.profile() + for facet in BUILT_FACETS: + getattr(composite, facet) + finally: + sys.path[:] = saved + _refuse_an_openxdox_from_elsewhere() + return composite + + +@_pytest.fixture(autouse=True) +def _the_governed_host_for_the_governed_suites(request): + if plane_host_for(request.node.path.name) != "governed": + yield + return + with a_hosts_plane(governed_host): + yield diff --git a/tests/test_host_plane.py b/tests/test_host_plane.py index 5045cb5..9520b83 100644 --- a/tests/test_host_plane.py +++ b/tests/test_host_plane.py @@ -1,24 +1,36 @@ """The suites that read the console token run on a host's plane (plan 034 T086, -step (b); openDox-code's T104, RULED openxFactory#656 comment 5963851934). +step (b); openDox-code's T104, RULED openxFactory#656 comment 5963851934), and +the governed suites that drive openXdox's gate verbs run under the governed +host (plan 038 T020, U-1; R2Q8 (a)). `tests/conftest.py`'s `_a_hosts_plane_for_the_token_reading_suites` registers a stand-in host for the length of each test of the modules `HOST_PLANE_SUITES` names, because from T104 openDox publishes the console token on -`/capabilities` only on a host's plane. These cases hold that list and that -host to their reasons, and they run in a lone checkout, so in F9.1. +`/capabilities` only on a host's plane. Its +`_the_governed_host_for_the_governed_suites` registers the governed host, the +same way, for the modules `GOVERNED_HOST_SUITES` names. These cases hold both +lists and both hosts to their reasons, and they run in a lone checkout, so in +F9.1. None of them needs openxFactory: its host is stood in where a case needs +one. A CREATED file: no manifest row (RULED OQ-C). """ from __future__ import annotations +import ast import importlib.util import inspect import re +import sys +import types from pathlib import Path +import pytest + import conftest as tests_conftest from opendox import default_profile +from opendox import domain_profile as registry TESTS = Path(__file__).resolve().parent @@ -81,3 +93,191 @@ def test_on_the_stand_in_hosts_plane_the_token_stays_on_capabilities() -> None: source = inspect.getsource(serve_mod.build_server) assert "capabilities[CONSOLE_TOKEN_FIELD] = console_token" in source assert "delivery_for" not in source + + +# --------------------------------------------------------------------------- +# THE GOVERNED HOST (plan 038 T020, U-1). +# --------------------------------------------------------------------------- + +SOURCE = TESTS.parent / "src" + +#: A `gate` verb, as a test module spells it: the string `gate` and, after the +#: comma, the verb, the way an argument vector names a subcommand. +_GATE_VERB = re.compile(r"""["']gate["']\s*,\s*["']([a-z][a-z-]*)["']""") + + +def _gate_verbs() -> frozenset[str]: + """The verbs openXdox's `gate` command declares: every name + `src/openxdox/cli_gate.py` hands to `add_parser`, less `gate` itself. Read + from the source, so a lone checkout can read it without `doc_health`.""" + tree = ast.parse((SOURCE / "openxdox" / "cli_gate.py").read_text(encoding="utf-8")) + names = {node.args[0].value for node in ast.walk(tree) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "add_parser" and node.args + and isinstance(node.args[0], ast.Constant) + and isinstance(node.args[0].value, str)} + return frozenset(names - {"gate"}) + + +def _drives_a_gate_verb(path: Path, verbs: frozenset[str]) -> bool: + return any(verb in verbs + for verb in _GATE_VERB.findall(path.read_text(encoding="utf-8"))) + + +def test_the_governed_list_is_every_governed_suite_that_drives_a_gate_verb() -> None: + verbs = _gate_verbs() + assert {"create-document", "edit-document", "abandon-session"} <= verbs + governed = tests_conftest.governed_suites() + assert len(governed) >= 16, "12.5's floor: fewer means a proof vanished" + scanned = {name for name in governed + if _drives_a_gate_verb(TESTS / name, verbs)} + listed = set(tests_conftest.GOVERNED_HOST_SUITES) + assert scanned == listed, ( + f"drive a gate verb but are not listed: {sorted(scanned - listed)}; " + f"listed but drive none: {sorted(listed - scanned)}. A governed suite " + "(12.5's set) that drives openXdox's gate verbs needs a host carrying " + "openXdox's columns, so it joins tests/conftest.py's " + "GOVERNED_HOST_SUITES") + # The scan is 12.5's own set, and not every module that drives a verb: + # outside the governed set a suite registers its own host, or none. + assert any(_drives_a_gate_verb(path, verbs) + for path in TESTS.glob("test_*.py") if path.name not in governed) + + +def test_a_module_on_both_lists_runs_under_the_governed_host() -> None: + governed = tests_conftest.GOVERNED_HOST_SUITES + plane = tests_conftest.HOST_PLANE_SUITES + assert governed & plane, "no module is on both lists, so nothing tests the order" + for name in governed: + assert tests_conftest.plane_host_for(name) == "governed", name + for name in plane - governed: + assert tests_conftest.plane_host_for(name) == "stand-in", name + assert tests_conftest.plane_host_for(Path(__file__).name) is None + + +def _stand_in_columns(monkeypatch) -> list[str]: + """A module object for each of openXdox's governed columns, standing in for + the real one, which reaches `doc_health`. Each carries its named class, and + the harness's imports of them are recorded, in order, in the list returned. + Nothing else is imported differently, and `sys.modules` is not touched.""" + modules: dict[str, types.ModuleType] = {} + for _, module, name in tests_conftest.GOVERNED_COLUMNS: + stand_in = modules.setdefault(module, types.ModuleType(module)) + setattr(stand_in, name, type(name, (), {})) + imported: list[str] = [] + real_import = importlib.import_module + + def import_module(name: str, package: str | None = None): + if name in modules: + imported.append(name) + return modules[name] + return real_import(name, package) + + monkeypatch.setattr(tests_conftest, "importlib", + types.SimpleNamespace(import_module=import_module)) + return imported + + +def test_the_governed_stand_in_carries_openxdoxs_three_columns(monkeypatch) -> None: + """openXdox's share of openxFactory's profile: its gate verbs, and its gate + and projection routes. Each named class is defined where it is named.""" + assert tests_conftest.GOVERNED_COLUMNS == ( + ("SUBCOMMAND_EXTENSIONS", "openxdox.cli_gate", "GateSubcommands"), + ("ROUTE_EXTENSIONS", "openxdox.serve_gate", "GateRoutesExtension"), + ("ROUTE_EXTENSIONS", "openxdox.serve_projection", "ProjectionRoutesExtension"), + ) + for _, module, name in tests_conftest.GOVERNED_COLUMNS: + source = (SOURCE / Path(*module.split("."))).with_suffix(".py") + tree = ast.parse(source.read_text(encoding="utf-8")) + assert name in {node.name for node in tree.body + if isinstance(node, ast.ClassDef)}, (module, name) + + _stand_in_columns(monkeypatch) + monkeypatch.setitem(sys.modules, "opendox_host", None) # not composed + host = tests_conftest.governed_host() + assert isinstance(host, tests_conftest.GovernedStandInHost) + assert [type(c).__name__ for c in host.SUBCOMMAND_EXTENSIONS] == ["GateSubcommands"] + assert [type(c).__name__ for c in host.ROUTE_EXTENSIONS] == [ + "GateRoutesExtension", "ProjectionRoutesExtension"] + assert host.ROUTE_EXTENSIONS is host.ROUTE_EXTENSIONS + assert host is not default_profile + if importlib.util.find_spec("opendox.console_access") is not None: + from opendox import console_access + + assert (console_access.delivery_for(host) + == console_access.DELIVERY_CAPABILITIES) + + +class _Composite: + """openxFactory's composite, stood in. Reading a built facet records + whether openXdox's columns were already imported, and then does what + importing openxFactory's lane column does: puts a pinned leg at the head + of `sys.path`.""" + + def __init__(self, leg: Path, imported: list[str], on_read=None) -> None: + self.leg, self.imported, self.on_read, self.reads = leg, imported, on_read, [] + + def __getattr__(self, name: str): + if name not in tests_conftest.BUILT_FACETS: + raise AttributeError(name) + columns = {module for _, module, _ in tests_conftest.GOVERNED_COLUMNS} + self.reads.append((name, columns <= set(self.imported))) + if str(self.leg) not in sys.path: + sys.path.insert(0, str(self.leg)) + if self.on_read is not None: + self.on_read() + return () + + +def _composed(monkeypatch, composite: _Composite) -> None: + host_module = types.ModuleType("opendox_host") + host_module.profile = lambda: composite + monkeypatch.setitem(sys.modules, "opendox_host", host_module) + + +def test_composed_the_governed_host_is_the_composite_and_the_reach_is_put_back( + monkeypatch, tmp_path) -> None: + imported = _stand_in_columns(monkeypatch) + composite = _Composite(tmp_path / "openXdox" / "code" / "src", imported) + _composed(monkeypatch, composite) + before = list(sys.path) + earlier = tests_conftest.StandInHost() + with tests_conftest.a_hosts_plane(lambda: earlier): + with tests_conftest.a_hosts_plane(tests_conftest.governed_host) as host: + assert host is composite + assert registry.current() is composite + assert sys.path == before + assert registry.current() is earlier + assert [name for name, _ in composite.reads] == list(tests_conftest.BUILT_FACETS) + assert all(imported for _, imported in composite.reads), ( + "a facet was read before openXdox's columns were imported from this " + "checkout") + + +def test_an_openxdox_module_from_elsewhere_refuses_the_case(monkeypatch, tmp_path) -> None: + imported = _stand_in_columns(monkeypatch) + elsewhere = types.ModuleType("openxdox.from_a_pinned_leg") + elsewhere.__file__ = str(tmp_path / "openxdox" / "from_a_pinned_leg.py") + + def import_from_the_leg(): + monkeypatch.setitem(sys.modules, elsewhere.__name__, elsewhere) + + composite = _Composite(tmp_path / "leg", imported, on_read=import_from_the_leg) + _composed(monkeypatch, composite) + before = list(sys.path) + with pytest.raises(RuntimeError, match=r"openxdox\.from_a_pinned_leg"): + tests_conftest.governed_host() + assert sys.path == before + + +def test_openxfactorys_host_is_absent_only_when_its_own_module_is(monkeypatch, tmp_path) -> None: + monkeypatch.setitem(sys.modules, "opendox_host", None) + assert tests_conftest.openxfactory_host() is None + monkeypatch.delitem(sys.modules, "opendox_host") + (tmp_path / "opendox_host.py").write_text( + "import a_dependency_the_host_lacks\n", encoding="utf-8") + monkeypatch.syspath_prepend(str(tmp_path)) + with pytest.raises(ModuleNotFoundError) as refused: + tests_conftest.openxfactory_host() + assert refused.value.name == "a_dependency_the_host_lacks"