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)