diff --git a/tests/conftest.py b/tests/conftest.py index 51b83b4..44397eb 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 @@ -800,3 +802,320 @@ 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`). 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-" + + +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 + 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") + 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 f254922..19f4537 100644 --- a/tests/test_host_plane.py +++ b/tests/test_host_plane.py @@ -288,3 +288,438 @@ 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 + assert 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, name + assert 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 _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() + 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"): + _raise_inside(workbench, "register_health_check", _health_check) + 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": "", + }) + 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(reach, 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 + 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") == { + "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_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": ""}) + _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 == {}