From f436cbaab5821988499230b52e5beef1201490e1 Mon Sep 17 00:00:00 2001 From: Tien Le Date: Fri, 2 Oct 2026 19:36:31 +0000 Subject: [PATCH 1/6] AIGTWY-4878 Extract reusable native agent integration helpers --- tests/README.md | 6 + tests/integration/README.md | 9 + tests/integration/utils/agents/__init__.py | 10 + tests/integration/utils/agents/claude.py | 65 ++++++ tests/integration/utils/agents/codex.py | 66 ++++++ tests/integration/utils/evidence.py | 70 ++++-- tests/integration/utils/http.py | 63 ++++++ tests/integration/utils/managed.py | 10 + tests/integration/utils/model_discovery.py | 60 +++++ tests/integration/utils/provider_catalog.py | 124 ++++------ tests/integration/utils/terminal.py | 25 ++ tests/test_integration_agent_helpers.py | 86 +++++++ tests/test_integration_evidence.py | 239 ++++++++++++++++++++ tests/test_integration_http.py | 116 ++++++++++ tests/test_integration_managed_config.py | 96 ++++++++ tests/test_integration_model_discovery.py | 131 ++++++++++- tests/test_integration_provider_catalog.py | 227 ++++++++++++++++++- 17 files changed, 1297 insertions(+), 106 deletions(-) create mode 100644 tests/integration/utils/agents/__init__.py create mode 100644 tests/integration/utils/agents/claude.py create mode 100644 tests/integration/utils/agents/codex.py create mode 100644 tests/integration/utils/http.py create mode 100644 tests/test_integration_agent_helpers.py create mode 100644 tests/test_integration_http.py create mode 100644 tests/test_integration_managed_config.py diff --git a/tests/README.md b/tests/README.md index dcfc7497f..c7f15d07e 100644 --- a/tests/README.md +++ b/tests/README.md @@ -139,6 +139,12 @@ Pi's token-command tests in `test_agent_pi.py` exercise Windows executable paths through POSIX parsing, including spaces, apostrophes, profile names, and PAT mode. They do not launch Pi or Git Bash on Windows. +Shared integration helpers have component coverage in `test_integration_agent_helpers.py`, +`test_integration_evidence.py`, `test_integration_model_discovery.py`, +`test_integration_provider_catalog.py`, `test_integration_http.py`, and +`test_integration_managed_config.py`. They check native transcript parsing, model attribution, +picker matching, scoped catalog requests, and persisted managed input; they are not live passes. + ## CUJ coverage matrix These are **implemented assertions**, not a claim that every version passes. diff --git a/tests/integration/README.md b/tests/integration/README.md index c6497d503..7862ce050 100644 --- a/tests/integration/README.md +++ b/tests/integration/README.md @@ -99,6 +99,15 @@ when an explicit model is supplied. The only live OpenCode journey is the headle prompt case in `test_ug_opencode_headless.py`; there is no dedicated live OpenCode model-selection journey. +## Shared agent helpers + +`utils/agents/claude.py` and `utils/agents/codex.py` own native transcript parsing and +discovery adapters. `utils/evidence.py` dispatches shared file-task evidence through those +adapters for existing journeys. Shared HTTP, catalog, picker, and terminal mechanics stay +in `utils/`; workspace-specific expectations belong in the consuming journey. +Claude model evidence is response-reported; Codex evidence is client-selected and joined +to the completed turn. Neither proves the gateway's backing destination. + ## Run a specific combination Prerequisites: Python 3.12+, uv, and Node/npm. Live runs also require Databricks diff --git a/tests/integration/utils/agents/__init__.py b/tests/integration/utils/agents/__init__.py new file mode 100644 index 000000000..ae5f60cc1 --- /dev/null +++ b/tests/integration/utils/agents/__init__.py @@ -0,0 +1,10 @@ +"""Shared validation for agent-specific integration evidence.""" + +import re + + +def _evidence_id(value: object, field: str) -> str: + assert isinstance(value, str) and re.fullmatch(r"\S+", value), ( + f"Missing or malformed completed-task {field}" + ) + return value diff --git a/tests/integration/utils/agents/claude.py b/tests/integration/utils/agents/claude.py new file mode 100644 index 000000000..2bcff696e --- /dev/null +++ b/tests/integration/utils/agents/claude.py @@ -0,0 +1,65 @@ +"""Claude-specific integration evidence and model discovery helpers.""" + +from ..model_discovery import claude_discovery_model_id as discovery_model_id +from ..model_discovery import claude_model_in_picker as model_in_picker +from ..model_discovery import claude_model_service_id as model_service_id +from ..provider_catalog import fetch_anthropic_parent_catalog as fetch_parent_catalog +from . import _evidence_id + +__all__ = [ + "EVIDENCE_KIND", + "assistant_answers", + "completed_task_models", + "discovery_model_id", + "fetch_parent_catalog", + "model_in_picker", + "model_service_id", +] + +SESSION_DIRECTORY = ".claude/projects" +EVIDENCE_KIND = "response-reported model" + + +def is_child_session(path: str, records: list[dict]) -> bool: + del records + return "/subagents/" in path + + +def assistant_answers(records: list[dict]) -> list[str]: + answers = [] + for record in records: + if record.get("type") == "assistant": + message = record.get("message", {}) + if message.get("role") == "assistant": + answers.extend( + part["text"] + for part in message.get("content", []) + if part.get("type") == "text" and isinstance(part.get("text"), str) + ) + return answers + + +def completed_task_models(records: list[dict], answer_value: str) -> set[str]: + """Parse Claude's response-reported model on the assistant's matching answer.""" + assert isinstance(answer_value, str) and answer_value.strip(), ( + "Expected a nonempty answer value" + ) + found: set[str] = set() + for record in records: + if record.get("type") != "assistant": + continue + message = record.get("message") + assert isinstance(message, dict), "Malformed Claude assistant message" + assert message.get("role") == "assistant", "Missing Claude assistant role" + content = message.get("content") + assert isinstance(content, list), "Malformed Claude assistant content" + for part in content: + assert isinstance(part, dict), "Malformed Claude assistant content block" + if part.get("type") != "text": + continue + text = part.get("text") + assert isinstance(text, str), "Malformed Claude assistant text" + if answer_value in text: + model = _evidence_id(message.get("model"), "model") + found.add(_evidence_id(model.removesuffix("[1m]"), "model")) + return found diff --git a/tests/integration/utils/agents/codex.py b/tests/integration/utils/agents/codex.py new file mode 100644 index 000000000..99e505420 --- /dev/null +++ b/tests/integration/utils/agents/codex.py @@ -0,0 +1,66 @@ +"""Codex-specific integration evidence and model discovery helpers.""" + +from ..model_discovery import codex_model_in_picker as model_in_picker +from ..provider_catalog import fetch_codex_parent_catalog as fetch_parent_catalog +from . import _evidence_id + +__all__ = [ + "EVIDENCE_KIND", + "assistant_answers", + "completed_task_models", + "fetch_parent_catalog", + "model_in_picker", +] + +SESSION_DIRECTORY = ".codex/sessions" +EVIDENCE_KIND = "client-selected model" + + +def is_child_session(path: str, records: list[dict]) -> bool: + del path + return any( + record.get("type") == "session_meta" + and isinstance(record.get("payload", {}).get("source"), dict) + and "subagent" in record["payload"]["source"] + for record in records + ) + + +def assistant_answers(records: list[dict]) -> list[str]: + answers = [] + for record in records: + if record.get("type") == "event_msg": + payload = record.get("payload", {}) + if payload.get("type") == "task_complete" and payload.get("last_agent_message"): + answers.append(payload["last_agent_message"]) + return answers + + +def completed_task_models(records: list[dict], answer_value: str) -> set[str]: + """Join Codex's completed answer turn to its client-selected context model.""" + assert isinstance(answer_value, str) and answer_value.strip(), ( + "Expected a nonempty answer value" + ) + completed: set[str] = set() + for record in records: + if record.get("type") != "event_msg": + continue + payload = record.get("payload") + assert isinstance(payload, dict), "Malformed Codex event payload" + if payload.get("type") != "task_complete": + continue + answer = payload.get("last_agent_message") + assert isinstance(answer, str), "Missing or malformed Codex completed answer" + if answer_value in answer: + completed.add(_evidence_id(payload.get("turn_id"), "turn ID")) + models: dict[str, set[str]] = {turn_id: set() for turn_id in completed} + for record in records: + if record.get("type") != "turn_context": + continue + payload = record.get("payload") + assert isinstance(payload, dict), "Malformed Codex turn context" + turn_id = _evidence_id(payload.get("turn_id"), "turn ID") + if turn_id in completed: + models[turn_id].add(_evidence_id(payload.get("model"), "model")) + assert all(models.values()), "Missing Codex model context for completed answer turn" + return {model for turn_models in models.values() for model in turn_models} diff --git a/tests/integration/utils/evidence.py b/tests/integration/utils/evidence.py index 4c813d671..06510f820 100644 --- a/tests/integration/utils/evidence.py +++ b/tests/integration/utils/evidence.py @@ -5,6 +5,17 @@ import uuid from pathlib import Path +from .agents import _evidence_id, claude, codex + +_EVIDENCE_ADAPTERS = {"claude": claude, "codex": codex} + + +def _evidence_adapter(agent: str): + try: + return _EVIDENCE_ADAPTERS[agent] + except (KeyError, TypeError): + raise AssertionError(f"Unsupported evidence agent: {agent}") from None + def assert_no_terminal_api_error(screen: str) -> None: """Fail on definitive client errors, not an in-progress transient retry.""" @@ -37,39 +48,52 @@ def read_jsonl(path: Path) -> list[dict]: def agent_sessions(session, agent: str) -> dict[str, list[dict]]: - directory = session.home / (".claude/projects" if agent == "claude" else ".codex/sessions") + directory = session.home / _evidence_adapter(agent).SESSION_DIRECTORY return { str(path.relative_to(directory)): read_jsonl(path) for path in directory.rglob("*.jsonl") } def assistant_answers(agent: str, records: list[dict]) -> list[str]: - answers = [] - for record in records: - if agent == "claude" and record.get("type") == "assistant": - message = record.get("message", {}) - if message.get("role") == "assistant": - answers.extend( - part["text"] - for part in message.get("content", []) - if part.get("type") == "text" and isinstance(part.get("text"), str) - ) - if agent == "codex" and record.get("type") == "event_msg": - payload = record.get("payload", {}) - if payload.get("type") == "task_complete" and payload.get("last_agent_message"): - answers.append(payload["last_agent_message"]) - return answers + return _evidence_adapter(agent).assistant_answers(records) def is_child_session(agent: str, path: str, records: list[dict]) -> bool: - if agent == "claude": - return "/subagents/" in path - return any( - record.get("type") == "session_meta" - and isinstance(record.get("payload", {}).get("source"), dict) - and "subagent" in record["payload"]["source"] - for record in records + return _evidence_adapter(agent).is_child_session(path, records) + + +claude_completed_task_models = claude.completed_task_models +codex_completed_task_models = codex.completed_task_models + + +def completed_task_models(session, agent: str, answer_value: str) -> set[str]: + """Read parent task model evidence, not proof of the gateway's destination.""" + adapter = _evidence_adapter(agent) + assert isinstance(answer_value, str) and answer_value.strip(), ( + "Expected a nonempty answer value" + ) + return { + model + for path, records in agent_sessions(session, agent).items() + if not is_child_session(agent, path, records) + for model in adapter.completed_task_models(records, answer_value) + } + + +def assert_completed_task_model(session, agent: str, answer_value: str, expected: str) -> None: + expected = _evidence_id(expected, "expected model") + adapter = _evidence_adapter(agent) + observed = completed_task_models(session, agent, answer_value) + session.record( + f"completed-task-model-{agent}-{answer_value[:12]}.json", + { + "expected": expected, + "observed": sorted(observed), + "evidence_kind": adapter.EVIDENCE_KIND, + "gateway_destination_proven": False, + }, ) + assert observed == {expected}, {"expected": expected, "observed": sorted(observed)} def assistant_answer_contains(session, agent: str, value: str, *, child: bool = False) -> bool: diff --git a/tests/integration/utils/http.py b/tests/integration/utils/http.py new file mode 100644 index 000000000..1c35cff5e --- /dev/null +++ b/tests/integration/utils/http.py @@ -0,0 +1,63 @@ +"""Read-only workspace JSON transport with strict origins and no redirects.""" + +from __future__ import annotations + +import json +import re +import urllib.error +import urllib.parse +import urllib.request + + +class NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +def workspace_origin(workspace: str, token: str) -> str: + """Validate an explicit HTTPS origin and bearer before constructing a client.""" + assert isinstance(workspace, str), "Expected an HTTPS workspace origin" + try: + origin = urllib.parse.urlsplit(workspace) + port = origin.port + except ValueError: + raise AssertionError("Expected a valid HTTPS workspace origin") from None + assert ( + origin.scheme == "https" + and origin.hostname + and origin.username is None + and origin.password is None + and origin.path in {"", "/"} + and not origin.query + and not origin.fragment + and (port is None or 0 < port <= 65535) + and not re.search(r"[\s\\]", workspace) + ), "Expected an HTTPS workspace origin without credentials" + assert isinstance(token, str) and token.strip() and not re.search(r"[\r\n]", token), ( + "Expected an explicit workspace bearer" + ) + return f"https://{origin.netloc}" + + +def safe_https_json_get( + workspace: str, token: str, path: str, *, headers: dict[str, str] | None = None +) -> object: + """GET JSON from an explicit workspace origin without redirects or secret diagnostics.""" + origin = workspace_origin(workspace, token) + assert path.startswith("/") and not path.startswith("//"), "Expected a workspace API path" + request = urllib.request.Request( + f"{origin}{path}", + headers={ + **(headers or {}), + "Authorization": f"Bearer {token}", + "Accept": "application/json", + }, + method="GET", + ) + try: + with urllib.request.build_opener(NoRedirect()).open(request, timeout=30) as response: + return json.load(response) + except urllib.error.HTTPError as error: + raise AssertionError(f"Workspace JSON GET returned HTTP {error.code}") from None + except (urllib.error.URLError, OSError, ValueError): + raise AssertionError("Workspace JSON GET failed or returned invalid JSON") from None diff --git a/tests/integration/utils/managed.py b/tests/integration/utils/managed.py index 3599d38f9..997d96948 100644 --- a/tests/integration/utils/managed.py +++ b/tests/integration/utils/managed.py @@ -13,6 +13,16 @@ MANAGED_CONFIGS_PATH = "/api/ai-gateway/v2/coding-agent-configs" +def read_persisted_managed_config(session, workspace: str) -> dict: + """Read the configure result without imposing a journey-specific policy.""" + path = session.home / ".ucode" / "managed-config.json" + assert path.is_file(), path + persisted = json.loads(path.read_text()) + assert isinstance(persisted, dict) and persisted.get("workspace") == workspace, persisted + assert isinstance(persisted.get("config"), dict), persisted + return persisted + + def assert_no_managed_config(payload: object) -> None: """Validate the real List response before claiming unmanaged-workspace coverage.""" configs = payload.get("coding_agent_configs", []) if isinstance(payload, dict) else payload diff --git a/tests/integration/utils/model_discovery.py b/tests/integration/utils/model_discovery.py index 36171de96..bfb338e94 100644 --- a/tests/integration/utils/model_discovery.py +++ b/tests/integration/utils/model_discovery.py @@ -1,5 +1,6 @@ """Validation helpers for model-discovery evidence.""" +import hashlib import re _CLAUDE_GATEWAY_ALIAS = re.compile(r"^anthropic-aigw-[0-9a-fA-F]{8}-(?P.*)$") @@ -12,6 +13,25 @@ _SYSTEM_AI_PREFIX = "system.ai." +def claude_discovery_model_id(model_id: str) -> str: + """Encode the exact wire ID expected by Claude's gateway discovery feature.""" + if "claude" in model_id.lower() or "anthropic" in model_id.lower(): + return model_id + checksum = hashlib.sha256(model_id.encode("utf-8")).hexdigest()[:8] + return f"anthropic-aigw-{checksum}-{model_id}" + + +def claude_model_service_id(model_id: str) -> str: + """Recover service identity only from the gateway's checksum-valid discovery alias.""" + alias = re.fullmatch(r"anthropic-aigw-([0-9a-f]{8})-(.+)", model_id) + if alias is None: + return model_id + checksum, original = alias.groups() + if hashlib.sha256(original.encode("utf-8")).hexdigest()[:8] != checksum: + return model_id + return original + + def claude_model_in_picker(screen: str, model_id: str, display_name: str | None) -> bool: """Return whether a Claude model appears in a numbered picker row. @@ -70,3 +90,43 @@ def claude_system_model_ids(models: list[dict]) -> list[str]: ), f"Expected a system.ai model id: {model_id!r}" canonical_ids.append(canonical_id) return canonical_ids + + +def codex_model_in_picker(screen: str, model_id: str, display_name: str) -> bool: + """Match native numbered picker rows, not banners, prompt text, or footers.""" + rows = re.findall(r"(?m)^[ \t]*(?:[❯›>][ \t]*)?\d+[.)][ \t]+([^\n]+)$", screen) + return any(_picker_row_matches(row, model_id, display_name) for row in rows) + + +def _picker_row_matches(label: str, model_id: str, display_name: str | None) -> bool: + return any( + re.search(rf"(? None: + """Require exactly one unambiguous numbered native row per expected model.""" + assert agent in {"claude", "codex"}, agent + assert display_names, "Expected a nonempty provider model inventory" + rows = re.findall(r"(?m)^[ \t]*(?:[❯›>][ \t]*)?\d+[.)](?:[ \t]+([^\n]*))?$", screen) + assert rows, f"No numbered {agent} picker rows:\n{screen}" + observed: list[str] = [] + for label in rows: + matched = { + model + for model, display_name in display_names.items() + if _picker_row_matches(label, model, display_name) + } + assert len(matched) == 1, ( + f"Unmatched or ambiguous {agent} picker row {label!r}: {sorted(matched)}" + ) + model = matched.pop() + assert model not in observed, f"Duplicate {agent} picker model row: {model}" + observed.append(model) + assert set(observed) == set(display_names), { + "agent": agent, + "expected": sorted(display_names), + "observed": observed, + } diff --git a/tests/integration/utils/provider_catalog.py b/tests/integration/utils/provider_catalog.py index d0a5c00fc..f87a9b95e 100644 --- a/tests/integration/utils/provider_catalog.py +++ b/tests/integration/utils/provider_catalog.py @@ -2,17 +2,16 @@ from __future__ import annotations -import json -import urllib.error import urllib.parse -import urllib.request -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import NoReturn +from .http import safe_https_json_get + _ANTHROPIC_MODELS_PATH = "/ai-gateway/anthropic/v1/models" _CODEX_MODELS_PATH = "/ai-gateway/codex/v1/models" _PROVIDER_HEADER = "Databricks-Model-Provider-Service" -_PARENT_SCHEMA_HEADER = "Databricks-Model-Service-Parent-Schema" +MODEL_SERVICE_PARENT_SCHEMA_HEADER = "Databricks-Model-Service-Parent-Schema" _ANTHROPIC_VERSION = "2023-06-01" _ANTHROPIC_PAGE_SIZE = 1000 _ANTHROPIC_MAX_PAGES = 20 @@ -33,6 +32,7 @@ class AnthropicProviderCatalog: model_ids: tuple[str, ...] display_names: dict[str, str | None] + payloads: tuple[dict, ...] = () @dataclass(frozen=True) @@ -40,6 +40,8 @@ class CodexProviderCatalog: """The list-visible model slugs from a Codex provider catalog.""" model_ids: tuple[str, ...] + payloads: tuple[dict, ...] = () + display_names: dict[str, str | None] = field(default_factory=dict) def _fail(message: str) -> NoReturn: @@ -110,8 +112,6 @@ def parse_codex_provider_catalog(payload: object) -> tuple[str, ...]: if not isinstance(visibility, str): _fail(f"Codex provider model {index} had an invalid visibility") if visibility != "list": - # Hidden/native aliases are part of the provider response but are not exposed by - # Codex's model/list request used as the observable catalog below. continue slug = raw_model.get("slug") if not isinstance(slug, str) or not slug.strip(): @@ -126,78 +126,48 @@ def parse_codex_provider_catalog(payload: object) -> tuple[str, ...]: return tuple(model_ids) -def _get_json(url: str, headers: dict[str, str]) -> object: - request = urllib.request.Request(url, headers=headers) - try: - with urllib.request.urlopen(request, timeout=30) as response: - getcode = getattr(response, "getcode", None) - status = getcode() if callable(getcode) else getattr(response, "status", None) - if status is not None and not 200 <= status < 300: - _fail(f"GET {url} returned HTTP {status}") - raw_body = response.read() - except urllib.error.HTTPError as error: - raise AssertionError(f"GET {url} returned HTTP {error.code} {error.reason}") from error - except (urllib.error.URLError, OSError) as error: - raise AssertionError(f"GET {url} failed: {error}") from error - - if isinstance(raw_body, bytes): - try: - raw_body = raw_body.decode("utf-8") - except UnicodeDecodeError as error: - raise AssertionError(f"GET {url} returned non-UTF-8 JSON") from error - if not isinstance(raw_body, str): - _fail(f"GET {url} returned a non-text response body") - try: - return json.loads(raw_body) - except json.JSONDecodeError as error: - raise AssertionError(f"GET {url} returned invalid JSON: {error.msg}") from error - - -def _validate_request_inputs(workspace: str, token: str, provider_service: str) -> str: - if not isinstance(workspace, str) or not workspace.rstrip("/"): - _fail("provider catalog request requires a workspace URL") - if not isinstance(token, str) or not token: - _fail("provider catalog request requires an explicit bearer token") - if not isinstance(provider_service, str) or not provider_service.strip(): - _fail("provider catalog request requires a provider service") - return workspace.rstrip("/") +def _validate_scope(scope: str) -> None: + if not isinstance(scope, str) or not scope.strip(): + _fail("provider catalog request requires an explicit catalog scope") def fetch_anthropic_provider_catalog( workspace: str, token: str, provider_service: str ) -> AnthropicProviderCatalog: """Fetch and validate the complete Anthropic provider catalog with bounded pagination.""" - return _fetch_anthropic_catalog(workspace, token, provider_service, _PROVIDER_HEADER) + _validate_scope(provider_service) + return _fetch_anthropic_catalog(workspace, token, {_PROVIDER_HEADER: provider_service}) def fetch_anthropic_parent_catalog( workspace: str, token: str, parent_schema: str ) -> AnthropicProviderCatalog: - """Fetch the Anthropic-compatible models advertised for a Unity Catalog parent schema.""" - return _fetch_anthropic_catalog(workspace, token, parent_schema, _PARENT_SCHEMA_HEADER) + """Fetch the complete Anthropic catalog scoped to a Unity Catalog parent schema.""" + _validate_scope(parent_schema) + return _fetch_anthropic_catalog( + workspace, token, {MODEL_SERVICE_PARENT_SCHEMA_HEADER: parent_schema} + ) def _fetch_anthropic_catalog( - workspace: str, token: str, source: str, source_header: str + workspace: str, token: str, scope_headers: dict[str, str] ) -> AnthropicProviderCatalog: - base_url = _validate_request_inputs(workspace, token, source) - headers = { - "Authorization": f"Bearer {token}", - "Accept": "application/json", - "Anthropic-Version": _ANTHROPIC_VERSION, - source_header: source, - } + headers = {"Anthropic-Version": _ANTHROPIC_VERSION, **scope_headers} model_ids: list[str] = [] display_names: dict[str, str | None] = {} + payloads: list[dict] = [] cursor: str | None = None cursors: set[str] = set() for _page_number in range(_ANTHROPIC_MAX_PAGES): query = {"limit": str(_ANTHROPIC_PAGE_SIZE)} if cursor is not None: query["after_id"] = cursor - url = f"{base_url}{_ANTHROPIC_MODELS_PATH}?{urllib.parse.urlencode(query)}" - page = parse_anthropic_provider_page(_get_json(url, headers)) + path = f"{_ANTHROPIC_MODELS_PATH}?{urllib.parse.urlencode(query)}" + payload = safe_https_json_get(workspace, token, path, headers=headers) + page = parse_anthropic_provider_page(payload) + assert isinstance(payload, dict), "Expected an Anthropic catalog object" + payloads.append(payload) for model_id, display_name in page.models: if model_id in display_names: _fail(f"Anthropic provider catalog repeated model id {model_id!r}") @@ -205,8 +175,8 @@ def _fetch_anthropic_catalog( display_names[model_id] = display_name if not page.has_more: - return AnthropicProviderCatalog(tuple(model_ids), display_names) - assert page.last_id is not None # checked by parse_anthropic_provider_page + return AnthropicProviderCatalog(tuple(model_ids), display_names, tuple(payloads)) + assert page.last_id is not None if page.last_id in cursors: _fail(f"Anthropic provider catalog repeated pagination cursor {page.last_id!r}") cursors.add(page.last_id) @@ -219,29 +189,31 @@ def fetch_codex_provider_catalog( workspace: str, token: str, provider_service: str ) -> CodexProviderCatalog: """Fetch and validate the list-visible Codex provider catalog.""" - base_url = _validate_request_inputs(workspace, token, provider_service) - payload = _get_json( - f"{base_url}{_CODEX_MODELS_PATH}", - { - "Authorization": f"Bearer {token}", - "Accept": "application/json", - _PROVIDER_HEADER: provider_service, - }, - ) - return CodexProviderCatalog(parse_codex_provider_catalog(payload)) + _validate_scope(provider_service) + return _fetch_codex_catalog(workspace, token, {_PROVIDER_HEADER: provider_service}) + + +def _fetch_codex_catalog( + workspace: str, token: str, scope_headers: dict[str, str] +) -> CodexProviderCatalog: + payload = safe_https_json_get(workspace, token, _CODEX_MODELS_PATH, headers=scope_headers) + model_ids = parse_codex_provider_catalog(payload) + assert isinstance(payload, dict), "Expected a Codex catalog object" + display_names: dict[str, str | None] = {} + for entry in payload["models"]: + if entry.get("visibility") == "list" and entry.get("slug") in model_ids: + label = entry.get("display_name") + if label is not None and (not isinstance(label, str) or not label.strip()): + _fail("Invalid Codex catalog display name") + display_names[entry["slug"]] = label + return CodexProviderCatalog(model_ids, (payload,), display_names) def fetch_codex_parent_catalog( workspace: str, token: str, parent_schema: str ) -> CodexProviderCatalog: """Fetch the API-compatible models advertised for a Unity Catalog parent schema.""" - base_url = _validate_request_inputs(workspace, token, parent_schema) - payload = _get_json( - f"{base_url}{_CODEX_MODELS_PATH}", - { - "Authorization": f"Bearer {token}", - "Accept": "application/json", - _PARENT_SCHEMA_HEADER: parent_schema, - }, + _validate_scope(parent_schema) + return _fetch_codex_catalog( + workspace, token, {MODEL_SERVICE_PARENT_SCHEMA_HEADER: parent_schema} ) - return CodexProviderCatalog(parse_codex_provider_catalog(payload)) diff --git a/tests/integration/utils/terminal.py b/tests/integration/utils/terminal.py index 9f4c4fda7..b0175037e 100644 --- a/tests/integration/utils/terminal.py +++ b/tests/integration/utils/terminal.py @@ -318,6 +318,31 @@ def open_model_picker(self, *, model_visible=None): ) return screen + def open_codex_model_picker(self, *, model_visible): + """Capture Codex's native numbered /model menu, then dismiss it with Escape.""" + assert self.agent == "codex", self.agent + self.submit("/model") + self.wait_for( + lambda text: ( + "select model" in text.lower() + and re.search(r"(?m)^[ \t]*(?:[❯›>][ \t]*)?\d+[.)][ \t]+\S", text) + ), + "Codex's numbered model picker", + timeout=60, + ) + self.wait_for(model_visible, "all scoped models in Codex's picker", timeout=60) + screen = self.visible + self.actions.append({"reason": "codex-model-picker-visible", "screen": screen}) + self.send("\x1b", "close Codex's model picker without changing its default") + self.wait_for( + lambda text: ( + "select model" not in text.lower() + and re.search(r"(?m)^\s*[❯›>]\s*(?!\d+[.)])", text) + ), + "the Codex prompt after closing the model picker", + ) + return screen + def wait_for_task(self, task, timeout=180): permission_in_progress = False diff --git a/tests/test_integration_agent_helpers.py b/tests/test_integration_agent_helpers.py new file mode 100644 index 000000000..28dabd3ac --- /dev/null +++ b/tests/test_integration_agent_helpers.py @@ -0,0 +1,86 @@ +"""Component checks for the reusable integration agent helper contracts.""" + +from types import SimpleNamespace + +import pytest + +from tests.integration.utils import evidence, model_discovery, provider_catalog +from tests.integration.utils.agents import claude, codex + + +def test_evidence_keeps_legacy_completed_task_names_as_adapter_aliases(): + assert evidence.claude_completed_task_models is claude.completed_task_models + assert evidence.codex_completed_task_models is codex.completed_task_models + + +def test_agent_modules_expose_direct_model_discovery_helpers(): + assert claude.fetch_parent_catalog is provider_catalog.fetch_anthropic_parent_catalog + assert claude.discovery_model_id is model_discovery.claude_discovery_model_id + assert claude.model_service_id is model_discovery.claude_model_service_id + assert claude.model_in_picker is model_discovery.claude_model_in_picker + assert codex.fetch_parent_catalog is provider_catalog.fetch_codex_parent_catalog + assert codex.model_in_picker is model_discovery.codex_model_in_picker + + +def test_agent_modules_preserve_exact_completion_attribution(): + claude_records = [ + { + "type": "user", + "message": {"model": "catalog.other_models.decoy", "content": []}, + }, + { + "type": "assistant", + "message": { + "role": "assistant", + "model": "catalog.models.claude_sonnet[1m]", + "content": [{"type": "text", "text": "value"}], + }, + }, + ] + codex_records = [ + { + "type": "turn_context", + "payload": {"turn_id": "other", "model": "catalog.other_models.decoy"}, + }, + { + "type": "turn_context", + "payload": {"turn_id": "matching", "model": "catalog.models.gpt_luna"}, + }, + { + "type": "event_msg", + "payload": { + "type": "task_complete", + "turn_id": "matching", + "last_agent_message": "value", + }, + }, + ] + + assert claude.completed_task_models(claude_records, "value") == {"catalog.models.claude_sonnet"} + assert codex.completed_task_models(codex_records, "value") == {"catalog.models.gpt_luna"} + + +def test_evidence_dispatches_assistant_answers_to_agent_modules(): + claude_records = [ + { + "type": "assistant", + "message": {"role": "assistant", "content": [{"type": "text", "text": "answer"}]}, + } + ] + codex_records = [ + { + "type": "event_msg", + "payload": {"type": "task_complete", "last_agent_message": "answer"}, + } + ] + + assert evidence.assistant_answers("claude", claude_records) == ["answer"] + assert evidence.assistant_answers("codex", codex_records) == ["answer"] + + +def test_unsupported_evidence_agents_fail_without_codex_fallback(): + session = SimpleNamespace(home=None, record=lambda *_: None) + with pytest.raises(AssertionError, match="Unsupported evidence agent: opencode"): + evidence.completed_task_models(session, "opencode", "value") + with pytest.raises(AssertionError, match="Unsupported evidence agent: opencode"): + evidence.assert_completed_task_model(session, "opencode", "value", "model") diff --git a/tests/test_integration_evidence.py b/tests/test_integration_evidence.py index f618304d1..4b6b31e1f 100644 --- a/tests/test_integration_evidence.py +++ b/tests/test_integration_evidence.py @@ -1,9 +1,11 @@ """Unit checks for interpreting terminal evidence, not live agent substitutes.""" import json +from types import SimpleNamespace import pytest +from tests.integration.utils import evidence from tests.integration.utils.evidence import ( SubagentCalculation, assert_no_terminal_api_error, @@ -89,3 +91,240 @@ def test_tagged_calculation_requires_the_native_child_answer(tmp_path, agent): assert task.marker in task.prompt assert f'task name "{task.marker}"' in task.prompt assert "1+1" in task.prompt + + +def test_codex_model_identity_uses_only_the_completed_answer_turn(monkeypatch): + records = [ + { + "type": "turn_context", + "payload": {"turn_id": "other", "model": "catalog.other_models.codex_decoy"}, + }, + { + "type": "turn_context", + "payload": {"turn_id": "matching", "model": "catalog.models.gpt_luna"}, + }, + { + "type": "event_msg", + "payload": { + "type": "task_complete", + "turn_id": "matching", + "last_agent_message": "withheld-file-value", + }, + }, + ] + monkeypatch.setattr(evidence, "agent_sessions", lambda *_: {"transcript": records}) + assert evidence.completed_task_models(None, "codex", "withheld-file-value") == { + "catalog.models.gpt_luna" + } + + +def test_codex_model_identity_rejects_prompt_only_evidence(monkeypatch): + records = [ + { + "type": "turn_context", + "payload": {"turn_id": "matching", "model": "catalog.models.gpt_luna"}, + }, + { + "type": "response_item", + "payload": {"type": "message", "role": "user", "content": "withheld-file-value"}, + }, + ] + monkeypatch.setattr(evidence, "agent_sessions", lambda *_: {"transcript": records}) + assert evidence.completed_task_models(None, "codex", "withheld-file-value") == set() + + +def test_claude_model_identity_uses_assistant_answer_not_tool_output(monkeypatch): + records = [ + { + "type": "user", + "message": { + "model": "catalog.other_models.claude_decoy", + "content": [{"type": "text", "text": "value"}], + }, + }, + { + "type": "assistant", + "message": { + "role": "assistant", + "model": "catalog.models.claude_sonnet", + "content": [{"type": "text", "text": "value"}], + }, + }, + ] + monkeypatch.setattr(evidence, "agent_sessions", lambda *_: {"transcript": records}) + assert evidence.completed_task_models(None, "claude", "value") == { + "catalog.models.claude_sonnet" + } + + +def completed_records(agent, model, answer="value", turn_id="matching"): + if agent == "claude": + return [ + { + "type": "assistant", + "message": { + "role": "assistant", + "model": model, + "content": [{"type": "text", "text": answer}], + }, + } + ] + return [ + {"type": "turn_context", "payload": {"turn_id": turn_id, "model": model}}, + { + "type": "event_msg", + "payload": {"type": "task_complete", "turn_id": turn_id, "last_agent_message": answer}, + }, + ] + + +@pytest.mark.parametrize("agent", ["claude", "codex"]) +@pytest.mark.parametrize("parent_answer", [None, "value", "unrelated-answer"]) +def test_completed_task_models_excludes_child_only_and_conflicting_child_evidence( + monkeypatch, agent, parent_answer +): + child = completed_records(agent, "child-model") + if agent == "codex": + child.insert(0, {"type": "session_meta", "payload": {"source": {"subagent": "spawn"}}}) + sessions = {"project/subagents/child.jsonl": child} + if parent_answer is not None: + sessions["project/parent.jsonl"] = completed_records(agent, "parent-model", parent_answer) + monkeypatch.setattr(evidence, "agent_sessions", lambda *_: sessions) + expected = {"parent-model"} if parent_answer == "value" else set() + assert evidence.completed_task_models(None, agent, "value") == expected + + +@pytest.mark.parametrize("agent", ["claude", "codex"]) +@pytest.mark.parametrize("model", [None, "", " ", 5, {}, [], "model with whitespace"]) +def test_completed_task_models_fails_closed_on_malformed_model(monkeypatch, agent, model): + records = completed_records(agent, model) + monkeypatch.setattr(evidence, "agent_sessions", lambda *_: {"parent": records}) + with pytest.raises(AssertionError, match="model"): + evidence.completed_task_models(None, agent, "value") + + +@pytest.mark.parametrize("agent", ["claude", "codex"]) +def test_completed_task_models_fails_closed_on_missing_model(monkeypatch, agent): + records = completed_records(agent, "model") + del records[0]["message" if agent == "claude" else "payload"]["model"] + monkeypatch.setattr(evidence, "agent_sessions", lambda *_: {"parent": records}) + with pytest.raises(AssertionError, match="model"): + evidence.completed_task_models(None, agent, "value") + + +@pytest.mark.parametrize("turn_id", [None, "", " ", 5, {}, [], "turn with whitespace"]) +@pytest.mark.parametrize("record_index", [0, 1]) +def test_codex_completed_task_models_fails_closed_on_malformed_turn_ids(turn_id, record_index): + records = completed_records("codex", "model") + records[record_index]["payload"]["turn_id"] = turn_id + with pytest.raises(AssertionError, match="turn ID"): + evidence.codex_completed_task_models(records, "value") + + +@pytest.mark.parametrize("record_index", [0, 1]) +def test_codex_completed_task_models_fails_closed_on_missing_turn_ids(record_index): + records = completed_records("codex", "model") + del records[record_index]["payload"]["turn_id"] + with pytest.raises(AssertionError, match="turn ID"): + evidence.codex_completed_task_models(records, "value") + + +def test_codex_completed_task_models_requires_context_in_same_session(monkeypatch): + context, completion = completed_records("codex", "model") + monkeypatch.setattr( + evidence, + "agent_sessions", + lambda *_: {"context-session": [context], "answer-session": [completion]}, + ) + with pytest.raises(AssertionError, match="Missing Codex model context"): + evidence.completed_task_models(None, "codex", "value") + + +@pytest.mark.parametrize("agent", ["claude", "codex"]) +@pytest.mark.parametrize("models", [[], ["unexpected"], ["expected", "conflicting"]]) +def test_completed_task_model_assertion_requires_exact_singleton(monkeypatch, agent, models): + records = [record for model in models for record in completed_records(agent, model)] + monkeypatch.setattr(evidence, "agent_sessions", lambda *_: {"parent": records}) + artifacts = {} + session = SimpleNamespace(record=lambda name, payload: artifacts.update({name: payload})) + with pytest.raises(AssertionError): + evidence.assert_completed_task_model(session, agent, "value", "expected") + assert next(iter(artifacts.values()))["observed"] == sorted(models) + + +@pytest.mark.parametrize("agent", ["claude", "codex"]) +def test_completed_task_model_records_evidence_limits(monkeypatch, agent): + monkeypatch.setattr( + evidence, "agent_sessions", lambda *_: {"parent": completed_records(agent, "expected")} + ) + artifacts = {} + session = SimpleNamespace(record=lambda name, payload: artifacts.update({name: payload})) + evidence.assert_completed_task_model(session, agent, "value", "expected") + assert artifacts == { + f"completed-task-model-{agent}-value.json": { + "expected": "expected", + "observed": ["expected"], + "evidence_kind": ( + "response-reported model" if agent == "claude" else "client-selected model" + ), + "gateway_destination_proven": False, + } + } + + +@pytest.mark.parametrize("agent", ["claude", "codex"]) +@pytest.mark.parametrize("answer_value", [None, "", " \t", 1, [], {}]) +def test_completed_task_models_rejects_empty_or_malformed_expected_answers( + monkeypatch, agent, answer_value +): + def unexpected_read(*args): + pytest.fail("Invalid answer must not read session evidence") + + monkeypatch.setattr(evidence, "agent_sessions", unexpected_read) + with pytest.raises(AssertionError, match="nonempty answer value"): + evidence.completed_task_models(None, agent, answer_value) + adapter = ( + evidence.claude_completed_task_models + if agent == "claude" + else evidence.codex_completed_task_models + ) + with pytest.raises(AssertionError, match="nonempty answer value"): + adapter([], answer_value) + + +@pytest.mark.parametrize( + "message", + [None, [], {}, {"role": "assistant"}, {"role": "assistant", "content": "value"}], +) +def test_claude_completed_task_models_rejects_malformed_message_metadata(message): + with pytest.raises(AssertionError): + evidence.claude_completed_task_models([{"type": "assistant", "message": message}], "value") + + +@pytest.mark.parametrize("content", [[None], [{"type": "text"}], [{"type": "text", "text": 4}]]) +def test_claude_completed_task_models_rejects_malformed_answer_blocks(content): + records = completed_records("claude", "model") + records[0]["message"]["content"] = content + with pytest.raises(AssertionError): + evidence.claude_completed_task_models(records, "value") + + +@pytest.mark.parametrize("payload", [None, [], "value"]) +@pytest.mark.parametrize("record_type", ["event_msg", "turn_context"]) +def test_codex_completed_task_models_rejects_malformed_payloads(payload, record_type): + with pytest.raises(AssertionError): + evidence.codex_completed_task_models([{"type": record_type, "payload": payload}], "value") + + +@pytest.mark.parametrize("answer", [None, [], {}, 7]) +def test_codex_completed_task_models_rejects_malformed_completed_answers(answer): + records = completed_records("codex", "model") + records[1]["payload"]["last_agent_message"] = answer + with pytest.raises(AssertionError, match="completed answer"): + evidence.codex_completed_task_models(records, "value") + + +def test_claude_completed_task_models_normalizes_only_context_window_suffix(): + assert evidence.claude_completed_task_models( + completed_records("claude", "catalog.models.claude_sonnet[1m]"), "value" + ) == {"catalog.models.claude_sonnet"} diff --git a/tests/test_integration_http.py b/tests/test_integration_http.py new file mode 100644 index 000000000..ef6df7c8e --- /dev/null +++ b/tests/test_integration_http.py @@ -0,0 +1,116 @@ +"""Offline component checks for read-only workspace HTTP boundaries.""" + +import io +import urllib.error +import urllib.response + +import pytest + +from tests.integration.utils import http, provider_catalog + +FETCHERS = [ + provider_catalog.fetch_anthropic_parent_catalog, + provider_catalog.fetch_codex_parent_catalog, + provider_catalog.fetch_anthropic_provider_catalog, + provider_catalog.fetch_codex_provider_catalog, +] + + +def fetch_inventory(fetcher, workspace="https://workspace.invalid", token="test-bearer"): + return fetcher(workspace, token, "catalog.models") + + +@pytest.mark.parametrize("fetcher", FETCHERS) +@pytest.mark.parametrize( + "workspace", + [ + None, + 7, + "", + "http://workspace.invalid", + "https://user:secret@workspace.invalid", + "https://workspace.invalid/path", + "https://workspace.invalid?secret=value", + "https://workspace.invalid#fragment", + "https:///missing-host", + "https://workspace.invalid:bad", + "https://workspace.invalid:0", + "https://workspace.invalid:70000", + "https://workspace.invalid\n", + "https://workspace.invalid\\@other.invalid", + ], +) +def test_catalog_gets_reject_invalid_workspace_origins(monkeypatch, fetcher, workspace): + def unexpected_open(*args, **kwargs): + pytest.fail("Invalid workspace must not reach the HTTP boundary") + + monkeypatch.setattr(http.urllib.request, "build_opener", unexpected_open) + with pytest.raises(AssertionError, match="HTTPS workspace origin") as failure: + fetch_inventory(fetcher, workspace) + assert "secret" not in str(failure.value) + + +@pytest.mark.parametrize("fetcher", FETCHERS) +@pytest.mark.parametrize("token", [None, "", " ", "bearer\nsecret", "bearer\rsecret"]) +def test_catalog_gets_require_explicit_valid_bearers(monkeypatch, fetcher, token): + def unexpected_open(*args, **kwargs): + pytest.fail("Invalid bearer must not reach the HTTP boundary") + + monkeypatch.setattr(http.urllib.request, "build_opener", unexpected_open) + with pytest.raises(AssertionError, match="explicit workspace bearer"): + fetch_inventory(fetcher, token=token) + + +@pytest.mark.parametrize("fetcher", FETCHERS) +@pytest.mark.parametrize("failure_kind", ["http", "url", "os", "json", "encoding"]) +def test_catalog_get_errors_are_sanitized(monkeypatch, fetcher, failure_kind): + secret = "private-bearer-and-response-secret" + + class FailedOpener: + def open(self, request, *, timeout): + assert timeout == 30 + assert request.get_method() == "GET" + assert request.get_header("Authorization") == f"Bearer {secret}" + if failure_kind == "http": + raise urllib.error.HTTPError(request.full_url, 403, secret, {}, None) + if failure_kind == "url": + raise urllib.error.URLError(secret) + if failure_kind == "os": + raise OSError(secret) + return io.BytesIO(secret.encode() if failure_kind == "json" else b"\xff") + + monkeypatch.setattr(http.urllib.request, "build_opener", lambda *_: FailedOpener()) + with pytest.raises(AssertionError, match="Workspace JSON GET") as failure: + fetch_inventory(fetcher, token=secret) + assert secret not in str(failure.value) + assert "workspace.invalid" not in str(failure.value) + assert failure.value.__suppress_context__ + + +@pytest.mark.parametrize("fetcher", FETCHERS) +@pytest.mark.parametrize("status", [301, 302, 303, 307, 308]) +@pytest.mark.parametrize("destination", ["https://other.invalid/secret", "/redirected"]) +def test_catalog_gets_deny_redirects(monkeypatch, fetcher, status, destination): + requests = [] + build_opener = http.urllib.request.build_opener + + class BoundaryHTTPS(http.urllib.request.HTTPSHandler): + def https_open(self, request): + requests.append(request) + response = urllib.response.addinfourl( + io.BytesIO(b"{}"), {"Location": destination}, request.full_url, status + ) + response.msg = "private-redirect-reason" + return response + + monkeypatch.setattr( + http.urllib.request, + "build_opener", + lambda handler: build_opener(handler, BoundaryHTTPS()), + ) + with pytest.raises(AssertionError, match=f"HTTP {status}") as failure: + fetch_inventory(fetcher) + assert len(requests) == 1 + assert requests[0].get_header("Authorization") == "Bearer test-bearer" + assert destination not in str(failure.value) + assert "private-redirect-reason" not in str(failure.value) diff --git a/tests/test_integration_managed_config.py b/tests/test_integration_managed_config.py new file mode 100644 index 000000000..111fd1e82 --- /dev/null +++ b/tests/test_integration_managed_config.py @@ -0,0 +1,96 @@ +"""Component checks for reading persisted managed-config evidence.""" + +import json +from types import SimpleNamespace + +import pytest + +from tests.integration.utils.managed import read_persisted_managed_config + +WORKSPACE = "https://workspace.invalid" + + +@pytest.fixture +def session(tmp_path): + return SimpleNamespace(home=tmp_path / "session-home") + + +def _write_persisted_config(session, envelope): + path = session.home / ".ucode" / "managed-config.json" + path.parent.mkdir(parents=True) + path.write_text(json.dumps(envelope), encoding="utf-8") + return path + + +@pytest.mark.parametrize( + "config", + [ + {}, + { + "default_agent": "unrelated-agent", + "custom_policy": {"labels": ["independent", "policy"], "enabled": True}, + }, + ], +) +def test_read_persisted_managed_config_returns_envelope_without_enforcing_policy(session, config): + envelope = {"workspace": WORKSPACE, "config": config, "metadata": {"revision": 7}} + path = _write_persisted_config(session, envelope) + original = path.read_text(encoding="utf-8") + + assert read_persisted_managed_config(session, WORKSPACE) == envelope + assert path.read_text(encoding="utf-8") == original + + +@pytest.mark.parametrize("directory_instead", [False, True]) +def test_read_persisted_managed_config_requires_a_file(session, directory_instead): + if directory_instead: + (session.home / ".ucode" / "managed-config.json").mkdir(parents=True) + + with pytest.raises(AssertionError): + read_persisted_managed_config(session, WORKSPACE) + + +@pytest.mark.parametrize("contents", ["", "{", '{"workspace":', "{} trailing-data"]) +def test_read_persisted_managed_config_rejects_invalid_json(session, contents): + path = _write_persisted_config(session, {}) + path.write_text(contents, encoding="utf-8") + + with pytest.raises(json.JSONDecodeError): + read_persisted_managed_config(session, WORKSPACE) + + +@pytest.mark.parametrize("envelope", [None, [], ["config"], "config", False, 7]) +def test_read_persisted_managed_config_requires_a_dict_envelope(session, envelope): + _write_persisted_config(session, envelope) + + with pytest.raises(AssertionError): + read_persisted_managed_config(session, WORKSPACE) + + +@pytest.mark.parametrize( + "workspace", + [None, "", "https://other-workspace.invalid", WORKSPACE + "/", WORKSPACE.upper(), 7, {}, []], +) +def test_read_persisted_managed_config_requires_an_exact_workspace_match(session, workspace): + _write_persisted_config(session, {"workspace": workspace, "config": {}}) + + with pytest.raises(AssertionError): + read_persisted_managed_config(session, WORKSPACE) + + +@pytest.mark.parametrize("config", [None, [], ["policy"], "policy", False, 7]) +def test_read_persisted_managed_config_requires_a_dict_config(session, config): + _write_persisted_config(session, {"workspace": WORKSPACE, "config": config}) + + with pytest.raises(AssertionError): + read_persisted_managed_config(session, WORKSPACE) + + +@pytest.mark.parametrize("field", ["workspace", "config"]) +def test_read_persisted_managed_config_rejects_missing_fields(session, field): + envelope = {"workspace": WORKSPACE, "config": {}} + del envelope[field] + _write_persisted_config(session, envelope) + + with pytest.raises(AssertionError): + read_persisted_managed_config(session, WORKSPACE) diff --git a/tests/test_integration_model_discovery.py b/tests/test_integration_model_discovery.py index fd8860e7f..27bf981fb 100644 --- a/tests/test_integration_model_discovery.py +++ b/tests/test_integration_model_discovery.py @@ -2,7 +2,21 @@ import pytest -from tests.integration.utils.model_discovery import claude_model_in_picker, claude_system_model_ids +from tests.integration.utils.model_discovery import ( + assert_picker_inventory, + claude_discovery_model_id, + claude_model_in_picker, + claude_model_service_id, + claude_system_model_ids, + codex_model_in_picker, +) + +PICKER_MODELS = { + "claude": frozenset( + {"catalog.models.claude_sonnet", "catalog.models.claude_haiku", "catalog.models.kimi"} + ), + "codex": frozenset({"catalog.models.gpt_luna", "catalog.models.kimi"}), +} @pytest.mark.parametrize("cursor", ["", "❯ ", "› ", "> "]) @@ -131,3 +145,118 @@ def test_claude_system_model_ids_rejects_invalid_or_out_of_scope_ids(model_id): def test_claude_system_model_ids_rejects_empty_missing_and_duplicate_ids(models): with pytest.raises(AssertionError): claude_system_model_ids(models) + + +@pytest.mark.parametrize( + ("wire_id", "expected"), + [ + ("catalog.models.kimi", "catalog.models.kimi"), + ("anthropic-aigw-69e2a9a0-catalog.models.kimi", "catalog.models.kimi"), + ("catalog.models.claude_sonnet", "catalog.models.claude_sonnet"), + ( + "anthropic-aigw-00000000-catalog.models.kimi", + "anthropic-aigw-00000000-catalog.models.kimi", + ), + ( + "anthropic-aigw-69e2a9a0-catalog.models.kimi_v2", + "anthropic-aigw-69e2a9a0-catalog.models.kimi_v2", + ), + ( + "anthropic-aigw-69e2a9a0-catalog.other_models.kimi", + "anthropic-aigw-69e2a9a0-catalog.other_models.kimi", + ), + ( + "anthropic-aigw-69E2A9A0-catalog.models.kimi", + "anthropic-aigw-69E2A9A0-catalog.models.kimi", + ), + ], +) +def test_claude_service_identity_requires_exact_gateway_alias_checksum(wire_id, expected): + assert claude_model_service_id(wire_id) == expected + + +def test_claude_discovery_requires_exact_gateway_wire_ids(): + assert {claude_discovery_model_id(model) for model in PICKER_MODELS["claude"]} == { + "catalog.models.claude_haiku", + "catalog.models.claude_sonnet", + "anthropic-aigw-69e2a9a0-catalog.models.kimi", + } + assert ( + claude_discovery_model_id("catalog.other_models.claude_decoy") + == "catalog.other_models.claude_decoy" + ) + + +@pytest.mark.parametrize("cursor", ["", "❯ ", "› ", "> "]) +@pytest.mark.parametrize("label", ["catalog.models.gpt_luna", "GPT Luna"]) +def test_codex_picker_matches_model_only_in_numbered_rows(cursor, label): + screen = f"Select Model and Effort\n {cursor}1. {label} (current)\nEnter to confirm" + assert codex_model_in_picker(screen, "catalog.models.gpt_luna", "GPT Luna") + + +@pytest.mark.parametrize( + "screen", + [ + "GPT Luna\nSelect Model and Effort\n 1. Other model", + "Select Model and Effort\n 1. Other model\nCurrent model: GPT Luna", + " 1. catalog.models.gpt_luna_v2", + " 1.\nGPT Luna", + "", + ], +) +def test_codex_picker_rejects_nonrows_and_different_model_ids(screen): + assert not codex_model_in_picker(screen, "catalog.models.gpt_luna", "GPT Luna") + + +@pytest.mark.parametrize("agent", ["claude", "codex"]) +def test_picker_inventory_accepts_exact_rows_and_ignores_banner_text(agent): + models = PICKER_MODELS["claude"] if agent == "claude" else PICKER_MODELS["codex"] + labels = {model: model.rsplit(".", 1)[-1].replace("_", " ").title() for model in models} + rows = "\n".join( + f" {position}. {label}" for position, label in enumerate(labels.values(), start=1) + ) + screen = f"Banner: Gemini Flash\nSelect model\n{rows}\nEnter to confirm" + assert_picker_inventory(screen, agent, labels) + + +@pytest.mark.parametrize("agent", ["claude", "codex"]) +@pytest.mark.parametrize("excluded_label", ["Gemini Flash", "Friendly out-of-scope model"]) +def test_picker_inventory_rejects_extra_friendly_label_rows(agent, excluded_label): + models = PICKER_MODELS["claude"] if agent == "claude" else PICKER_MODELS["codex"] + labels = dict.fromkeys(models, None) + rows = "\n".join(f" {position}. {model}" for position, model in enumerate(models, start=1)) + screen = f"Select model\n{rows}\n {len(models) + 1}. {excluded_label}" + with pytest.raises(AssertionError): + assert_picker_inventory(screen, agent, labels) + + +@pytest.mark.parametrize("agent", ["claude", "codex"]) +def test_picker_inventory_rejects_duplicate_and_missing_rows(agent): + models = sorted(PICKER_MODELS["claude"] if agent == "claude" else PICKER_MODELS["codex"]) + labels = dict.fromkeys(models, None) + rows = "\n".join(f" {position}. {model}" for position, model in enumerate(models, start=1)) + with pytest.raises(AssertionError): + assert_picker_inventory(f"{rows}\n {len(models) + 1}. {models[0]}", agent, labels) + with pytest.raises(AssertionError): + assert_picker_inventory(f" 1. {models[0]}", agent, labels) + + +@pytest.mark.parametrize("agent", ["claude", "codex"]) +def test_picker_inventory_rejects_ambiguous_labels_and_split_line_evidence(agent): + models = sorted(PICKER_MODELS["claude"] if agent == "claude" else PICKER_MODELS["codex"]) + with pytest.raises(AssertionError): + assert_picker_inventory(" 1. Shared label", agent, dict.fromkeys(models, "Shared label")) + with pytest.raises(AssertionError): + assert_picker_inventory(f" 1.\n{models[0]}", agent, dict.fromkeys(models, None)) + + +@pytest.mark.parametrize("agent", ["claude", "codex"]) +@pytest.mark.parametrize("suffix", ["_v2", "-decoy", ".other"]) +def test_picker_inventory_rejects_model_id_prefix_matches(agent, suffix): + models = sorted(PICKER_MODELS["claude"] if agent == "claude" else PICKER_MODELS["codex"]) + rows = "\n".join( + f" {position}. {model}{suffix if position == 1 else ''}" + for position, model in enumerate(models, start=1) + ) + with pytest.raises(AssertionError): + assert_picker_inventory(rows, agent, dict.fromkeys(models, None)) diff --git a/tests/test_integration_provider_catalog.py b/tests/test_integration_provider_catalog.py index 41aa1f66d..39e2e0ad8 100644 --- a/tests/test_integration_provider_catalog.py +++ b/tests/test_integration_provider_catalog.py @@ -7,9 +7,32 @@ import pytest +from tests.integration.utils import http from tests.integration.utils import provider_catalog as catalog +def _mock_json_boundary(monkeypatch, get_json): + class CatalogOpener: + def open(self, request, *, timeout): + assert request.get_method() == "GET" + assert timeout == 30 + names = ( + "Authorization", + "Accept", + "Anthropic-Version", + "Databricks-Model-Provider-Service", + catalog.MODEL_SERVICE_PARENT_SCHEMA_HEADER, + ) + headers = { + name: request.get_header(name.capitalize()) + for name in names + if request.has_header(name.capitalize()) + } + return io.BytesIO(json.dumps(get_json(request.full_url, headers)).encode()) + + monkeypatch.setattr(http.urllib.request, "build_opener", lambda *_: CatalogOpener()) + + def _page(*ids, has_more=False, last_id=None): return { "data": [{"id": model_id, "display_name": model_id} for model_id in ids], @@ -83,7 +106,7 @@ def get_json(url, headers): requests.append((url, headers)) return next(pages) - monkeypatch.setattr(catalog, "_get_json", get_json) + _mock_json_boundary(monkeypatch, get_json) result = catalog.fetch_anthropic_provider_catalog("https://workspace/", "token", "c.s.mps") assert result.model_ids == ("a", "b") assert result.display_names == {"a": "a", "b": "b"} @@ -93,6 +116,8 @@ def get_json(url, headers): for _, headers in requests: assert headers["Authorization"] == "Bearer token" assert headers["Databricks-Model-Provider-Service"] == "c.s.mps" + assert headers["Anthropic-Version"] == "2023-06-01" + assert catalog.MODEL_SERVICE_PARENT_SCHEMA_HEADER not in headers pages = iter([_page("a", has_more=True, last_id="a"), _page("a")]) with pytest.raises(AssertionError, match="repeated model id"): @@ -103,14 +128,14 @@ def test_anthropic_provider_fetch_rejects_repeated_cursor(monkeypatch): pages = iter( [_page("a", has_more=True, last_id="cursor"), _page("b", has_more=True, last_id="cursor")] ) - monkeypatch.setattr(catalog, "_get_json", lambda *args: next(pages)) + _mock_json_boundary(monkeypatch, lambda *args: next(pages)) with pytest.raises(AssertionError, match="repeated pagination cursor"): catalog.fetch_anthropic_provider_catalog("https://workspace", "token", "c.s.mps") def test_anthropic_provider_fetch_has_a_page_bound(monkeypatch): pages = iter(_page(str(i), has_more=True, last_id=str(i)) for i in range(20)) - monkeypatch.setattr(catalog, "_get_json", lambda *args: next(pages)) + _mock_json_boundary(monkeypatch, lambda *args: next(pages)) with pytest.raises(AssertionError, match="exceeded 20 pages"): catalog.fetch_anthropic_provider_catalog("https://workspace", "token", "c.s.mps") @@ -127,10 +152,15 @@ def urlopen(request, timeout): assert request.full_url == "https://workspace/ai-gateway/codex/v1/models" assert request.get_header("Authorization") == "Bearer token" assert request.get_header("Databricks-model-provider-service") == "c.s.mps" + assert not request.has_header(catalog.MODEL_SERVICE_PARENT_SCHEMA_HEADER.capitalize()) assert timeout == 30 return Response(json.dumps(payload).encode()) - monkeypatch.setattr(catalog.urllib.request, "urlopen", urlopen) + monkeypatch.setattr( + http.urllib.request, + "build_opener", + lambda *_: type("CatalogOpener", (), {"open": staticmethod(urlopen)})(), + ) result = catalog.fetch_codex_provider_catalog("https://workspace/", "token", "c.s.mps") assert result.model_ids == ("gpt-a",) @@ -144,7 +174,7 @@ def get_json(url, headers): assert "Databricks-Model-Provider-Service" not in headers return {"models": [{"slug": model, "visibility": "list"} for model in ids]} - monkeypatch.setattr(catalog, "_get_json", get_json) + _mock_json_boundary(monkeypatch, get_json) result = catalog.fetch_codex_parent_catalog("https://workspace/", "token", "c.s") assert result.model_ids == tuple(ids) @@ -154,6 +184,191 @@ def test_provider_fetch_does_not_hide_http_failures(monkeypatch, status): def urlopen(request, timeout): raise urllib.error.HTTPError(request.full_url, status, "failure", {}, None) - monkeypatch.setattr(catalog.urllib.request, "urlopen", urlopen) + monkeypatch.setattr( + http.urllib.request, + "build_opener", + lambda *_: type("CatalogOpener", (), {"open": staticmethod(urlopen)})(), + ) with pytest.raises(AssertionError, match=f"HTTP {status}"): catalog.fetch_codex_provider_catalog("https://workspace", "token", "c.s.mps") + + +def test_anthropic_parent_catalog_retains_pages_labels_and_scoped_requests(monkeypatch): + pages = [ + { + "data": [{"id": "catalog.models.claude_sonnet", "display_name": "Sonnet"}], + "has_more": True, + "last_id": "cursor with / and ?", + }, + { + "data": [{"id": "anthropic-aigw-12345678-catalog.models.kimi"}], + "has_more": False, + }, + ] + requests = [] + + class CatalogOpener: + def open(self, request, *, timeout): + requests.append(request) + assert timeout == 30 + assert request.get_header("Authorization") == "Bearer test-bearer" + assert ( + request.get_header(catalog.MODEL_SERVICE_PARENT_SCHEMA_HEADER.capitalize()) + == "catalog.models" + ) + assert request.get_header("Anthropic-version") == "2023-06-01" + assert not request.has_header("Databricks-model-provider-service") + return io.BytesIO(json.dumps(pages[len(requests) - 1]).encode()) + + monkeypatch.setattr(http.urllib.request, "build_opener", lambda *_: CatalogOpener()) + result = catalog.fetch_anthropic_parent_catalog( + "https://workspace.invalid/", "test-bearer", "catalog.models" + ) + assert result.payloads == tuple(pages) + assert result.model_ids == (pages[0]["data"][0]["id"], pages[1]["data"][0]["id"]) + assert result.display_names == {result.model_ids[0]: "Sonnet", result.model_ids[1]: None} + assert http.urllib.parse.parse_qs(http.urllib.parse.urlsplit(requests[1].full_url).query) == { + "limit": ["1000"], + "after_id": ["cursor with / and ?"], + } + + +@pytest.mark.parametrize("failure_kind", ["duplicate", "cursor", "limit"]) +def test_anthropic_parent_catalog_rejects_duplicate_or_unbounded_pagination( + monkeypatch, failure_kind +): + requests = [] + + class CatalogOpener: + def open(self, request, *, timeout): + requests.append(request) + position = len(requests) + payload = { + "data": [ + {"id": "duplicate" if failure_kind == "duplicate" else f"model-{position}"} + ], + "has_more": True, + "last_id": "repeated" if failure_kind == "cursor" else f"cursor-{position}", + } + return io.BytesIO(json.dumps(payload).encode()) + + monkeypatch.setattr(http.urllib.request, "build_opener", lambda *_: CatalogOpener()) + with pytest.raises(AssertionError): + catalog.fetch_anthropic_parent_catalog( + "https://workspace.invalid", "test-bearer", "catalog.models" + ) + assert len(requests) == (20 if failure_kind == "limit" else 2) + + +def test_codex_parent_catalog_retains_unfiltered_payload_and_validates_duplicates(monkeypatch): + payload = { + "models": [ + {"slug": "catalog.models.gpt_luna", "display_name": "Luna", "visibility": "list"}, + {"slug": "catalog.other_models.codex_decoy", "visibility": "hidden"}, + ] + } + + class CatalogOpener: + def open(self, request, *, timeout): + assert timeout == 30 + assert ( + request.get_header(catalog.MODEL_SERVICE_PARENT_SCHEMA_HEADER.capitalize()) + == "catalog.models" + ) + assert not request.has_header("Databricks-model-provider-service") + return io.BytesIO(json.dumps(payload).encode()) + + monkeypatch.setattr(http.urllib.request, "build_opener", lambda *_: CatalogOpener()) + result = catalog.fetch_codex_parent_catalog( + "https://workspace.invalid", "test-bearer", "catalog.models" + ) + assert result.payloads == (payload,) + assert result.model_ids == ("catalog.models.gpt_luna",) + assert result.display_names == {"catalog.models.gpt_luna": "Luna"} + payload["models"].append(payload["models"][0]) + with pytest.raises(AssertionError, match="repeated"): + catalog.fetch_codex_parent_catalog( + "https://workspace.invalid", "test-bearer", "catalog.models" + ) + + +def test_provider_catalog_constructors_preserve_existing_arguments_and_isolate_defaults(): + anthropic = catalog.AnthropicProviderCatalog(("claude-a",), {"claude-a": None}) + codex = catalog.CodexProviderCatalog(("gpt-a",)) + other_codex = catalog.CodexProviderCatalog(("gpt-b",)) + assert anthropic.payloads == () + assert codex.payloads == () + assert codex.display_names == {} + codex.display_names["gpt-a"] = "Model A" + assert other_codex.display_names == {} + + +@pytest.mark.parametrize( + "fetcher", [catalog.fetch_anthropic_provider_catalog, catalog.fetch_anthropic_parent_catalog] +) +def test_anthropic_catalog_preserves_raw_pages_and_optional_labels(monkeypatch, fetcher): + payloads = [ + { + "data": [{"id": "b", "visibility": "hidden", "extra": [1, 2]}], + "has_more": True, + "last_id": "cursor", + "metadata": {"raw": True}, + }, + {"data": [{"id": "a", "display_name": "Model A"}], "has_more": False}, + ] + pages = iter(payloads) + _mock_json_boundary(monkeypatch, lambda *args: next(pages)) + result = fetcher("https://workspace", "token", "catalog.models") + assert result.payloads == tuple(payloads) + assert result.model_ids == ("b", "a") + assert result.display_names == {"b": None, "a": "Model A"} + + +@pytest.mark.parametrize( + "fetcher", [catalog.fetch_codex_provider_catalog, catalog.fetch_codex_parent_catalog] +) +def test_codex_catalog_preserves_hidden_entries_order_and_optional_labels(monkeypatch, fetcher): + payload = { + "models": [ + {"slug": "b", "visibility": "list"}, + {"slug": "b", "visibility": "hidden", "display_name": 7}, + {"visibility": "hidden", "extra": {"raw": True}}, + {"slug": "a", "visibility": "list", "display_name": "Model A"}, + ], + "metadata": {"raw": True}, + } + _mock_json_boundary(monkeypatch, lambda *args: payload) + result = fetcher("https://workspace", "token", "catalog.models") + assert result.payloads == (payload,) + assert result.model_ids == ("b", "a") + assert result.display_names == {"b": None, "a": "Model A"} + + +@pytest.mark.parametrize( + "fetcher", [catalog.fetch_codex_provider_catalog, catalog.fetch_codex_parent_catalog] +) +@pytest.mark.parametrize("display_name", ["", " ", 7, [], {}]) +def test_codex_catalog_rejects_invalid_optional_display_names(monkeypatch, fetcher, display_name): + payload = {"models": [{"slug": "a", "visibility": "list", "display_name": display_name}]} + _mock_json_boundary(monkeypatch, lambda *args: payload) + with pytest.raises(AssertionError, match="display name"): + fetcher("https://workspace", "token", "catalog.models") + + +@pytest.mark.parametrize( + "fetcher", + [ + catalog.fetch_anthropic_provider_catalog, + catalog.fetch_anthropic_parent_catalog, + catalog.fetch_codex_provider_catalog, + catalog.fetch_codex_parent_catalog, + ], +) +@pytest.mark.parametrize("scope", [None, "", " ", 7]) +def test_provider_catalog_requests_require_an_explicit_scope(monkeypatch, fetcher, scope): + def unexpected_open(*args, **kwargs): + pytest.fail("Invalid catalog scope must not reach the HTTP boundary") + + monkeypatch.setattr(http.urllib.request, "build_opener", unexpected_open) + with pytest.raises(AssertionError, match="explicit catalog scope"): + fetcher("https://workspace", "token", scope) From 47312ea2b7a65b69d4ba195a2f9dc20691fa650a Mon Sep 17 00:00:00 2001 From: Tien Le Date: Fri, 2 Oct 2026 19:49:52 +0000 Subject: [PATCH 2/6] AIGTWY-4878 Limit helper parent to existing transcript extraction --- tests/README.md | 8 +- tests/integration/README.md | 9 +- tests/integration/utils/agents/__init__.py | 11 +- tests/integration/utils/agents/claude.py | 45 +--- tests/integration/utils/agents/codex.py | 45 +--- tests/integration/utils/evidence.py | 52 +---- tests/integration/utils/http.py | 63 ------ tests/integration/utils/managed.py | 10 - tests/integration/utils/model_discovery.py | 60 ----- tests/integration/utils/provider_catalog.py | 124 ++++++---- tests/integration/utils/terminal.py | 25 -- tests/test_integration_agent_helpers.py | 104 +++------ tests/test_integration_evidence.py | 239 -------------------- tests/test_integration_http.py | 116 ---------- tests/test_integration_managed_config.py | 96 -------- tests/test_integration_model_discovery.py | 131 +---------- tests/test_integration_provider_catalog.py | 227 +------------------ 17 files changed, 134 insertions(+), 1231 deletions(-) delete mode 100644 tests/integration/utils/http.py delete mode 100644 tests/test_integration_http.py delete mode 100644 tests/test_integration_managed_config.py diff --git a/tests/README.md b/tests/README.md index c7f15d07e..66d24fa4f 100644 --- a/tests/README.md +++ b/tests/README.md @@ -139,11 +139,9 @@ Pi's token-command tests in `test_agent_pi.py` exercise Windows executable paths through POSIX parsing, including spaces, apostrophes, profile names, and PAT mode. They do not launch Pi or Git Bash on Windows. -Shared integration helpers have component coverage in `test_integration_agent_helpers.py`, -`test_integration_evidence.py`, `test_integration_model_discovery.py`, -`test_integration_provider_catalog.py`, `test_integration_http.py`, and -`test_integration_managed_config.py`. They check native transcript parsing, model attribution, -picker matching, scoped catalog requests, and persisted managed input; they are not live passes. +Shared native transcript parsing is checked by `test_integration_agent_helpers.py` and +`test_integration_evidence.py`. These component checks protect existing journey evidence; +they do not establish live agent coverage. ## CUJ coverage matrix diff --git a/tests/integration/README.md b/tests/integration/README.md index 7862ce050..8cdd60186 100644 --- a/tests/integration/README.md +++ b/tests/integration/README.md @@ -101,12 +101,9 @@ model-selection journey. ## Shared agent helpers -`utils/agents/claude.py` and `utils/agents/codex.py` own native transcript parsing and -discovery adapters. `utils/evidence.py` dispatches shared file-task evidence through those -adapters for existing journeys. Shared HTTP, catalog, picker, and terminal mechanics stay -in `utils/`; workspace-specific expectations belong in the consuming journey. -Claude model evidence is response-reported; Codex evidence is client-selected and joined -to the completed turn. Neither proves the gateway's backing destination. +`utils/agents/claude.py` and `utils/agents/codex.py` own the existing native answer +parsing, session paths, and child-session detection. `utils/evidence.py` dispatches +through them without changing existing journey commands or completion assertions. ## Run a specific combination diff --git a/tests/integration/utils/agents/__init__.py b/tests/integration/utils/agents/__init__.py index ae5f60cc1..1a952fc10 100644 --- a/tests/integration/utils/agents/__init__.py +++ b/tests/integration/utils/agents/__init__.py @@ -1,10 +1 @@ -"""Shared validation for agent-specific integration evidence.""" - -import re - - -def _evidence_id(value: object, field: str) -> str: - assert isinstance(value, str) and re.fullmatch(r"\S+", value), ( - f"Missing or malformed completed-task {field}" - ) - return value +"""Native agent transcript helpers used by the shared evidence layer.""" diff --git a/tests/integration/utils/agents/claude.py b/tests/integration/utils/agents/claude.py index 2bcff696e..2c79f661d 100644 --- a/tests/integration/utils/agents/claude.py +++ b/tests/integration/utils/agents/claude.py @@ -1,23 +1,6 @@ -"""Claude-specific integration evidence and model discovery helpers.""" - -from ..model_discovery import claude_discovery_model_id as discovery_model_id -from ..model_discovery import claude_model_in_picker as model_in_picker -from ..model_discovery import claude_model_service_id as model_service_id -from ..provider_catalog import fetch_anthropic_parent_catalog as fetch_parent_catalog -from . import _evidence_id - -__all__ = [ - "EVIDENCE_KIND", - "assistant_answers", - "completed_task_models", - "discovery_model_id", - "fetch_parent_catalog", - "model_in_picker", - "model_service_id", -] +"""Claude transcript helpers for integration journeys.""" SESSION_DIRECTORY = ".claude/projects" -EVIDENCE_KIND = "response-reported model" def is_child_session(path: str, records: list[dict]) -> bool: @@ -37,29 +20,3 @@ def assistant_answers(records: list[dict]) -> list[str]: if part.get("type") == "text" and isinstance(part.get("text"), str) ) return answers - - -def completed_task_models(records: list[dict], answer_value: str) -> set[str]: - """Parse Claude's response-reported model on the assistant's matching answer.""" - assert isinstance(answer_value, str) and answer_value.strip(), ( - "Expected a nonempty answer value" - ) - found: set[str] = set() - for record in records: - if record.get("type") != "assistant": - continue - message = record.get("message") - assert isinstance(message, dict), "Malformed Claude assistant message" - assert message.get("role") == "assistant", "Missing Claude assistant role" - content = message.get("content") - assert isinstance(content, list), "Malformed Claude assistant content" - for part in content: - assert isinstance(part, dict), "Malformed Claude assistant content block" - if part.get("type") != "text": - continue - text = part.get("text") - assert isinstance(text, str), "Malformed Claude assistant text" - if answer_value in text: - model = _evidence_id(message.get("model"), "model") - found.add(_evidence_id(model.removesuffix("[1m]"), "model")) - return found diff --git a/tests/integration/utils/agents/codex.py b/tests/integration/utils/agents/codex.py index 99e505420..fea968ce8 100644 --- a/tests/integration/utils/agents/codex.py +++ b/tests/integration/utils/agents/codex.py @@ -1,19 +1,6 @@ -"""Codex-specific integration evidence and model discovery helpers.""" - -from ..model_discovery import codex_model_in_picker as model_in_picker -from ..provider_catalog import fetch_codex_parent_catalog as fetch_parent_catalog -from . import _evidence_id - -__all__ = [ - "EVIDENCE_KIND", - "assistant_answers", - "completed_task_models", - "fetch_parent_catalog", - "model_in_picker", -] +"""Codex transcript helpers for integration journeys.""" SESSION_DIRECTORY = ".codex/sessions" -EVIDENCE_KIND = "client-selected model" def is_child_session(path: str, records: list[dict]) -> bool: @@ -34,33 +21,3 @@ def assistant_answers(records: list[dict]) -> list[str]: if payload.get("type") == "task_complete" and payload.get("last_agent_message"): answers.append(payload["last_agent_message"]) return answers - - -def completed_task_models(records: list[dict], answer_value: str) -> set[str]: - """Join Codex's completed answer turn to its client-selected context model.""" - assert isinstance(answer_value, str) and answer_value.strip(), ( - "Expected a nonempty answer value" - ) - completed: set[str] = set() - for record in records: - if record.get("type") != "event_msg": - continue - payload = record.get("payload") - assert isinstance(payload, dict), "Malformed Codex event payload" - if payload.get("type") != "task_complete": - continue - answer = payload.get("last_agent_message") - assert isinstance(answer, str), "Missing or malformed Codex completed answer" - if answer_value in answer: - completed.add(_evidence_id(payload.get("turn_id"), "turn ID")) - models: dict[str, set[str]] = {turn_id: set() for turn_id in completed} - for record in records: - if record.get("type") != "turn_context": - continue - payload = record.get("payload") - assert isinstance(payload, dict), "Malformed Codex turn context" - turn_id = _evidence_id(payload.get("turn_id"), "turn ID") - if turn_id in completed: - models[turn_id].add(_evidence_id(payload.get("model"), "model")) - assert all(models.values()), "Missing Codex model context for completed answer turn" - return {model for turn_models in models.values() for model in turn_models} diff --git a/tests/integration/utils/evidence.py b/tests/integration/utils/evidence.py index 06510f820..b5dff142a 100644 --- a/tests/integration/utils/evidence.py +++ b/tests/integration/utils/evidence.py @@ -5,16 +5,9 @@ import uuid from pathlib import Path -from .agents import _evidence_id, claude, codex +from .agents import claude, codex -_EVIDENCE_ADAPTERS = {"claude": claude, "codex": codex} - - -def _evidence_adapter(agent: str): - try: - return _EVIDENCE_ADAPTERS[agent] - except (KeyError, TypeError): - raise AssertionError(f"Unsupported evidence agent: {agent}") from None +_AGENT_HELPERS = {"claude": claude, "codex": codex} def assert_no_terminal_api_error(screen: str) -> None: @@ -48,52 +41,19 @@ def read_jsonl(path: Path) -> list[dict]: def agent_sessions(session, agent: str) -> dict[str, list[dict]]: - directory = session.home / _evidence_adapter(agent).SESSION_DIRECTORY + directory = session.home / _AGENT_HELPERS.get(agent, codex).SESSION_DIRECTORY return { str(path.relative_to(directory)): read_jsonl(path) for path in directory.rglob("*.jsonl") } def assistant_answers(agent: str, records: list[dict]) -> list[str]: - return _evidence_adapter(agent).assistant_answers(records) + helper = _AGENT_HELPERS.get(agent) + return helper.assistant_answers(records) if helper is not None else [] def is_child_session(agent: str, path: str, records: list[dict]) -> bool: - return _evidence_adapter(agent).is_child_session(path, records) - - -claude_completed_task_models = claude.completed_task_models -codex_completed_task_models = codex.completed_task_models - - -def completed_task_models(session, agent: str, answer_value: str) -> set[str]: - """Read parent task model evidence, not proof of the gateway's destination.""" - adapter = _evidence_adapter(agent) - assert isinstance(answer_value, str) and answer_value.strip(), ( - "Expected a nonempty answer value" - ) - return { - model - for path, records in agent_sessions(session, agent).items() - if not is_child_session(agent, path, records) - for model in adapter.completed_task_models(records, answer_value) - } - - -def assert_completed_task_model(session, agent: str, answer_value: str, expected: str) -> None: - expected = _evidence_id(expected, "expected model") - adapter = _evidence_adapter(agent) - observed = completed_task_models(session, agent, answer_value) - session.record( - f"completed-task-model-{agent}-{answer_value[:12]}.json", - { - "expected": expected, - "observed": sorted(observed), - "evidence_kind": adapter.EVIDENCE_KIND, - "gateway_destination_proven": False, - }, - ) - assert observed == {expected}, {"expected": expected, "observed": sorted(observed)} + return _AGENT_HELPERS.get(agent, codex).is_child_session(path, records) def assistant_answer_contains(session, agent: str, value: str, *, child: bool = False) -> bool: diff --git a/tests/integration/utils/http.py b/tests/integration/utils/http.py deleted file mode 100644 index 1c35cff5e..000000000 --- a/tests/integration/utils/http.py +++ /dev/null @@ -1,63 +0,0 @@ -"""Read-only workspace JSON transport with strict origins and no redirects.""" - -from __future__ import annotations - -import json -import re -import urllib.error -import urllib.parse -import urllib.request - - -class NoRedirect(urllib.request.HTTPRedirectHandler): - def redirect_request(self, req, fp, code, msg, headers, newurl): - return None - - -def workspace_origin(workspace: str, token: str) -> str: - """Validate an explicit HTTPS origin and bearer before constructing a client.""" - assert isinstance(workspace, str), "Expected an HTTPS workspace origin" - try: - origin = urllib.parse.urlsplit(workspace) - port = origin.port - except ValueError: - raise AssertionError("Expected a valid HTTPS workspace origin") from None - assert ( - origin.scheme == "https" - and origin.hostname - and origin.username is None - and origin.password is None - and origin.path in {"", "/"} - and not origin.query - and not origin.fragment - and (port is None or 0 < port <= 65535) - and not re.search(r"[\s\\]", workspace) - ), "Expected an HTTPS workspace origin without credentials" - assert isinstance(token, str) and token.strip() and not re.search(r"[\r\n]", token), ( - "Expected an explicit workspace bearer" - ) - return f"https://{origin.netloc}" - - -def safe_https_json_get( - workspace: str, token: str, path: str, *, headers: dict[str, str] | None = None -) -> object: - """GET JSON from an explicit workspace origin without redirects or secret diagnostics.""" - origin = workspace_origin(workspace, token) - assert path.startswith("/") and not path.startswith("//"), "Expected a workspace API path" - request = urllib.request.Request( - f"{origin}{path}", - headers={ - **(headers or {}), - "Authorization": f"Bearer {token}", - "Accept": "application/json", - }, - method="GET", - ) - try: - with urllib.request.build_opener(NoRedirect()).open(request, timeout=30) as response: - return json.load(response) - except urllib.error.HTTPError as error: - raise AssertionError(f"Workspace JSON GET returned HTTP {error.code}") from None - except (urllib.error.URLError, OSError, ValueError): - raise AssertionError("Workspace JSON GET failed or returned invalid JSON") from None diff --git a/tests/integration/utils/managed.py b/tests/integration/utils/managed.py index 997d96948..3599d38f9 100644 --- a/tests/integration/utils/managed.py +++ b/tests/integration/utils/managed.py @@ -13,16 +13,6 @@ MANAGED_CONFIGS_PATH = "/api/ai-gateway/v2/coding-agent-configs" -def read_persisted_managed_config(session, workspace: str) -> dict: - """Read the configure result without imposing a journey-specific policy.""" - path = session.home / ".ucode" / "managed-config.json" - assert path.is_file(), path - persisted = json.loads(path.read_text()) - assert isinstance(persisted, dict) and persisted.get("workspace") == workspace, persisted - assert isinstance(persisted.get("config"), dict), persisted - return persisted - - def assert_no_managed_config(payload: object) -> None: """Validate the real List response before claiming unmanaged-workspace coverage.""" configs = payload.get("coding_agent_configs", []) if isinstance(payload, dict) else payload diff --git a/tests/integration/utils/model_discovery.py b/tests/integration/utils/model_discovery.py index bfb338e94..36171de96 100644 --- a/tests/integration/utils/model_discovery.py +++ b/tests/integration/utils/model_discovery.py @@ -1,6 +1,5 @@ """Validation helpers for model-discovery evidence.""" -import hashlib import re _CLAUDE_GATEWAY_ALIAS = re.compile(r"^anthropic-aigw-[0-9a-fA-F]{8}-(?P.*)$") @@ -13,25 +12,6 @@ _SYSTEM_AI_PREFIX = "system.ai." -def claude_discovery_model_id(model_id: str) -> str: - """Encode the exact wire ID expected by Claude's gateway discovery feature.""" - if "claude" in model_id.lower() or "anthropic" in model_id.lower(): - return model_id - checksum = hashlib.sha256(model_id.encode("utf-8")).hexdigest()[:8] - return f"anthropic-aigw-{checksum}-{model_id}" - - -def claude_model_service_id(model_id: str) -> str: - """Recover service identity only from the gateway's checksum-valid discovery alias.""" - alias = re.fullmatch(r"anthropic-aigw-([0-9a-f]{8})-(.+)", model_id) - if alias is None: - return model_id - checksum, original = alias.groups() - if hashlib.sha256(original.encode("utf-8")).hexdigest()[:8] != checksum: - return model_id - return original - - def claude_model_in_picker(screen: str, model_id: str, display_name: str | None) -> bool: """Return whether a Claude model appears in a numbered picker row. @@ -90,43 +70,3 @@ def claude_system_model_ids(models: list[dict]) -> list[str]: ), f"Expected a system.ai model id: {model_id!r}" canonical_ids.append(canonical_id) return canonical_ids - - -def codex_model_in_picker(screen: str, model_id: str, display_name: str) -> bool: - """Match native numbered picker rows, not banners, prompt text, or footers.""" - rows = re.findall(r"(?m)^[ \t]*(?:[❯›>][ \t]*)?\d+[.)][ \t]+([^\n]+)$", screen) - return any(_picker_row_matches(row, model_id, display_name) for row in rows) - - -def _picker_row_matches(label: str, model_id: str, display_name: str | None) -> bool: - return any( - re.search(rf"(? None: - """Require exactly one unambiguous numbered native row per expected model.""" - assert agent in {"claude", "codex"}, agent - assert display_names, "Expected a nonempty provider model inventory" - rows = re.findall(r"(?m)^[ \t]*(?:[❯›>][ \t]*)?\d+[.)](?:[ \t]+([^\n]*))?$", screen) - assert rows, f"No numbered {agent} picker rows:\n{screen}" - observed: list[str] = [] - for label in rows: - matched = { - model - for model, display_name in display_names.items() - if _picker_row_matches(label, model, display_name) - } - assert len(matched) == 1, ( - f"Unmatched or ambiguous {agent} picker row {label!r}: {sorted(matched)}" - ) - model = matched.pop() - assert model not in observed, f"Duplicate {agent} picker model row: {model}" - observed.append(model) - assert set(observed) == set(display_names), { - "agent": agent, - "expected": sorted(display_names), - "observed": observed, - } diff --git a/tests/integration/utils/provider_catalog.py b/tests/integration/utils/provider_catalog.py index f87a9b95e..d0a5c00fc 100644 --- a/tests/integration/utils/provider_catalog.py +++ b/tests/integration/utils/provider_catalog.py @@ -2,16 +2,17 @@ from __future__ import annotations +import json +import urllib.error import urllib.parse -from dataclasses import dataclass, field +import urllib.request +from dataclasses import dataclass from typing import NoReturn -from .http import safe_https_json_get - _ANTHROPIC_MODELS_PATH = "/ai-gateway/anthropic/v1/models" _CODEX_MODELS_PATH = "/ai-gateway/codex/v1/models" _PROVIDER_HEADER = "Databricks-Model-Provider-Service" -MODEL_SERVICE_PARENT_SCHEMA_HEADER = "Databricks-Model-Service-Parent-Schema" +_PARENT_SCHEMA_HEADER = "Databricks-Model-Service-Parent-Schema" _ANTHROPIC_VERSION = "2023-06-01" _ANTHROPIC_PAGE_SIZE = 1000 _ANTHROPIC_MAX_PAGES = 20 @@ -32,7 +33,6 @@ class AnthropicProviderCatalog: model_ids: tuple[str, ...] display_names: dict[str, str | None] - payloads: tuple[dict, ...] = () @dataclass(frozen=True) @@ -40,8 +40,6 @@ class CodexProviderCatalog: """The list-visible model slugs from a Codex provider catalog.""" model_ids: tuple[str, ...] - payloads: tuple[dict, ...] = () - display_names: dict[str, str | None] = field(default_factory=dict) def _fail(message: str) -> NoReturn: @@ -112,6 +110,8 @@ def parse_codex_provider_catalog(payload: object) -> tuple[str, ...]: if not isinstance(visibility, str): _fail(f"Codex provider model {index} had an invalid visibility") if visibility != "list": + # Hidden/native aliases are part of the provider response but are not exposed by + # Codex's model/list request used as the observable catalog below. continue slug = raw_model.get("slug") if not isinstance(slug, str) or not slug.strip(): @@ -126,48 +126,78 @@ def parse_codex_provider_catalog(payload: object) -> tuple[str, ...]: return tuple(model_ids) -def _validate_scope(scope: str) -> None: - if not isinstance(scope, str) or not scope.strip(): - _fail("provider catalog request requires an explicit catalog scope") +def _get_json(url: str, headers: dict[str, str]) -> object: + request = urllib.request.Request(url, headers=headers) + try: + with urllib.request.urlopen(request, timeout=30) as response: + getcode = getattr(response, "getcode", None) + status = getcode() if callable(getcode) else getattr(response, "status", None) + if status is not None and not 200 <= status < 300: + _fail(f"GET {url} returned HTTP {status}") + raw_body = response.read() + except urllib.error.HTTPError as error: + raise AssertionError(f"GET {url} returned HTTP {error.code} {error.reason}") from error + except (urllib.error.URLError, OSError) as error: + raise AssertionError(f"GET {url} failed: {error}") from error + + if isinstance(raw_body, bytes): + try: + raw_body = raw_body.decode("utf-8") + except UnicodeDecodeError as error: + raise AssertionError(f"GET {url} returned non-UTF-8 JSON") from error + if not isinstance(raw_body, str): + _fail(f"GET {url} returned a non-text response body") + try: + return json.loads(raw_body) + except json.JSONDecodeError as error: + raise AssertionError(f"GET {url} returned invalid JSON: {error.msg}") from error + + +def _validate_request_inputs(workspace: str, token: str, provider_service: str) -> str: + if not isinstance(workspace, str) or not workspace.rstrip("/"): + _fail("provider catalog request requires a workspace URL") + if not isinstance(token, str) or not token: + _fail("provider catalog request requires an explicit bearer token") + if not isinstance(provider_service, str) or not provider_service.strip(): + _fail("provider catalog request requires a provider service") + return workspace.rstrip("/") def fetch_anthropic_provider_catalog( workspace: str, token: str, provider_service: str ) -> AnthropicProviderCatalog: """Fetch and validate the complete Anthropic provider catalog with bounded pagination.""" - _validate_scope(provider_service) - return _fetch_anthropic_catalog(workspace, token, {_PROVIDER_HEADER: provider_service}) + return _fetch_anthropic_catalog(workspace, token, provider_service, _PROVIDER_HEADER) def fetch_anthropic_parent_catalog( workspace: str, token: str, parent_schema: str ) -> AnthropicProviderCatalog: - """Fetch the complete Anthropic catalog scoped to a Unity Catalog parent schema.""" - _validate_scope(parent_schema) - return _fetch_anthropic_catalog( - workspace, token, {MODEL_SERVICE_PARENT_SCHEMA_HEADER: parent_schema} - ) + """Fetch the Anthropic-compatible models advertised for a Unity Catalog parent schema.""" + return _fetch_anthropic_catalog(workspace, token, parent_schema, _PARENT_SCHEMA_HEADER) def _fetch_anthropic_catalog( - workspace: str, token: str, scope_headers: dict[str, str] + workspace: str, token: str, source: str, source_header: str ) -> AnthropicProviderCatalog: - headers = {"Anthropic-Version": _ANTHROPIC_VERSION, **scope_headers} + base_url = _validate_request_inputs(workspace, token, source) + headers = { + "Authorization": f"Bearer {token}", + "Accept": "application/json", + "Anthropic-Version": _ANTHROPIC_VERSION, + source_header: source, + } model_ids: list[str] = [] display_names: dict[str, str | None] = {} - payloads: list[dict] = [] cursor: str | None = None cursors: set[str] = set() for _page_number in range(_ANTHROPIC_MAX_PAGES): query = {"limit": str(_ANTHROPIC_PAGE_SIZE)} if cursor is not None: query["after_id"] = cursor - path = f"{_ANTHROPIC_MODELS_PATH}?{urllib.parse.urlencode(query)}" - payload = safe_https_json_get(workspace, token, path, headers=headers) - page = parse_anthropic_provider_page(payload) - assert isinstance(payload, dict), "Expected an Anthropic catalog object" - payloads.append(payload) + url = f"{base_url}{_ANTHROPIC_MODELS_PATH}?{urllib.parse.urlencode(query)}" + page = parse_anthropic_provider_page(_get_json(url, headers)) for model_id, display_name in page.models: if model_id in display_names: _fail(f"Anthropic provider catalog repeated model id {model_id!r}") @@ -175,8 +205,8 @@ def _fetch_anthropic_catalog( display_names[model_id] = display_name if not page.has_more: - return AnthropicProviderCatalog(tuple(model_ids), display_names, tuple(payloads)) - assert page.last_id is not None + return AnthropicProviderCatalog(tuple(model_ids), display_names) + assert page.last_id is not None # checked by parse_anthropic_provider_page if page.last_id in cursors: _fail(f"Anthropic provider catalog repeated pagination cursor {page.last_id!r}") cursors.add(page.last_id) @@ -189,31 +219,29 @@ def fetch_codex_provider_catalog( workspace: str, token: str, provider_service: str ) -> CodexProviderCatalog: """Fetch and validate the list-visible Codex provider catalog.""" - _validate_scope(provider_service) - return _fetch_codex_catalog(workspace, token, {_PROVIDER_HEADER: provider_service}) - - -def _fetch_codex_catalog( - workspace: str, token: str, scope_headers: dict[str, str] -) -> CodexProviderCatalog: - payload = safe_https_json_get(workspace, token, _CODEX_MODELS_PATH, headers=scope_headers) - model_ids = parse_codex_provider_catalog(payload) - assert isinstance(payload, dict), "Expected a Codex catalog object" - display_names: dict[str, str | None] = {} - for entry in payload["models"]: - if entry.get("visibility") == "list" and entry.get("slug") in model_ids: - label = entry.get("display_name") - if label is not None and (not isinstance(label, str) or not label.strip()): - _fail("Invalid Codex catalog display name") - display_names[entry["slug"]] = label - return CodexProviderCatalog(model_ids, (payload,), display_names) + base_url = _validate_request_inputs(workspace, token, provider_service) + payload = _get_json( + f"{base_url}{_CODEX_MODELS_PATH}", + { + "Authorization": f"Bearer {token}", + "Accept": "application/json", + _PROVIDER_HEADER: provider_service, + }, + ) + return CodexProviderCatalog(parse_codex_provider_catalog(payload)) def fetch_codex_parent_catalog( workspace: str, token: str, parent_schema: str ) -> CodexProviderCatalog: """Fetch the API-compatible models advertised for a Unity Catalog parent schema.""" - _validate_scope(parent_schema) - return _fetch_codex_catalog( - workspace, token, {MODEL_SERVICE_PARENT_SCHEMA_HEADER: parent_schema} + base_url = _validate_request_inputs(workspace, token, parent_schema) + payload = _get_json( + f"{base_url}{_CODEX_MODELS_PATH}", + { + "Authorization": f"Bearer {token}", + "Accept": "application/json", + _PARENT_SCHEMA_HEADER: parent_schema, + }, ) + return CodexProviderCatalog(parse_codex_provider_catalog(payload)) diff --git a/tests/integration/utils/terminal.py b/tests/integration/utils/terminal.py index b0175037e..9f4c4fda7 100644 --- a/tests/integration/utils/terminal.py +++ b/tests/integration/utils/terminal.py @@ -318,31 +318,6 @@ def open_model_picker(self, *, model_visible=None): ) return screen - def open_codex_model_picker(self, *, model_visible): - """Capture Codex's native numbered /model menu, then dismiss it with Escape.""" - assert self.agent == "codex", self.agent - self.submit("/model") - self.wait_for( - lambda text: ( - "select model" in text.lower() - and re.search(r"(?m)^[ \t]*(?:[❯›>][ \t]*)?\d+[.)][ \t]+\S", text) - ), - "Codex's numbered model picker", - timeout=60, - ) - self.wait_for(model_visible, "all scoped models in Codex's picker", timeout=60) - screen = self.visible - self.actions.append({"reason": "codex-model-picker-visible", "screen": screen}) - self.send("\x1b", "close Codex's model picker without changing its default") - self.wait_for( - lambda text: ( - "select model" not in text.lower() - and re.search(r"(?m)^\s*[❯›>]\s*(?!\d+[.)])", text) - ), - "the Codex prompt after closing the model picker", - ) - return screen - def wait_for_task(self, task, timeout=180): permission_in_progress = False diff --git a/tests/test_integration_agent_helpers.py b/tests/test_integration_agent_helpers.py index 28dabd3ac..40e9626a8 100644 --- a/tests/test_integration_agent_helpers.py +++ b/tests/test_integration_agent_helpers.py @@ -1,86 +1,54 @@ -"""Component checks for the reusable integration agent helper contracts.""" +"""Regression checks for extracted native transcript parsing.""" +import json from types import SimpleNamespace import pytest -from tests.integration.utils import evidence, model_discovery, provider_catalog +from tests.integration.utils import evidence from tests.integration.utils.agents import claude, codex -def test_evidence_keeps_legacy_completed_task_names_as_adapter_aliases(): - assert evidence.claude_completed_task_models is claude.completed_task_models - assert evidence.codex_completed_task_models is codex.completed_task_models - - -def test_agent_modules_expose_direct_model_discovery_helpers(): - assert claude.fetch_parent_catalog is provider_catalog.fetch_anthropic_parent_catalog - assert claude.discovery_model_id is model_discovery.claude_discovery_model_id - assert claude.model_service_id is model_discovery.claude_model_service_id - assert claude.model_in_picker is model_discovery.claude_model_in_picker - assert codex.fetch_parent_catalog is provider_catalog.fetch_codex_parent_catalog - assert codex.model_in_picker is model_discovery.codex_model_in_picker - - -def test_agent_modules_preserve_exact_completion_attribution(): - claude_records = [ - { - "type": "user", - "message": {"model": "catalog.other_models.decoy", "content": []}, - }, +@pytest.mark.parametrize("agent,helper", [("claude", claude), ("codex", codex)]) +def test_agent_helpers_preserve_native_answers(agent, helper): + records = [ + {"type": "user", "message": {"content": [{"type": "text", "text": "prompt"}]}}, { "type": "assistant", "message": { "role": "assistant", - "model": "catalog.models.claude_sonnet[1m]", - "content": [{"type": "text", "text": "value"}], + "content": [{"type": "text", "text": "claude-answer"}, {"type": "tool_use"}], }, }, - ] - codex_records = [ - { - "type": "turn_context", - "payload": {"turn_id": "other", "model": "catalog.other_models.decoy"}, - }, - { - "type": "turn_context", - "payload": {"turn_id": "matching", "model": "catalog.models.gpt_luna"}, - }, + {"type": "event_msg", "payload": {"type": "task_started"}}, { "type": "event_msg", - "payload": { - "type": "task_complete", - "turn_id": "matching", - "last_agent_message": "value", - }, + "payload": {"type": "task_complete", "last_agent_message": "codex-answer"}, }, ] - - assert claude.completed_task_models(claude_records, "value") == {"catalog.models.claude_sonnet"} - assert codex.completed_task_models(codex_records, "value") == {"catalog.models.gpt_luna"} - - -def test_evidence_dispatches_assistant_answers_to_agent_modules(): - claude_records = [ - { - "type": "assistant", - "message": {"role": "assistant", "content": [{"type": "text", "text": "answer"}]}, - } - ] - codex_records = [ - { - "type": "event_msg", - "payload": {"type": "task_complete", "last_agent_message": "answer"}, - } - ] - - assert evidence.assistant_answers("claude", claude_records) == ["answer"] - assert evidence.assistant_answers("codex", codex_records) == ["answer"] - - -def test_unsupported_evidence_agents_fail_without_codex_fallback(): - session = SimpleNamespace(home=None, record=lambda *_: None) - with pytest.raises(AssertionError, match="Unsupported evidence agent: opencode"): - evidence.completed_task_models(session, "opencode", "value") - with pytest.raises(AssertionError, match="Unsupported evidence agent: opencode"): - evidence.assert_completed_task_model(session, "opencode", "value", "model") + expected = [f"{agent}-answer"] + assert helper.assistant_answers(records) == expected + assert evidence.assistant_answers(agent, records) == expected + + +@pytest.mark.parametrize("agent,helper", [("claude", claude), ("codex", codex)]) +def test_agent_helpers_preserve_session_paths_and_child_detection(tmp_path, agent, helper): + expected_directory = {"claude": ".claude/projects", "codex": ".codex/sessions"}[agent] + assert helper.SESSION_DIRECTORY == expected_directory + directory = tmp_path / expected_directory + directory.mkdir(parents=True) + parent = [{"type": "session_meta", "payload": {"source": "cli"}}] + child = [{"type": "session_meta", "payload": {"source": {"subagent": "spawn"}}}] + child_path = directory / "project/subagents/child.jsonl" + child_path.parent.mkdir(parents=True) + (directory / "parent.jsonl").write_text(json.dumps(parent[0]) + "\n") + child_path.write_text(json.dumps(child[0]) + "\n") + session = SimpleNamespace(home=tmp_path) + assert evidence.agent_sessions(session, agent) == { + "parent.jsonl": parent, + "project/subagents/child.jsonl": child, + } + assert not helper.is_child_session("parent.jsonl", parent) + assert not evidence.is_child_session(agent, "parent.jsonl", parent) + assert helper.is_child_session("project/subagents/child.jsonl", child) + assert evidence.is_child_session(agent, "project/subagents/child.jsonl", child) diff --git a/tests/test_integration_evidence.py b/tests/test_integration_evidence.py index 4b6b31e1f..f618304d1 100644 --- a/tests/test_integration_evidence.py +++ b/tests/test_integration_evidence.py @@ -1,11 +1,9 @@ """Unit checks for interpreting terminal evidence, not live agent substitutes.""" import json -from types import SimpleNamespace import pytest -from tests.integration.utils import evidence from tests.integration.utils.evidence import ( SubagentCalculation, assert_no_terminal_api_error, @@ -91,240 +89,3 @@ def test_tagged_calculation_requires_the_native_child_answer(tmp_path, agent): assert task.marker in task.prompt assert f'task name "{task.marker}"' in task.prompt assert "1+1" in task.prompt - - -def test_codex_model_identity_uses_only_the_completed_answer_turn(monkeypatch): - records = [ - { - "type": "turn_context", - "payload": {"turn_id": "other", "model": "catalog.other_models.codex_decoy"}, - }, - { - "type": "turn_context", - "payload": {"turn_id": "matching", "model": "catalog.models.gpt_luna"}, - }, - { - "type": "event_msg", - "payload": { - "type": "task_complete", - "turn_id": "matching", - "last_agent_message": "withheld-file-value", - }, - }, - ] - monkeypatch.setattr(evidence, "agent_sessions", lambda *_: {"transcript": records}) - assert evidence.completed_task_models(None, "codex", "withheld-file-value") == { - "catalog.models.gpt_luna" - } - - -def test_codex_model_identity_rejects_prompt_only_evidence(monkeypatch): - records = [ - { - "type": "turn_context", - "payload": {"turn_id": "matching", "model": "catalog.models.gpt_luna"}, - }, - { - "type": "response_item", - "payload": {"type": "message", "role": "user", "content": "withheld-file-value"}, - }, - ] - monkeypatch.setattr(evidence, "agent_sessions", lambda *_: {"transcript": records}) - assert evidence.completed_task_models(None, "codex", "withheld-file-value") == set() - - -def test_claude_model_identity_uses_assistant_answer_not_tool_output(monkeypatch): - records = [ - { - "type": "user", - "message": { - "model": "catalog.other_models.claude_decoy", - "content": [{"type": "text", "text": "value"}], - }, - }, - { - "type": "assistant", - "message": { - "role": "assistant", - "model": "catalog.models.claude_sonnet", - "content": [{"type": "text", "text": "value"}], - }, - }, - ] - monkeypatch.setattr(evidence, "agent_sessions", lambda *_: {"transcript": records}) - assert evidence.completed_task_models(None, "claude", "value") == { - "catalog.models.claude_sonnet" - } - - -def completed_records(agent, model, answer="value", turn_id="matching"): - if agent == "claude": - return [ - { - "type": "assistant", - "message": { - "role": "assistant", - "model": model, - "content": [{"type": "text", "text": answer}], - }, - } - ] - return [ - {"type": "turn_context", "payload": {"turn_id": turn_id, "model": model}}, - { - "type": "event_msg", - "payload": {"type": "task_complete", "turn_id": turn_id, "last_agent_message": answer}, - }, - ] - - -@pytest.mark.parametrize("agent", ["claude", "codex"]) -@pytest.mark.parametrize("parent_answer", [None, "value", "unrelated-answer"]) -def test_completed_task_models_excludes_child_only_and_conflicting_child_evidence( - monkeypatch, agent, parent_answer -): - child = completed_records(agent, "child-model") - if agent == "codex": - child.insert(0, {"type": "session_meta", "payload": {"source": {"subagent": "spawn"}}}) - sessions = {"project/subagents/child.jsonl": child} - if parent_answer is not None: - sessions["project/parent.jsonl"] = completed_records(agent, "parent-model", parent_answer) - monkeypatch.setattr(evidence, "agent_sessions", lambda *_: sessions) - expected = {"parent-model"} if parent_answer == "value" else set() - assert evidence.completed_task_models(None, agent, "value") == expected - - -@pytest.mark.parametrize("agent", ["claude", "codex"]) -@pytest.mark.parametrize("model", [None, "", " ", 5, {}, [], "model with whitespace"]) -def test_completed_task_models_fails_closed_on_malformed_model(monkeypatch, agent, model): - records = completed_records(agent, model) - monkeypatch.setattr(evidence, "agent_sessions", lambda *_: {"parent": records}) - with pytest.raises(AssertionError, match="model"): - evidence.completed_task_models(None, agent, "value") - - -@pytest.mark.parametrize("agent", ["claude", "codex"]) -def test_completed_task_models_fails_closed_on_missing_model(monkeypatch, agent): - records = completed_records(agent, "model") - del records[0]["message" if agent == "claude" else "payload"]["model"] - monkeypatch.setattr(evidence, "agent_sessions", lambda *_: {"parent": records}) - with pytest.raises(AssertionError, match="model"): - evidence.completed_task_models(None, agent, "value") - - -@pytest.mark.parametrize("turn_id", [None, "", " ", 5, {}, [], "turn with whitespace"]) -@pytest.mark.parametrize("record_index", [0, 1]) -def test_codex_completed_task_models_fails_closed_on_malformed_turn_ids(turn_id, record_index): - records = completed_records("codex", "model") - records[record_index]["payload"]["turn_id"] = turn_id - with pytest.raises(AssertionError, match="turn ID"): - evidence.codex_completed_task_models(records, "value") - - -@pytest.mark.parametrize("record_index", [0, 1]) -def test_codex_completed_task_models_fails_closed_on_missing_turn_ids(record_index): - records = completed_records("codex", "model") - del records[record_index]["payload"]["turn_id"] - with pytest.raises(AssertionError, match="turn ID"): - evidence.codex_completed_task_models(records, "value") - - -def test_codex_completed_task_models_requires_context_in_same_session(monkeypatch): - context, completion = completed_records("codex", "model") - monkeypatch.setattr( - evidence, - "agent_sessions", - lambda *_: {"context-session": [context], "answer-session": [completion]}, - ) - with pytest.raises(AssertionError, match="Missing Codex model context"): - evidence.completed_task_models(None, "codex", "value") - - -@pytest.mark.parametrize("agent", ["claude", "codex"]) -@pytest.mark.parametrize("models", [[], ["unexpected"], ["expected", "conflicting"]]) -def test_completed_task_model_assertion_requires_exact_singleton(monkeypatch, agent, models): - records = [record for model in models for record in completed_records(agent, model)] - monkeypatch.setattr(evidence, "agent_sessions", lambda *_: {"parent": records}) - artifacts = {} - session = SimpleNamespace(record=lambda name, payload: artifacts.update({name: payload})) - with pytest.raises(AssertionError): - evidence.assert_completed_task_model(session, agent, "value", "expected") - assert next(iter(artifacts.values()))["observed"] == sorted(models) - - -@pytest.mark.parametrize("agent", ["claude", "codex"]) -def test_completed_task_model_records_evidence_limits(monkeypatch, agent): - monkeypatch.setattr( - evidence, "agent_sessions", lambda *_: {"parent": completed_records(agent, "expected")} - ) - artifacts = {} - session = SimpleNamespace(record=lambda name, payload: artifacts.update({name: payload})) - evidence.assert_completed_task_model(session, agent, "value", "expected") - assert artifacts == { - f"completed-task-model-{agent}-value.json": { - "expected": "expected", - "observed": ["expected"], - "evidence_kind": ( - "response-reported model" if agent == "claude" else "client-selected model" - ), - "gateway_destination_proven": False, - } - } - - -@pytest.mark.parametrize("agent", ["claude", "codex"]) -@pytest.mark.parametrize("answer_value", [None, "", " \t", 1, [], {}]) -def test_completed_task_models_rejects_empty_or_malformed_expected_answers( - monkeypatch, agent, answer_value -): - def unexpected_read(*args): - pytest.fail("Invalid answer must not read session evidence") - - monkeypatch.setattr(evidence, "agent_sessions", unexpected_read) - with pytest.raises(AssertionError, match="nonempty answer value"): - evidence.completed_task_models(None, agent, answer_value) - adapter = ( - evidence.claude_completed_task_models - if agent == "claude" - else evidence.codex_completed_task_models - ) - with pytest.raises(AssertionError, match="nonempty answer value"): - adapter([], answer_value) - - -@pytest.mark.parametrize( - "message", - [None, [], {}, {"role": "assistant"}, {"role": "assistant", "content": "value"}], -) -def test_claude_completed_task_models_rejects_malformed_message_metadata(message): - with pytest.raises(AssertionError): - evidence.claude_completed_task_models([{"type": "assistant", "message": message}], "value") - - -@pytest.mark.parametrize("content", [[None], [{"type": "text"}], [{"type": "text", "text": 4}]]) -def test_claude_completed_task_models_rejects_malformed_answer_blocks(content): - records = completed_records("claude", "model") - records[0]["message"]["content"] = content - with pytest.raises(AssertionError): - evidence.claude_completed_task_models(records, "value") - - -@pytest.mark.parametrize("payload", [None, [], "value"]) -@pytest.mark.parametrize("record_type", ["event_msg", "turn_context"]) -def test_codex_completed_task_models_rejects_malformed_payloads(payload, record_type): - with pytest.raises(AssertionError): - evidence.codex_completed_task_models([{"type": record_type, "payload": payload}], "value") - - -@pytest.mark.parametrize("answer", [None, [], {}, 7]) -def test_codex_completed_task_models_rejects_malformed_completed_answers(answer): - records = completed_records("codex", "model") - records[1]["payload"]["last_agent_message"] = answer - with pytest.raises(AssertionError, match="completed answer"): - evidence.codex_completed_task_models(records, "value") - - -def test_claude_completed_task_models_normalizes_only_context_window_suffix(): - assert evidence.claude_completed_task_models( - completed_records("claude", "catalog.models.claude_sonnet[1m]"), "value" - ) == {"catalog.models.claude_sonnet"} diff --git a/tests/test_integration_http.py b/tests/test_integration_http.py deleted file mode 100644 index ef6df7c8e..000000000 --- a/tests/test_integration_http.py +++ /dev/null @@ -1,116 +0,0 @@ -"""Offline component checks for read-only workspace HTTP boundaries.""" - -import io -import urllib.error -import urllib.response - -import pytest - -from tests.integration.utils import http, provider_catalog - -FETCHERS = [ - provider_catalog.fetch_anthropic_parent_catalog, - provider_catalog.fetch_codex_parent_catalog, - provider_catalog.fetch_anthropic_provider_catalog, - provider_catalog.fetch_codex_provider_catalog, -] - - -def fetch_inventory(fetcher, workspace="https://workspace.invalid", token="test-bearer"): - return fetcher(workspace, token, "catalog.models") - - -@pytest.mark.parametrize("fetcher", FETCHERS) -@pytest.mark.parametrize( - "workspace", - [ - None, - 7, - "", - "http://workspace.invalid", - "https://user:secret@workspace.invalid", - "https://workspace.invalid/path", - "https://workspace.invalid?secret=value", - "https://workspace.invalid#fragment", - "https:///missing-host", - "https://workspace.invalid:bad", - "https://workspace.invalid:0", - "https://workspace.invalid:70000", - "https://workspace.invalid\n", - "https://workspace.invalid\\@other.invalid", - ], -) -def test_catalog_gets_reject_invalid_workspace_origins(monkeypatch, fetcher, workspace): - def unexpected_open(*args, **kwargs): - pytest.fail("Invalid workspace must not reach the HTTP boundary") - - monkeypatch.setattr(http.urllib.request, "build_opener", unexpected_open) - with pytest.raises(AssertionError, match="HTTPS workspace origin") as failure: - fetch_inventory(fetcher, workspace) - assert "secret" not in str(failure.value) - - -@pytest.mark.parametrize("fetcher", FETCHERS) -@pytest.mark.parametrize("token", [None, "", " ", "bearer\nsecret", "bearer\rsecret"]) -def test_catalog_gets_require_explicit_valid_bearers(monkeypatch, fetcher, token): - def unexpected_open(*args, **kwargs): - pytest.fail("Invalid bearer must not reach the HTTP boundary") - - monkeypatch.setattr(http.urllib.request, "build_opener", unexpected_open) - with pytest.raises(AssertionError, match="explicit workspace bearer"): - fetch_inventory(fetcher, token=token) - - -@pytest.mark.parametrize("fetcher", FETCHERS) -@pytest.mark.parametrize("failure_kind", ["http", "url", "os", "json", "encoding"]) -def test_catalog_get_errors_are_sanitized(monkeypatch, fetcher, failure_kind): - secret = "private-bearer-and-response-secret" - - class FailedOpener: - def open(self, request, *, timeout): - assert timeout == 30 - assert request.get_method() == "GET" - assert request.get_header("Authorization") == f"Bearer {secret}" - if failure_kind == "http": - raise urllib.error.HTTPError(request.full_url, 403, secret, {}, None) - if failure_kind == "url": - raise urllib.error.URLError(secret) - if failure_kind == "os": - raise OSError(secret) - return io.BytesIO(secret.encode() if failure_kind == "json" else b"\xff") - - monkeypatch.setattr(http.urllib.request, "build_opener", lambda *_: FailedOpener()) - with pytest.raises(AssertionError, match="Workspace JSON GET") as failure: - fetch_inventory(fetcher, token=secret) - assert secret not in str(failure.value) - assert "workspace.invalid" not in str(failure.value) - assert failure.value.__suppress_context__ - - -@pytest.mark.parametrize("fetcher", FETCHERS) -@pytest.mark.parametrize("status", [301, 302, 303, 307, 308]) -@pytest.mark.parametrize("destination", ["https://other.invalid/secret", "/redirected"]) -def test_catalog_gets_deny_redirects(monkeypatch, fetcher, status, destination): - requests = [] - build_opener = http.urllib.request.build_opener - - class BoundaryHTTPS(http.urllib.request.HTTPSHandler): - def https_open(self, request): - requests.append(request) - response = urllib.response.addinfourl( - io.BytesIO(b"{}"), {"Location": destination}, request.full_url, status - ) - response.msg = "private-redirect-reason" - return response - - monkeypatch.setattr( - http.urllib.request, - "build_opener", - lambda handler: build_opener(handler, BoundaryHTTPS()), - ) - with pytest.raises(AssertionError, match=f"HTTP {status}") as failure: - fetch_inventory(fetcher) - assert len(requests) == 1 - assert requests[0].get_header("Authorization") == "Bearer test-bearer" - assert destination not in str(failure.value) - assert "private-redirect-reason" not in str(failure.value) diff --git a/tests/test_integration_managed_config.py b/tests/test_integration_managed_config.py deleted file mode 100644 index 111fd1e82..000000000 --- a/tests/test_integration_managed_config.py +++ /dev/null @@ -1,96 +0,0 @@ -"""Component checks for reading persisted managed-config evidence.""" - -import json -from types import SimpleNamespace - -import pytest - -from tests.integration.utils.managed import read_persisted_managed_config - -WORKSPACE = "https://workspace.invalid" - - -@pytest.fixture -def session(tmp_path): - return SimpleNamespace(home=tmp_path / "session-home") - - -def _write_persisted_config(session, envelope): - path = session.home / ".ucode" / "managed-config.json" - path.parent.mkdir(parents=True) - path.write_text(json.dumps(envelope), encoding="utf-8") - return path - - -@pytest.mark.parametrize( - "config", - [ - {}, - { - "default_agent": "unrelated-agent", - "custom_policy": {"labels": ["independent", "policy"], "enabled": True}, - }, - ], -) -def test_read_persisted_managed_config_returns_envelope_without_enforcing_policy(session, config): - envelope = {"workspace": WORKSPACE, "config": config, "metadata": {"revision": 7}} - path = _write_persisted_config(session, envelope) - original = path.read_text(encoding="utf-8") - - assert read_persisted_managed_config(session, WORKSPACE) == envelope - assert path.read_text(encoding="utf-8") == original - - -@pytest.mark.parametrize("directory_instead", [False, True]) -def test_read_persisted_managed_config_requires_a_file(session, directory_instead): - if directory_instead: - (session.home / ".ucode" / "managed-config.json").mkdir(parents=True) - - with pytest.raises(AssertionError): - read_persisted_managed_config(session, WORKSPACE) - - -@pytest.mark.parametrize("contents", ["", "{", '{"workspace":', "{} trailing-data"]) -def test_read_persisted_managed_config_rejects_invalid_json(session, contents): - path = _write_persisted_config(session, {}) - path.write_text(contents, encoding="utf-8") - - with pytest.raises(json.JSONDecodeError): - read_persisted_managed_config(session, WORKSPACE) - - -@pytest.mark.parametrize("envelope", [None, [], ["config"], "config", False, 7]) -def test_read_persisted_managed_config_requires_a_dict_envelope(session, envelope): - _write_persisted_config(session, envelope) - - with pytest.raises(AssertionError): - read_persisted_managed_config(session, WORKSPACE) - - -@pytest.mark.parametrize( - "workspace", - [None, "", "https://other-workspace.invalid", WORKSPACE + "/", WORKSPACE.upper(), 7, {}, []], -) -def test_read_persisted_managed_config_requires_an_exact_workspace_match(session, workspace): - _write_persisted_config(session, {"workspace": workspace, "config": {}}) - - with pytest.raises(AssertionError): - read_persisted_managed_config(session, WORKSPACE) - - -@pytest.mark.parametrize("config", [None, [], ["policy"], "policy", False, 7]) -def test_read_persisted_managed_config_requires_a_dict_config(session, config): - _write_persisted_config(session, {"workspace": WORKSPACE, "config": config}) - - with pytest.raises(AssertionError): - read_persisted_managed_config(session, WORKSPACE) - - -@pytest.mark.parametrize("field", ["workspace", "config"]) -def test_read_persisted_managed_config_rejects_missing_fields(session, field): - envelope = {"workspace": WORKSPACE, "config": {}} - del envelope[field] - _write_persisted_config(session, envelope) - - with pytest.raises(AssertionError): - read_persisted_managed_config(session, WORKSPACE) diff --git a/tests/test_integration_model_discovery.py b/tests/test_integration_model_discovery.py index 27bf981fb..fd8860e7f 100644 --- a/tests/test_integration_model_discovery.py +++ b/tests/test_integration_model_discovery.py @@ -2,21 +2,7 @@ import pytest -from tests.integration.utils.model_discovery import ( - assert_picker_inventory, - claude_discovery_model_id, - claude_model_in_picker, - claude_model_service_id, - claude_system_model_ids, - codex_model_in_picker, -) - -PICKER_MODELS = { - "claude": frozenset( - {"catalog.models.claude_sonnet", "catalog.models.claude_haiku", "catalog.models.kimi"} - ), - "codex": frozenset({"catalog.models.gpt_luna", "catalog.models.kimi"}), -} +from tests.integration.utils.model_discovery import claude_model_in_picker, claude_system_model_ids @pytest.mark.parametrize("cursor", ["", "❯ ", "› ", "> "]) @@ -145,118 +131,3 @@ def test_claude_system_model_ids_rejects_invalid_or_out_of_scope_ids(model_id): def test_claude_system_model_ids_rejects_empty_missing_and_duplicate_ids(models): with pytest.raises(AssertionError): claude_system_model_ids(models) - - -@pytest.mark.parametrize( - ("wire_id", "expected"), - [ - ("catalog.models.kimi", "catalog.models.kimi"), - ("anthropic-aigw-69e2a9a0-catalog.models.kimi", "catalog.models.kimi"), - ("catalog.models.claude_sonnet", "catalog.models.claude_sonnet"), - ( - "anthropic-aigw-00000000-catalog.models.kimi", - "anthropic-aigw-00000000-catalog.models.kimi", - ), - ( - "anthropic-aigw-69e2a9a0-catalog.models.kimi_v2", - "anthropic-aigw-69e2a9a0-catalog.models.kimi_v2", - ), - ( - "anthropic-aigw-69e2a9a0-catalog.other_models.kimi", - "anthropic-aigw-69e2a9a0-catalog.other_models.kimi", - ), - ( - "anthropic-aigw-69E2A9A0-catalog.models.kimi", - "anthropic-aigw-69E2A9A0-catalog.models.kimi", - ), - ], -) -def test_claude_service_identity_requires_exact_gateway_alias_checksum(wire_id, expected): - assert claude_model_service_id(wire_id) == expected - - -def test_claude_discovery_requires_exact_gateway_wire_ids(): - assert {claude_discovery_model_id(model) for model in PICKER_MODELS["claude"]} == { - "catalog.models.claude_haiku", - "catalog.models.claude_sonnet", - "anthropic-aigw-69e2a9a0-catalog.models.kimi", - } - assert ( - claude_discovery_model_id("catalog.other_models.claude_decoy") - == "catalog.other_models.claude_decoy" - ) - - -@pytest.mark.parametrize("cursor", ["", "❯ ", "› ", "> "]) -@pytest.mark.parametrize("label", ["catalog.models.gpt_luna", "GPT Luna"]) -def test_codex_picker_matches_model_only_in_numbered_rows(cursor, label): - screen = f"Select Model and Effort\n {cursor}1. {label} (current)\nEnter to confirm" - assert codex_model_in_picker(screen, "catalog.models.gpt_luna", "GPT Luna") - - -@pytest.mark.parametrize( - "screen", - [ - "GPT Luna\nSelect Model and Effort\n 1. Other model", - "Select Model and Effort\n 1. Other model\nCurrent model: GPT Luna", - " 1. catalog.models.gpt_luna_v2", - " 1.\nGPT Luna", - "", - ], -) -def test_codex_picker_rejects_nonrows_and_different_model_ids(screen): - assert not codex_model_in_picker(screen, "catalog.models.gpt_luna", "GPT Luna") - - -@pytest.mark.parametrize("agent", ["claude", "codex"]) -def test_picker_inventory_accepts_exact_rows_and_ignores_banner_text(agent): - models = PICKER_MODELS["claude"] if agent == "claude" else PICKER_MODELS["codex"] - labels = {model: model.rsplit(".", 1)[-1].replace("_", " ").title() for model in models} - rows = "\n".join( - f" {position}. {label}" for position, label in enumerate(labels.values(), start=1) - ) - screen = f"Banner: Gemini Flash\nSelect model\n{rows}\nEnter to confirm" - assert_picker_inventory(screen, agent, labels) - - -@pytest.mark.parametrize("agent", ["claude", "codex"]) -@pytest.mark.parametrize("excluded_label", ["Gemini Flash", "Friendly out-of-scope model"]) -def test_picker_inventory_rejects_extra_friendly_label_rows(agent, excluded_label): - models = PICKER_MODELS["claude"] if agent == "claude" else PICKER_MODELS["codex"] - labels = dict.fromkeys(models, None) - rows = "\n".join(f" {position}. {model}" for position, model in enumerate(models, start=1)) - screen = f"Select model\n{rows}\n {len(models) + 1}. {excluded_label}" - with pytest.raises(AssertionError): - assert_picker_inventory(screen, agent, labels) - - -@pytest.mark.parametrize("agent", ["claude", "codex"]) -def test_picker_inventory_rejects_duplicate_and_missing_rows(agent): - models = sorted(PICKER_MODELS["claude"] if agent == "claude" else PICKER_MODELS["codex"]) - labels = dict.fromkeys(models, None) - rows = "\n".join(f" {position}. {model}" for position, model in enumerate(models, start=1)) - with pytest.raises(AssertionError): - assert_picker_inventory(f"{rows}\n {len(models) + 1}. {models[0]}", agent, labels) - with pytest.raises(AssertionError): - assert_picker_inventory(f" 1. {models[0]}", agent, labels) - - -@pytest.mark.parametrize("agent", ["claude", "codex"]) -def test_picker_inventory_rejects_ambiguous_labels_and_split_line_evidence(agent): - models = sorted(PICKER_MODELS["claude"] if agent == "claude" else PICKER_MODELS["codex"]) - with pytest.raises(AssertionError): - assert_picker_inventory(" 1. Shared label", agent, dict.fromkeys(models, "Shared label")) - with pytest.raises(AssertionError): - assert_picker_inventory(f" 1.\n{models[0]}", agent, dict.fromkeys(models, None)) - - -@pytest.mark.parametrize("agent", ["claude", "codex"]) -@pytest.mark.parametrize("suffix", ["_v2", "-decoy", ".other"]) -def test_picker_inventory_rejects_model_id_prefix_matches(agent, suffix): - models = sorted(PICKER_MODELS["claude"] if agent == "claude" else PICKER_MODELS["codex"]) - rows = "\n".join( - f" {position}. {model}{suffix if position == 1 else ''}" - for position, model in enumerate(models, start=1) - ) - with pytest.raises(AssertionError): - assert_picker_inventory(rows, agent, dict.fromkeys(models, None)) diff --git a/tests/test_integration_provider_catalog.py b/tests/test_integration_provider_catalog.py index 39e2e0ad8..41aa1f66d 100644 --- a/tests/test_integration_provider_catalog.py +++ b/tests/test_integration_provider_catalog.py @@ -7,32 +7,9 @@ import pytest -from tests.integration.utils import http from tests.integration.utils import provider_catalog as catalog -def _mock_json_boundary(monkeypatch, get_json): - class CatalogOpener: - def open(self, request, *, timeout): - assert request.get_method() == "GET" - assert timeout == 30 - names = ( - "Authorization", - "Accept", - "Anthropic-Version", - "Databricks-Model-Provider-Service", - catalog.MODEL_SERVICE_PARENT_SCHEMA_HEADER, - ) - headers = { - name: request.get_header(name.capitalize()) - for name in names - if request.has_header(name.capitalize()) - } - return io.BytesIO(json.dumps(get_json(request.full_url, headers)).encode()) - - monkeypatch.setattr(http.urllib.request, "build_opener", lambda *_: CatalogOpener()) - - def _page(*ids, has_more=False, last_id=None): return { "data": [{"id": model_id, "display_name": model_id} for model_id in ids], @@ -106,7 +83,7 @@ def get_json(url, headers): requests.append((url, headers)) return next(pages) - _mock_json_boundary(monkeypatch, get_json) + monkeypatch.setattr(catalog, "_get_json", get_json) result = catalog.fetch_anthropic_provider_catalog("https://workspace/", "token", "c.s.mps") assert result.model_ids == ("a", "b") assert result.display_names == {"a": "a", "b": "b"} @@ -116,8 +93,6 @@ def get_json(url, headers): for _, headers in requests: assert headers["Authorization"] == "Bearer token" assert headers["Databricks-Model-Provider-Service"] == "c.s.mps" - assert headers["Anthropic-Version"] == "2023-06-01" - assert catalog.MODEL_SERVICE_PARENT_SCHEMA_HEADER not in headers pages = iter([_page("a", has_more=True, last_id="a"), _page("a")]) with pytest.raises(AssertionError, match="repeated model id"): @@ -128,14 +103,14 @@ def test_anthropic_provider_fetch_rejects_repeated_cursor(monkeypatch): pages = iter( [_page("a", has_more=True, last_id="cursor"), _page("b", has_more=True, last_id="cursor")] ) - _mock_json_boundary(monkeypatch, lambda *args: next(pages)) + monkeypatch.setattr(catalog, "_get_json", lambda *args: next(pages)) with pytest.raises(AssertionError, match="repeated pagination cursor"): catalog.fetch_anthropic_provider_catalog("https://workspace", "token", "c.s.mps") def test_anthropic_provider_fetch_has_a_page_bound(monkeypatch): pages = iter(_page(str(i), has_more=True, last_id=str(i)) for i in range(20)) - _mock_json_boundary(monkeypatch, lambda *args: next(pages)) + monkeypatch.setattr(catalog, "_get_json", lambda *args: next(pages)) with pytest.raises(AssertionError, match="exceeded 20 pages"): catalog.fetch_anthropic_provider_catalog("https://workspace", "token", "c.s.mps") @@ -152,15 +127,10 @@ def urlopen(request, timeout): assert request.full_url == "https://workspace/ai-gateway/codex/v1/models" assert request.get_header("Authorization") == "Bearer token" assert request.get_header("Databricks-model-provider-service") == "c.s.mps" - assert not request.has_header(catalog.MODEL_SERVICE_PARENT_SCHEMA_HEADER.capitalize()) assert timeout == 30 return Response(json.dumps(payload).encode()) - monkeypatch.setattr( - http.urllib.request, - "build_opener", - lambda *_: type("CatalogOpener", (), {"open": staticmethod(urlopen)})(), - ) + monkeypatch.setattr(catalog.urllib.request, "urlopen", urlopen) result = catalog.fetch_codex_provider_catalog("https://workspace/", "token", "c.s.mps") assert result.model_ids == ("gpt-a",) @@ -174,7 +144,7 @@ def get_json(url, headers): assert "Databricks-Model-Provider-Service" not in headers return {"models": [{"slug": model, "visibility": "list"} for model in ids]} - _mock_json_boundary(monkeypatch, get_json) + monkeypatch.setattr(catalog, "_get_json", get_json) result = catalog.fetch_codex_parent_catalog("https://workspace/", "token", "c.s") assert result.model_ids == tuple(ids) @@ -184,191 +154,6 @@ def test_provider_fetch_does_not_hide_http_failures(monkeypatch, status): def urlopen(request, timeout): raise urllib.error.HTTPError(request.full_url, status, "failure", {}, None) - monkeypatch.setattr( - http.urllib.request, - "build_opener", - lambda *_: type("CatalogOpener", (), {"open": staticmethod(urlopen)})(), - ) + monkeypatch.setattr(catalog.urllib.request, "urlopen", urlopen) with pytest.raises(AssertionError, match=f"HTTP {status}"): catalog.fetch_codex_provider_catalog("https://workspace", "token", "c.s.mps") - - -def test_anthropic_parent_catalog_retains_pages_labels_and_scoped_requests(monkeypatch): - pages = [ - { - "data": [{"id": "catalog.models.claude_sonnet", "display_name": "Sonnet"}], - "has_more": True, - "last_id": "cursor with / and ?", - }, - { - "data": [{"id": "anthropic-aigw-12345678-catalog.models.kimi"}], - "has_more": False, - }, - ] - requests = [] - - class CatalogOpener: - def open(self, request, *, timeout): - requests.append(request) - assert timeout == 30 - assert request.get_header("Authorization") == "Bearer test-bearer" - assert ( - request.get_header(catalog.MODEL_SERVICE_PARENT_SCHEMA_HEADER.capitalize()) - == "catalog.models" - ) - assert request.get_header("Anthropic-version") == "2023-06-01" - assert not request.has_header("Databricks-model-provider-service") - return io.BytesIO(json.dumps(pages[len(requests) - 1]).encode()) - - monkeypatch.setattr(http.urllib.request, "build_opener", lambda *_: CatalogOpener()) - result = catalog.fetch_anthropic_parent_catalog( - "https://workspace.invalid/", "test-bearer", "catalog.models" - ) - assert result.payloads == tuple(pages) - assert result.model_ids == (pages[0]["data"][0]["id"], pages[1]["data"][0]["id"]) - assert result.display_names == {result.model_ids[0]: "Sonnet", result.model_ids[1]: None} - assert http.urllib.parse.parse_qs(http.urllib.parse.urlsplit(requests[1].full_url).query) == { - "limit": ["1000"], - "after_id": ["cursor with / and ?"], - } - - -@pytest.mark.parametrize("failure_kind", ["duplicate", "cursor", "limit"]) -def test_anthropic_parent_catalog_rejects_duplicate_or_unbounded_pagination( - monkeypatch, failure_kind -): - requests = [] - - class CatalogOpener: - def open(self, request, *, timeout): - requests.append(request) - position = len(requests) - payload = { - "data": [ - {"id": "duplicate" if failure_kind == "duplicate" else f"model-{position}"} - ], - "has_more": True, - "last_id": "repeated" if failure_kind == "cursor" else f"cursor-{position}", - } - return io.BytesIO(json.dumps(payload).encode()) - - monkeypatch.setattr(http.urllib.request, "build_opener", lambda *_: CatalogOpener()) - with pytest.raises(AssertionError): - catalog.fetch_anthropic_parent_catalog( - "https://workspace.invalid", "test-bearer", "catalog.models" - ) - assert len(requests) == (20 if failure_kind == "limit" else 2) - - -def test_codex_parent_catalog_retains_unfiltered_payload_and_validates_duplicates(monkeypatch): - payload = { - "models": [ - {"slug": "catalog.models.gpt_luna", "display_name": "Luna", "visibility": "list"}, - {"slug": "catalog.other_models.codex_decoy", "visibility": "hidden"}, - ] - } - - class CatalogOpener: - def open(self, request, *, timeout): - assert timeout == 30 - assert ( - request.get_header(catalog.MODEL_SERVICE_PARENT_SCHEMA_HEADER.capitalize()) - == "catalog.models" - ) - assert not request.has_header("Databricks-model-provider-service") - return io.BytesIO(json.dumps(payload).encode()) - - monkeypatch.setattr(http.urllib.request, "build_opener", lambda *_: CatalogOpener()) - result = catalog.fetch_codex_parent_catalog( - "https://workspace.invalid", "test-bearer", "catalog.models" - ) - assert result.payloads == (payload,) - assert result.model_ids == ("catalog.models.gpt_luna",) - assert result.display_names == {"catalog.models.gpt_luna": "Luna"} - payload["models"].append(payload["models"][0]) - with pytest.raises(AssertionError, match="repeated"): - catalog.fetch_codex_parent_catalog( - "https://workspace.invalid", "test-bearer", "catalog.models" - ) - - -def test_provider_catalog_constructors_preserve_existing_arguments_and_isolate_defaults(): - anthropic = catalog.AnthropicProviderCatalog(("claude-a",), {"claude-a": None}) - codex = catalog.CodexProviderCatalog(("gpt-a",)) - other_codex = catalog.CodexProviderCatalog(("gpt-b",)) - assert anthropic.payloads == () - assert codex.payloads == () - assert codex.display_names == {} - codex.display_names["gpt-a"] = "Model A" - assert other_codex.display_names == {} - - -@pytest.mark.parametrize( - "fetcher", [catalog.fetch_anthropic_provider_catalog, catalog.fetch_anthropic_parent_catalog] -) -def test_anthropic_catalog_preserves_raw_pages_and_optional_labels(monkeypatch, fetcher): - payloads = [ - { - "data": [{"id": "b", "visibility": "hidden", "extra": [1, 2]}], - "has_more": True, - "last_id": "cursor", - "metadata": {"raw": True}, - }, - {"data": [{"id": "a", "display_name": "Model A"}], "has_more": False}, - ] - pages = iter(payloads) - _mock_json_boundary(monkeypatch, lambda *args: next(pages)) - result = fetcher("https://workspace", "token", "catalog.models") - assert result.payloads == tuple(payloads) - assert result.model_ids == ("b", "a") - assert result.display_names == {"b": None, "a": "Model A"} - - -@pytest.mark.parametrize( - "fetcher", [catalog.fetch_codex_provider_catalog, catalog.fetch_codex_parent_catalog] -) -def test_codex_catalog_preserves_hidden_entries_order_and_optional_labels(monkeypatch, fetcher): - payload = { - "models": [ - {"slug": "b", "visibility": "list"}, - {"slug": "b", "visibility": "hidden", "display_name": 7}, - {"visibility": "hidden", "extra": {"raw": True}}, - {"slug": "a", "visibility": "list", "display_name": "Model A"}, - ], - "metadata": {"raw": True}, - } - _mock_json_boundary(monkeypatch, lambda *args: payload) - result = fetcher("https://workspace", "token", "catalog.models") - assert result.payloads == (payload,) - assert result.model_ids == ("b", "a") - assert result.display_names == {"b": None, "a": "Model A"} - - -@pytest.mark.parametrize( - "fetcher", [catalog.fetch_codex_provider_catalog, catalog.fetch_codex_parent_catalog] -) -@pytest.mark.parametrize("display_name", ["", " ", 7, [], {}]) -def test_codex_catalog_rejects_invalid_optional_display_names(monkeypatch, fetcher, display_name): - payload = {"models": [{"slug": "a", "visibility": "list", "display_name": display_name}]} - _mock_json_boundary(monkeypatch, lambda *args: payload) - with pytest.raises(AssertionError, match="display name"): - fetcher("https://workspace", "token", "catalog.models") - - -@pytest.mark.parametrize( - "fetcher", - [ - catalog.fetch_anthropic_provider_catalog, - catalog.fetch_anthropic_parent_catalog, - catalog.fetch_codex_provider_catalog, - catalog.fetch_codex_parent_catalog, - ], -) -@pytest.mark.parametrize("scope", [None, "", " ", 7]) -def test_provider_catalog_requests_require_an_explicit_scope(monkeypatch, fetcher, scope): - def unexpected_open(*args, **kwargs): - pytest.fail("Invalid catalog scope must not reach the HTTP boundary") - - monkeypatch.setattr(http.urllib.request, "build_opener", unexpected_open) - with pytest.raises(AssertionError, match="explicit catalog scope"): - fetcher("https://workspace", "token", scope) From 8a42c7e2873c4d1e7944f77058550e06e376e956 Mon Sep 17 00:00:00 2001 From: Tien Le Date: Tue, 6 Oct 2026 13:42:35 +0000 Subject: [PATCH 3/6] Remove redundant agent helper extraction tests --- tests/test_integration_agent_helpers.py | 54 ------------------------- 1 file changed, 54 deletions(-) delete mode 100644 tests/test_integration_agent_helpers.py diff --git a/tests/test_integration_agent_helpers.py b/tests/test_integration_agent_helpers.py deleted file mode 100644 index 40e9626a8..000000000 --- a/tests/test_integration_agent_helpers.py +++ /dev/null @@ -1,54 +0,0 @@ -"""Regression checks for extracted native transcript parsing.""" - -import json -from types import SimpleNamespace - -import pytest - -from tests.integration.utils import evidence -from tests.integration.utils.agents import claude, codex - - -@pytest.mark.parametrize("agent,helper", [("claude", claude), ("codex", codex)]) -def test_agent_helpers_preserve_native_answers(agent, helper): - records = [ - {"type": "user", "message": {"content": [{"type": "text", "text": "prompt"}]}}, - { - "type": "assistant", - "message": { - "role": "assistant", - "content": [{"type": "text", "text": "claude-answer"}, {"type": "tool_use"}], - }, - }, - {"type": "event_msg", "payload": {"type": "task_started"}}, - { - "type": "event_msg", - "payload": {"type": "task_complete", "last_agent_message": "codex-answer"}, - }, - ] - expected = [f"{agent}-answer"] - assert helper.assistant_answers(records) == expected - assert evidence.assistant_answers(agent, records) == expected - - -@pytest.mark.parametrize("agent,helper", [("claude", claude), ("codex", codex)]) -def test_agent_helpers_preserve_session_paths_and_child_detection(tmp_path, agent, helper): - expected_directory = {"claude": ".claude/projects", "codex": ".codex/sessions"}[agent] - assert helper.SESSION_DIRECTORY == expected_directory - directory = tmp_path / expected_directory - directory.mkdir(parents=True) - parent = [{"type": "session_meta", "payload": {"source": "cli"}}] - child = [{"type": "session_meta", "payload": {"source": {"subagent": "spawn"}}}] - child_path = directory / "project/subagents/child.jsonl" - child_path.parent.mkdir(parents=True) - (directory / "parent.jsonl").write_text(json.dumps(parent[0]) + "\n") - child_path.write_text(json.dumps(child[0]) + "\n") - session = SimpleNamespace(home=tmp_path) - assert evidence.agent_sessions(session, agent) == { - "parent.jsonl": parent, - "project/subagents/child.jsonl": child, - } - assert not helper.is_child_session("parent.jsonl", parent) - assert not evidence.is_child_session(agent, "parent.jsonl", parent) - assert helper.is_child_session("project/subagents/child.jsonl", child) - assert evidence.is_child_session(agent, "project/subagents/child.jsonl", child) From c3e9880182a020841f35925706e6a26460b1289b Mon Sep 17 00:00:00 2001 From: Tien Le Date: Wed, 7 Oct 2026 04:00:30 +0000 Subject: [PATCH 4/6] Wait for Codex directory trust before treating the TUI as ready --- tests/integration/utils/terminal.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/integration/utils/terminal.py b/tests/integration/utils/terminal.py index 9f4c4fda7..010ed0042 100644 --- a/tests/integration/utils/terminal.py +++ b/tests/integration/utils/terminal.py @@ -265,6 +265,7 @@ def boot(self, timeout=120): handled.add(label) break if matched: + ready_since = None continue assert "Select login method:" not in text, ( "Configured ug launched Claude's account-login flow instead of its gateway session:\n" @@ -278,6 +279,7 @@ def boot(self, timeout=120): if ( title in text and "loading" not in text.lower() + and not re.search(r"(?m)^\s*>\s+You are in\b", text) and re.search(r"(?m)^\s*[❯›>]\s*(?!\d+[.)])", text) ): ready_since = ready_since or time.monotonic() From 9d53e18973e14874f232685faef90074999273a8 Mon Sep 17 00:00:00 2001 From: Tien Le Date: Wed, 7 Oct 2026 05:12:21 +0000 Subject: [PATCH 5/6] Handle Codex trust prompts after startup redraw --- tests/integration/utils/terminal.py | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/tests/integration/utils/terminal.py b/tests/integration/utils/terminal.py index 010ed0042..72c517c97 100644 --- a/tests/integration/utils/terminal.py +++ b/tests/integration/utils/terminal.py @@ -250,9 +250,18 @@ def boot(self, timeout=120): ), ( "trust-directory", - self.session.cwd.name in text - and "trust" in text.lower() - and bool(re.search(r"1[.)]\s+Yes, (?:continue|proceed)", text)), + self.agent == "codex" + and bool( + re.search( + rf"(?m)^\s*>\s+You are in " + rf"{re.escape(str(self.session.cwd.parent))}/[^/\r\n]*$", + text, + ) + ) + and "Do you trust the contents of this directory?" in text + and bool(re.search(r"(?m)^\s*[›❯>]\s*1[.)]\s+Yes, continue\s*$", text)) + and bool(re.search(r"(?m)^\s*2[.)]\s+No, quit\s*$", text)) + and "Press enter to continue" in text, "\r", ), ] From a4c46edbce6464fa780aab09edf12a6ab74e95f7 Mon Sep 17 00:00:00 2001 From: Tien Le Date: Wed, 7 Oct 2026 05:26:45 +0000 Subject: [PATCH 6/6] Recognize Codex trust menu after question redraw --- tests/integration/utils/terminal.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/integration/utils/terminal.py b/tests/integration/utils/terminal.py index 72c517c97..3f7b626c0 100644 --- a/tests/integration/utils/terminal.py +++ b/tests/integration/utils/terminal.py @@ -258,7 +258,11 @@ def boot(self, timeout=120): text, ) ) - and "Do you trust the contents of this directory?" in text + and ( + "Do you trust the contents of this directory?" in text + or "directory allows project-local config, hooks, and exec policies to load." + in text + ) and bool(re.search(r"(?m)^\s*[›❯>]\s*1[.)]\s+Yes, continue\s*$", text)) and bool(re.search(r"(?m)^\s*2[.)]\s+No, quit\s*$", text)) and "Press enter to continue" in text,