From c52900e9eb80324fcecceaf6705a06c057a4d243 Mon Sep 17 00:00:00 2001 From: Brett Heap <1513478+brettheap@users.noreply.github.com> Date: Tue, 6 Oct 2026 10:39:02 +0000 Subject: [PATCH 1/7] T020, U-1: the governed host for the governed suites that drive the gate verbs (plan 038) tests/conftest.py registers THE GOVERNED HOST for the length of each test of the eight governed suites that drive openXdox's gate verbs (GOVERNED_HOST_SUITES): openxFactory's composite, opendox_host.profile(), where the run is composed (R1Q23 (a)), and GovernedStandInHost, which carries openXdox's three governed columns, elsewhere. The composite is registered with openDox's registry alone, so no openxFactory seam outlives the test. Its facets are read with sys.path saved and restored, because importing openxFactory's lane column runs carved_reach.install(), which would otherwise put openxFactory's pinned openXdox leg ahead of this checkout's namespace package. An openxdox module from anywhere else refuses the case by name. tests/test_host_plane.py holds the list to its scan (12.5's governed set that names a gate verb of cli_gate.py) and the mechanism to its word, all in a lone checkout. Both lists share one registration helper, a_hosts_plane; a module on both runs under the governed host. Composed, 12.5's sixteen suites go from 174 red to 109: host-reg 58 and one-off group H 7 turn green, and no green node turns red. Refs opensoft/openxFactory#656 (CLAIMED 6013608299). Lane: openXfactory-3 Co-Authored-By: Claude Opus 5.5 --- tests/conftest.py | 213 +++++++++++++++++++++++++++++++++++++-- tests/test_host_plane.py | 206 ++++++++++++++++++++++++++++++++++++- 2 files changed, 409 insertions(+), 10 deletions(-) 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" From 073615822921fe7b562f0256607a39d5deb44116 Mon Sep 17 00:00:00 2001 From: Brett Heap <1513478+brettheap@users.noreply.github.com> Date: Tue, 6 Oct 2026 11:01:10 +0000 Subject: [PATCH 2/7] T021, U-2: the gate console's schemas, each read where its owner keeps it, nothing vendored (plan 038) gate_console._validate_contract_document read both of its schemas from parents[2] / "contracts" / "schemas", which in this leg is the checkout root and holds no contracts/. Now each schema is read where its owner keeps it: * gate-action-record.schema.yaml, openXdox-spec's, from this distribution's own packaged copy, through contracts.verified_bytes(), which checks it against copies.yaml's digest first; * demotion-execution-receipt.schema.yaml, openxFactory's, from a source the HOST registers with the new register_contract_schema_source(). The same directory again is a no-op, a different one is refused, and a directory without the schema is refused when it is registered. unregister_contract_schema_source() and contract_schema_source_registered() complete the seam. With nothing registered, a receipt refuses, naming the call (R1Q27 (a)). Nothing is added to copies.yaml, and test_packaged_validator.py is unchanged and green. The seam sits near the end of the module, so that no line before it moves. tests/test_no_hardcoded_status_words.py pins seven prose f-strings in this module by line number. tests/conftest.py registers openxFactory's contracts/schemas/ from the composed tree, at process start, where the run is composed; a lone checkout registers none. tests/test_gate_console_schema_source.py runs the console in a child interpreter beside a stand-in doc_health, so it runs in a lone checkout. Composed, the 16 schema nodes and test_explicit_release_has_cli_and_http_parity turn green on T020's host. 12.5's suites read 92 red of the 174, and no green node turns red. Refs opensoft/openxFactory#656 (CLAIMED 6013608628). Lane: openXfactory-3 Co-Authored-By: Claude Opus 5.5 --- src/openxdox/gate_console.py | 138 ++++++++++++-- tests/conftest.py | 16 ++ tests/test_gate_console_schema_source.py | 217 +++++++++++++++++++++++ 3 files changed, 359 insertions(+), 12 deletions(-) create mode 100644 tests/test_gate_console_schema_source.py diff --git a/src/openxdox/gate_console.py b/src/openxdox/gate_console.py index 640de0c..b6dbe45 100644 --- a/src/openxdox/gate_console.py +++ b/src/openxdox/gate_console.py @@ -631,22 +631,22 @@ def build_gate_action_record( def _validate_contract_document( - document: Mapping[str, Any], *, schema_filename: str, label: str, + document: Mapping[str, Any], *, schema_text: str, label: str, ) -> None: - """Validate one document against a repository-pinned released schema.""" + """Validate one document against a released schema's text, which its + caller reads where the schema's owner keeps it: `_packaged_schema_text` + or `_host_schema_text`, under "THE CONTRACT SCHEMAS THE CONSOLE VALIDATES + AGAINST" near the end of this module (plan 038 T021).""" try: from jsonschema import Draft202012Validator, FormatChecker except ImportError as exc: # pragma: no cover - release dependency guard raise GateRefused( f"jsonschema is required to validate {label}") from exc - schema_path = (Path(__file__).resolve().parents[2] / "contracts" / "schemas" - / schema_filename) try: - schema = yaml.safe_load(schema_path.read_text(encoding="utf-8")) - except (OSError, yaml.YAMLError) as exc: + schema = yaml.safe_load(schema_text) + except yaml.YAMLError as exc: raise GateRefused( - f"the {label} schema could not be loaded from {schema_path}: " - f"{exc}") from exc + f"the {label} schema could not be parsed: {exc}") from exc errors = sorted( Draft202012Validator(schema, format_checker=FormatChecker()).iter_errors( dict(document)), @@ -661,8 +661,8 @@ def _validate_contract_document( def validate_gate_action_record(record: Mapping[str, Any]) -> None: """Validate one record against the repository-pinned released schema.""" - _validate_contract_document( - record, schema_filename="gate-action-record.schema.yaml", + _validate_contract_document(record, schema_text=_packaged_schema_text( + GATE_ACTION_RECORD_SCHEMA, label="gate-action record"), label="gate-action record") if record.get("action") != ACTION_CLEANUP_ABANDONED_BRANCH: return @@ -707,8 +707,8 @@ def validate_gate_action_record(record: Mapping[str, Any]) -> None: def validate_demotion_execution_receipt(receipt: Mapping[str, Any]) -> None: - _validate_contract_document( - receipt, schema_filename="demotion-execution-receipt.schema.yaml", + _validate_contract_document(receipt, schema_text=_host_schema_text( + DEMOTION_RECEIPT_SCHEMA, label="demotion execution receipt"), label="demotion execution receipt") destination = receipt.get("destination") if not isinstance(destination, Mapping) or str(destination.get("path") or "") \ @@ -2348,6 +2348,120 @@ def edit_project(self, project_id: str, *, add=None, remove=None, at=at, records_dir=self.records_dir, provenance=provenance, **extra) +# -------------------------------------------------------------------------- +# THE CONTRACT SCHEMAS THE CONSOLE VALIDATES AGAINST, AND WHERE EACH IS READ +# (plan 038 T021, U-2; R2Q8 (a); P4F-4, ruled with plan 038 on +# openxFactory#656 comment 6013547504; R1Q27 (a), comment 5851950767). +# +# Two documents are validated here, against two schemas with two owners: +# * a gate-action record, against openXdox-spec's +# `gate-action-record.schema.yaml`. This distribution packages it in +# `openxdox.contracts`, as one of the consumer validator's own three kinds, +# held to its digest by `copies.yaml`. So it is read from there, through +# `contracts.verified_bytes()`, wherever this package runs; +# * a demotion execution receipt, against openxFactory's +# `demotion-execution-receipt.schema.yaml`. That schema is the HOST's +# contract: openXdox-spec carries only its examples, and the copy record +# admits the validator's three kinds and no fourth +# (`tests/test_packaged_validator.py`), so nothing of it is vendored here. +# A host that writes receipts registers, once, at process start, the +# directory its schema is in, with `register_contract_schema_source()`. +# With none registered, validating a receipt refuses, naming that call: +# R1Q27 (a)'s "only where the tree it runs from supplies their schemas". +# +# Both used to be read from `parents[2] / "contracts" / "schemas"` beside +# this module. That was openxFactory's own tree before the carve. In this +# leg it is the checkout's root, which holds no `contracts/`. +# -------------------------------------------------------------------------- + +#: The gate-action record schema's copy id in `openxdox.contracts`. +GATE_ACTION_RECORD_SCHEMA = "gate-action-record" +#: The demotion execution receipt schema's file name in the host's source. +DEMOTION_RECEIPT_SCHEMA = "demotion-execution-receipt.schema.yaml" + +_contract_schema_source: Path | None = None + + +class ContractSchemaSourceRefused(RuntimeError): + """A contract schema source the gate console cannot take, refused when it + is registered rather than when a receipt is first validated.""" + + +def register_contract_schema_source(source: str | Path) -> Path: + """Register the host's contract schema directory. Returns it, resolved. + + The directory must hold `demotion-execution-receipt.schema.yaml`, the one + schema the console reads from a host. Registering the same directory again + is a no-op, so an idempotent host start is not punished. A DIFFERENT one is + refused: one process whose receipts are validated against two schemas, + depending on which registration a caller reached, is the failure one + registration exists to prevent. Call `unregister_contract_schema_source()` + first if the swap is deliberate.""" + global _contract_schema_source + directory = Path(source).resolve() + if not (directory / DEMOTION_RECEIPT_SCHEMA).is_file(): + raise ContractSchemaSourceRefused( + f"{directory} holds no {DEMOTION_RECEIPT_SCHEMA}, so the gate " + "console cannot read a demotion execution receipt's schema from " + "it. Register the directory the host keeps its contract schemas " + "in, its contracts/schemas/.") + if (_contract_schema_source is not None + and _contract_schema_source != directory): + raise ContractSchemaSourceRefused( + f"a contract schema source is already registered " + f"({_contract_schema_source}), and {directory} would replace it. " + "Registration happens once, at process start. Call " + "openxdox.gate_console.unregister_contract_schema_source() first " + "if the swap is deliberate.") + _contract_schema_source = directory + return directory + + +def unregister_contract_schema_source() -> None: + """Drop the registration. For test isolation and for a host tearing down.""" + global _contract_schema_source + _contract_schema_source = None + + +def contract_schema_source_registered() -> bool: + """Is a contract schema source registered, without reading anything?""" + return _contract_schema_source is not None + + +def _packaged_schema_text(copy_id: str, *, label: str) -> str: + """A schema this distribution packages, once its digest has been checked.""" + from . import contracts + + try: + return contracts.verified_bytes(copy_id).decode("utf-8") + except (contracts.CopyRefused, UnicodeDecodeError) as exc: + raise GateRefused( + f"the {label} schema could not be read from openXdox's packaged " + f"copy: {exc}") from exc + + +def _host_schema_text(filename: str, *, label: str) -> str: + """A schema the host supplies, from the source it registered.""" + source = _contract_schema_source + if source is None: + raise GateRefused( + f"the {label} cannot be validated: its schema, {filename}, is the " + "host's contract, and no contract schema source is registered. A " + "host registers the directory it keeps its contract schemas in, " + "once, at process start, with " + "openxdox.gate_console.register_contract_schema_source()" + " (R1Q27 (a): a kind is validated only where the tree it runs " + "from supplies its schema).") + path = source / filename + try: + return path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError) as exc: + raise GateRefused( + f"the {label} schema could not be loaded from {path}: " + f"{exc}") from exc + + + # --------------------------------------------------------------------------- # BACKWARD-COMPATIBLE MODULE ATTRIBUTES, RESOLVED ON FIRST ACCESS (PEP 562). # diff --git a/tests/conftest.py b/tests/conftest.py index 44e1b4f..5c0ad6a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -775,3 +775,19 @@ def _the_governed_host_for_the_governed_suites(request): return with a_hosts_plane(governed_host): yield + + +# THE GATE CONSOLE'S CONTRACT SCHEMA SOURCE, REGISTERED AS A HOST REGISTERS IT +# (plan 038 T021, U-2; P4F-4). A demotion execution receipt's schema is +# openxFactory's contract, and the gate console reads it only from a source a +# host registers (`gate_console.register_contract_schema_source`). Where the +# run is composed, this registers openxFactory's `contracts/schemas/` from the +# composed tree, at process start, as openxFactory's own host wiring does from +# T030. A lone checkout registers none, and there a receipt refuses by name +# (R1Q27 (a)). +_OPENXFACTORY_HOST = openxfactory_host() +if _OPENXFACTORY_HOST is not None: + from openxdox import gate_console as _gate_console # noqa: E402 + + _gate_console.register_contract_schema_source( + Path(_OPENXFACTORY_HOST.__file__).resolve().parents[1] / "contracts" / "schemas") diff --git a/tests/test_gate_console_schema_source.py b/tests/test_gate_console_schema_source.py new file mode 100644 index 0000000..79152db --- /dev/null +++ b/tests/test_gate_console_schema_source.py @@ -0,0 +1,217 @@ +"""The gate console's contract schemas, each read where its owner keeps it +(plan 038 T021, U-2; R2Q8 (a); P4F-4, ruled with plan 038 on openxFactory#656 +comment 6013547504; R1Q27 (a), comment 5851950767). + +`openxdox.gate_console` validates two documents against two schemas with two +owners: + * a gate-action record, against openXdox-spec's schema, which this + distribution packages (`openxdox.contracts`, held to its digest by + `copies.yaml`). The console reads it from there; + * a demotion execution receipt, against openxFactory's schema, which nothing + here vendors. The console reads it from the source a host registers with + `register_contract_schema_source()`, and with none registered it refuses, + naming that call. + +THESE CASES RUN IN A LONE CHECKOUT, so in F9.1 and in the required job. The +console imports openxFactory's `doc_health` when it loads (R1Q6 (d)'s reason), +though nothing these cases run calls it. So each case runs the console in a +CHILD interpreter, beside a stand-in `doc_health` that answers the three names +the console's import reads and nothing else. This process never imports the +console through the stand-in. The child's import path is set whole, so in a +composed run openxFactory's `doc_health` is not on it either, and no source is +registered there unless the case registers one. + +A CREATED file: no manifest row (RULED OQ-C). +""" + +from __future__ import annotations + +import ast +import json +import os +import subprocess +import sys +import textwrap +from pathlib import Path + +SOURCE = Path(__file__).resolve().parents[1] / "src" +GATE_CONSOLE = SOURCE / "openxdox" / "gate_console.py" + +#: The names the console's import reads from `doc_health`: `corpus` for +#: `STATUS_SCAN_LINES` (gate_console.py), and `lines` for `split_keepends` and +#: `join_rows` (round_trip.py). Nothing these cases run calls either function. +_STAND_IN = { + "__init__.py": "", + "corpus.py": "STATUS_SCAN_LINES = 0\n", + "lines.py": ("def split_keepends(text):\n" + " raise AssertionError('the doc_health stand-in')\n\n\n" + "def join_rows(rows):\n" + " raise AssertionError('the doc_health stand-in')\n"), +} + +#: Run first in every child: the console, a valid gate-action record and a +#: receipt, and `refusal()`, which answers None where the call passes and the +#: refusal's type and words where it refuses. +_PRELUDE = ''' +import json +import shutil +import sys +from pathlib import Path + +from openxdox import contracts +from openxdox import gate_console as gc + +out = {} +RECORD = gc.build_gate_action_record( + actor="brett", action="demote", at="2026-10-06T00:00:00Z", reason="why", + change_id="add-x", artifacts=[{"kind": "transition-manifest", + "reference": "ideation/staging/x/m.yaml"}]) +RECEIPT = {"kind": "from-a", "destination": {"id": "x", "path": "ideation/staging/x"}, + "transition_manifest": "ideation/staging/x/m.yaml"} + + +def refusal(call, *args): + try: + call(*args) + except (gc.GateRefused, gc.ContractSchemaSourceRefused) as exc: + return f"{type(exc).__name__}: {exc}" + return None +''' + + +def _in_a_child(tmp_path: Path, body: str) -> dict: + """Run `body` after the prelude in a fresh interpreter whose import path is + the `doc_health` stand-in and this checkout's `src/`, and answer the + `out` mapping it prints.""" + stand_in = tmp_path / "stand-in" / "doc_health" + stand_in.mkdir(parents=True) + for name, text in _STAND_IN.items(): + (stand_in / name).write_text(text, encoding="utf-8") + script = tmp_path / "case.py" + script.write_text(_PRELUDE + textwrap.dedent(body) + + "\nprint(json.dumps(out))\n", encoding="utf-8") + env = {key: value for key, value in os.environ.items() + if key != "PYTHONPATH"} + env["PYTHONPATH"] = os.pathsep.join((str(stand_in.parent), str(SOURCE))) + env["PYTHONDONTWRITEBYTECODE"] = "1" + run = subprocess.run([sys.executable, str(script)], cwd=tmp_path, env=env, + capture_output=True, text=True, check=False) + assert run.returncode == 0, run.stderr + return json.loads(run.stdout.splitlines()[-1]) + + +def _receipt_source(directory: Path, *, kind: str, extra: dict | None = None) -> Path: + """A host's contract schema directory, stood in: its receipt schema admits + one `kind`. `extra` adds further schema files beside it.""" + directory.mkdir(parents=True) + schema = {"type": "object", "required": ["kind", "destination"], + "properties": {"kind": {"const": kind}}} + (directory / "demotion-execution-receipt.schema.yaml").write_text( + json.dumps(schema), encoding="utf-8") + for name, text in (extra or {}).items(): + (directory / name).write_text(text, encoding="utf-8") + return directory + + +def test_a_receipt_refuses_by_name_where_no_host_registered_its_schema(tmp_path) -> None: + out = _in_a_child(tmp_path, ''' + out["registered"] = gc.contract_schema_source_registered() + out["refusal"] = refusal(gc.validate_demotion_execution_receipt, RECEIPT) + ''') + assert out["registered"] is False + assert out["refusal"].startswith("GateRefused: the demotion execution receipt cannot be validated") + assert "demotion-execution-receipt.schema.yaml" in out["refusal"] + assert ("openxdox.gate_console.register_contract_schema_source()" + in out["refusal"]) + + +def test_a_gate_action_record_is_read_from_the_packaged_copy(tmp_path) -> None: + """With no source registered the record validates; a record the schema + does not admit refuses; a registered source whose own record schema admits + nothing is never read for a record; and a packaged copy that drifted from + its digest refuses.""" + hostile = _receipt_source(tmp_path / "hostile", kind="from-a", extra={ + "gate-action-record.schema.yaml": json.dumps({"not": {}})}) + out = _in_a_child(tmp_path, f''' + out["valid"] = refusal(gc.validate_gate_action_record, RECORD) + out["not_admitted"] = refusal(gc.validate_gate_action_record, + dict(RECORD, kind="another-kind")) + gc.register_contract_schema_source({str(hostile)!r}) + out["beside_a_hostile_source"] = refusal(gc.validate_gate_action_record, RECORD) + staged = Path({str(tmp_path / "staged")!r}) + shutil.copytree(contracts.package_dir(), staged, + ignore=shutil.ignore_patterns("__pycache__", "*.py")) + drifted = staged / "schemas" / "gate-action-record.schema.yaml" + drifted.write_bytes(drifted.read_bytes() + b"\\n") + contracts.package_dir = lambda: staged + out["drifted"] = refusal(gc.validate_gate_action_record, RECORD) + ''') + assert out["valid"] is None + assert out["not_admitted"].startswith( + "GateRefused: the gate-action record failed schema validation: kind") + assert out["beside_a_hostile_source"] is None + assert out["drifted"].startswith( + "GateRefused: the gate-action record schema could not be read from " + "openXdox's packaged copy") + assert "is not the spec leg's file" in out["drifted"] + + +def test_a_receipt_is_validated_against_the_registered_sources_schema(tmp_path) -> None: + source = _receipt_source(tmp_path / "host" / "contracts" / "schemas", kind="from-a") + out = _in_a_child(tmp_path, f''' + out["returned"] = str(gc.register_contract_schema_source({str(source)!r})) + out["registered"] = gc.contract_schema_source_registered() + out["valid"] = refusal(gc.validate_demotion_execution_receipt, RECEIPT) + out["not_admitted"] = refusal(gc.validate_demotion_execution_receipt, + dict(RECEIPT, kind="from-b")) + ''') + assert out["returned"] == str(source.resolve()) + assert out["registered"] is True + assert out["valid"] is None + assert out["not_admitted"].startswith( + "GateRefused: the demotion execution receipt failed schema validation: kind") + + +def test_the_seam_holds_one_source_and_refuses_a_second(tmp_path) -> None: + first = _receipt_source(tmp_path / "a", kind="from-a") + second = _receipt_source(tmp_path / "b", kind="from-b") + empty = tmp_path / "c" + empty.mkdir() + out = _in_a_child(tmp_path, f''' + out["first"] = str(gc.register_contract_schema_source(Path({str(first)!r}))) + out["again"] = str(gc.register_contract_schema_source({str(first)!r} + "/.")) + out["second"] = refusal(gc.register_contract_schema_source, {str(second)!r}) + out["still_first"] = refusal(gc.validate_demotion_execution_receipt, RECEIPT) + gc.unregister_contract_schema_source() + out["after_unregister"] = gc.contract_schema_source_registered() + out["swapped"] = str(gc.register_contract_schema_source({str(second)!r})) + out["now_second"] = refusal(gc.validate_demotion_execution_receipt, + dict(RECEIPT, kind="from-b")) + gc.unregister_contract_schema_source() + out["no_schema"] = refusal(gc.register_contract_schema_source, {str(empty)!r}) + out["after_no_schema"] = gc.contract_schema_source_registered() + ''') + assert out["first"] == out["again"] == str(first.resolve()) + assert out["second"].startswith( + "ContractSchemaSourceRefused: a contract schema source is already registered") + assert str(first.resolve()) in out["second"] and str(second.resolve()) in out["second"] + assert out["still_first"] is None + assert out["after_unregister"] is False + assert out["swapped"] == str(second.resolve()) + assert out["now_second"] is None + assert out["no_schema"].startswith( + f"ContractSchemaSourceRefused: {empty.resolve()} holds no " + "demotion-execution-receipt.schema.yaml") + assert out["after_no_schema"] is False + + +def test_the_console_reads_no_schema_from_beside_the_checkout() -> None: + """Before T021 both schemas were read from `parents[2]` of the module, + which in this leg is the checkout root. No `parents` index is read in the + console's code any more (its comments may still say where it was).""" + tree = ast.parse(GATE_CONSOLE.read_text(encoding="utf-8")) + reaches = [node.lineno for node in ast.walk(tree) + if isinstance(node, ast.Subscript) + and isinstance(node.value, ast.Attribute) + and node.value.attr == "parents"] + assert reaches == [], f"gate_console.py reads a parents index at lines {reaches}" From ecf0ba9bd4f2b54af66b7022ac1fc8c68005a89c Mon Sep 17 00:00:00 2001 From: Brett Heap <1513478+brettheap@users.noreply.github.com> Date: Tue, 6 Oct 2026 11:20:19 +0000 Subject: [PATCH 3/7] T021: a host schema the console cannot apply refuses as GateRefused (Copilot r4194497045) A registered source's receipt schema was checked only for existence. A document that is valid YAML but no valid JSON Schema escaped as a raw exception: an empty file raised AttributeError, `required: 1` raised TypeError, and an unresolvable $ref raised jsonschema's referencing error. cli_gate's demotion reports only GateRefused, BoundaryViolation or OSError as "executed but receipt not written", after files have moved, so any of these ended the verb with a traceback. _schema_errors now parses the schema, checks it against the validator's own meta-schema, and applies it. A failure at any of the three refuses as GateRefused. _validate_contract_document keeps its line count, so no line that test_no_hardcoded_status_words.py's PROSE_EXEMPT pins moves. A new lone case covers the three schemas, and two more mutants are killed. Refs opensoft/openxFactory#656 (CLAIMED 6013608628). Lane: openXfactory-3 Co-Authored-By: Claude Opus 5.5 --- src/openxdox/gate_console.py | 49 ++++++++++++++++++++---- tests/test_gate_console_schema_source.py | 35 +++++++++++++++++ 2 files changed, 76 insertions(+), 8 deletions(-) diff --git a/src/openxdox/gate_console.py b/src/openxdox/gate_console.py index b6dbe45..eba3773 100644 --- a/src/openxdox/gate_console.py +++ b/src/openxdox/gate_console.py @@ -636,20 +636,20 @@ def _validate_contract_document( """Validate one document against a released schema's text, which its caller reads where the schema's owner keeps it: `_packaged_schema_text` or `_host_schema_text`, under "THE CONTRACT SCHEMAS THE CONSOLE VALIDATES - AGAINST" near the end of this module (plan 038 T021).""" + AGAINST" near the end of this module (plan 038 T021). + + A schema that is not YAML, is not a valid JSON Schema, or cannot be + applied is refused there too, by `_schema_errors`, as `GateRefused`. So + a host's broken schema reaches a caller as a failing document does, and + never as another exception.""" try: from jsonschema import Draft202012Validator, FormatChecker except ImportError as exc: # pragma: no cover - release dependency guard raise GateRefused( f"jsonschema is required to validate {label}") from exc - try: - schema = yaml.safe_load(schema_text) - except yaml.YAMLError as exc: - raise GateRefused( - f"the {label} schema could not be parsed: {exc}") from exc errors = sorted( - Draft202012Validator(schema, format_checker=FormatChecker()).iter_errors( - dict(document)), + _schema_errors(document, schema_text, label=label, + validator=Draft202012Validator, formats=FormatChecker), key=lambda error: tuple(str(part) for part in error.absolute_path)) if errors: detail = "; ".join( @@ -2440,6 +2440,39 @@ def _packaged_schema_text(copy_id: str, *, label: str) -> str: f"copy: {exc}") from exc +def _schema_errors(document: Mapping[str, Any], schema_text: str, *, + label: str, validator: Any, formats: Any) -> list: + """`document`'s errors against the schema `schema_text` holds. + + The schema itself is checked first, and refused as `GateRefused` where it + is not YAML, where it is not a valid JSON Schema under the validator's own + meta-schema (an empty document, `required: 1`), and where applying it + raises (an unresolvable `$ref`). A host supplies the receipt schema, and + the demotion verb writes its receipt after it has moved files: it reports + a `GateRefused` there as an executed demotion whose receipt was not + written, where any other exception would end it with a traceback.""" + from jsonschema.exceptions import SchemaError + + try: + schema = yaml.safe_load(schema_text) + except yaml.YAMLError as exc: + raise GateRefused( + f"the {label} schema could not be parsed: {exc}") from exc + try: + validator.check_schema(schema) + except SchemaError as exc: + raise GateRefused( + f"the {label} schema is not a valid JSON Schema: {exc.message}" + ) from exc + try: + return list(validator(schema, format_checker=formats()).iter_errors( + dict(document))) + except Exception as exc: # noqa: BLE001 - any failure to apply it refuses + raise GateRefused( + f"the {label} schema could not be applied: " + f"{type(exc).__name__}: {exc}") from exc + + def _host_schema_text(filename: str, *, label: str) -> str: """A schema the host supplies, from the source it registered.""" source = _contract_schema_source diff --git a/tests/test_gate_console_schema_source.py b/tests/test_gate_console_schema_source.py index 79152db..2427cf0 100644 --- a/tests/test_gate_console_schema_source.py +++ b/tests/test_gate_console_schema_source.py @@ -205,6 +205,41 @@ def test_the_seam_holds_one_source_and_refuses_a_second(tmp_path) -> None: assert out["after_no_schema"] is False +def test_a_host_schema_the_console_cannot_apply_refuses_as_gate_refused(tmp_path) -> None: + """A registered source whose receipt schema is empty, is not a valid JSON + Schema, or cannot be applied refuses as `GateRefused`, never as another + exception. The demotion verb writes its receipt after it has moved files, + and reports only a `GateRefused` there as a receipt not written (Copilot + on #40, r4194497045).""" + sources = {} + for name, text in {"empty": "", "not_a_schema": "required: 1\n", + "unresolvable": json.dumps({"$ref": "#/$defs/absent"})}.items(): + directory = tmp_path / name + directory.mkdir() + (directory / "demotion-execution-receipt.schema.yaml").write_text( + text, encoding="utf-8") + sources[name] = str(directory) + out = _in_a_child(tmp_path, f''' + for name, source in {sources!r}.items(): + gc.unregister_contract_schema_source() + gc.register_contract_schema_source(source) + try: + gc.validate_demotion_execution_receipt(RECEIPT) + out[name] = None + except gc.GateRefused as exc: + out[name] = f"GateRefused: {{exc}}" + except Exception as exc: + out[name] = f"ESCAPED {{type(exc).__name__}}: {{exc}}" + ''') + for name in ("empty", "not_a_schema"): + assert out[name].startswith( + "GateRefused: the demotion execution receipt schema is not a valid " + "JSON Schema"), out[name] + assert out["unresolvable"].startswith( + "GateRefused: the demotion execution receipt schema could not be " + "applied"), out["unresolvable"] + + def test_the_console_reads_no_schema_from_beside_the_checkout() -> None: """Before T021 both schemas were read from `parents[2]` of the module, which in this leg is the checkout root. No `parents` index is read in the From 542a5c401c0384b267d20295c25b216b202c45f6 Mon Sep 17 00:00:00 2001 From: Brett Heap <1513478+brettheap@users.noreply.github.com> Date: Tue, 6 Oct 2026 17:27:00 +0000 Subject: [PATCH 4/7] T021: a host schema the safe loader cannot read refuses as GateRefused (Copilot r4197914632) yaml.safe_load raises RecursionError, for a nesting deeper than the interpreter allows, and ValueError, for an impossible date such as 2001-02-30, past YAMLError. _schema_errors caught only YAMLError, so a registered receipt schema holding either payload escaped as a raw exception after the demotion had moved files. It now catches the same three classes openxdox.contracts' record reader does, and refuses as GateRefused, "could not be parsed". tests/test_gate_console_schema_source.py adds three payloads to its cannot-apply case: an unclosed flow sequence (a YAMLError, the control), a 5000-deep sequence, and the impossible date. Before the fix the case reads "ESCAPED RecursionError: maximum recursion depth exceeded". With it, all six cases pass. Dropping RecursionError from the tuple, or dropping ValueError, is killed by that same case. Refs opensoft/openxFactory#656 (CLAIMED 6013608628). Lane: openXfactory-3 Co-Authored-By: Claude Opus 5.5 --- src/openxdox/gate_console.py | 12 ++++++++---- tests/test_gate_console_schema_source.py | 20 ++++++++++++++------ 2 files changed, 22 insertions(+), 10 deletions(-) diff --git a/src/openxdox/gate_console.py b/src/openxdox/gate_console.py index eba3773..5cc71e5 100644 --- a/src/openxdox/gate_console.py +++ b/src/openxdox/gate_console.py @@ -2445,9 +2445,13 @@ def _schema_errors(document: Mapping[str, Any], schema_text: str, *, """`document`'s errors against the schema `schema_text` holds. The schema itself is checked first, and refused as `GateRefused` where it - is not YAML, where it is not a valid JSON Schema under the validator's own - meta-schema (an empty document, `required: 1`), and where applying it - raises (an unresolvable `$ref`). A host supplies the receipt schema, and + is not YAML the safe loader can read (a `YAMLError`, or the + `RecursionError` of a nesting deeper than the interpreter allows, or the + `ValueError` of an impossible date, which the loader raises past + `YAMLError`, as `openxdox.contracts` reads its record), where it is not a + valid JSON Schema under the validator's own meta-schema (an empty + document, `required: 1`), and where applying it raises (an unresolvable + `$ref`). A host supplies the receipt schema, and the demotion verb writes its receipt after it has moved files: it reports a `GateRefused` there as an executed demotion whose receipt was not written, where any other exception would end it with a traceback.""" @@ -2455,7 +2459,7 @@ def _schema_errors(document: Mapping[str, Any], schema_text: str, *, try: schema = yaml.safe_load(schema_text) - except yaml.YAMLError as exc: + except (yaml.YAMLError, RecursionError, ValueError) as exc: raise GateRefused( f"the {label} schema could not be parsed: {exc}") from exc try: diff --git a/tests/test_gate_console_schema_source.py b/tests/test_gate_console_schema_source.py index 2427cf0..6499cfe 100644 --- a/tests/test_gate_console_schema_source.py +++ b/tests/test_gate_console_schema_source.py @@ -206,14 +206,18 @@ def test_the_seam_holds_one_source_and_refuses_a_second(tmp_path) -> None: def test_a_host_schema_the_console_cannot_apply_refuses_as_gate_refused(tmp_path) -> None: - """A registered source whose receipt schema is empty, is not a valid JSON - Schema, or cannot be applied refuses as `GateRefused`, never as another - exception. The demotion verb writes its receipt after it has moved files, - and reports only a `GateRefused` there as a receipt not written (Copilot - on #40, r4194497045).""" + """A registered source whose receipt schema the safe loader cannot read, + is empty, is not a valid JSON Schema, or cannot be applied refuses as + `GateRefused`, never as another exception. The demotion verb writes its + receipt after it has moved files, and reports only a `GateRefused` there + as a receipt not written (Copilot on #40, r4194497045 and r4197914632: + the loader raises `RecursionError` and `ValueError` past `YAMLError`).""" sources = {} for name, text in {"empty": "", "not_a_schema": "required: 1\n", - "unresolvable": json.dumps({"$ref": "#/$defs/absent"})}.items(): + "unresolvable": json.dumps({"$ref": "#/$defs/absent"}), + "not_yaml": "type: [object\n", + "too_deep": "[" * 5000 + "]" * 5000, + "impossible_date": "type: object\nx: 2001-02-30\n"}.items(): directory = tmp_path / name directory.mkdir() (directory / "demotion-execution-receipt.schema.yaml").write_text( @@ -238,6 +242,10 @@ def test_a_host_schema_the_console_cannot_apply_refuses_as_gate_refused(tmp_path assert out["unresolvable"].startswith( "GateRefused: the demotion execution receipt schema could not be " "applied"), out["unresolvable"] + for name in ("not_yaml", "too_deep", "impossible_date"): + assert out[name].startswith( + "GateRefused: the demotion execution receipt schema could not be " + "parsed"), out[name] def test_the_console_reads_no_schema_from_beside_the_checkout() -> None: From 7ad77ae8e918ca4dc84fecd83f8bcda180db2ecc Mon Sep 17 00:00:00 2001 From: Brett Heap <1513478+brettheap@users.noreply.github.com> Date: Tue, 6 Oct 2026 17:53:11 +0000 Subject: [PATCH 5/7] T095: the composed harness for the declared integration tests (plan 038; T094's map R2-INV-R9) The composed conftest T094's map names (R2-INV-R9, slice T095), under the holder's rulings on #656 comment 6021830531 (OQ-R9-1, OQ-R9-2 and OQ-R9-4 at their defaults; W2). Every piece is composed-only: where opendox_host is absent, which is every lone checkout, nothing is registered, built or set, so a lone run and tests/test_declared_exclusion.py's evidence are unchanged. The interface the other slices code against, in tests/conftest.py: - HR: DECLARED_HOST_SUITES (17 declared files), declared_host_for(), and the autouse fixture that registers the governed host for them, set up inside the stand-in host's plane; - SR: HOST_SEAM_SUITES, HEALTH_CHECK_SEAM, BINDING_TRUST_SEAM, openxfactory_seams() and a_hosts_seam(): openxFactory's health check for test_authoring_agent.py and its binding-trust policy for test_doxchat_model_intake.py, read off opendox_host.seams() and put back after each test; - CS: CONTRACTS_DIR, the contract family's composed farm (55 schemas through carved_reach.sources_under, 139 examples from the three owners less openDox's opendox-snapshot-*), built outside the checkout at import and exported in the environment; build_contracts_farm() and compose_the_contracts_farm(); - OQ-R9-4: openxfactory_root(); - LI (W2): LOCAL_INSTALL_SUITES and runs_a_local_install(), which T086's launch-suite fixture now reads. tests/test_host_plane.py holds each list to its rules (declared entries, outside 12.5's governed set, test_column_contributions_governed.py and any file registering openXdox's columns kept off the declared-host list) and each piece to its behaviour, in a lone checkout. Lane: openXfactory-3 Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/conftest.py | 313 +++++++++++++++++++++++++++++- tests/test_host_plane.py | 410 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 722 insertions(+), 1 deletion(-) diff --git a/tests/conftest.py b/tests/conftest.py index 5c0ad6a..eb66870 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -454,7 +454,9 @@ def unregistered_profile(): @_pytest.fixture(autouse=True) def _the_launch_suite_runs_a_local_install(request, monkeypatch): - if request.node.path.name != _LAUNCH_SUITE: + # W2 (plan 038 T095) extends this premise, composed, to the suites + # `runs_a_local_install` names (LOCAL_INSTALL_SUITES, at the end of file). + if not runs_a_local_install(request.node.path.name): yield return from opendox import cli as _cli @@ -791,3 +793,312 @@ def _the_governed_host_for_the_governed_suites(request): _gate_console.register_contract_schema_source( Path(_OPENXFACTORY_HOST.__file__).resolve().parents[1] / "contracts" / "schemas") + + +# --------------------------------------------------------------------------- +# THE COMPOSED HARNESS FOR THE DECLARED INTEGRATION TESTS (plan 038 T095; T094's +# map, R2-INV-R9, and the holder's rulings on it, openxFactory#656 comment +# 6021830531: OQ-R9-1, OQ-R9-2 and OQ-R9-4 at their defaults, and W2). +# +# ARC-Q2 (a) (comment 6016982816) made the composed workflow the permanent home +# of the declared exclusion: each file `tests/declared_exclusion.yaml` lists +# runs there alone, with openxFactory's `scripts/` on the path. T094 mapped +# every composed red of those files outside 12.5's set. Four of its means are +# this harness's, and the test files' own respellings code against them: +# * HR: the governed host, for the files DECLARED_HOST_SUITES names; +# * SR: two of openxFactory's host seams, for the files HOST_SEAM_SUITES +# names; +# * CS: the contract family's composed source, a farm outside the checkout, +# named by CONTRACTS_DIR; +# * LI: the launch suite's local install, for LOCAL_INSTALL_SUITES too. +# And OQ-R9-4's `openxfactory_root()`, for the files that read openxFactory's +# own tooling or documents in the composed tree. +# +# ALL OF IT IS COMPOSED-ONLY. Each piece acts only where `opendox_host` is +# importable, which no lone checkout supplies. There `openxfactory_root()` and +# CONTRACTS_DIR are None, and nothing is registered, built or set. So a lone +# run, and the evidence `tests/test_declared_exclusion.py` holds each listed +# file to, are what they were. `tests/test_host_plane.py` holds each list to +# its rules and each piece to its behaviour, in a lone checkout. +# --------------------------------------------------------------------------- + +import atexit as _atexit # noqa: E402 +import os as _os # noqa: E402 +import shutil as _shutil # noqa: E402 +import tempfile as _tempfile # noqa: E402 + + +def openxfactory_root() -> Path | None: + """The composed openxFactory tree's root where the run is composed, and + None where it is not (OQ-R9-4): the directory above openxFactory's host + module, as T021's registration above reaches its `contracts/schemas/`.""" + host_module = openxfactory_host() + if host_module is None: + return None + return Path(host_module.__file__).resolve().parents[1] + + +# HR. THE GOVERNED HOST FOR THE DECLARED INTEGRATION TESTS THAT NEED A HOST'S +# COLUMNS (OQ-R9-1). Each of these files builds openDox's parser, server or +# canvas through what the REGISTERED host contributes: openXdox's `gate` verbs +# and gate and projection routes, openxFactory's lane column +# (`POST /actions/refresh`), the plane `document-abstract` asks for, or the +# profile and display a host serves. With no host carrying them, T094 measured +# them composed reading `invalid choice: 'gate'`, `unknown_action`, +# `model_capability_unavailable` or `ProfileNotRegistered`. They are not 12.5's +# governed suites, so `GOVERNED_HOST_SUITES` cannot carry them: this list sits +# beside it. +# +# WHERE THE RUN IS COMPOSED, for the length of each of their tests, this +# registers THE GOVERNED HOST, `governed_host()` above, and puts the registry +# back afterwards. It is set up after the stand-in host's fixture, which it +# requests, so a file on both lists runs under the governed host, which is a +# host's plane too, and the stand-in is restored after it. Where the run is not +# composed it registers nothing, so a lone run keeps the stand-in or nothing, +# as before. +# +# THE LIST IS T094's MEASUREMENT, not a scan. No reading of a file's text +# separates these from declared files that name the same routes in a static +# check, or build a plane that needs no host's column (`test_renderer.py`, +# `test_doxbench_transport.py`), or bring their own (`test_hermeticity_gate_verbs.py`). +# So `tests/test_host_plane.py` holds it to the rules a list of this kind can be +# held to: every member is a declared entry, none is a governed suite, and +# `test_column_contributions_governed.py` is never on it. That file registers +# openXdox's columns itself, and T094 measured two of its cases red under the +# governed host, where two bindings claim `POST /actions/gate/` (OQ-R9-1). +# `test_explorer_viewer.py` and `test_doxbench_scope.py` are on it for the +# display a host serves (VR, under R9-R1 (a)); `test_gate_console.py` for its +# two cases that parse `gate`. +DECLARED_HOST_SUITES = frozenset({ + "test_canvas.py", + "test_create_document_cli.py", + "test_create_project.py", + "test_doxbench_abstract_route.py", + "test_doxbench_request_handling.py", + "test_doxbench_scope.py", + "test_edit_project.py", + "test_explorer_viewer.py", + "test_gate_console.py", + "test_gate_failure_diagnostics.py", + "test_generated_at_anchor.py", + "test_notebook_action.py", + "test_register_edit_lane.py", + "test_repo_selector.py", + "test_session_document_ownership.py", + "test_trust_gaps.py", + "test_wheel_verbs_cli.py", +}) + + +def declared_host_for(suite: str): + """The host a declared-host suite's cases run under: `governed_host` + where the run is composed and `suite` is on DECLARED_HOST_SUITES, and + None otherwise, which registers nothing.""" + if suite in DECLARED_HOST_SUITES and openxfactory_host() is not None: + return governed_host + return None + + +@_pytest.fixture(autouse=True) +def _the_governed_host_for_the_declared_host_suites( + request, _a_hosts_plane_for_the_token_reading_suites): + host = declared_host_for(request.node.path.name) + if host is None: + yield + return + with a_hosts_plane(host): + yield + + +# SR. TWO OF openxFactory'S HOST SEAMS, FOR THE FILES THAT MEET THEM. openDox +# answers its scoped health check and its model-binding trust from seams a +# host fills at process start. openxFactory's host fills both, and with neither +# filled T094 measured `test_authoring_agent.py`'s scaffold case reading the +# health check `not-available`, and `test_doxchat_model_intake.py`'s approval +# case reading its approved binding "not trusted on this machine", which the +# governed policy decides by the declaration instead. +# +# WHERE THE RUN IS COMPOSED, for the length of each test of a file +# HOST_SEAM_SUITES names, this registers what openxFactory's host registers at +# each named seam, and puts the seam back afterwards, as T073 registers the +# rail. What is registered is read off `opendox_host.seams()`, the table the +# host's process start registers from, so a later host pin is followed. A +# registration held before the test is dropped for it, and a host's is +# registered again after it. openDox's own default is not, since its consumers +# register it again, as `a_hosts_plane` treats the default profile. +HEALTH_CHECK_SEAM = ("opendox.workbench", "register_health_check") +BINDING_TRUST_SEAM = ("opendox.doxbench_trust", "register") + +HOST_SEAM_SUITES: dict[str, tuple[tuple[str, str], ...]] = { + "test_authoring_agent.py": (HEALTH_CHECK_SEAM,), + "test_doxchat_model_intake.py": (BINDING_TRUST_SEAM,), +} + +#: For each seam: the global that holds its registration, the global that marks +#: it openDox's default (None where the seam has no default), and the call that +#: empties it. +HOST_SEAM_HOLDERS: dict[tuple[str, str], tuple[str, str | None, str]] = { + HEALTH_CHECK_SEAM: ("_health_check", None, "unregister_health_check"), + BINDING_TRUST_SEAM: ("_registered", "_is_default", "unregister"), +} + + +def openxfactory_seams(wanted) -> tuple: + """The rows of openxFactory's host seam table for the seams `wanted` names, + each `(module, registration call, what it registers)`, where the run is + composed, and none where it is not.""" + host_module = openxfactory_host() + if host_module is None or not wanted: + return () + saved = list(sys.path) + try: + table = tuple(host_module.seams()) + finally: + sys.path[:] = saved + _refuse_an_openxdox_from_elsewhere() + rows = [] + for seam in wanted: + found = [row for row in table if (row[0].__name__, row[1]) == seam] + if len(found) != 1: + raise RuntimeError( + f"openxFactory's host seam table names {len(found)} rows for " + f"{'.'.join(seam)}, so what its host registers there is not one " + "thing this harness can register for it") + rows.append(found[0]) + return tuple(rows) + + +@contextlib.contextmanager +def a_hosts_seam(module, call: str, what): + """Register `what` at the seam `module.call` for the length of the block, + then put the seam back as it was, less openDox's default.""" + held, is_default, unregister = HOST_SEAM_HOLDERS[(module.__name__, call)] + previous = getattr(module, held, None) + a_hosts = previous is not None and not ( + is_default is not None and getattr(module, is_default, False)) + getattr(module, unregister)() + try: + yield getattr(module, call)(what) + finally: + getattr(module, unregister)() + if a_hosts: + getattr(module, call)(previous) + + +@_pytest.fixture(autouse=True) +def _the_hosts_seams_for_the_seam_suites(request): + rows = openxfactory_seams(HOST_SEAM_SUITES.get(request.node.path.name, ())) + if not rows: + yield + return + with contextlib.ExitStack() as stack: + for module, call, what in rows: + stack.enter_context(a_hosts_seam(module, call, what)) + yield + + +# CS. THE CONTRACT FAMILY'S COMPOSED SOURCE (OQ-R9-2). The family's validator +# is this leg's (`scripts/validate-ideation-dashboard-contracts.py` and its +# packaged copy), and its schemas and examples sit with three owners in the +# composed tree: openxFactory, openDox-spec and openXdox-spec. The validator +# takes "the family's other seven" schemas from the ONE directory +# `CONTRACTS_DIR` names, and its examples from beside it. With nothing named, +# T094 measured the family's tests unresolvable composed. +# +# WHERE THE RUN IS COMPOSED, the farm is built once, at this conftest's import, +# and CONTRACTS_DIR is set to it in this process's environment, before any test +# imports the validator (it reads the variable at import) or starts it (a +# subprocess inherits it). The module's CONTRACTS_DIR names the same directory. +# The farm is laid out openxFactory-style, as openxFactory's own composed +# validator farm is (`doxbench_contracts._composed_validator`): +# * `/contracts/schemas/`: a link to each schema the carve manifest +# resolves under `contracts/schemas/`, wherever the shed left it +# (`carved_reach.sources_under`), then to each of openxFactory's own +# schemas that is not linked yet; +# * `/examples/ideation-dashboard/`: a link to each YAML example the +# three owners keep under `examples/ideation-dashboard/`, at the same +# relative path, less openDox's own `opendox-snapshot-*` examples, whose +# kind the family's validator does not serve. Two owners naming one path +# is refused, never resolved. +# It is OUTSIDE the checkout, in a temporary directory removed at exit: the +# validator refuses a `contracts/` that links out of its tree (`:157-163`). +# The reach is read with `sys.path` saved and put back, and it must be the +# composed tree's own. +CONTRACT_EXAMPLE_LEFT_OUT = "opendox-snapshot-" + + +def build_contracts_farm(reach, farm: Path) -> Path: + """Lay the family's composed source out under `farm`, from openxFactory's + reach `reach` (`carved_reach`), and answer `/contracts`, the + directory CONTRACTS_DIR names.""" + schemas = farm / "contracts" / "schemas" + schemas.mkdir(parents=True) + for key, resolved in sorted(reach.sources_under("contracts/schemas").items()): + name = key.rsplit("/", 1)[-1] + if key == f"contracts/schemas/{name}": + (schemas / name).symlink_to(resolved) + for path in sorted((reach.REPO_ROOT / "contracts" / "schemas").glob("*.yaml")): + if not (schemas / path.name).exists(): + (schemas / path.name).symlink_to(path) + examples = farm / "examples" / "ideation-dashboard" + examples.mkdir(parents=True) + linked: dict[Path, Path] = {} + for owner in (reach.REPO_ROOT, reach.MOUNTS["opendox_spec"], + reach.MOUNTS["openxdox_spec"]): + base = owner / "examples" / "ideation-dashboard" + for path in sorted(base.rglob("*.yaml")): + if path.name.startswith(CONTRACT_EXAMPLE_LEFT_OUT): + continue + relative = path.relative_to(base) + if relative in linked: + raise RuntimeError( + f"two owners keep the example {relative}: {linked[relative]} " + f"and {path}. The farm links one file per path, so it " + "refuses rather than choose") + linked[relative] = path + (examples / relative).parent.mkdir(parents=True, exist_ok=True) + (examples / relative).symlink_to(path) + return farm / "contracts" + + +def compose_the_contracts_farm(environ, *, under: str | None = None) -> Path | None: + """Where the run is composed: build the farm in a new temporary directory + (in `under`, or the system's), set `environ["CONTRACTS_DIR"]` to it, and + answer it. Where it is not: answer None, and touch nothing.""" + root = openxfactory_root() + if root is None: + return None + saved = list(sys.path) + try: + reach = importlib.import_module("carved_reach") + finally: + sys.path[:] = saved + if Path(reach.REPO_ROOT).resolve() != root: + raise RuntimeError( + f"carved_reach was imported from {reach.__file__}, not from the " + f"composed tree at {root}, so the farm would read another tree") + farm = Path(_tempfile.mkdtemp(prefix="openxdox-contracts-farm-", dir=under)) + _atexit.register(_shutil.rmtree, farm, ignore_errors=True) + contracts = build_contracts_farm(reach, farm) + environ["CONTRACTS_DIR"] = str(contracts) + return contracts + + +CONTRACTS_DIR = compose_the_contracts_farm(_os.environ) + + +# LI. THE LAUNCH SUITE'S LOCAL INSTALL, EXTENDED (W2). `test_renderer.py` and +# `test_repo_root_guard.py` drive `generate-and-open` too, and T094 measured +# four of their cases composed reading "generate-and-open refused: +# OPENDOX_OIDC_ISSUER is required ... this install is HOSTED". W2 extends T086's +# Q4 (a) premise to them: `_the_launch_suite_runs_a_local_install` above runs +# for each file `runs_a_local_install` answers. For these two, only where the +# run is composed; for the launch suite, everywhere, as before. +LOCAL_INSTALL_SUITES = frozenset({"test_renderer.py", "test_repo_root_guard.py"}) + + +def runs_a_local_install(suite: str) -> bool: + """Whether `suite`'s cases run on the local install the launch suite's + fixture stands in.""" + return suite == _LAUNCH_SUITE or ( + suite in LOCAL_INSTALL_SUITES and openxfactory_host() is not None) diff --git a/tests/test_host_plane.py b/tests/test_host_plane.py index 9520b83..a8a6590 100644 --- a/tests/test_host_plane.py +++ b/tests/test_host_plane.py @@ -281,3 +281,413 @@ def test_openxfactorys_host_is_absent_only_when_its_own_module_is(monkeypatch, t with pytest.raises(ModuleNotFoundError) as refused: tests_conftest.openxfactory_host() assert refused.value.name == "a_dependency_the_host_lacks" + + +# --------------------------------------------------------------------------- +# THE COMPOSED HARNESS FOR THE DECLARED INTEGRATION TESTS (plan 038 T095; T094's +# map R2-INV-R9, ruled at openxFactory#656 comment 6021830531). Its lists are +# held to their rules, and each of its pieces to its behaviour, here, in a +# lone checkout: openxFactory's host, its reach and its composite are stood in +# where a case needs them. +# --------------------------------------------------------------------------- + +import contextlib # noqa: E402 +import os # noqa: E402 + +import yaml # noqa: E402 + +from opendox import doxbench_trust, workbench # noqa: E402 + +#: A test module that registers openXdox's columns itself names one of these. +_OWN_COLUMNS = re.compile(r"\b(?:column_contributions|column_seams)\b") + + +def _declared_entries() -> frozenset[str]: + declaration = yaml.safe_load( + (TESTS / "declared_exclusion.yaml").read_text(encoding="utf-8")) + return frozenset(Path(entry["path"]).name for entry in declaration["entries"]) + + +def _composed_lists() -> dict[str, frozenset[str]]: + return { + "DECLARED_HOST_SUITES": tests_conftest.DECLARED_HOST_SUITES, + "HOST_SEAM_SUITES": frozenset(tests_conftest.HOST_SEAM_SUITES), + "LOCAL_INSTALL_SUITES": tests_conftest.LOCAL_INSTALL_SUITES, + } + + +def test_each_composed_list_names_declared_files_outside_the_governed_set() -> None: + """Each list names test files the declaration lists, which run composed + and never in a lone checkout, and none of 12.5's governed suites, which + have their own host list and their own scan.""" + declared = _declared_entries() + governed = tests_conftest.governed_suites() + for name, members in _composed_lists().items(): + assert members, f"{name} is empty" + assert members <= declared, ( + f"{name} names files the declaration does not list: " + f"{sorted(members - declared)}. Its pieces are for the declared " + "integration tests, which run only composed") + assert not members & governed, ( + f"{name} names 12.5's governed suites: {sorted(members & governed)}") + for member in members: + assert (TESTS / member).is_file(), (name, member) + + +def test_a_file_that_registers_openxdoxs_columns_never_gets_the_declared_host() -> None: + """OQ-R9-1: `test_column_contributions_governed.py` registers openXdox's + columns itself, and T094 measured two of its cases red under the governed + host, where two bindings claim `POST /actions/gate/`. So it stays off the + list, as does any file that registers the columns itself.""" + assert "test_column_contributions_governed.py" in _declared_entries() + assert "test_column_contributions_governed.py" not in tests_conftest.DECLARED_HOST_SUITES + for member in tests_conftest.DECLARED_HOST_SUITES: + assert not _OWN_COLUMNS.search((TESTS / member).read_text(encoding="utf-8")), member + + +def test_the_seam_suites_name_seams_opendox_carries() -> None: + """Each seam named is a registration call openDox carries, with the + holder and the call that empties it that `a_hosts_seam` reads.""" + named = {seam for seams in tests_conftest.HOST_SEAM_SUITES.values() for seam in seams} + assert named == set(tests_conftest.HOST_SEAM_HOLDERS) + for (module_name, call), (held, is_default, unregister) in ( + tests_conftest.HOST_SEAM_HOLDERS.items()): + module = importlib.import_module(module_name) + assert callable(getattr(module, call)), (module_name, call) + assert callable(getattr(module, unregister)), (module_name, unregister) + assert hasattr(module, held), (module_name, held) + assert is_default is None or hasattr(module, is_default), (module_name, is_default) + + +def test_the_local_install_suites_drive_generate_and_open() -> None: + """W2's premise: a suite on the list drives `generate-and-open`, the verb + a hosted install refuses.""" + for member in tests_conftest.LOCAL_INSTALL_SUITES | {tests_conftest._LAUNCH_SUITE}: + assert "generate-and-open" in (TESTS / member).read_text(encoding="utf-8"), member + + +def _not_composed(monkeypatch) -> None: + monkeypatch.setitem(sys.modules, "opendox_host", None) + + +def _a_host_module(monkeypatch, root: Path, **calls) -> types.ModuleType: + """openxFactory's host module, stood in at `root/scripts/opendox_host.py`.""" + host_module = types.ModuleType("opendox_host") + host_module.__file__ = str(root / "scripts" / "opendox_host.py") + for name, call in calls.items(): + setattr(host_module, name, call) + monkeypatch.setitem(sys.modules, "opendox_host", host_module) + return host_module + + +def test_lone_nothing_of_the_composed_harness_acts(monkeypatch) -> None: + """Where `opendox_host` is absent, which is every lone checkout, no list + gets a host, a seam, a farm or a local install, and the environment is + untouched.""" + _not_composed(monkeypatch) + assert tests_conftest.openxfactory_root() is None + for member in tests_conftest.DECLARED_HOST_SUITES: + assert tests_conftest.declared_host_for(member) is None, member + for member, seams in tests_conftest.HOST_SEAM_SUITES.items(): + assert tests_conftest.openxfactory_seams(seams) == (), member + for member in tests_conftest.LOCAL_INSTALL_SUITES: + assert not tests_conftest.runs_a_local_install(member), member + assert tests_conftest.runs_a_local_install(tests_conftest._LAUNCH_SUITE) + environ: dict[str, str] = {} + assert tests_conftest.compose_the_contracts_farm(environ) is None + assert environ == {} + + +def test_this_process_has_a_farm_exactly_where_it_is_composed() -> None: + """CONTRACTS_DIR is None in a lone checkout. Where the run is composed it + names the farm this conftest built at import, and so does the + environment every subprocess inherits.""" + if tests_conftest.openxfactory_root() is None: + assert tests_conftest.CONTRACTS_DIR is None + return + farm = tests_conftest.CONTRACTS_DIR + assert farm is not None and farm.name == "contracts" + assert (farm / "schemas").is_dir() + assert (farm.parent / "examples" / "ideation-dashboard").is_dir() + assert os.environ["CONTRACTS_DIR"] == str(farm) + assert not farm.resolve().is_relative_to(TESTS.parent.resolve()) + + +def test_composed_openxfactorys_root_is_the_directory_above_its_host_module( + monkeypatch, tmp_path) -> None: + _a_host_module(monkeypatch, tmp_path / "openxFactory") + assert tests_conftest.openxfactory_root() == (tmp_path / "openxFactory").resolve() + + +def test_composed_a_declared_host_suite_runs_under_the_governed_host( + monkeypatch, tmp_path) -> None: + """Composed, each listed file's cases get the governed host, inside the + stand-in host's plane where the file is on both lists, and the stand-in + is back afterwards. A file off the list gets nothing.""" + imported = _stand_in_columns(monkeypatch) + composite = _Composite(tmp_path / "openXdox" / "code" / "src", imported) + _composed(monkeypatch, composite) + for member in tests_conftest.DECLARED_HOST_SUITES: + assert tests_conftest.declared_host_for(member) is tests_conftest.governed_host + for other in ("test_column_contributions_governed.py", Path(__file__).name, + "test_session_gates.py"): + assert tests_conftest.declared_host_for(other) is None, other + both = tests_conftest.DECLARED_HOST_SUITES & tests_conftest.HOST_PLANE_SUITES + assert both, "no listed file is on the stand-in's list too, so nothing tests the order" + before = list(sys.path) + earlier = tests_conftest.StandInHost() + with tests_conftest.a_hosts_plane(lambda: earlier): + with tests_conftest.a_hosts_plane( + tests_conftest.declared_host_for(sorted(both)[0])) as host: + assert host is composite + assert registry.current() is composite + assert sys.path == before + assert registry.current() is earlier + + +def _fixture_definition(request, name: str): + manager = request.config.pluginmanager.get_plugin("funcmanage") + definitions = manager.getfixturedefs(name, request.node) + assert definitions and len(definitions) == 1, name + return definitions[0] + + +def test_the_declared_host_is_set_up_inside_the_stand_in_hosts_plane(request) -> None: + """The declared host's fixture requests the stand-in's, so pytest sets the + stand-in up first, and the governed host is the one registered while a + file on both lists runs.""" + definition = _fixture_definition( + request, "_the_governed_host_for_the_declared_host_suites") + assert "_a_hosts_plane_for_the_token_reading_suites" in definition.argnames + + +def test_the_launch_suites_fixture_reads_the_extended_list(request, monkeypatch, + tmp_path) -> None: + """W2: T086's fixture runs for each file `runs_a_local_install` answers, + and that is the launch suite everywhere and LOCAL_INSTALL_SUITES composed.""" + definition = _fixture_definition(request, "_the_launch_suite_runs_a_local_install") + assert "runs_a_local_install(request.node.path.name)" in inspect.getsource(definition.func) + _a_host_module(monkeypatch, tmp_path / "openxFactory") + for member in tests_conftest.LOCAL_INSTALL_SUITES: + assert tests_conftest.runs_a_local_install(member), member + assert tests_conftest.runs_a_local_install(tests_conftest._LAUNCH_SUITE) + assert not tests_conftest.runs_a_local_install("test_canvas.py") + + +class _Policy: + """A trust policy, stood in.""" + + def verdict(self, binding, *, root): # noqa: D102 + return None + + def record(self, binding, *, root): # noqa: D102 + return None + + +def _health_check(*args, **kwargs): # noqa: D103 + return None + + +def _another_check(*args, **kwargs): # noqa: D103 + return None + + +@contextlib.contextmanager +def _the_seams_kept(monkeypatch): + """Whatever this case does at the two seams, they are as they were after.""" + for attribute in ("_health_check",): + monkeypatch.setattr(workbench, attribute, getattr(workbench, attribute)) + for attribute in ("_registered", "_is_default", "_default_read"): + monkeypatch.setattr(doxbench_trust, attribute, getattr(doxbench_trust, attribute)) + yield + + +def test_composed_the_seam_rows_are_read_off_the_hosts_table(monkeypatch, tmp_path) -> None: + policy = _Policy() + table = ((workbench, "register_session_notebook_scope", "documents"), + (workbench, "register_health_check", _health_check), + (doxbench_trust, "register", policy)) + _a_host_module(monkeypatch, tmp_path, seams=lambda: table) + assert tests_conftest.openxfactory_seams((tests_conftest.HEALTH_CHECK_SEAM,)) == ( + (workbench, "register_health_check", _health_check),) + assert tests_conftest.openxfactory_seams( + (tests_conftest.HEALTH_CHECK_SEAM, tests_conftest.BINDING_TRUST_SEAM)) == ( + (workbench, "register_health_check", _health_check), + (doxbench_trust, "register", policy)) + _a_host_module(monkeypatch, tmp_path, seams=lambda: table + table[1:2]) + with pytest.raises(RuntimeError, match="names 2 rows"): + tests_conftest.openxfactory_seams((tests_conftest.HEALTH_CHECK_SEAM,)) + _a_host_module(monkeypatch, tmp_path, seams=lambda: table[:1]) + with pytest.raises(RuntimeError, match="names 0 rows"): + tests_conftest.openxfactory_seams((tests_conftest.HEALTH_CHECK_SEAM,)) + + +def test_composed_the_seam_table_is_read_with_the_reach_put_back(monkeypatch, tmp_path) -> None: + """Reading the table may put a pinned leg on `sys.path` and import an + `openxdox` module from it, as the composite's facets can. The path is put + back, and a module from elsewhere refuses the case, by name.""" + elsewhere = types.ModuleType("openxdox.from_the_seam_table") + elsewhere.__file__ = str(tmp_path / "openxdox" / "from_the_seam_table.py") + + def seams(): + sys.path.insert(0, str(tmp_path / "leg")) + return ((workbench, "register_health_check", _health_check),) + + _a_host_module(monkeypatch, tmp_path, seams=seams) + before = list(sys.path) + tests_conftest.openxfactory_seams((tests_conftest.HEALTH_CHECK_SEAM,)) + assert sys.path == before + monkeypatch.setitem(sys.modules, elsewhere.__name__, elsewhere) + with pytest.raises(RuntimeError, match=r"openxdox\.from_the_seam_table"): + tests_conftest.openxfactory_seams((tests_conftest.HEALTH_CHECK_SEAM,)) + assert sys.path == before + + +def test_a_hosts_health_check_is_registered_for_the_block_and_put_back(monkeypatch) -> None: + with _the_seams_kept(monkeypatch): + workbench.unregister_health_check() + with tests_conftest.a_hosts_seam(workbench, "register_health_check", _health_check): + assert workbench._health_check is _health_check + assert not workbench.health_check_registered() + workbench.register_health_check(_another_check) + with tests_conftest.a_hosts_seam(workbench, "register_health_check", _health_check): + assert workbench._health_check is _health_check + assert workbench._health_check is _another_check + with pytest.raises(RuntimeError, match="raised in the block"): + with tests_conftest.a_hosts_seam(workbench, "register_health_check", + _health_check): + raise RuntimeError("raised in the block") + assert workbench._health_check is _another_check + + +def test_a_hosts_trust_policy_is_registered_for_the_block_and_put_back(monkeypatch) -> None: + """A host's policy held before is registered again after; openDox's own + default is not, since its consumers register it again.""" + policy, earlier = _Policy(), _Policy() + with _the_seams_kept(monkeypatch): + doxbench_trust.unregister() + doxbench_trust.policy() # openDox's default, registered and read + with tests_conftest.a_hosts_seam(doxbench_trust, "register", policy): + assert doxbench_trust.current() is policy + assert not doxbench_trust.is_registered() + doxbench_trust.register(earlier) + with tests_conftest.a_hosts_seam(doxbench_trust, "register", policy): + assert doxbench_trust.current() is policy + assert doxbench_trust.current() is earlier + + +def _a_tree(root: Path, files: dict[str, str]) -> None: + for relative, text in files.items(): + (root / relative).parent.mkdir(parents=True, exist_ok=True) + (root / relative).write_text(text, encoding="utf-8") + + +def _a_reach(root: Path, rows: dict[str, Path]) -> types.SimpleNamespace: + """openxFactory's reach (`carved_reach`), stood in over the tree `root`.""" + return types.SimpleNamespace( + __file__=str(root / "scripts" / "carved_reach.py"), + REPO_ROOT=root, + MOUNTS={"opendox_spec": root / "openDox" / "spec", + "openxdox_spec": root / "openXdox" / "spec"}, + sources_under=lambda prefix: {key: path for key, path in rows.items() + if key.startswith(prefix.rstrip("/") + "/")}) + + +def _links(base: Path) -> dict[str, Path]: + return {path.relative_to(base).as_posix(): Path(os.readlink(path)) + for path in sorted(base.rglob("*")) if path.is_symlink()} + + +def test_the_farm_links_the_familys_schemas_and_examples(tmp_path) -> None: + """The schemas: each carve row under `contracts/schemas/` at its file today + (nested rows are not the family's), then openxFactory's own not linked yet. + The examples: each owner's YAML at its relative path, less openDox's own + snapshot examples.""" + root = tmp_path / "openxFactory" + _a_tree(root, { + "contracts/schemas/gate-intent.schema.yaml": "", + "contracts/schemas/project-register.schema.yaml": "", + "contracts/schemas/notes.md": "", + "openXdox/spec/contracts/schemas/gate-action-record.schema.yaml": "", + "openXdox/spec/contracts/schemas/nested/inner.schema.yaml": "", + "openDox/spec/contracts/schemas/project-register.schema.yaml": "", + "examples/ideation-dashboard/gate-intent-x.example.yaml": "", + "examples/ideation-dashboard/transitions/t.example.yaml": "", + "openDox/spec/examples/ideation-dashboard/workbench-x.example.yaml": "", + "openDox/spec/examples/ideation-dashboard/opendox-snapshot-x.example.yaml": "", + "openDox/spec/examples/ideation-dashboard/negative/opendox-snapshot-y.negative.yaml": "", + "openDox/spec/examples/ideation-dashboard/negative/pair/turn.negative.yaml": "", + "openXdox/spec/examples/ideation-dashboard/README.md": "", + "openXdox/spec/examples/ideation-dashboard/negative/record.negative.yaml": "", + }) + rows = { + "contracts/schemas/gate-action-record.schema.yaml": + root / "openXdox/spec/contracts/schemas/gate-action-record.schema.yaml", + "contracts/schemas/project-register.schema.yaml": + root / "openDox/spec/contracts/schemas/project-register.schema.yaml", + "contracts/schemas/nested/inner.schema.yaml": + root / "openXdox/spec/contracts/schemas/nested/inner.schema.yaml", + "examples/ideation-dashboard/gate-intent-x.example.yaml": root / "elsewhere.yaml", + } + farm = tmp_path / "farm" + contracts = tests_conftest.build_contracts_farm(_a_reach(root, rows), farm) + assert contracts == farm / "contracts" + assert _links(contracts / "schemas") == { + "gate-action-record.schema.yaml": + root / "openXdox/spec/contracts/schemas/gate-action-record.schema.yaml", + "gate-intent.schema.yaml": root / "contracts/schemas/gate-intent.schema.yaml", + "project-register.schema.yaml": + root / "openDox/spec/contracts/schemas/project-register.schema.yaml", + } + assert _links(farm / "examples" / "ideation-dashboard") == { + "gate-intent-x.example.yaml": root / "examples/ideation-dashboard/gate-intent-x.example.yaml", + "transitions/t.example.yaml": root / "examples/ideation-dashboard/transitions/t.example.yaml", + "workbench-x.example.yaml": + root / "openDox/spec/examples/ideation-dashboard/workbench-x.example.yaml", + "negative/pair/turn.negative.yaml": + root / "openDox/spec/examples/ideation-dashboard/negative/pair/turn.negative.yaml", + "negative/record.negative.yaml": + root / "openXdox/spec/examples/ideation-dashboard/negative/record.negative.yaml", + } + + +def test_the_farm_refuses_an_example_two_owners_keep(tmp_path) -> None: + root = tmp_path / "openxFactory" + _a_tree(root, { + "examples/ideation-dashboard/negative/same.negative.yaml": "", + "openXdox/spec/examples/ideation-dashboard/negative/same.negative.yaml": "", + }) + with pytest.raises(RuntimeError, match=r"two owners keep the example negative/same"): + tests_conftest.build_contracts_farm(_a_reach(root, {}), tmp_path / "farm") + + +def test_composed_the_farm_is_built_outside_the_checkout_and_named_in_the_environment( + monkeypatch, tmp_path) -> None: + root = tmp_path / "openxFactory" + _a_tree(root, {"contracts/schemas/gate-intent.schema.yaml": ""}) + _a_host_module(monkeypatch, root) + monkeypatch.setitem(sys.modules, "carved_reach", _a_reach(root, {})) + removed = [] + monkeypatch.setattr(tests_conftest, "_atexit", types.SimpleNamespace( + register=lambda call, *args, **kwargs: removed.append((call, args, kwargs)))) + (tmp_path / "t").mkdir() + environ = {"CONTRACTS_DIR": "a directory the environment named before"} + contracts = tests_conftest.compose_the_contracts_farm(environ, under=str(tmp_path / "t")) + assert contracts is not None and contracts.parent.parent == tmp_path / "t" + assert contracts.parent.name.startswith("openxdox-contracts-farm-") + assert environ == {"CONTRACTS_DIR": str(contracts)} + assert _links(contracts / "schemas") == { + "gate-intent.schema.yaml": root / "contracts/schemas/gate-intent.schema.yaml"} + assert removed == [(tests_conftest._shutil.rmtree, (contracts.parent,), + {"ignore_errors": True})] + + +def test_composed_a_reach_from_another_tree_is_refused(monkeypatch, tmp_path) -> None: + root = tmp_path / "openxFactory" + _a_tree(root, {"contracts/schemas/gate-intent.schema.yaml": ""}) + _a_host_module(monkeypatch, root) + monkeypatch.setitem(sys.modules, "carved_reach", _a_reach(tmp_path / "another", {})) + environ: dict[str, str] = {} + with pytest.raises(RuntimeError, match="not from the composed tree"): + tests_conftest.compose_the_contracts_farm(environ, under=str(tmp_path)) + assert environ == {} From c2192e3e0484a8c4c86cca648af1dd581169fff9 Mon Sep 17 00:00:00 2001 From: Brett Heap <1513478+brettheap@users.noreply.github.com> Date: Tue, 6 Oct 2026 19:06:05 +0000 Subject: [PATCH 6/7] T095: split the guard's composite assertions; one raising call per pytest.raises SonarCloud at 70ed40c6 read five code smells in tests/test_host_plane.py (python:S9073 at three asserts, python:S5778 at two pytest.raises blocks). Each assertion is split, and each pytest.raises block now holds one call that may raise. What the cases assert is unchanged: 27 passed lone. Lane: openXfactory-3 Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/test_host_plane.py | 21 ++++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) diff --git a/tests/test_host_plane.py b/tests/test_host_plane.py index a8a6590..0f7d748 100644 --- a/tests/test_host_plane.py +++ b/tests/test_host_plane.py @@ -406,7 +406,8 @@ def test_this_process_has_a_farm_exactly_where_it_is_composed() -> None: assert tests_conftest.CONTRACTS_DIR is None return farm = tests_conftest.CONTRACTS_DIR - assert farm is not None and farm.name == "contracts" + assert farm is not None + assert farm.name == "contracts" assert (farm / "schemas").is_dir() assert (farm.parent / "examples" / "ideation-dashboard").is_dir() assert os.environ["CONTRACTS_DIR"] == str(farm) @@ -448,7 +449,8 @@ def test_composed_a_declared_host_suite_runs_under_the_governed_host( def _fixture_definition(request, name: str): manager = request.config.pluginmanager.get_plugin("funcmanage") definitions = manager.getfixturedefs(name, request.node) - assert definitions and len(definitions) == 1, name + assert definitions, name + assert len(definitions) == 1, name return definitions[0] @@ -543,6 +545,11 @@ def seams(): assert sys.path == before +def _raise_inside(module, call, what) -> None: + with tests_conftest.a_hosts_seam(module, call, what): + raise RuntimeError("raised in the block") + + def test_a_hosts_health_check_is_registered_for_the_block_and_put_back(monkeypatch) -> None: with _the_seams_kept(monkeypatch): workbench.unregister_health_check() @@ -554,9 +561,7 @@ def test_a_hosts_health_check_is_registered_for_the_block_and_put_back(monkeypat assert workbench._health_check is _health_check assert workbench._health_check is _another_check with pytest.raises(RuntimeError, match="raised in the block"): - with tests_conftest.a_hosts_seam(workbench, "register_health_check", - _health_check): - raise RuntimeError("raised in the block") + _raise_inside(workbench, "register_health_check", _health_check) assert workbench._health_check is _another_check @@ -657,8 +662,9 @@ def test_the_farm_refuses_an_example_two_owners_keep(tmp_path) -> None: "examples/ideation-dashboard/negative/same.negative.yaml": "", "openXdox/spec/examples/ideation-dashboard/negative/same.negative.yaml": "", }) + reach, farm = _a_reach(root, {}), tmp_path / "farm" with pytest.raises(RuntimeError, match=r"two owners keep the example negative/same"): - tests_conftest.build_contracts_farm(_a_reach(root, {}), tmp_path / "farm") + tests_conftest.build_contracts_farm(reach, farm) def test_composed_the_farm_is_built_outside_the_checkout_and_named_in_the_environment( @@ -673,7 +679,8 @@ def test_composed_the_farm_is_built_outside_the_checkout_and_named_in_the_enviro (tmp_path / "t").mkdir() environ = {"CONTRACTS_DIR": "a directory the environment named before"} contracts = tests_conftest.compose_the_contracts_farm(environ, under=str(tmp_path / "t")) - assert contracts is not None and contracts.parent.parent == tmp_path / "t" + assert contracts is not None + assert contracts.parent.parent == tmp_path / "t" assert contracts.parent.name.startswith("openxdox-contracts-farm-") assert environ == {"CONTRACTS_DIR": str(contracts)} assert _links(contracts / "schemas") == { From c6982e345db8b61c8dfd31ac242358bbba1a84b8 Mon Sep 17 00:00:00 2001 From: Brett Heap <1513478+brettheap@users.noreply.github.com> Date: Tue, 6 Oct 2026 22:20:25 +0000 Subject: [PATCH 7/7] T095: refuse a temporary root inside the checkout before building the farm Copilot r4200978505 at c5d07c6b: mkdtemp() honours TMPDIR and an explicit under, so a temporary root beneath the checkout would build the contracts farm inside the working tree, against the harness's own invariant. compose_the_contracts_farm() now resolves the parent first and refuses one inside the checkout, before anything is created or set; a new case covers both an explicit under and TMPDIR. Lane: openXfactory-3 Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/conftest.py | 10 +++++++++- tests/test_host_plane.py | 18 ++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/tests/conftest.py b/tests/conftest.py index eb66870..21ced7c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1021,7 +1021,9 @@ def _the_hosts_seams_for_the_seam_suites(request): # kind the family's validator does not serve. Two owners naming one path # is refused, never resolved. # It is OUTSIDE the checkout, in a temporary directory removed at exit: the -# validator refuses a `contracts/` that links out of its tree (`:157-163`). +# validator refuses a `contracts/` that links out of its tree (`:157-163`). A +# temporary root inside the checkout (a TMPDIR, or `under`, beneath it) is +# refused before anything is created. # The reach is read with `sys.path` saved and put back, and it must be the # composed tree's own. CONTRACT_EXAMPLE_LEFT_OUT = "opendox-snapshot-" @@ -1068,6 +1070,12 @@ def compose_the_contracts_farm(environ, *, under: str | None = None) -> Path | N root = openxfactory_root() if root is None: return None + parent = Path(under if under is not None else _tempfile.gettempdir()).resolve() + if parent.is_relative_to(REPO_ROOT): + raise RuntimeError( + f"the contracts farm would be built under {parent}, inside this " + f"checkout ({REPO_ROOT}), and it must sit outside it. Point TMPDIR " + "outside the checkout") saved = list(sys.path) try: reach = importlib.import_module("carved_reach") diff --git a/tests/test_host_plane.py b/tests/test_host_plane.py index 0f7d748..e371417 100644 --- a/tests/test_host_plane.py +++ b/tests/test_host_plane.py @@ -689,6 +689,24 @@ def test_composed_the_farm_is_built_outside_the_checkout_and_named_in_the_enviro {"ignore_errors": True})] +def test_composed_a_temporary_root_inside_the_checkout_is_refused(monkeypatch, tmp_path) -> None: + """The farm sits outside the checkout. A temporary root beneath it, from + TMPDIR or `under`, is refused before anything is created or set.""" + root = tmp_path / "openxFactory" + _a_tree(root, {"contracts/schemas/gate-intent.schema.yaml": ""}) + _a_host_module(monkeypatch, root) + monkeypatch.setitem(sys.modules, "carved_reach", _a_reach(root, {})) + inside = tests_conftest.REPO_ROOT / ".a-temporary-root-never-created" + environ: dict[str, str] = {} + with pytest.raises(RuntimeError, match="inside this checkout"): + tests_conftest.compose_the_contracts_farm(environ, under=str(inside)) + monkeypatch.setattr(tests_conftest._tempfile, "gettempdir", lambda: str(inside)) + with pytest.raises(RuntimeError, match="inside this checkout"): + tests_conftest.compose_the_contracts_farm(environ) + assert environ == {} + assert not inside.exists() + + def test_composed_a_reach_from_another_tree_is_refused(monkeypatch, tmp_path) -> None: root = tmp_path / "openxFactory" _a_tree(root, {"contracts/schemas/gate-intent.schema.yaml": ""})