feat(admin-config-tool): task 2.2 - xcecommon.schema

Load controller.capnp once per process via a cached loader that
resolves identically under editable and pinned installs. Missing or
corrupted schema content converts a pycapnp uncaught-C++-abort into a
clean SchemaNotFoundError via isolated subprocess pre-validation.

src/xceadmin/tests/test_schema.py

@@ -0,0 +1,270 @@
+"""Tests for xcecommon.schema (task 2.2).
+
+Covers Requirements 5.1, 5.5, 7.4: loading the bundled ``controller.capnp``
+wire-protocol schema once per process and exposing every message type
+xceadmin needs to build or decode, resolving identically under both
+editable and pinned (non-editable) installs, and producing a clear,
+actionable error -- never a crash -- when the bundled schema copy is
+missing or corrupted.
+
+Fixture strategy mirrors task 2.1's ``test_validation.py``: the "happy
+path" tests exercise the *real* bundled schema (``default_schema_path()``)
+so we prove the actual shipped ``controller.capnp``/``c.capnp`` pair loads
+and exposes real top-level types, not just a synthetic fixture. The
+missing/corrupted-schema error-path tests build small throwaway ``.capnp``
+files under ``tmp_path`` instead, since those failure modes must not touch
+the real bundled copy.
+
+Design-vs-reality note (see schema.py module docstring for the full
+explanation): design.md's Service Interface postcondition lists a
+``Reply`` top-level type, but the real ``controller.capnp`` names this
+struct ``RequestReply`` (``RequestReply`` is the reply *message*;
+``request.body.replymsg`` embeds it). Verified directly against
+``capnp.load()`` output rather than assumed -- see
+``test_load_schema_exposes_expected_top_level_types`` below, which asserts
+against the real name.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+from xcecommon.errors import SchemaNotFoundError
+from xcecommon.schema import (
+    SCHEMA_PATH_ENV_VAR,
+    default_schema_path,
+    load_schema,
+)
+
+# ---------------------------------------------------------------------------
+# default_schema_path() -- resolution under the active install
+# ---------------------------------------------------------------------------
+
+
+def test_default_schema_path_resolves_to_real_bundled_schema() -> None:
+    """default_schema_path() must resolve to a real, existing
+    controller.capnp, with c.capnp sitting alongside it (Requirement 5.5).
+    This is the assertion that matters identically whether xceadmin is
+    installed editable or pinned -- see module docstring and the task's
+    Observable bullet; the *editable-vs-pinned* half of that guarantee is
+    proven by running this same test suite under both
+    `xce-build.sh xceadmin` and `xce-build.sh xceadmin release` profiles
+    (see TESTS_RUN in the task status report), not by anything expressible
+    inside a single test process.
+    """
+    schema_path = Path(default_schema_path())
+
+    assert schema_path.is_file()
+    assert schema_path.name == "controller.capnp"
+    assert (schema_path.parent / "c.capnp").is_file()
+
+
+def test_default_schema_path_env_var_override(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """XCEADMIN_SCHEMA_PATH overrides the packaged default, per design.md's
+    '--schema-path (CLI) / XCEADMIN_SCHEMA_PATH (env var) override the
+    default for troubleshooting' (7.4)."""
+    custom = tmp_path / "custom.capnp"
+    monkeypatch.setenv(SCHEMA_PATH_ENV_VAR, str(custom))
+
+    assert default_schema_path() == str(custom)
+
+
+def test_default_schema_path_no_env_var_uses_packaged_default(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv(SCHEMA_PATH_ENV_VAR, raising=False)
+
+    resolved = Path(default_schema_path())
+
+    assert resolved.is_file()
+
+
+# ---------------------------------------------------------------------------
+# load_schema() -- happy path against the real bundled schema
+# ---------------------------------------------------------------------------
+
+
+def test_load_schema_default_exposes_expected_top_level_types() -> None:
+    """load_schema() with no argument must load the real bundled schema
+    and expose every top-level type xceadmin needs to build or decode
+    (Requirement 5.1): ClusterConfig plus its nested list-section types,
+    Request, the real reply struct name RequestReply (not "Reply" -- see
+    module docstring), and the Upgrade/ProbeConfigTag request/reply pairs
+    used by config push / --dry-run."""
+    mod = load_schema()
+
+    for type_name in (
+        "ClusterConfig",
+        "LimitConfig",
+        "HostConfig",
+        "ScalerConfig",
+        "Resource",
+        "QueueConf",
+        "GroupConfig",
+        "Request",
+        "RequestReply",
+        "UpgradeConfigRequest",
+        "UpgradeConfigReply",
+        "ProbeConfigTagRequest",
+        "ProbeConfigTagReply",
+        "OPCode",
+    ):
+        assert hasattr(mod, type_name), f"loaded schema module missing {type_name}"
+
+
+def test_load_schema_default_types_are_usable_messages() -> None:
+    """The exposed types must be real, constructible capnp struct types --
+    not just names -- proving the schema actually parsed and linked
+    correctly against the real bundled controller.capnp/c.capnp pair."""
+    mod = load_schema()
+
+    cluster_config = mod.ClusterConfig.new_message()
+    cluster_config.clusterName = "test-cluster"
+    assert cluster_config.clusterName == "test-cluster"
+
+    assert "upgradeConfig" in mod.OPCode.schema.enumerants
+
+
+def test_load_schema_caches_default_load_returns_same_module_object() -> None:
+    """'Single load point ... called once per process; other modules
+    import the returned module object, never re-load' (design.md,
+    xcecommon.schema Responsibilities & Constraints). capnp.load() itself
+    returns a *new* module object (and new type objects) on every call --
+    verified directly -- so callers that didn't go through a shared cache
+    would silently get non-identical ClusterConfig/Request type objects
+    across modules. load_schema() must paper over that."""
+    first = load_schema()
+    second = load_schema()
+
+    assert first is second
+    assert first.ClusterConfig is second.ClusterConfig
+
+
+def test_load_schema_explicit_path_matching_default_is_usable() -> None:
+    """An explicit schema_path pointing at the same real bundled file
+    (the --schema-path override path, at the xcecommon.schema layer)
+    loads just as well as the implicit default."""
+    mod = load_schema(default_schema_path())
+
+    assert hasattr(mod, "ClusterConfig")
+
+
+# ---------------------------------------------------------------------------
+# load_schema() -- missing schema (Requirement 7.4)
+# ---------------------------------------------------------------------------
+
+
+def test_load_schema_missing_file_raises_schema_not_found_error(tmp_path: Path) -> None:
+    """A nonexistent controller.capnp path must raise SchemaNotFoundError
+    with the missing path attached (Requirement 7.4) -- and must never
+    reach capnp.load() itself: an absent file passed to capnp.load()
+    (without the c.capnp sibling present either) is a second, compounding
+    failure mode covered separately below."""
+    missing = tmp_path / "does-not-exist.capnp"
+
+    with pytest.raises(SchemaNotFoundError) as exc_info:
+        load_schema(str(missing))
+
+    assert exc_info.value.path == str(missing)
+    assert str(missing) in exc_info.value.message
+
+
+def test_load_schema_missing_c_capnp_import_raises_schema_not_found_error(
+    tmp_path: Path,
+) -> None:
+    """controller.capnp present but its c.capnp sibling import missing --
+    e.g. a broken/partial installation -- must raise SchemaNotFoundError,
+    not the uncatchable process abort pycapnp/kj otherwise produces for a
+    failed import (verified empirically, see module docstring). This is
+    the 'installation is broken' reading of the task's 'corrupted schema
+    copy' bullet."""
+    real_schema = Path(default_schema_path())
+    broken = tmp_path / "controller.capnp"
+    broken.write_text(real_schema.read_text(encoding="utf-8"), encoding="utf-8")
+    # Deliberately do NOT copy c.capnp alongside it.
+
+    with pytest.raises(SchemaNotFoundError) as exc_info:
+        load_schema(str(broken))
+
+    assert "c.capnp" in exc_info.value.message
+
+
+def test_load_schema_env_var_override_missing_raises_schema_not_found_error(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    missing = tmp_path / "nope.capnp"
+    monkeypatch.setenv(SCHEMA_PATH_ENV_VAR, str(missing))
+
+    with pytest.raises(SchemaNotFoundError):
+        load_schema()
+
+
+# ---------------------------------------------------------------------------
+# load_schema() -- corrupted (present but invalid) schema content
+# ---------------------------------------------------------------------------
+
+
+def test_load_schema_corrupted_content_raises_schema_not_found_error_not_crash(
+    tmp_path: Path,
+) -> None:
+    """A present-but-syntactically-invalid controller.capnp must raise a
+    clear SchemaNotFoundError, not crash the process.
+
+    This is the test that would fail loudly (by taking the whole pytest
+    process down with it, exit via SIGABRT) if load_schema() called
+    capnp.load() directly on unvalidated content: pycapnp 2.0.0 /
+    capnproto 1.2.0 propagate a genuine capnp parse error as an uncaught
+    C++ 'kj::ExceptionImpl', which calls std::terminate() before it ever
+    reaches Python's exception machinery -- confirmed by direct
+    experimentation, not assumed. load_schema() must pre-validate in an
+    isolated subprocess before ever loading in-process."""
+    broken = tmp_path / "controller.capnp"
+    broken.write_text("this is not { valid capnp syntax at all ]]]\n", encoding="utf-8")
+    real_schema = Path(default_schema_path())
+    (tmp_path / "c.capnp").write_text(
+        (real_schema.parent / "c.capnp").read_text(encoding="utf-8"), encoding="utf-8"
+    )
+
+    with pytest.raises(SchemaNotFoundError) as exc_info:
+        load_schema(str(broken))
+
+    assert str(broken) in exc_info.value.message
+
+
+def test_load_schema_corrupted_content_subprocess_never_aborts(tmp_path: Path) -> None:
+    """Belt-and-suspenders proof, fully isolated from pytest's own
+    process: run load_schema() against a corrupted schema inside a
+    throwaway subprocess and assert it exits with an ordinary Python
+    exception (returncode 1 from an uncaught SchemaNotFoundError formatted
+    by the interpreter), never a negative returncode (== killed by a
+    signal, e.g. -6/SIGABRT from the raw capnp.load() abort this task
+    exists to prevent)."""
+    broken = tmp_path / "controller.capnp"
+    broken.write_text("also not valid capnp {{{\n", encoding="utf-8")
+    real_schema = Path(default_schema_path())
+    (tmp_path / "c.capnp").write_text(
+        (real_schema.parent / "c.capnp").read_text(encoding="utf-8"), encoding="utf-8"
+    )
+
+    snippet = (
+        "from xcecommon.schema import load_schema\n"
+        "import sys\n"
+        "load_schema(sys.argv[1])\n"
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", snippet, str(broken)],
+        capture_output=True,
+        text=True,
+        timeout=30,
+    )
+
+    assert result.returncode >= 0, (
+        f"load_schema() let a raw process abort escape (returncode="
+        f"{result.returncode}); stderr={result.stderr!r}"
+    )
+    assert result.returncode != 0
+    assert "SchemaNotFoundError" in result.stderr