#!/usr/bin/env python3"""Behavioral and packaging checks for the bundled factory-files skill. The skill no longer carries a copy of the Factory file format. warp-server ownsthe format and decides whether a tree is valid, so there is nothing here thatasserts which documents the format accepts - those cases live beside the parserin warp-server (logic/factoryfile). What is left to check is everything the client is still responsible for: - which files it selects and uploads, and which it refuses to read- that it reports the server's verdict as the server worded it- that every way of not reaching a verdict is reported as "not validated" rather than as a pass That last one is the point of the whole design. A stale local copy of theformat used to answer confidently and wrongly; the replacement must neverimply a tree was checked when it was not. Run directly, or via script/presubmit.""" from __future__ import annotations import contextlibimport http.serverimport jsonimport osimport shutilimport subprocessimport sysimport tempfileimport threadingfrom pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parentSKILL = REPO_ROOT / "resources" / "bundled" / "skills" / "factory-files"VALIDATOR = SKILL / "scripts" / "validate_factory_files.py" EXIT_VALID = 0EXIT_DIAGNOSTICS = 1EXIT_NOT_VALIDATED = 2 DISCLOSURE = "Validated with the warp-server parser"NOT_VALIDATED = "was NOT validated" FACTORY = """schemaVersion: v1alpha1name: demorepositories: - owner: warpdotdev name: warpagentDefaults: model: auto""" MAIN_AGENT = "---\nagentType: MAIN\n---\nDo the thing.\n" CLEAN_RESPONSE = {"schema_version": "v1alpha1", "valid": True, "diagnostics": []} class _FakeServer: """A warp-server stand-in. Records what the validator submitted so a test can assert the tree it sent, not just the verdict it printed. """ def __init__(self, httpd): self._httpd = httpd self.url = f"http://127.0.0.1:{httpd.server_address[1]}" def submitted_files(self) -> list[dict]: return self._httpd.submitted_files def authorization(self) -> str: return self._httpd.authorization def request_count(self) -> int: return self._httpd.request_count @contextlib.contextmanagerdef fake_server( validate_body, validate_status: int = 200, raw_body: bytes | None = None, reject_authenticated_status: int | None = None,): """reject_authenticated_status answers that status whenever a request carries an Authorization header, regardless of validate_status, so a test can prove a credentialed request is retried without one.""" class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, *_args): # keep the corpus output readable pass def do_POST(self): # noqa: N802 - BaseHTTPRequestHandler's naming length = int(self.headers.get("Content-Length", "0")) payload = json.loads(self.rfile.read(length) or b"{}") self.server.submitted_files.extend(payload.get("files", [])) authorization = self.headers.get("Authorization", "") self.server.authorization = authorization self.server.request_count += 1 status = validate_status if reject_authenticated_status is not None and authorization: status = reject_authenticated_status if raw_body is not None: encoded = raw_body else: encoded = json.dumps(validate_body).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(encoded))) self.end_headers() self.wfile.write(encoded) httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) httpd.submitted_files = [] httpd.authorization = "" httpd.request_count = 0 thread = threading.Thread(target=httpd.serve_forever, daemon=True) thread.start() try: yield _FakeServer(httpd) finally: httpd.shutdown() httpd.server_close() thread.join(timeout=5) SERVER_ROOT_VARIABLES = ("WARP_SERVER_ROOT", "WARP_SERVER_ROOT_URL") def run_validator( root: Path, server_url: str | None = None, api_key: str = "", extra: list[str] | None = None, extra_env: dict[str, str] | None = None,): """Run the validator against a server the caller named. server_url is optional only so a check can name the server through the environment instead. One of the two is required: with neither, the validator would fall through to its production default and post a real tree to app.warp.dev from a test run. """ if server_url is None and not set(extra_env or {}) & set(SERVER_ROOT_VARIABLES): raise RuntimeError( "a check must name the fake server, via server_url or a server-root " "variable in extra_env; otherwise the validator reaches production" ) environment = os.environ.copy() environment.pop("WARP_API_KEY", None) for variable in SERVER_ROOT_VARIABLES: environment.pop(variable, None) if api_key: environment["WARP_API_KEY"] = api_key if extra_env: environment.update(extra_env) args = [sys.executable, str(VALIDATOR), str(root)] if server_url is not None: args += ["--server-root", server_url] args += extra or [] return subprocess.run( args, capture_output=True, text=True, check=False, env=environment, ) @contextlib.contextmanagerdef factory_tree(**files: str): """A minimal Factory root, plus any extra files, cleaned up afterwards.""" root = Path(tempfile.mkdtemp(prefix="factory-files-")) try: contents = {"factory.yaml": FACTORY, "agents/main/agent.md": MAIN_AGENT} contents.update(files) for relative, content in contents.items(): path = root / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") yield root finally: shutil.rmtree(root, ignore_errors=True) def assert_submits_resource_files() -> None: """Canonical resources and invalid benchmark YAML paths are uploaded. Skills can hold anything a repository wants to give an agent, and unrelated files are none of the server's business, so neither is sent. Invalid benchmark YAML must reach the server so it can report the unsupported path. """ with factory_tree( **{ "automations/nightly/automation.md": "---\nagent: main\n---\nrun\n", "runners/linux.yaml": "platform:\n linux:\n dockerImage: ubuntu:24.04\n", "scorers/tests/scorer.md": "---\nagents: [main]\n---\nRubric.\n", "benchmarks/terminal-bench/suite.yaml": "name: Terminal Bench\nagent: main\n", "benchmarks/terminal-bench/tasks/build.yaml": ( "title: Build\nprompt: Build it.\nsuccessCriteria: It builds.\n" ), "webhooks/ci.yaml": "authMode: token\nsecretName: CI_TOKEN\n", "skills/helper/SKILL.md": "---\nname: helper\n---\nhelp\n", "agents/main/skills/inner/SKILL.md": "---\nname: inner\n---\nhelp\n", "README.md": "unrelated", "agents/main/notes.txt": "unrelated", "benchmarks/README.md": "unrelated", "benchmarks/terminal-bench/tasks/paused.yaml.disabled": "unrelated", "benchmarks/terminal-bench/notes.yaml": "unsupported benchmark resource path", "benchmarks/terminal-bench/tasks/CON.yaml": "invalid benchmark task slug", "benchmarks/terminal-bench/tasks/wrong.yml": "unsupported benchmark extension", "benchmarks/terminal-bench/tasks/nested/inner.yaml": ( "unsupported benchmark resource path" ), # Not a canonical webhook path: the server would not classify a # nested file as a source, so submitting it would be noise. "webhooks/nested/inner.yaml": "authMode: token\n", } ) as root: response = { "schema_version": "v1alpha1", "valid": False, "diagnostics": [ { "path": "benchmarks/terminal-bench/notes.yaml", "line": 1, "column": 1, "code": "FF_UNSUPPORTED_PATH", "message": "unsupported Factory resource path", } ], } with fake_server(response) as server: result = run_validator(root, server.url) if result.returncode != EXIT_DIAGNOSTICS: raise RuntimeError( f"expected an unsupported-path diagnostic, got: {result.stdout}{result.stderr}" ) if "FF_UNSUPPORTED_PATH" not in result.stderr: raise RuntimeError(f"the server diagnostic was not relayed: {result.stderr}") submitted = {entry["path"] for entry in server.submitted_files()} expected = { "factory.yaml", "agents/main/agent.md", "automations/nightly/automation.md", "runners/linux.yaml", "scorers/tests/scorer.md", "benchmarks/terminal-bench/suite.yaml", "benchmarks/terminal-bench/tasks/build.yaml", "benchmarks/terminal-bench/notes.yaml", "benchmarks/terminal-bench/tasks/CON.yaml", "benchmarks/terminal-bench/tasks/wrong.yml", "benchmarks/terminal-bench/tasks/nested/inner.yaml", "webhooks/ci.yaml", } if submitted != expected: raise RuntimeError( f"unexpected submitted tree:\n missing={sorted(expected - submitted)}" f"\n extra={sorted(submitted - expected)}" ) def assert_reports_the_servers_verdict() -> None: """Diagnostics are relayed as the server worded them, never reinterpreted.""" with factory_tree() as root: response = { "schema_version": "v1alpha1", "valid": False, "diagnostics": [ { "path": "factory.yaml", "line": 2, "column": 1, "code": "FF_UNKNOWN_FIELD", "message": 'unknown field "bogus"', } ], } with fake_server(response) as server: result = run_validator(root, server.url) if result.returncode != EXIT_DIAGNOSTICS: raise RuntimeError(f"expected exit {EXIT_DIAGNOSTICS}, got {result.returncode}") for fragment in ("FF_UNKNOWN_FIELD", 'unknown field "bogus"', "factory.yaml:2"): if fragment not in result.stderr: raise RuntimeError(f"the server diagnostic was not relayed: {result.stderr}") # An unrecognized schemaVersion is just a server diagnostic now. The # client has no opinion about versions because it no longer reads YAML. unsupported = { "schema_version": "v9alpha1", "valid": False, "diagnostics": [ { "path": "factory.yaml", "line": 1, "column": 1, "code": "FF_UNSUPPORTED_VERSION", "message": "unsupported schemaVersion v9alpha1", } ], } with fake_server(unsupported) as server: result = run_validator(root, server.url) if result.returncode != EXIT_DIAGNOSTICS or "FF_UNSUPPORTED_VERSION" not in result.stderr: raise RuntimeError(f"an unsupported version was not relayed: {result.stderr}") def assert_surfaces_deferred_resolutions() -> None: """A deferred provider alias is reported, not silently dropped. These are the values the endpoint deliberately did not prove. Hiding them would let a clean result read as a guarantee the tree will apply. """ with factory_tree() as root: response = { "schema_version": "v1alpha1", "valid": True, "diagnostics": [], "deferred_resolutions": [ { "path": "automations/t/automation.md", "field": "triggers[0].filter.teams", "kind": "linear_name_alias", } ], } with fake_server(response) as server: result = run_validator(root, server.url) if result.returncode != EXIT_VALID: raise RuntimeError("a deferred alias should not fail the tree") for fragment in ("deferred", "linear_name_alias", "triggers[0].filter.teams"): if fragment not in result.stdout: raise RuntimeError(f"deferred resolutions were not reported: {result.stdout}") def assert_unreached_verdicts_are_never_a_pass() -> None: """Every way of failing to reach the server exits 2 and says so. This is the check that matters most. The previous design answered from a bundled copy of the format when the server was unavailable, and a stale copy produced confident, wrong diagnostics. Silence is the safe failure, but only if it is loud about being silence. """ cases = { "http 401": dict(validate_status=401), "http 429": dict(validate_status=429), "http 500": dict(validate_status=500), "response is not JSON": dict(raw_body=b"<html>nope</html>"), "response is missing diagnostics": dict(validate_body={"schema_version": "v1alpha1"}), "diagnostics are not objects": dict( validate_body={"schema_version": "v1alpha1", "diagnostics": ["nope"]} ), } with factory_tree() as root: for name, options in cases.items(): body = options.pop("validate_body", CLEAN_RESPONSE) with fake_server(body, **options) as server: result = run_validator(root, server.url) if result.returncode != EXIT_NOT_VALIDATED: raise RuntimeError( f"{name}: expected exit {EXIT_NOT_VALIDATED}, got {result.returncode}" ) combined = result.stdout + result.stderr if NOT_VALIDATED not in combined: raise RuntimeError(f"{name}: did not report that nothing was validated") if DISCLOSURE in combined: raise RuntimeError(f"{name}: claimed a server verdict it never received") # An unreachable server is the same case, and must not hang. result = run_validator(root, "http://127.0.0.1:9") if result.returncode != EXIT_NOT_VALIDATED or NOT_VALIDATED not in result.stderr: raise RuntimeError(f"an unreachable server was not reported: {result.stderr}") # A directory that is not a Factory root is also not a verdict. empty = Path(tempfile.mkdtemp(prefix="factory-files-empty-")) try: result = run_validator(empty, "http://127.0.0.1:9") if result.returncode != EXIT_NOT_VALIDATED or "factory.yaml" not in result.stderr: raise RuntimeError(f"a non-Factory directory was not reported: {result.stderr}") finally: shutil.rmtree(empty, ignore_errors=True) def assert_credentials_are_optional_but_forwarded() -> None: """The endpoint needs no key; one is forwarded when the environment has it. Requiring a key would disable validation for local authoring agents, whose shell cannot see the Warp client's session. """ with factory_tree() as root: with fake_server(CLEAN_RESPONSE) as server: result = run_validator(root, server.url) if result.returncode != EXIT_VALID: raise RuntimeError("validation should not require a credential") if server.authorization(): raise RuntimeError("an Authorization header was sent without a key") with fake_server(CLEAN_RESPONSE) as server: result = run_validator(root, server.url, api_key="wk-1.abc") if result.returncode != EXIT_VALID: raise RuntimeError("validation failed when a credential was present") if server.authorization() != "Bearer wk-1.abc": raise RuntimeError(f"the key was not forwarded: {server.authorization()!r}") def assert_401_is_retried_without_the_credential() -> None: """A stale or mismatched WARP_API_KEY must not be able to block validation. The endpoint needs no credential. When a forwarded key draws a 401 or 403, the request is retried once, without it, before giving up. The drop is reported, so a rejected key is still visible on a run that then passes. """ with factory_tree() as root: with fake_server(CLEAN_RESPONSE, reject_authenticated_status=401) as server: result = run_validator(root, server.url, api_key="wk-1.stale") if result.returncode != EXIT_VALID: raise RuntimeError(f"a stale key should not block validation: {result.stderr}") if server.request_count() != 2: raise RuntimeError( f"expected a retry without the credential, got {server.request_count()}" ) if server.authorization(): raise RuntimeError("the retry still carried an Authorization header") if "WARP_API_KEY was rejected" not in result.stderr: raise RuntimeError(f"the dropped credential was not reported: {result.stderr!r}") if DISCLOSURE not in result.stdout: raise RuntimeError("the note displaced the verdict the agent must repeat") with fake_server(CLEAN_RESPONSE, reject_authenticated_status=403) as server: result = run_validator(root, server.url, api_key="wk-1.stale") if result.returncode != EXIT_VALID: raise RuntimeError(f"a 403 credential should also be retried: {result.stderr}") # No credential is in play, so a 401 is not retried; it is reported. with fake_server(CLEAN_RESPONSE, validate_status=401) as server: result = run_validator(root, server.url) if result.returncode != EXIT_NOT_VALIDATED: raise RuntimeError("an unauthenticated 401 should not be retried into a pass") if server.request_count() != 1: raise RuntimeError( f"an unauthenticated request should not be retried: {server.request_count()}" ) # If the retry also fails, the tree is still not validated, and only # one retry is attempted rather than a loop. with fake_server(CLEAN_RESPONSE, validate_status=401) as server: result = run_validator(root, server.url, api_key="wk-1.stale") if result.returncode != EXIT_NOT_VALIDATED: raise RuntimeError("a server that always answers 401 must still report failure") if server.request_count() != 2: raise RuntimeError(f"expected exactly one retry, got {server.request_count()}") def assert_server_root_falls_back_to_the_sandbox_url() -> None: """WARP_SERVER_ROOT_URL is used when WARP_SERVER_ROOT is unset. An Oz sandbox exports the server root under this name, not the older WARP_SERVER_ROOT. Without this fallback the validator silently targets production instead of the server the sandbox actually belongs to. """ with factory_tree() as root: with fake_server(CLEAN_RESPONSE) as server: result = run_validator(root, extra_env={"WARP_SERVER_ROOT_URL": server.url}) if result.returncode != EXIT_VALID: raise RuntimeError( f"WARP_SERVER_ROOT_URL was not honored: {result.stdout}{result.stderr}" ) # WARP_SERVER_ROOT still wins when both are set. with fake_server(CLEAN_RESPONSE) as server: result = run_validator( root, extra_env={ "WARP_SERVER_ROOT": server.url, "WARP_SERVER_ROOT_URL": "http://127.0.0.1:9", }, ) if result.returncode != EXIT_VALID: raise RuntimeError("WARP_SERVER_ROOT should take precedence over WARP_SERVER_ROOT_URL") # --server-root still wins over both environment variables. with fake_server(CLEAN_RESPONSE) as server: result = run_validator( root, server.url, extra_env={ "WARP_SERVER_ROOT": "http://127.0.0.1:9", "WARP_SERVER_ROOT_URL": "http://127.0.0.1:9", }, ) if result.returncode != EXIT_VALID: raise RuntimeError( "--server-root should take precedence over both environment variables" ) Lines 1–500 of 670. Load more
#!/usr/bin/env python3 """Behavioral and packaging checks for the bundled factory-files skill. The skill no longer carries a copy of the Factory file format. warp-server owns the format and decides whether a tree is valid, so there is nothing here that asserts which documents the format accepts - those cases live beside the parser in warp-server (logic/factoryfile). What is left to check is everything the client is still responsible for: - which files it selects and uploads, and which it refuses to read - that it reports the server's verdict as the server worded it - that every way of not reaching a verdict is reported as "not validated" rather than as a pass That last one is the point of the whole design. A stale local copy of the format used to answer confidently and wrongly; the replacement must never imply a tree was checked when it was not. Run directly, or via script/presubmit. """ from __future__ import annotations import contextlib import http.server import json import os import shutil import subprocess import sys import tempfile import threading from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent SKILL = REPO_ROOT / "resources" / "bundled" / "skills" / "factory-files" VALIDATOR = SKILL / "scripts" / "validate_factory_files.py" EXIT_VALID = 0 EXIT_DIAGNOSTICS = 1 EXIT_NOT_VALIDATED = 2 DISCLOSURE = "Validated with the warp-server parser" NOT_VALIDATED = "was NOT validated" FACTORY = """schemaVersion: v1alpha1 name: demo repositories: - owner: warpdotdev name: warp agentDefaults: model: auto """ MAIN_AGENT = "---\nagentType: MAIN\n---\nDo the thing.\n" CLEAN_RESPONSE = {"schema_version": "v1alpha1", "valid": True, "diagnostics": []} class _FakeServer: """A warp-server stand-in. Records what the validator submitted so a test can assert the tree it sent, not just the verdict it printed. """ def __init__(self, httpd): self._httpd = httpd self.url = f"http://127.0.0.1:{httpd.server_address[1]}" def submitted_files(self) -> list[dict]: return self._httpd.submitted_files def authorization(self) -> str: return self._httpd.authorization def request_count(self) -> int: return self._httpd.request_count @contextlib.contextmanager def fake_server( validate_body, validate_status: int = 200, raw_body: bytes | None = None, reject_authenticated_status: int | None = None, ): """reject_authenticated_status answers that status whenever a request carries an Authorization header, regardless of validate_status, so a test can prove a credentialed request is retried without one.""" class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, *_args): # keep the corpus output readable pass def do_POST(self): # noqa: N802 - BaseHTTPRequestHandler's naming length = int(self.headers.get("Content-Length", "0")) payload = json.loads(self.rfile.read(length) or b"{}") self.server.submitted_files.extend(payload.get("files", [])) authorization = self.headers.get("Authorization", "") self.server.authorization = authorization self.server.request_count += 1 status = validate_status if reject_authenticated_status is not None and authorization: status = reject_authenticated_status if raw_body is not None: encoded = raw_body else: encoded = json.dumps(validate_body).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(encoded))) self.end_headers() self.wfile.write(encoded) httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) httpd.submitted_files = [] httpd.authorization = "" httpd.request_count = 0 thread = threading.Thread(target=httpd.serve_forever, daemon=True) thread.start() try: yield _FakeServer(httpd) finally: httpd.shutdown() httpd.server_close() thread.join(timeout=5) SERVER_ROOT_VARIABLES = ("WARP_SERVER_ROOT", "WARP_SERVER_ROOT_URL") def run_validator( root: Path, server_url: str | None = None, api_key: str = "", extra: list[str] | None = None, extra_env: dict[str, str] | None = None, ): """Run the validator against a server the caller named. server_url is optional only so a check can name the server through the environment instead. One of the two is required: with neither, the validator would fall through to its production default and post a real tree to app.warp.dev from a test run. """ if server_url is None and not set(extra_env or {}) & set(SERVER_ROOT_VARIABLES): raise RuntimeError( "a check must name the fake server, via server_url or a server-root " "variable in extra_env; otherwise the validator reaches production" ) environment = os.environ.copy() environment.pop("WARP_API_KEY", None) for variable in SERVER_ROOT_VARIABLES: environment.pop(variable, None) if api_key: environment["WARP_API_KEY"] = api_key if extra_env: environment.update(extra_env) args = [sys.executable, str(VALIDATOR), str(root)] if server_url is not None: args += ["--server-root", server_url] args += extra or [] return subprocess.run( args, capture_output=True, text=True, check=False, env=environment, ) @contextlib.contextmanager def factory_tree(**files: str): """A minimal Factory root, plus any extra files, cleaned up afterwards.""" root = Path(tempfile.mkdtemp(prefix="factory-files-")) try: contents = {"factory.yaml": FACTORY, "agents/main/agent.md": MAIN_AGENT} contents.update(files) for relative, content in contents.items(): path = root / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") yield root finally: shutil.rmtree(root, ignore_errors=True) def assert_submits_resource_files() -> None: """Canonical resources and invalid benchmark YAML paths are uploaded. Skills can hold anything a repository wants to give an agent, and unrelated files are none of the server's business, so neither is sent. Invalid benchmark YAML must reach the server so it can report the unsupported path. """ with factory_tree( **{ "automations/nightly/automation.md": "---\nagent: main\n---\nrun\n", "runners/linux.yaml": "platform:\n linux:\n dockerImage: ubuntu:24.04\n", "scorers/tests/scorer.md": "---\nagents: [main]\n---\nRubric.\n", "benchmarks/terminal-bench/suite.yaml": "name: Terminal Bench\nagent: main\n", "benchmarks/terminal-bench/tasks/build.yaml": ( "title: Build\nprompt: Build it.\nsuccessCriteria: It builds.\n" ), "webhooks/ci.yaml": "authMode: token\nsecretName: CI_TOKEN\n", "skills/helper/SKILL.md": "---\nname: helper\n---\nhelp\n", "agents/main/skills/inner/SKILL.md": "---\nname: inner\n---\nhelp\n", "README.md": "unrelated", "agents/main/notes.txt": "unrelated", "benchmarks/README.md": "unrelated", "benchmarks/terminal-bench/tasks/paused.yaml.disabled": "unrelated", "benchmarks/terminal-bench/notes.yaml": "unsupported benchmark resource path", "benchmarks/terminal-bench/tasks/CON.yaml": "invalid benchmark task slug", "benchmarks/terminal-bench/tasks/wrong.yml": "unsupported benchmark extension", "benchmarks/terminal-bench/tasks/nested/inner.yaml": ( "unsupported benchmark resource path" ), # Not a canonical webhook path: the server would not classify a # nested file as a source, so submitting it would be noise. "webhooks/nested/inner.yaml": "authMode: token\n", } ) as root: response = { "schema_version": "v1alpha1", "valid": False, "diagnostics": [ { "path": "benchmarks/terminal-bench/notes.yaml", "line": 1, "column": 1, "code": "FF_UNSUPPORTED_PATH", "message": "unsupported Factory resource path", } ], } with fake_server(response) as server: result = run_validator(root, server.url) if result.returncode != EXIT_DIAGNOSTICS: raise RuntimeError( f"expected an unsupported-path diagnostic, got: {result.stdout}{result.stderr}" ) if "FF_UNSUPPORTED_PATH" not in result.stderr: raise RuntimeError(f"the server diagnostic was not relayed: {result.stderr}") submitted = {entry["path"] for entry in server.submitted_files()} expected = { "factory.yaml", "agents/main/agent.md", "automations/nightly/automation.md", "runners/linux.yaml", "scorers/tests/scorer.md", "benchmarks/terminal-bench/suite.yaml", "benchmarks/terminal-bench/tasks/build.yaml", "benchmarks/terminal-bench/notes.yaml", "benchmarks/terminal-bench/tasks/CON.yaml", "benchmarks/terminal-bench/tasks/wrong.yml", "benchmarks/terminal-bench/tasks/nested/inner.yaml", "webhooks/ci.yaml", } if submitted != expected: raise RuntimeError( f"unexpected submitted tree:\n missing={sorted(expected - submitted)}" f"\n extra={sorted(submitted - expected)}" ) def assert_reports_the_servers_verdict() -> None: """Diagnostics are relayed as the server worded them, never reinterpreted.""" with factory_tree() as root: response = { "schema_version": "v1alpha1", "valid": False, "diagnostics": [ { "path": "factory.yaml", "line": 2, "column": 1, "code": "FF_UNKNOWN_FIELD", "message": 'unknown field "bogus"', } ], } with fake_server(response) as server: result = run_validator(root, server.url) if result.returncode != EXIT_DIAGNOSTICS: raise RuntimeError(f"expected exit {EXIT_DIAGNOSTICS}, got {result.returncode}") for fragment in ("FF_UNKNOWN_FIELD", 'unknown field "bogus"', "factory.yaml:2"): if fragment not in result.stderr: raise RuntimeError(f"the server diagnostic was not relayed: {result.stderr}") # An unrecognized schemaVersion is just a server diagnostic now. The # client has no opinion about versions because it no longer reads YAML. unsupported = { "schema_version": "v9alpha1", "valid": False, "diagnostics": [ { "path": "factory.yaml", "line": 1, "column": 1, "code": "FF_UNSUPPORTED_VERSION", "message": "unsupported schemaVersion v9alpha1", } ], } with fake_server(unsupported) as server: result = run_validator(root, server.url) if result.returncode != EXIT_DIAGNOSTICS or "FF_UNSUPPORTED_VERSION" not in result.stderr: raise RuntimeError(f"an unsupported version was not relayed: {result.stderr}") def assert_surfaces_deferred_resolutions() -> None: """A deferred provider alias is reported, not silently dropped. These are the values the endpoint deliberately did not prove. Hiding them would let a clean result read as a guarantee the tree will apply. """ with factory_tree() as root: response = { "schema_version": "v1alpha1", "valid": True, "diagnostics": [], "deferred_resolutions": [ { "path": "automations/t/automation.md", "field": "triggers[0].filter.teams", "kind": "linear_name_alias", } ], } with fake_server(response) as server: result = run_validator(root, server.url) if result.returncode != EXIT_VALID: raise RuntimeError("a deferred alias should not fail the tree") for fragment in ("deferred", "linear_name_alias", "triggers[0].filter.teams"): if fragment not in result.stdout: raise RuntimeError(f"deferred resolutions were not reported: {result.stdout}") def assert_unreached_verdicts_are_never_a_pass() -> None: """Every way of failing to reach the server exits 2 and says so. This is the check that matters most. The previous design answered from a bundled copy of the format when the server was unavailable, and a stale copy produced confident, wrong diagnostics. Silence is the safe failure, but only if it is loud about being silence. """ cases = { "http 401": dict(validate_status=401), "http 429": dict(validate_status=429), "http 500": dict(validate_status=500), "response is not JSON": dict(raw_body=b"<html>nope</html>"), "response is missing diagnostics": dict(validate_body={"schema_version": "v1alpha1"}), "diagnostics are not objects": dict( validate_body={"schema_version": "v1alpha1", "diagnostics": ["nope"]} ), } with factory_tree() as root: for name, options in cases.items(): body = options.pop("validate_body", CLEAN_RESPONSE) with fake_server(body, **options) as server: result = run_validator(root, server.url) if result.returncode != EXIT_NOT_VALIDATED: raise RuntimeError( f"{name}: expected exit {EXIT_NOT_VALIDATED}, got {result.returncode}" ) combined = result.stdout + result.stderr if NOT_VALIDATED not in combined: raise RuntimeError(f"{name}: did not report that nothing was validated") if DISCLOSURE in combined: raise RuntimeError(f"{name}: claimed a server verdict it never received") # An unreachable server is the same case, and must not hang. result = run_validator(root, "http://127.0.0.1:9") if result.returncode != EXIT_NOT_VALIDATED or NOT_VALIDATED not in result.stderr: raise RuntimeError(f"an unreachable server was not reported: {result.stderr}") # A directory that is not a Factory root is also not a verdict. empty = Path(tempfile.mkdtemp(prefix="factory-files-empty-")) try: result = run_validator(empty, "http://127.0.0.1:9") if result.returncode != EXIT_NOT_VALIDATED or "factory.yaml" not in result.stderr: raise RuntimeError(f"a non-Factory directory was not reported: {result.stderr}") finally: shutil.rmtree(empty, ignore_errors=True) def assert_credentials_are_optional_but_forwarded() -> None: """The endpoint needs no key; one is forwarded when the environment has it. Requiring a key would disable validation for local authoring agents, whose shell cannot see the Warp client's session. """ with factory_tree() as root: with fake_server(CLEAN_RESPONSE) as server: result = run_validator(root, server.url) if result.returncode != EXIT_VALID: raise RuntimeError("validation should not require a credential") if server.authorization(): raise RuntimeError("an Authorization header was sent without a key") with fake_server(CLEAN_RESPONSE) as server: result = run_validator(root, server.url, api_key="wk-1.abc") if result.returncode != EXIT_VALID: raise RuntimeError("validation failed when a credential was present") if server.authorization() != "Bearer wk-1.abc": raise RuntimeError(f"the key was not forwarded: {server.authorization()!r}") def assert_401_is_retried_without_the_credential() -> None: """A stale or mismatched WARP_API_KEY must not be able to block validation. The endpoint needs no credential. When a forwarded key draws a 401 or 403, the request is retried once, without it, before giving up. The drop is reported, so a rejected key is still visible on a run that then passes. """ with factory_tree() as root: with fake_server(CLEAN_RESPONSE, reject_authenticated_status=401) as server: result = run_validator(root, server.url, api_key="wk-1.stale") if result.returncode != EXIT_VALID: raise RuntimeError(f"a stale key should not block validation: {result.stderr}") if server.request_count() != 2: raise RuntimeError( f"expected a retry without the credential, got {server.request_count()}" ) if server.authorization(): raise RuntimeError("the retry still carried an Authorization header") if "WARP_API_KEY was rejected" not in result.stderr: raise RuntimeError(f"the dropped credential was not reported: {result.stderr!r}") if DISCLOSURE not in result.stdout: raise RuntimeError("the note displaced the verdict the agent must repeat") with fake_server(CLEAN_RESPONSE, reject_authenticated_status=403) as server: result = run_validator(root, server.url, api_key="wk-1.stale") if result.returncode != EXIT_VALID: raise RuntimeError(f"a 403 credential should also be retried: {result.stderr}") # No credential is in play, so a 401 is not retried; it is reported. with fake_server(CLEAN_RESPONSE, validate_status=401) as server: result = run_validator(root, server.url) if result.returncode != EXIT_NOT_VALIDATED: raise RuntimeError("an unauthenticated 401 should not be retried into a pass") if server.request_count() != 1: raise RuntimeError( f"an unauthenticated request should not be retried: {server.request_count()}" ) # If the retry also fails, the tree is still not validated, and only # one retry is attempted rather than a loop. with fake_server(CLEAN_RESPONSE, validate_status=401) as server: result = run_validator(root, server.url, api_key="wk-1.stale") if result.returncode != EXIT_NOT_VALIDATED: raise RuntimeError("a server that always answers 401 must still report failure") if server.request_count() != 2: raise RuntimeError(f"expected exactly one retry, got {server.request_count()}") def assert_server_root_falls_back_to_the_sandbox_url() -> None: """WARP_SERVER_ROOT_URL is used when WARP_SERVER_ROOT is unset. An Oz sandbox exports the server root under this name, not the older WARP_SERVER_ROOT. Without this fallback the validator silently targets production instead of the server the sandbox actually belongs to. """ with factory_tree() as root: with fake_server(CLEAN_RESPONSE) as server: result = run_validator(root, extra_env={"WARP_SERVER_ROOT_URL": server.url}) if result.returncode != EXIT_VALID: raise RuntimeError( f"WARP_SERVER_ROOT_URL was not honored: {result.stdout}{result.stderr}" ) # WARP_SERVER_ROOT still wins when both are set. with fake_server(CLEAN_RESPONSE) as server: result = run_validator( root, extra_env={ "WARP_SERVER_ROOT": server.url, "WARP_SERVER_ROOT_URL": "http://127.0.0.1:9", }, ) if result.returncode != EXIT_VALID: raise RuntimeError("WARP_SERVER_ROOT should take precedence over WARP_SERVER_ROOT_URL") # --server-root still wins over both environment variables. with fake_server(CLEAN_RESPONSE) as server: result = run_validator( root, server.url, extra_env={ "WARP_SERVER_ROOT": "http://127.0.0.1:9", "WARP_SERVER_ROOT_URL": "http://127.0.0.1:9", }, ) if result.returncode != EXIT_VALID: raise RuntimeError( "--server-root should take precedence over both environment variables" )