From a4659a303c1789673943a0aafbbb527dd2b04119 Mon Sep 17 00:00:00 2001 From: David Liu Date: Tue, 1 Sep 2026 23:26:16 +0000 Subject: [PATCH] Add a sandbox that exercises every `ucode configure` path end to end The configure flow branches on three axes at once: whether the managed-config opt-in and backend are live, whether the workspace publishes a config, and whether the caller is an admin (including the case where the SCIM check cannot say). The unit tests cover each branch with the collaborators stubbed, which is where the branch logic belongs, but nothing exercised a whole run: the state file it leaves behind, the settings it writes, the order the messages come out in, or whether the server was touched at all. Add a sandbox that runs the real CLI in a subprocess per scenario, inside a throwaway HOME, against a fake workspace. Only three seams are faked: the HTTP layer in `ucode.databricks`, the subprocesses that shell out to the databricks CLI or an agent binary, and the questionary prompts, which are driven by a scripted answer queue that fails loudly on an unexpected question. Everything above them (cli, managed_wizard, managed_config, managed_publish, agents/*, config_io, state) runs for real, and any endpoint the fake does not model is reported instead of quietly degrading. The 27 scenarios cover the gates, a published config for each persona, an empty workspace, the draft lifecycle, the pre-v2 migration, and the admin-only sections. `draft/survives-a-launch` is the load-bearing one: it fails if the draft and published slots are ever collapsed back into one, since a launch has to overwrite the fetched snapshot while leaving the admin's authored draft alone. Expectations live in `tests/test_configure_sandbox.py`, which drives the table concurrently and adds about two seconds to the suite. Verified by mutation: dropping the apply of a freshly authored draft fails four scenarios, and making a published write clobber the draft slot fails twelve. Co-authored-by: Isaac --- AGENTS.md | 1 + tests/sandbox_harness.py | 229 ++++++++++++++ tests/sandbox_scenarios.py | 505 ++++++++++++++++++++++++++++++ tests/sandbox_workspace.py | 326 +++++++++++++++++++ tests/test_configure_sandbox.py | 532 ++++++++++++++++++++++++++++++++ 5 files changed, 1593 insertions(+) create mode 100644 tests/sandbox_harness.py create mode 100644 tests/sandbox_scenarios.py create mode 100644 tests/sandbox_workspace.py create mode 100644 tests/test_configure_sandbox.py diff --git a/AGENTS.md b/AGENTS.md index f9353018..92317a09 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -12,6 +12,7 @@ Tests live in `tests/`. - Run the full test suite with `uv run pytest`. - Run focused tests with `uv run pytest tests/.py`. - Run e2e tests with `UCODE_TEST_WORKSPACE= uv run pytest tests/test_e2e.py -v`. +- Run the `ucode configure` sandbox with `uv run pytest tests/test_configure_sandbox.py` (each scenario is a `test_configure_scenario[]` case). To drive one scenario by hand, run `python -m tests.sandbox_scenarios ` from the repo root. - Run lint with `uv run ruff check .`. - Run the CLI from the current checkout with `uv run ucode ...`. - Reinstall the local checkout as the `ucode` tool with `uv tool install --reinstall .`. diff --git a/tests/sandbox_harness.py b/tests/sandbox_harness.py new file mode 100644 index 00000000..68035c57 --- /dev/null +++ b/tests/sandbox_harness.py @@ -0,0 +1,229 @@ +"""Sandbox harness for `ucode configure`. + +Runs the real CLI against a fake Databricks workspace inside a throwaway HOME, so every +file write, state transition and printed line is real. Only two layers are faked: + + * the HTTP layer in ucode.databricks (`_http_get_json` / `_http_send_json`), backed by + FakeWorkspace, plus the subprocess seams that shell out to the databricks CLI; + * the questionary prompts in ucode.ui, driven by a scripted answer queue. + +Everything above those seams (cli, managed_wizard, managed_config, managed_publish, +agents/*, config_io, state) executes for real. + +Import order matters: HOME must be set before ucode is imported (module-level +`Path.home()`), and ui/databricks must be patched before ucode.cli binds +`from ... import name` references. +""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent + + +def make_home() -> Path: + """A throwaway HOME, exported before ucode is imported. + + Honors SANDBOX_HOME so the caller (pytest's tmp_path) can own the directory and clean it + up; falls back to a temp dir when the module is driven by hand. + """ + override = os.environ.get("SANDBOX_HOME") + home = Path(override) if override else Path(tempfile.mkdtemp(prefix="ucode-sandbox-home-")) + home.mkdir(parents=True, exist_ok=True) + os.environ["HOME"] = str(home) + os.environ["USERPROFILE"] = str(home) + os.environ["DATABRICKS_CONFIG_FILE"] = str(home / ".databrickscfg") + (home / ".databrickscfg").write_text( + "[DEFAULT]\nhost = https://sandbox.cloud.databricks.com\ntoken = sandbox-pat\n", + encoding="utf-8", + ) + for var in ("ENABLE_MANAGED_AGENT_CONFIG", "ENABLE_CLAUDE_CODE_GATEWAY_MODEL_DISCOVERY"): + os.environ.pop(var, None) + for path in (str(REPO_ROOT / "src"), str(REPO_ROOT)): + if path not in sys.path: + sys.path.insert(0, path) + return home + + +class PromptScript: + """Scripted answers for the ui.prompt_* primitives. + + Each entry is (kind, value). A prompt pops the next entry and asserts its kind, so a + flow that asks something unexpected fails loudly instead of silently taking a default. + """ + + def __init__(self, answers: list[tuple[str, object]], *, discover: bool = False): + self.pending = list(answers) + self.asked: list[tuple[str, str]] = [] + self.discover = discover + + def take(self, kind: str, question: str, options=None) -> object: + self.asked.append((kind, question)) + if not self.pending: + if self.discover: + return self._auto(kind, options) + raise AssertionError( + f"unscripted {kind} prompt: {question!r}\nasked so far: {self.asked}" + ) + want, value = self.pending.pop(0) + if want != kind: + if self.discover: + self.pending.insert(0, (want, value)) + return self._auto(kind, options) + raise AssertionError( + f"prompt order mismatch: scripted {want!r} but flow asked {kind!r}: {question!r}" + ) + return value + + @staticmethod + def _auto(kind, options): + """Discovery-mode default: keep the flow moving so one run reveals the whole sequence.""" + first = None + if options: + head = list(options)[0] + first = head[0] if isinstance(head, (tuple, list)) else head + return { + "yes_no": False, + "select": first, + "multi": [first] if first is not None else [], + "text": "", + "number": 50.0, + "workspace": ("https://sandbox.cloud.databricks.com", "DEFAULT"), + }[kind] + + def drained(self) -> bool: + return not self.pending + + +CURRENT: PromptScript | None = None + + +def set_script(script: PromptScript) -> None: + global CURRENT + CURRENT = script + + +def install_prompts(script: PromptScript) -> None: + """Patch ucode.ui's prompt primitives. Must run before ucode.cli is imported.""" + import ucode.ui as ui + + set_script(script) + + def take(kind, question, options=None): + assert CURRENT is not None, "no prompt script installed" + return CURRENT.take(kind, question, options) + + def yes_no(question, default=True): + return take("yes_no", question) + + def yes_no_default(question, default=True): + return take("yes_no", question) + + def selection(prompt, options, **kwargs): + return take("select", prompt, options) + + def multi(prompt, options, **kwargs): + return take("multi", prompt, options) + + def tools(available, preselected=None, prompt="Select coding agents to configure:"): + return take("multi", prompt, available) + + def text(prompt, **kwargs): + return take("text", prompt) + + def percentage(prompt, **kwargs): + return take("number", prompt) + + def workspace(*args, **kwargs): + return take("workspace", "workspace") + + for name, fn in ( + ("prompt_yes_no", yes_no), + ("prompt_yes_no_default", yes_no_default), + ("prompt_for_selection", selection), + ("prompt_for_multi_selection", multi), + ("prompt_for_tools", tools), + ("prompt_for_text", text), + ("prompt_for_percentage", percentage), + ("prompt_for_workspace", workspace), + ): + if hasattr(ui, name): + setattr(ui, name, fn) + + +def run_cli(argv: list[str], *, env: dict[str, str] | None = None) -> tuple[int, str]: + """Invoke the real typer app in-process and return (exit_code, output).""" + from typer.testing import CliRunner + + from ucode.cli import app + + previous = {} + for key, value in (env or {}).items(): + previous[key] = os.environ.get(key) + os.environ[key] = value + try: + result = CliRunner().invoke(app, argv) + finally: + for key, value in previous.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + output = result.output + if result.exception is not None and not isinstance(result.exception, SystemExit): + import traceback + + output += "\n[EXCEPTION] " + "".join( + traceback.format_exception( + type(result.exception), result.exception, result.exception.__traceback__ + ) + ) + return result.exit_code, output + + +def managed_state(home: Path) -> dict: + path = home / ".ucode" / "managed-state.json" + return json.loads(path.read_text(encoding="utf-8")) if path.exists() else {} + + +def _entry(home: Path, workspace: str) -> dict: + """The slots for one workspace, read straight off disk without going through ucode. + + Mirrors the v1 -> v2 read migration so a scenario that leaves a pre-v2 file on disk reports + the slots ucode would see, not "absent". + """ + state = managed_state(home) + if state.get("version") == 2: + return (state.get("workspaces") or {}).get(workspace) or {} + if state.get("workspace") == workspace: + return {"published": state.get("config") or {}} + return {} + + +def state_shape(home: Path) -> str: + state = managed_state(home) + if not state: + return "absent" + return "v2" if state.get("version") == 2 else "pre-v2 (unmigrated on disk)" + + +def draft_slot(home: Path, workspace: str) -> dict | None: + return _entry(home, workspace).get("draft") + + +def published_slot(home: Path, workspace: str) -> dict | None: + return _entry(home, workspace).get("published") + + +def home_tree(home: Path) -> list[str]: + """Every file under the sandbox HOME, relative and sorted, for eyeballing writes.""" + return sorted( + str(p.relative_to(home)) + for p in home.rglob("*") + if p.is_file() and "__pycache__" not in str(p) + ) diff --git a/tests/sandbox_scenarios.py b/tests/sandbox_scenarios.py new file mode 100644 index 00000000..153b8dff --- /dev/null +++ b/tests/sandbox_scenarios.py @@ -0,0 +1,505 @@ +"""Scenario table for the `ucode configure` sandbox. + +Each scenario runs in its own subprocess (fresh HOME, fresh module state, fresh caches). +`tests/test_configure_sandbox.py` drives the whole table and holds the expectations; to look +at one scenario by hand, from the repo root: + + python -m tests.sandbox_scenarios # print its JSON verdict + SANDBOX_DISCOVER=1 python -m tests.sandbox_scenarios # auto-answer new prompts + +A scenario declares the fake workspace, plus one or more CLI steps with their scripted prompt +answers. It is judged on the real on-disk state and the real printed output. +""" + +from __future__ import annotations + +import json +import os +import sys +import traceback + +from tests import sandbox_harness as harness +from tests import sandbox_workspace as gw + +ON = {"ENABLE_MANAGED_AGENT_CONFIG": "1"} + +SCENARIOS: dict[str, dict] = {} + +CONFIGURE_SUBCOMMANDS = {"tracing", "spend-tiers"} + + +def scenario(name, *, persona, desc, ws, steps=None, os_managed=None, seed=None, **single): + """Register a scenario. `steps` for multi-command flows, else argv/env/prompts inline.""" + if steps is None: + steps = [single] + SCENARIOS[name] = { + "persona": persona, + "desc": desc, + "ws": ws, + "steps": steps, + "os_managed": os_managed, + "seed": seed, + } + + +AUTHOR_CLAUDE = [ + ("multi", ["claude"]), + ("select", "system.ai.claude-opus-4-8"), + ("select", "system.ai.claude-sonnet-4-6"), + ("select", "system.ai.claude-haiku-4-5"), + ("select", "system.ai.claude-opus-4-8"), +] + +# Same picks, but an overall default that differs from what the workspace publishes, so a flow +# that reads the wrong slot produces visibly different output. +AUTHOR_CLAUDE_SONNET_DEFAULT = [*AUTHOR_CLAUDE[:-1], ("select", "system.ai.claude-sonnet-4-6")] + + +scenario( + "gate/flag-off", + persona="either", + desc="ENABLE_MANAGED_AGENT_CONFIG unset: never reads the workspace config at all", + ws={"admin": True, "published": "PUBLISHED"}, + env={}, + argv=["configure", "--agents", "claude", "--skip-validate"], + prompts=[], +) + +scenario( + "gate/feature-disabled", + persona="either", + desc="backend feature off: local configure runs, no publish advice", + ws={"admin": True, "published": None, "feature_disabled": True}, + env=ON, + argv=["configure", "--agents", "claude", "--skip-validate"], + prompts=[], +) + +scenario( + "gate/not-found-is-no-config", + persona="either", + desc="a NOT_FOUND read is 'no config defined', not a failure: local configure runs quietly", + ws={"admin": False, "published": None, "config_read_error": "HTTP 404 Not Found: NOT_FOUND"}, + env=ON, + argv=["configure", "--agents", "claude", "--skip-validate"], + prompts=[], +) + +scenario( + "gate/read-error-no-cache", + persona="either", + desc="config read fails, nothing cached: warns and configures locally", + ws={ + "admin": False, + "published": "PUBLISHED", + "config_read_error": "HTTP 503 Service Unavailable", + }, + env=ON, + argv=["configure", "--agents", "claude", "--skip-validate"], + prompts=[], +) + +scenario( + "gate/read-error-with-cache", + persona="non-admin", + desc="config read fails but a published slot is cached: warns, applies the cached copy", + ws={ + "admin": False, + "published": "PUBLISHED", + "config_read_error": "HTTP 503 Service Unavailable", + }, + env=ON, + seed="published", + argv=["configure"], + prompts=[], +) + + +scenario( + "published/non-admin", + persona="non-admin", + desc="developer with a published config: applies it and exits, never offered authoring", + ws={"admin": False, "published": "PUBLISHED"}, + env=ON, + argv=["configure"], + prompts=[], +) + +scenario( + "published/admin-declines", + persona="admin", + desc="admin declines the update offer: published applied, no draft written", + ws={"admin": True, "published": "PUBLISHED"}, + env=ON, + argv=["configure"], + prompts=[("yes_no", False)], +) + +scenario( + "published/admin-accepts", + persona="admin", + desc="admin accepts: authors a draft, is advised to publish, server untouched", + ws={"admin": True, "published": "PUBLISHED"}, + env=ON, + argv=["configure"], + prompts=[("yes_no", True), *AUTHOR_CLAUDE], +) + +scenario( + "published/admin-unknown", + persona="admin?", + desc="SCIM check inconclusive: applies published, must not offer authoring", + ws={"admin": None, "published": "PUBLISHED"}, + env=ON, + argv=["configure"], + prompts=[], +) + +scenario( + "published/admin-non-interactive", + persona="admin", + desc="admin passing --agents: applies published, points at a flagless re-run", + ws={"admin": True, "published": "PUBLISHED"}, + env=ON, + argv=["configure", "--agents", "claude", "--skip-validate"], + prompts=[], +) + +scenario( + "published/os-managed-conflict", + persona="non-admin", + desc="OS-managed Claude settings block the apply: warns, yet still claims completion", + ws={"admin": False, "published": "PUBLISHED"}, + env=ON, + os_managed={ + "claude": json.dumps( + { + "apiKeyHelper": "/opt/corp/key.sh", + "env": {"ANTHROPIC_BASE_URL": "https://corp.example.com"}, + } + ) + }, + argv=["configure"], + prompts=[], +) + + +scenario( + "empty/non-admin", + persona="non-admin", + desc="developer, no workspace config: falls through to plain local configure", + ws={"admin": False, "published": None}, + env=ON, + argv=["configure", "--agents", "claude", "--skip-validate"], + prompts=[], +) + +scenario( + "empty/admin-interactive", + persona="admin", + desc="admin authors from scratch: draft saved and applied, publish advised, server untouched", + ws={"admin": True, "published": None}, + env=ON, + argv=["configure"], + prompts=[*AUTHOR_CLAUDE], +) + +scenario( + "empty/admin-non-interactive", + persona="admin", + desc="admin passing --agents with no workspace config: local configure, no authoring", + ws={"admin": True, "published": None}, + env=ON, + argv=["configure", "--agents", "claude", "--skip-validate"], + prompts=[], +) + +scenario( + "empty/admin-unknown", + persona="admin?", + desc="SCIM inconclusive, no workspace config: falls through to local configure", + ws={"admin": None, "published": None}, + env=ON, + argv=["configure", "--agents", "claude", "--skip-validate"], + prompts=[], +) + + +scenario( + "draft/survives-a-launch", + persona="admin", + desc="author a draft, an admin elsewhere republishes, then launch: the launch refresh must " + "overwrite the published slot and leave the draft alone", + ws={"admin": True, "published": "PUBLISHED"}, + env=ON, + steps=[ + {"argv": ["configure"], "env": ON, "prompts": [("yes_no", True), *AUTHOR_CLAUDE]}, + {"argv": ["claude"], "env": ON, "prompts": [], "server_default": "CODING_AGENT_CODEX"}, + {"argv": ["status"], "env": ON, "prompts": []}, + ], +) + +scenario( + "draft/publish-promotes", + persona="admin", + desc="author a draft then `ucode publish`: draft reaches the server, published slot updates", + ws={"admin": True, "published": None}, + env=ON, + steps=[ + {"argv": ["configure"], "env": ON, "prompts": [*AUTHOR_CLAUDE]}, + {"argv": ["publish"], "env": ON, "prompts": [("yes_no", True)]}, + ], +) + +scenario( + "draft/publish-declined", + persona="admin", + desc="`ucode publish` declined at the diff: server untouched, draft still on disk", + ws={"admin": True, "published": None}, + env=ON, + steps=[ + {"argv": ["configure"], "env": ON, "prompts": [*AUTHOR_CLAUDE]}, + {"argv": ["publish"], "env": ON, "prompts": [("yes_no", False)]}, + ], +) + +scenario( + "draft/non-admin-publish-refused", + persona="non-admin", + desc="a developer running `ucode publish` is refused before any write", + ws={"admin": False, "published": None}, + env=ON, + seed="draft", + argv=["publish"], + prompts=[], +) + +scenario( + "draft/export-reads-draft", + persona="admin", + desc="`ucode export` emits the authored draft, not the published copy", + ws={"admin": True, "published": "PUBLISHED"}, + env=ON, + steps=[ + { + "argv": ["configure"], + "env": ON, + "prompts": [("yes_no", True), *AUTHOR_CLAUDE_SONNET_DEFAULT], + }, + {"argv": ["export"], "env": ON, "prompts": []}, + ], +) + +scenario( + "draft/legacy-v1-migration", + persona="either", + desc="a pre-v2 managed-state.json migrates into the published slot, keeping a .pre-v2.bak", + ws={"admin": False, "published": "PUBLISHED"}, + env=ON, + seed="legacy-v1", + argv=["configure"], + prompts=[], +) + +scenario( + "draft/legacy-v1-was-a-draft", + persona="admin", + desc="the lossy migration case: a pre-v2 slot holding an unpublished draft lands in published, " + "so `ucode publish` sees nothing authored and .pre-v2.bak is the only copy left", + ws={"admin": True, "published": None}, + env=ON, + seed="legacy-v1", + argv=["publish"], + prompts=[], +) + +scenario( + "draft/legacy-v1-serves-as-fallback", + persona="non-admin", + desc="a pre-v2 file plus a failing read: the migrated published slot is what gets applied", + ws={ + "admin": False, + "published": "PUBLISHED", + "config_read_error": "HTTP 503 Service Unavailable", + }, + env=ON, + seed="legacy-v1", + argv=["configure"], + prompts=[], +) + + +scenario( + "section/non-admin-refused", + persona="non-admin", + desc="`ucode configure spend-tiers` refuses a non-admin", + ws={"admin": False, "published": None}, + env=ON, + seed="draft", + argv=["configure", "spend-tiers"], + prompts=[], +) + +scenario( + "section/tracing-is-local", + persona="non-admin", + desc="`ucode configure tracing` is a per-machine setting, not a managed-config section: a " + "non-admin can run it, unlike `configure spend-tiers`", + ws={"admin": False, "published": None}, + env=ON, + seed="draft", + steps=[ + {"argv": ["configure", "--agents", "claude", "--skip-validate"], "env": ON, "prompts": []}, + {"argv": ["configure", "tracing"], "env": ON, "prompts": []}, + ], +) + +scenario( + "multi/two-workspaces", + persona="admin", + desc="--workspaces with two entries skips the managed flow entirely: a managed config is " + "per-workspace, so both workspaces are configured locally instead", + ws={"admin": True, "published": "PUBLISHED"}, + env=ON, + argv=["configure", "--workspaces", f"{gw.WORKSPACE},https://other.cloud.databricks.com"], + prompts=[("multi", ["claude"])], +) + +scenario( + "section/no-authored-config", + persona="admin", + desc="`ucode configure spend-tiers` with nothing authored points back at `ucode configure`", + ws={"admin": True, "published": None}, + env=ON, + argv=["configure", "spend-tiers"], + prompts=[], +) + + +def _seed_disk(kind, home): + """Pre-seed ~/.ucode/managed-state.json before the CLI runs. + + Returns the seeded bytes for the pre-v2 file, whose backup must survive the migration + unchanged; None for the seeds ucode is free to rewrite. + """ + from ucode.managed_config import save_draft_config, save_published_config + + if kind == "published": + save_published_config(gw.WORKSPACE, _normalized()) + elif kind == "draft": + save_draft_config(gw.WORKSPACE, _normalized()) + elif kind == "legacy-v1": + path = home / ".ucode" / "managed-state.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps({"workspace": gw.WORKSPACE, "config": _normalized()}, indent=2) + "\n", + encoding="utf-8", + ) + return path.read_bytes() + return None + + +def _legacy_backup_verdict(home, seeded): + """How the .pre-v2.bak compares to the file that was on disk before the migration.""" + if seeded is None: + return None + backup = home / ".ucode" / "managed-state.json.pre-v2.bak" + if not backup.exists(): + return "absent" + return "byte-identical" if backup.read_bytes() == seeded else "rewritten" + + +def _normalized(): + from ucode.managed_config import normalize_managed_config + + return normalize_managed_config(gw.PUBLISHED_MANIFEST) + + +def _prompts_for_workspace(argv): + """True when the flow will ask which workspace to configure. + + `_run_managed_configure_flow` prompts whenever neither --workspaces nor --profiles was + passed, even though state.json already records one. Only bare `ucode configure` reaches + it: the `configure
` subcommands resolve the workspace from state themselves. + """ + if argv[0] != "configure": + return False + if any(token in argv for token in ("--workspaces", "--profiles")): + return False + return not any(token in CONFIGURE_SUBCOMMANDS for token in argv[1:]) + + +def _resolve_ws(spec_ws): + kwargs = dict(spec_ws) + if kwargs.get("published") == "PUBLISHED": + kwargs["published"] = gw.PUBLISHED_MANIFEST + return gw.FakeWorkspace(**kwargs) + + +def run_one(name: str) -> dict: + spec = SCENARIOS[name] + + home = harness.make_home() + + ws = _resolve_ws(spec["ws"]) + + import ucode.ui # noqa: F401 (imported before patching so cli binds the patched names) + + gw.install(ws) + gw.guard_privileged_writes(home, os_managed=spec.get("os_managed")) + + from ucode.state import save_state + + save_state({"workspace": gw.WORKSPACE, "profile": "DEFAULT"}) + seeded_legacy = _seed_disk(spec["seed"], home) if spec.get("seed") else None + + step_results = [] + installed = False + for step in spec["steps"]: + argv = step["argv"] + if step.get("server_default"): + ws.published["default_agent"] = step["server_default"] + answers = list(step.get("prompts") or []) + if _prompts_for_workspace(argv): + answers.insert(0, ("workspace", (gw.WORKSPACE, "DEFAULT"))) + script = harness.PromptScript(answers, discover=bool(os.environ.get("SANDBOX_DISCOVER"))) + if installed: + harness.set_script(script) + else: + harness.install_prompts(script) + installed = True + code, output = harness.run_cli(argv, env=step.get("env") or {}) + step_results.append( + { + "argv": argv, + "exit_code": code, + "output": output, + "prompts_asked": script.asked, + "prompts_left": script.pending, + "draft_after": harness.draft_slot(home, gw.WORKSPACE), + "published_after": harness.published_slot(home, gw.WORKSPACE), + } + ) + + return { + "name": name, + "persona": spec["persona"], + "desc": spec["desc"], + "steps": step_results, + "home": str(home), + "home_tree": harness.home_tree(home), + "managed_state": harness.managed_state(home), + "state_shape": harness.state_shape(home), + "legacy_backup": _legacy_backup_verdict(home, seeded_legacy), + "gateway": ws.report(), + } + + +if __name__ == "__main__": + target = sys.argv[1] + try: + result = run_one(target) + except Exception: + result = {"name": target, "harness_error": traceback.format_exc()} + print("---SANDBOX-JSON---") + print(json.dumps(result, indent=2, default=str)) diff --git a/tests/sandbox_workspace.py b/tests/sandbox_workspace.py new file mode 100644 index 00000000..6150965e --- /dev/null +++ b/tests/sandbox_workspace.py @@ -0,0 +1,326 @@ +"""A fake Databricks workspace for the `ucode configure` sandbox. + +Serves the REST surface `ucode configure` touches, from mutable in-memory state, and +records every call. Any URL the router does not recognize is recorded as UNROUTED and +returned as an error, so a flow that reaches an unmodelled endpoint is visible rather +than silently degraded. + +Server-shaped payloads only: manifests go out with proto enum names +(CODING_AGENT_CLAUDE_CODE, ...) so the real `normalize_managed_config` runs. +""" + +from __future__ import annotations + +import copy +import json +from urllib.parse import urlparse + +WORKSPACE = "https://sandbox.cloud.databricks.com" + +PUBLISHED_MANIFEST = { + "name": "coding-agent-configs/sandbox-1", + "workspace_id": 1653573648247579, + "default_agent": "CODING_AGENT_CLAUDE_CODE", + "enabled_agents": [ + { + "agent": "CODING_AGENT_CLAUDE_CODE", + "config": { + "model_config": { + "claude": { + "default_model": "system.ai.claude-opus-4-8", + "models": { + "default_opus_model": "system.ai.claude-opus-4-8", + "default_sonnet_model": "system.ai.claude-sonnet-4-6", + "default_haiku_model": "system.ai.claude-haiku-4-5", + }, + } + }, + }, + }, + ], + "mcp_servers": [{"name": "system.ai.github", "type": "MCP_SERVER_TYPE_UC_SERVICE"}], + "skills": {"names": ["system.ai.pdf-extraction"]}, + "tracing": {"table": "main.default.ucode_traces"}, +} + +CLAUDE_MODELS = [ + "system.ai.claude-opus-4-8", + "system.ai.claude-sonnet-4-6", + "system.ai.claude-haiku-4-5", +] + +CLAUDE_FAMILIES = { + "opus": CLAUDE_MODELS[0], + "sonnet": CLAUDE_MODELS[1], + "haiku": CLAUDE_MODELS[2], +} + +CONFIG_COLLECTION = "/coding-agent-configs" + +RECOMMENDATION = { + "agent": "CODING_AGENT_CLAUDE_CODE", + "model": CLAUDE_MODELS[0], + "current_spend": {"amount": "0", "currency": "USD"}, +} + +EXPERIMENT = { + "experiment_id": "4242", + "name": "/Users/sandbox@example.com/ucode-traces", + "tags": [ + { + "key": "mlflow.experiment.databricksTraceDestinationPath", + "value": "main.default.ucode_traces", + } + ], +} + +_AGENT_BINARIES = frozenset( + {"claude", "codex", "gemini", "opencode", "copilot", "pi", "cursor-agent"} +) + +MODEL_SERVICES = [ + *CLAUDE_MODELS, + "system.ai.gpt-5-codex", + "system.ai.gemini-2-5-pro", + "system.ai.llama-4-maverick", +] + + +class FakeWorkspace: + def __init__( + self, + *, + admin: bool | None = True, + published: dict | None = None, + config_read_error: str | None = None, + feature_disabled: bool = False, + write_error: str | None = None, + experiment: bool = True, + warehouses: list[dict] | None = None, + ): + self.experiment = experiment + self.warehouses = ( + warehouses + if warehouses is not None + else [{"id": "sandbox-warehouse-1", "name": "Sandbox", "state": "RUNNING"}] + ) + self.admin = admin + self.published = copy.deepcopy(published) if published else None + self.config_read_error = config_read_error + self.feature_disabled = feature_disabled + self.write_error = write_error + self.launches: list[list[str]] = [] + self.calls: list[tuple[str, str]] = [] + self.probes: list[str] = [] + self.recommendation = copy.deepcopy(RECOMMENDATION) + self.recommendations = 0 + self.writes: list[tuple[str, dict | None]] = [] + self.unrouted: list[tuple[str, str]] = [] + + def get(self, url, token, *, timeout=10, max_retries=0): + path = urlparse(url).path + self.calls.append(("GET", path)) + + if path == "/api/2.0/preview/scim/v2/Me": + if self.admin is None: + return None, "HTTP 500 Internal Server Error" + groups = [{"display": "admins"}] if self.admin else [{"display": "users"}] + return {"userName": "sandbox@example.com", "groups": groups}, None + + if "coding-agent-config" in path: + if self.feature_disabled: + return None, ( + 'HTTP 400 Bad Request: {"error_code": "FEATURE_DISABLED", "message": ' + '"Coding agent configs are not enabled for this workspace."}' + ) + if self.config_read_error: + return None, self.config_read_error + configs = [self.published] if self.published else [] + return {"coding_agent_configs": configs}, None + + if "workspace-metrics/budgets" in path: + return {"workspace_ai_gateway_budgets": []}, None + + if path == "/api/2.1/unity-catalog/model-services": + return {"model_services": [{"name": name} for name in MODEL_SERVICES]}, None + + if path == "/api/2.0/sql/warehouses": + return {"warehouses": list(self.warehouses)}, None + + return self._unrouted("GET", url) + + def send(self, method, url, token, payload, *, timeout=10, allow_empty_body=False): + path = urlparse(url).path + self.calls.append((method, path)) + + if path.endswith(":recommendModel"): + self.recommendations += 1 + return copy.deepcopy(self.recommendation), None + + if path == "/api/2.0/mlflow/experiments/search": + return {"experiments": [EXPERIMENT] if self.experiment else []}, None + + if path.endswith(CONFIG_COLLECTION) or "/coding-agent-configs/" in path: + self.writes.append((method, payload)) + if self.write_error: + return None, self.write_error + if method in ("POST", "PATCH"): + self.published = copy.deepcopy(payload or {}) + self.published.setdefault("name", "coding-agent-configs/sandbox-1") + return self.published, None + if method == "DELETE": + self.published = None + return None, None + return self._unrouted(method, url) + + def run(self, argv, **kwargs): + """Fake every subprocess ucode shells out to. Unknown invocations fail loudly.""" + import subprocess + + cmd = " ".join(argv) if isinstance(argv, (list, tuple)) else str(argv) + self.calls.append(("CLI", cmd)) + + def done(stdout="", code=0): + return subprocess.CompletedProcess(argv, code, stdout=stdout, stderr="") + + head = argv[0] if isinstance(argv, (list, tuple)) and argv else "" + if str(head).rsplit("/", 1)[-1] in _AGENT_BINARIES: + self.probes.append(cmd) + return done("sandbox ok") + + if "auth profiles" in cmd: + return done( + json.dumps( + { + "profiles": [ + { + "name": "DEFAULT", + "host": WORKSPACE, + "auth_type": "databricks-cli", + "valid": True, + } + ] + } + ) + ) + if "auth token" in cmd: + return done(json.dumps({"access_token": "sandbox-token", "token_type": "Bearer"})) + if "auth login" in cmd or "aitools install" in cmd: + return done() + if "--version" in cmd: + return done("Databricks CLI v0.240.0") + raise AssertionError(f"sandbox: unexpected subprocess call: {cmd}") + + def get_bytes(self, url, token, *, timeout=10): + self.calls.append(("GET-BYTES", urlparse(url).path)) + return None, "sandbox: binary fetch not modelled" + + def _unrouted(self, method, url): + self.unrouted.append((method, url)) + return None, f"sandbox: UNROUTED {method} {url}" + + def report(self) -> dict: + return { + "calls": [f"{m} {p}" for m, p in self.calls], + "writes": [{"method": m, "payload": p} for m, p in self.writes], + "unrouted": [f"{m} {u}" for m, u in self.unrouted], + "recommendations": self.recommendations, + "agent_probes": self.probes, + "launches": self.launches, + "server_published": self.published, + } + + +def install(ws: FakeWorkspace) -> None: + """Patch the HTTP and subprocess seams in ucode.databricks. Call before importing ucode.cli.""" + import ucode.databricks as db + + db._http_get_json = ws.get + db._http_send_json = ws.send + db._http_get_bytes = ws.get_bytes + + db.ensure_databricks_cli_version = lambda *a, **k: None + db.databricks_cli_version = lambda *a, **k: (0, 240, 0) + db.install_databricks_cli = lambda *a, **k: None + db.upgrade_databricks_cli = lambda *a, **k: False + db.install_ai_tools = lambda *a, **k: None + db.ensure_databricks_auth = lambda *a, **k: None + db.run_databricks_login = lambda *a, **k: None + db.has_valid_databricks_auth = lambda *a, **k: True + db.get_databricks_token = lambda *a, **k: "sandbox-token" + db.ensure_pat_bearer = lambda *a, **k: True + + db.run = ws.run + + import subprocess + + subprocess.run = ws.run + + import shutil + + import ucode.launcher as launcher + + launcher.exec_or_spawn = lambda argv: ws.launches.append(list(argv)) + + from ucode.agents import claude as claude_mod + + claude_mod._ensure_mlflow_cli = lambda *a, **k: True + claude_mod._uv_tool_mlflow_path = lambda *a, **k: "/usr/local/bin/mlflow" + _real_which = shutil.which + _faked_binaries = (*_AGENT_BINARIES, "databricks") + + def which(cmd, *args, **kwargs): + if cmd in _faked_binaries: + return f"/usr/local/bin/{cmd}" + return _real_which(cmd, *args, **kwargs) + + shutil.which = which + + db.list_model_services = lambda *a, **k: (list(MODEL_SERVICES), None) + db.discover_claude_models = lambda *a, **k: (dict(CLAUDE_FAMILIES), None) + db.discover_claude_models_unbucketed = lambda *a, **k: (list(CLAUDE_MODELS), None) + db.fetch_ai_gateway_claude_models = lambda *a, **k: dict(CLAUDE_FAMILIES) + db.discover_codex_models = lambda *a, **k: (["system.ai.gpt-5-codex"], None) + db.discover_gemini_models = lambda *a, **k: (["system.ai.gemini-2-5-pro"], None) + db.fetch_codex_models = lambda *a, **k: ["system.ai.gpt-5-codex"] + db.fetch_gemini_models = lambda *a, **k: ["system.ai.gemini-2-5-pro"] + db.ensure_ai_gateway = lambda *a, **k: None + db.list_model_provider_services = lambda *a, **k: ([], None) + db.list_mcp_services = lambda *a, **k: ([], None) + db.list_all_mcp_services = lambda *a, **k: ([], None) + db.list_workspace_budgets = lambda *a, **k: ([], None) + db.model_service_exists = lambda *a, **k: (True, None) + db.clear_model_services_cache() + + +def guard_privileged_writes(home, *, os_managed: dict | None = None) -> None: + """Refuse OS-managed (sudo) config writes and keep the real machine's out of the sandbox. + + The OS-managed settings paths are absolute (/etc/claude-code/..., /Library/...), so without + redirecting them a developer machine that really has enterprise-managed Claude settings leaks + into every scenario. Points them under the sandbox HOME instead; `os_managed` seeds one + deliberately for the scenario that exercises the conflict path. + """ + import ucode.managed_files as managed_files + from ucode.agents import claude as claude_mod + from ucode.agents import codex as codex_mod + + def reject(path, _text): + raise AssertionError(f"sandbox: attempted privileged write to {path}") + + managed_files._sudo_replace = reject + + os_dir = home / "os-managed" + os_dir.mkdir(parents=True, exist_ok=True) + claude_path = os_dir / "claude-managed-settings.json" + codex_path = os_dir / "codex-managed-config.toml" + claude_mod._managed_settings_path = lambda: claude_path + codex_mod._managed_config_path = lambda: None + + for tool, content in (os_managed or {}).items(): + target = {"claude": claude_path, "codex": codex_path}[tool] + target.write_text(content, encoding="utf-8") + + +def dump(obj) -> str: + return json.dumps(obj, indent=2, sort_keys=True, default=str) diff --git a/tests/test_configure_sandbox.py b/tests/test_configure_sandbox.py new file mode 100644 index 00000000..edc4fd18 --- /dev/null +++ b/tests/test_configure_sandbox.py @@ -0,0 +1,532 @@ +"""End-to-end expectations for every `ucode configure` path, admin and non-admin. + +Each scenario in `tests/sandbox_scenarios.py` runs the real CLI in its own subprocess against +a fake workspace inside a throwaway HOME (see `tests/sandbox_harness.py`). Only the HTTP layer, +the subprocess seams and the questionary prompts are faked, so what is asserted here is the +real on-disk state, the real printed output and the real set of server writes. + +This is the regression net for the draft/published split: `draft/survives-a-launch` is the +scenario that fails if the two slots are collapsed back into one. + +Expectations live in EXPECTATIONS below rather than in the scenario table, so the table stays a +plain description of what is being run. Every scenario must have an entry. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest + +from tests.sandbox_scenarios import SCENARIOS + +ROOT = Path(__file__).resolve().parent.parent + +ABSENT = "" +EMPTY = "" +SANDBOX_MODEL = "system.ai.claude-opus-4-8" +SANDBOX_SONNET_MODEL = "system.ai.claude-sonnet-4-6" +SANDBOX_TRACING_TABLE = "main.default.ucode_traces" + + +def _claude_slot( + *, default_agent="claude", tracing_table=SANDBOX_TRACING_TABLE, model=SANDBOX_MODEL +) -> dict: + """The `_slot` fingerprint of the sandbox workspace's single-agent config.""" + return { + "agents": ["claude"], + "default_agent": default_agent, + "models": {"claude": model}, + "tracing_table": tracing_table, + "budget_policy": None, + } + + +CLAUDE = _claude_slot() +CLAUDE_CODEX_DEFAULT = _claude_slot(default_agent="codex") +# Authored with a different overall default from the published config, so the two slots are +# distinguishable in output that reads only one of them. +CLAUDE_SONNET_DEFAULT = _claude_slot(model=SANDBOX_SONNET_MODEL) +# Authored with no published config to carry a tracing table forward from. +CLAUDE_UNTRACED = _claude_slot(tracing_table=None) + +APPLIED_PUBLISHED = ( + "Configuration complete — this machine now uses your workspace's managed config." +) +APPLIED_PUBLISHED_NOT_FINAL = "This machine now uses your workspace's managed config." +APPLIED_AUTHORED = "Configuration complete — this machine now uses your authored config." +AUTHORING_STARTED = "step 1 of 3" +DRAFT_SAVED = "Draft saved to ~/.ucode/managed-state.json" +PUBLISH_ADVICE = "Publish with ucode publish" +NOT_ADMIN = "You are not an admin of" +LOCAL_OPTIONS_NOTE = ( + "To update the workspace configuration, re-run `ucode configure` without local " + "configuration options." +) + +EXPECTATIONS: dict[str, dict] = { + "gate/flag-off": { + "steps": [ + { + "says": ["Configuration Complete", "Claude Code: configured"], + "silent": ["Applying managed config", AUTHORING_STARTED], + "draft": ABSENT, + "published": ABSENT, + } + ], + "state": "absent", + "forbid_calls": ["coding-agent-config"], + }, + "gate/feature-disabled": { + "steps": [ + { + "says": ["Configuration Complete"], + "silent": ["Applying managed config", AUTHORING_STARTED, "ucode publish"], + "draft": ABSENT, + "published": ABSENT, + } + ], + "state": "absent", + }, + "gate/not-found-is-no-config": { + "steps": [ + { + "says": ["Configuration Complete"], + "silent": ["Could not read", "Applying managed config"], + "draft": ABSENT, + "published": EMPTY, + } + ], + "state": "v2", + }, + "gate/read-error-no-cache": { + "steps": [ + { + "says": [ + "Could not read your workspace's managed config (HTTP 503 Service Unavailable)", + "configuring your own settings for now", + "Configuration Complete", + ], + "draft": ABSENT, + "published": ABSENT, + } + ], + "state": "absent", + }, + "gate/read-error-with-cache": { + "steps": [ + { + "says": [ + "applying the last one saved for this workspace", + APPLIED_PUBLISHED, + ], + "draft": ABSENT, + "published": CLAUDE, + } + ], + "state": "v2", + }, + "published/non-admin": { + "steps": [ + { + "says": [ + "Applying managed config", + "Claude Code: default model -> system.ai.claude-opus-4-8", + APPLIED_PUBLISHED, + ], + "silent": [AUTHORING_STARTED, "ucode publish"], + "draft": ABSENT, + "published": CLAUDE, + } + ], + }, + "published/admin-declines": { + "steps": [ + { + "says": [APPLIED_PUBLISHED_NOT_FINAL], + "silent": ["Configuration complete", AUTHORING_STARTED], + "draft": ABSENT, + "published": CLAUDE, + } + ], + }, + "published/admin-accepts": { + "steps": [ + { + "says": [ + APPLIED_PUBLISHED_NOT_FINAL, + AUTHORING_STARTED, + DRAFT_SAVED, + APPLIED_AUTHORED, + PUBLISH_ADVICE, + ], + "order": [APPLIED_AUTHORED, PUBLISH_ADVICE], + "draft": CLAUDE, + "published": CLAUDE, + } + ], + "writes": 0, + }, + "published/admin-unknown": { + "steps": [ + { + "says": [APPLIED_PUBLISHED], + "silent": [AUTHORING_STARTED], + "draft": ABSENT, + "published": CLAUDE, + } + ], + }, + "published/admin-non-interactive": { + "steps": [ + { + "says": [APPLIED_PUBLISHED, LOCAL_OPTIONS_NOTE], + "silent": [AUTHORING_STARTED], + "draft": ABSENT, + "published": CLAUDE, + } + ], + }, + "published/os-managed-conflict": { + "steps": [ + { + "says": [ + "Could not apply the managed config for Claude Code", + "override ucode values: apiKeyHelper", + "Could not apply your workspace's managed config to this machine", + ], + "silent": ["Configuration complete"], + "draft": ABSENT, + "published": CLAUDE, + } + ], + "files": ["os-managed/claude-managed-settings.json"], + }, + "empty/non-admin": { + "steps": [ + { + "says": ["Configuration Complete"], + "silent": [AUTHORING_STARTED, "ucode publish"], + "draft": ABSENT, + "published": EMPTY, + } + ], + "state": "v2", + }, + "empty/admin-interactive": { + "steps": [ + { + "says": [AUTHORING_STARTED, DRAFT_SAVED, APPLIED_AUTHORED, PUBLISH_ADVICE], + "order": [DRAFT_SAVED, APPLIED_AUTHORED, PUBLISH_ADVICE], + "draft": CLAUDE_UNTRACED, + "published": EMPTY, + } + ], + "writes": 0, + }, + "empty/admin-non-interactive": { + "steps": [ + { + "says": [ + "Using the requested local configuration options", + "re-run `ucode configure` without them", + ], + "silent": [AUTHORING_STARTED], + "draft": ABSENT, + "published": EMPTY, + } + ], + }, + "empty/admin-unknown": { + "steps": [ + { + "silent": [AUTHORING_STARTED, "Using the requested local configuration options"], + "draft": ABSENT, + "published": EMPTY, + } + ], + }, + "draft/survives-a-launch": { + "steps": [ + {"says": [APPLIED_AUTHORED], "draft": CLAUDE, "published": CLAUDE}, + { + "says": ["Applied your workspace's managed coding agent config", "Launching"], + "draft": CLAUDE, + "published": CLAUDE_CODEX_DEFAULT, + }, + { + "says": ["Workspace-managed config"], + "draft": CLAUDE, + "published": CLAUDE_CODEX_DEFAULT, + }, + ], + "writes": 0, + }, + "draft/publish-promotes": { + "steps": [ + {"says": [DRAFT_SAVED], "draft": CLAUDE_UNTRACED, "published": EMPTY}, + { + "says": [ + "Admin permissions verified", + "This will create a new managed config", + "Published coding-agent-configs/sandbox-1", + ], + "draft": CLAUDE_UNTRACED, + "published": CLAUDE_UNTRACED, + }, + ], + "writes": 1, + }, + "draft/publish-declined": { + "steps": [ + {"draft": CLAUDE_UNTRACED, "published": EMPTY}, + { + "exit": 1, + "says": ["Nothing was published"], + "draft": CLAUDE_UNTRACED, + "published": EMPTY, + }, + ], + "writes": 0, + }, + "draft/non-admin-publish-refused": { + "steps": [ + { + "exit": 1, + "says": [NOT_ADMIN, "restricted to workspace admins"], + "draft": CLAUDE, + "published": ABSENT, + } + ], + "writes": 0, + }, + "draft/export-reads-draft": { + "steps": [ + {"says": [APPLIED_AUTHORED], "draft": CLAUDE_SONNET_DEFAULT, "published": CLAUDE}, + { + "says": [ + '"spec_version": 1', + '"default_agent": "CODING_AGENT_CLAUDE_CODE"', + f'"default_model": "{SANDBOX_SONNET_MODEL}"', + ], + "silent": [ + '"name"', + "mcp_servers", + "skills", + f'"default_model": "{SANDBOX_MODEL}"', + ], + "draft": CLAUDE_SONNET_DEFAULT, + "published": CLAUDE, + }, + ], + }, + "draft/legacy-v1-migration": { + "steps": [{"says": [APPLIED_PUBLISHED], "draft": ABSENT, "published": CLAUDE}], + "state": "v2", + "files": [".ucode/managed-state.json.pre-v2.bak"], + "legacy_backup": "byte-identical", + }, + "draft/legacy-v1-was-a-draft": { + "steps": [ + { + "exit": 1, + "says": [ + "No managed config draft found locally", + "a fetched published config is not an authoring source", + ], + "draft": ABSENT, + "published": CLAUDE, + } + ], + "state": "pre-v2 (unmigrated on disk)", + }, + "draft/legacy-v1-serves-as-fallback": { + "steps": [ + { + "says": ["applying the last one saved for this workspace", APPLIED_PUBLISHED], + "draft": ABSENT, + "published": CLAUDE, + } + ], + }, + "section/non-admin-refused": { + "steps": [{"exit": 1, "says": [NOT_ADMIN], "draft": CLAUDE, "published": ABSENT}], + "writes": 0, + }, + "section/tracing-is-local": { + "steps": [ + {"says": ["Configuration Complete"]}, + { + "says": ["Unity Catalog: main.default.ucode_traces", "Tracing configured for"], + }, + ], + }, + "multi/two-workspaces": { + "steps": [ + { + "says": ["Configuration Complete"], + "silent": [AUTHORING_STARTED, APPLIED_PUBLISHED, LOCAL_OPTIONS_NOTE], + } + ], + "writes": 0, + "state": "absent", + "forbid_calls": ["coding-agent-config"], + }, + "section/no-authored-config": { + "steps": [ + { + "exit": 1, + "says": [ + "No managed config has been authored for", + "Run `ucode configure` first to pick the agents and models", + ], + "draft": ABSENT, + "published": ABSENT, + } + ], + "state": "absent", + }, +} + + +def _run_scenario(name: str, home: Path) -> dict: + env = dict(os.environ) + env["SANDBOX_HOME"] = str(home) + env["PYTHONPATH"] = os.pathsep.join([str(ROOT), str(ROOT / "src")]) + proc = subprocess.run( + [sys.executable, "-m", "tests.sandbox_scenarios", name], + cwd=ROOT, + capture_output=True, + text=True, + env=env, + timeout=300, + ) + marker = "---SANDBOX-JSON---" + if marker not in proc.stdout: + return { + "name": name, + "harness_error": ( + f"no JSON verdict\nstdout:\n{proc.stdout[-4000:]}\nstderr:\n{proc.stderr[-4000:]}" + ), + } + return json.loads(proc.stdout.split(marker)[-1]) + + +@pytest.fixture(scope="session") +def sandbox_results(tmp_path_factory) -> dict[str, dict]: + """Run the whole scenario table once, concurrently, and cache the verdicts. + + The scenarios are independent subprocesses with their own HOME, so they parallelize cleanly; + run serially the table costs about 15 seconds. + """ + names = list(SCENARIOS) + homes = {name: tmp_path_factory.mktemp("sandbox") for name in names} + with ThreadPoolExecutor(max_workers=8) as pool: + verdicts = pool.map(lambda name: _run_scenario(name, homes[name]), names) + return dict(zip(names, verdicts, strict=True)) + + +def _normalize(text: str) -> str: + """Collapse whitespace: Rich wraps output at the terminal width, mid-sentence.""" + return " ".join((text or "").split()) + + +def _slot(actual: dict | None) -> str | dict: + """Fingerprint a slot, including the fields a bad apply or carry-forward would silently drop.""" + if actual is None: + return ABSENT + if actual == {}: + return EMPTY + agents = actual.get("enabled_agents") or {} + return { + "agents": sorted(agents), + "default_agent": actual.get("default_agent"), + "models": { + tool: (config.get("model_config") or {}).get("default_model") + for tool, config in sorted(agents.items()) + }, + "tracing_table": actual.get("tracing_table"), + "budget_policy": actual.get("budget_policy"), + } + + +def _check_step(step: dict, expected: dict, label: str) -> list[str]: + problems = [] + output = _normalize(step["output"]) + if step["exit_code"] != expected.get("exit", 0): + problems.append(f"{label}: exit {step['exit_code']}, expected {expected.get('exit', 0)}") + for needle in expected.get("says", []): + if _normalize(needle) not in output: + problems.append(f"{label}: missing output {needle!r}") + for needle in expected.get("silent", []): + if _normalize(needle) in output: + problems.append(f"{label}: unexpected output {needle!r}") + positions = [output.find(_normalize(n)) for n in expected.get("order", [])] + if positions != sorted(positions): + problems.append(f"{label}: output out of order: {expected['order']}") + for slot in ("draft", "published"): + if slot in expected and _slot(step[f"{slot}_after"]) != expected[slot]: + problems.append( + f"{label}: {slot} slot is {_slot(step[f'{slot}_after'])}, expected {expected[slot]}" + ) + if step["prompts_left"]: + problems.append(f"{label}: scripted answers went unused: {step['prompts_left']}") + return problems + + +@pytest.mark.parametrize("name", list(SCENARIOS)) +def test_configure_scenario(name, sandbox_results): + result = sandbox_results[name] + expected = EXPECTATIONS[name] + assert not result.get("harness_error"), result.get("harness_error") + + problems = [] + steps = result["steps"] + if len(steps) != len(expected["steps"]): + pytest.fail(f"{name}: ran {len(steps)} steps, expected {len(expected['steps'])}") + for index, (step, step_expected) in enumerate(zip(steps, expected["steps"], strict=True), 1): + label = f"step {index} (ucode {' '.join(step['argv'])})" + problems += _check_step(step, step_expected, label) + + gateway = result["gateway"] + if gateway["unrouted"]: + problems.append(f"reached an unmodelled endpoint: {gateway['unrouted']}") + if len(gateway["writes"]) != expected.get("writes", 0): + problems.append( + f"{len(gateway['writes'])} server writes, expected {expected.get('writes', 0)}: " + f"{gateway['writes']}" + ) + for forbidden in expected.get("forbid_calls", []): + hit = [call for call in gateway["calls"] if forbidden in call] + if hit: + problems.append(f"called {forbidden!r}: {hit}") + if "state" in expected and result["state_shape"] != expected["state"]: + problems.append( + f"managed-state.json is {result['state_shape']!r}, expected {expected['state']!r}" + ) + for wanted in expected.get("files", []): + if not any(wanted in path for path in result["home_tree"]): + problems.append(f"missing file {wanted!r} in HOME: {result['home_tree']}") + if "legacy_backup" in expected and result["legacy_backup"] != expected["legacy_backup"]: + problems.append( + f"pre-v2 backup is {result['legacy_backup']!r} against the seeded file, " + f"expected {expected['legacy_backup']!r}" + ) + + if problems: + transcript = "\n\n".join( + f"$ ucode {' '.join(step['argv'])} -> exit {step['exit_code']}\n{step['output']}" + for step in steps + ) + pytest.fail( + f"{name}: {SCENARIOS[name]['desc']}\n\n" + "\n".join(problems) + f"\n\n{transcript}" + ) + + +def test_every_scenario_has_expectations(): + assert sorted(EXPECTATIONS) == sorted(SCENARIOS)