From e660c8c94853a6a27f933a4cb2a74185451b81d3 Mon Sep 17 00:00:00 2001 From: Zach Wentz <4832+zkwentz@users.noreply.github.com> Date: Sun, 12 Jul 2026 07:48:20 -0400 Subject: [PATCH] feat(validation): add execution-model-aware local profiles --- src/openenv/cli/_validation.py | 495 +-------------------- src/openenv/validation/__init__.py | 5 +- src/openenv/validation/executor.py | 12 + src/openenv/validation/local.py | 528 +++++++++++++++++++++++ src/openenv/validation/runtime_probe.py | 441 +++++++++++++++++++ src/openenv/validation/specs/__init__.py | 3 +- src/openenv/validation/specs/openenv.py | 14 + tests/test_cli/test_validate.py | 16 +- tests/test_validation/test_local.py | 257 +++++++++++ 9 files changed, 1285 insertions(+), 486 deletions(-) create mode 100644 src/openenv/validation/local.py create mode 100644 src/openenv/validation/runtime_probe.py create mode 100644 tests/test_validation/test_local.py diff --git a/src/openenv/cli/_validation.py b/src/openenv/cli/_validation.py index 0b6d95944..f53792e84 100644 --- a/src/openenv/cli/_validation.py +++ b/src/openenv/cli/_validation.py @@ -7,13 +7,25 @@ configured for multi-mode deployment (Docker, direct Python, notebooks, clusters). """ -import ast import re from pathlib import Path from typing import Any -from urllib.parse import urlparse -import requests +import requests as requests +from openenv.validation.runtime_probe import ( + _build_summary as _build_summary, + _make_criterion as _make_criterion, + _normalize_runtime_url as _normalize_runtime_url, + _runtime_standard_profile as _runtime_standard_profile, + validate_running_environment as validate_running_environment, +) +from openenv.validation.source_inspection import ( + _contains_main_call as _contains_main_call, + _dockerfile_installs_openenv_runtime as _dockerfile_installs_openenv_runtime, + _has_main_guard_call as _has_main_guard_call, + _is_main_guard as _is_main_guard, + _OPENENV_DOCKER_INSTALL_RE as _OPENENV_DOCKER_INSTALL_RE, +) try: import tomllib @@ -21,485 +33,8 @@ import tomli as tomllib -def _make_criterion( - criterion_id: str, - description: str, - passed: bool, - *, - required: bool = True, - details: str | None = None, - expected: Any | None = None, - actual: Any | None = None, -) -> dict[str, Any]: - """Create a standard criterion result payload.""" - criterion: dict[str, Any] = { - "id": criterion_id, - "description": description, - "passed": passed, - "required": required, - } - if details is not None: - criterion["details"] = details - if expected is not None: - criterion["expected"] = expected - if actual is not None: - criterion["actual"] = actual - return criterion - - -def _normalize_runtime_url(base_url: str) -> str: - """Normalize and validate a runtime target URL.""" - target = base_url.strip() - if not target: - raise ValueError("Runtime URL cannot be empty") - - if "://" not in target: - target = f"http://{target}" - - parsed = urlparse(target) - if not parsed.scheme or not parsed.netloc: - raise ValueError(f"Invalid runtime URL: {base_url}") - - return target.rstrip("/") - - -def _runtime_standard_profile(api_version: str) -> str: - """Resolve the runtime standard profile for an API version.""" - if api_version.startswith("1."): - return "openenv-http/1.x" - return "openenv-http/unknown" - - -def _build_summary(criteria: list[dict[str, Any]]) -> dict[str, Any]: - """Build a compact pass/fail summary for a criteria list.""" - total_count = len(criteria) - passed_count = sum(1 for criterion in criteria if criterion.get("passed", False)) - failed_criteria = [ - criterion.get("id", "unknown") - for criterion in criteria - if not criterion.get("passed", False) - ] - required_criteria = [ - criterion for criterion in criteria if criterion.get("required", True) - ] - required_total_count = len(required_criteria) - required_passed_count = sum( - 1 for criterion in required_criteria if criterion.get("passed", False) - ) - - return { - "passed_count": passed_count, - "total_count": total_count, - "failed_criteria": failed_criteria, - "required_passed_count": required_passed_count, - "required_total_count": required_total_count, - } - - -def validate_running_environment( - base_url: str, timeout_s: float = 5.0 -) -> dict[str, Any]: - """ - Validate a running OpenEnv server against runtime API standards. - - The returned JSON report contains an overall pass/fail result and - per-criterion outcomes that can be consumed in CI. - """ - normalized_url = _normalize_runtime_url(base_url) - criteria: list[dict[str, Any]] = [] - - report: dict[str, Any] = { - "target": normalized_url, - "validation_type": "running_environment", - "standard_version": "unknown", - "standard_profile": "openenv-http/unknown", - "mode": "unknown", - "passed": False, - "summary": {}, - "criteria": criteria, - } - - openapi_paths: dict[str, Any] = {} - api_version = "unknown" - - # Criterion: OpenAPI endpoint reachable with a declared version. - try: - openapi_response = requests.get( - f"{normalized_url}/openapi.json", timeout=timeout_s - ) - except requests.RequestException as exc: - criteria.append( - _make_criterion( - "openapi_version_available", - "GET /openapi.json returns OpenAPI info.version", - False, - details=f"Request failed: {type(exc).__name__}: {exc}", - expected={"status_code": 200, "info.version": "string"}, - ) - ) - else: - try: - openapi_json = openapi_response.json() - except ValueError: - openapi_json = None - - openapi_ok = ( - openapi_response.status_code == 200 - and isinstance(openapi_json, dict) - and isinstance(openapi_json.get("info"), dict) - and isinstance(openapi_json["info"].get("version"), str) - ) - - if openapi_ok: - api_version = str(openapi_json["info"]["version"]) - openapi_paths = openapi_json.get("paths", {}) - criteria.append( - _make_criterion( - "openapi_version_available", - "GET /openapi.json returns OpenAPI info.version", - True, - expected={"status_code": 200, "info.version": "string"}, - actual={ - "status_code": openapi_response.status_code, - "info.version": api_version, - }, - ) - ) - else: - criteria.append( - _make_criterion( - "openapi_version_available", - "GET /openapi.json returns OpenAPI info.version", - False, - details="Response missing required OpenAPI info.version field", - expected={"status_code": 200, "info.version": "string"}, - actual={ - "status_code": openapi_response.status_code, - "body_type": ( - type(openapi_json).__name__ - if openapi_json is not None - else "non_json" - ), - }, - ) - ) - - report["standard_version"] = api_version - report["standard_profile"] = _runtime_standard_profile(api_version) - - # Criterion: Health endpoint. - try: - health_response = requests.get(f"{normalized_url}/health", timeout=timeout_s) - except requests.RequestException as exc: - criteria.append( - _make_criterion( - "health_endpoint", - "GET /health returns healthy status", - False, - details=f"Request failed: {type(exc).__name__}: {exc}", - expected={"status_code": 200, "status": "healthy"}, - ) - ) - else: - try: - health_json = health_response.json() - except ValueError: - health_json = None - - health_ok = ( - health_response.status_code == 200 - and isinstance(health_json, dict) - and health_json.get("status") == "healthy" - ) - criteria.append( - _make_criterion( - "health_endpoint", - "GET /health returns healthy status", - health_ok, - expected={"status_code": 200, "status": "healthy"}, - actual={ - "status_code": health_response.status_code, - "status": ( - health_json.get("status") - if isinstance(health_json, dict) - else None - ), - }, - ) - ) - - # Criterion: Metadata endpoint has required fields. - try: - metadata_response = requests.get( - f"{normalized_url}/metadata", timeout=timeout_s - ) - except requests.RequestException as exc: - criteria.append( - _make_criterion( - "metadata_endpoint", - "GET /metadata returns name and description", - False, - details=f"Request failed: {type(exc).__name__}: {exc}", - expected={"status_code": 200, "fields": ["name", "description"]}, - ) - ) - else: - try: - metadata_json = metadata_response.json() - except ValueError: - metadata_json = None - - metadata_ok = ( - metadata_response.status_code == 200 - and isinstance(metadata_json, dict) - and isinstance(metadata_json.get("name"), str) - and isinstance(metadata_json.get("description"), str) - ) - criteria.append( - _make_criterion( - "metadata_endpoint", - "GET /metadata returns name and description", - metadata_ok, - expected={"status_code": 200, "fields": ["name", "description"]}, - actual={ - "status_code": metadata_response.status_code, - "name": ( - metadata_json.get("name") - if isinstance(metadata_json, dict) - else None - ), - "description": ( - metadata_json.get("description") - if isinstance(metadata_json, dict) - else None - ), - }, - ) - ) - - # Criterion: Schema endpoint returns action/observation/state. - try: - schema_response = requests.get(f"{normalized_url}/schema", timeout=timeout_s) - except requests.RequestException as exc: - criteria.append( - _make_criterion( - "schema_endpoint", - "GET /schema returns action, observation, and state schemas", - False, - details=f"Request failed: {type(exc).__name__}: {exc}", - expected={ - "status_code": 200, - "fields": ["action", "observation", "state"], - }, - ) - ) - else: - try: - schema_json = schema_response.json() - except ValueError: - schema_json = None - - schema_ok = ( - schema_response.status_code == 200 - and isinstance(schema_json, dict) - and isinstance(schema_json.get("action"), dict) - and isinstance(schema_json.get("observation"), dict) - and isinstance(schema_json.get("state"), dict) - ) - criteria.append( - _make_criterion( - "schema_endpoint", - "GET /schema returns action, observation, and state schemas", - schema_ok, - expected={ - "status_code": 200, - "fields": ["action", "observation", "state"], - }, - actual={ - "status_code": schema_response.status_code, - "has_action": ( - isinstance(schema_json.get("action"), dict) - if isinstance(schema_json, dict) - else False - ), - "has_observation": ( - isinstance(schema_json.get("observation"), dict) - if isinstance(schema_json, dict) - else False - ), - "has_state": ( - isinstance(schema_json.get("state"), dict) - if isinstance(schema_json, dict) - else False - ), - }, - ) - ) - - # Criterion: MCP endpoint is reachable. - try: - mcp_response = requests.post( - f"{normalized_url}/mcp", json={}, timeout=timeout_s - ) - except requests.RequestException as exc: - criteria.append( - _make_criterion( - "mcp_endpoint", - "POST /mcp is reachable and returns JSON-RPC payload", - False, - details=f"Request failed: {type(exc).__name__}: {exc}", - expected={"status_code": 200, "jsonrpc": "2.0"}, - ) - ) - else: - try: - mcp_json = mcp_response.json() - except ValueError: - mcp_json = None - - mcp_ok = ( - mcp_response.status_code == 200 - and isinstance(mcp_json, dict) - and mcp_json.get("jsonrpc") == "2.0" - ) - criteria.append( - _make_criterion( - "mcp_endpoint", - "POST /mcp is reachable and returns JSON-RPC payload", - mcp_ok, - expected={"status_code": 200, "jsonrpc": "2.0"}, - actual={ - "status_code": mcp_response.status_code, - "jsonrpc": ( - mcp_json.get("jsonrpc") if isinstance(mcp_json, dict) else None - ), - }, - ) - ) - - # Criterion: mode endpoint contract consistency via OpenAPI paths. - if isinstance(openapi_paths, dict) and openapi_paths: - has_reset = "/reset" in openapi_paths - has_step = "/step" in openapi_paths - has_state = "/state" in openapi_paths - - if has_reset: - report["mode"] = "simulation" - mode_ok = has_step and has_state - expected_paths = {"/reset": True, "/step": True, "/state": True} - else: - report["mode"] = "production" - mode_ok = not has_step and not has_state - expected_paths = {"/reset": False, "/step": False, "/state": False} - - criteria.append( - _make_criterion( - "mode_endpoint_consistency", - "OpenAPI endpoint set matches OpenEnv mode contract", - mode_ok, - expected=expected_paths, - actual={ - "/reset": has_reset, - "/step": has_step, - "/state": has_state, - }, - ) - ) - else: - criteria.append( - _make_criterion( - "mode_endpoint_consistency", - "OpenAPI endpoint set matches OpenEnv mode contract", - False, - details="Cannot determine mode without OpenAPI paths", - expected={"openapi.paths": "present"}, - actual={"openapi.paths": "missing"}, - ) - ) - - report["passed"] = all( - criterion["passed"] for criterion in criteria if criterion.get("required", True) - ) - report["summary"] = _build_summary(criteria) - return report - - -def _has_main_guard_call(app_content: str) -> bool: - """Return True when the module calls main() under a __main__ guard.""" - try: - tree = ast.parse(app_content) - except SyntaxError: - return ( - "__name__" in app_content - and "__main__" in app_content - and "main(" in app_content - ) - - for node in ast.iter_child_nodes(tree): - if not isinstance(node, ast.If) or not _is_main_guard(node.test): - continue - - if any(_contains_main_call(guarded_node) for guarded_node in node.body): - return True - - return False - - -def _is_main_guard(test: ast.expr) -> bool: - """Return True for `if __name__ == "__main__"` tests.""" - return ( - isinstance(test, ast.Compare) - and isinstance(test.left, ast.Name) - and test.left.id == "__name__" - and len(test.ops) == 1 - and isinstance(test.ops[0], ast.Eq) - and len(test.comparators) == 1 - and isinstance(test.comparators[0], ast.Constant) - and test.comparators[0].value == "__main__" - ) - - -def _contains_main_call(node: ast.AST) -> bool: - """Return True when an AST node contains a direct `main(...)` call.""" - return any( - isinstance(candidate, ast.Call) - and isinstance(candidate.func, ast.Name) - and candidate.func.id == "main" - for candidate in ast.walk(node) - ) - - _OPENENV_RUNTIME_DEP_RE = re.compile(r"^openenv(?:\s*(?:$|[<>=!~@;])|\[)") _LEGACY_OPENENV_CORE_DEP_RE = re.compile(r"^openenv-core(?:\s*(?:$|[<>=!~@;])|\[)") -_OPENENV_DOCKER_INSTALL_RE = re.compile( - r"(?=!~@;])|\[)" -) - - -def _dockerfile_installs_openenv_runtime(env_path: Path) -> bool: - """Return True when a Docker deployment installs OpenEnv outside pyproject.""" - for dockerfile_path in ( - env_path / "server" / "Dockerfile", - env_path / "Dockerfile", - ): - if not dockerfile_path.exists(): - continue - - try: - dockerfile = dockerfile_path.read_text(encoding="utf-8") - except OSError: - continue - - for line in dockerfile.splitlines(): - stripped = line.strip().lower() - if not stripped or stripped.startswith("#"): - continue - if _OPENENV_DOCKER_INSTALL_RE.search(stripped): - return True - if "openenv-core" in stripped: - return True - - return False def validate_multi_mode_deployment(env_path: Path) -> tuple[bool, list[str]]: diff --git a/src/openenv/validation/__init__.py b/src/openenv/validation/__init__.py index 37d1f75d9..dbb93bdf8 100644 --- a/src/openenv/validation/__init__.py +++ b/src/openenv/validation/__init__.py @@ -1,8 +1,9 @@ # SPDX-License-Identifier: BSD-3-Clause -"""Versioned, spec-neutral OpenEnv validation contracts.""" +"""Spec-neutral OpenEnv validation planning, execution, and local profiles.""" from .executor import execute_validation_plan +from .local import format_shared_validation_report, run_local_validation from .models import ( CheckOutcome, RunnerCapabilities, @@ -42,4 +43,6 @@ "ValidationStatus", "build_validation_plan", "execute_validation_plan", + "format_shared_validation_report", + "run_local_validation", ] diff --git a/src/openenv/validation/executor.py b/src/openenv/validation/executor.py index 59e9cd2fb..5f978fac3 100644 --- a/src/openenv/validation/executor.py +++ b/src/openenv/validation/executor.py @@ -66,6 +66,17 @@ def _execute_check( check = plan.checks[index] missing = check.capabilities - plan.capabilities.available if missing: + declared_reasons = context.discovered.get("capability_unavailable_reasons", {}) + reasons = ( + { + capability.value: declared_reasons[capability.value] + for capability in missing + if capability.value in declared_reasons + and isinstance(declared_reasons[capability.value], str) + } + if isinstance(declared_reasons, Mapping) + else {} + ) return ValidationResult( criterion_id=check.criterion_id, requirement=check.requirement, @@ -80,6 +91,7 @@ def _execute_check( "available": sorted( capability.value for capability in plan.capabilities.available ), + "unavailable_reasons": reasons, }, duration_s=0.0, timeout_s=check.timeout_s, diff --git a/src/openenv/validation/local.py b/src/openenv/validation/local.py new file mode 100644 index 000000000..0ea664eca --- /dev/null +++ b/src/openenv/validation/local.py @@ -0,0 +1,528 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Local/static and runtime-capable validation execution.""" + +from __future__ import annotations + +import errno +import os +import shutil +import signal +import socket +import subprocess +import sys +import time +from contextlib import contextmanager, suppress +from pathlib import Path +from tempfile import TemporaryDirectory +from typing import Any, Iterator +from urllib.parse import urlparse + +import requests +import yaml + +from .executor import execute_validation_plan +from .models import ( + RunnerCapabilities, + ValidationCapability, + ValidationPolicy, + ValidationProfile, + ValidationReport, +) +from .planner import build_validation_plan +from .runtime_probe import _normalize_runtime_url, validate_running_environment +from .specs import ( + DEFAULT_SPEC_REGISTRY, + ExecutionModel, + runtime_openenv_spec_load, + SpecLoad, + ValidationSpecRegistry, +) + + +def _looks_like_url(value: str) -> bool: + lowered = value.strip().lower() + return lowered.startswith("http://") or lowered.startswith("https://") + + +def _runtime_configuration(env_path: Path) -> tuple[str, int]: + manifest_path = env_path / "openenv.yaml" + try: + manifest = yaml.safe_load(manifest_path.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError) as exc: + raise RuntimeError("Unable to load openenv.yaml for local runtime") from exc + if not isinstance(manifest, dict): + raise RuntimeError("openenv.yaml must contain a mapping") + app = manifest.get("app") + port = manifest.get("port", 8000) + if not isinstance(app, str) or not app.strip(): + raise RuntimeError("openenv.yaml must declare an app module") + if not isinstance(port, int) or not 0 < port < 65536: + raise RuntimeError("openenv.yaml port must be a valid integer") + return app, port + + +def _runtime_environment( + env_path: Path, *, temporary_home: Path | None = None +) -> dict[str, str]: + """Build a minimal environment that does not forward ambient credentials.""" + safe_names = { + "HOME", + "LANG", + "LC_ALL", + "PATH", + "SSL_CERT_DIR", + "SSL_CERT_FILE", + "SYSTEMROOT", + "TEMP", + "TMP", + "TMPDIR", + "VIRTUAL_ENV", + } + process_env = { + name: value for name, value in os.environ.items() if name in safe_names + } + if temporary_home is not None: + process_env["HOME"] = str(temporary_home) + source_root = Path(__file__).resolve().parents[2] + python_path = [str(env_path), str(env_path.parent), str(source_root)] + existing_python_path = os.environ.get("PYTHONPATH") + if existing_python_path: + python_path.append(existing_python_path) + process_env["PYTHONPATH"] = os.pathsep.join(python_path) + process_env["OPENENV_VALIDATION"] = "1" + return process_env + + +def _ensure_port_available(port: int) -> None: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: + probe.settimeout(0.25) + result = probe.connect_ex(("127.0.0.1", port)) + if result == 0: + raise RuntimeError(f"Local runtime port {port} is already in use") + if result in {errno.EACCES, errno.EPERM}: + raise PermissionError(f"Cannot inspect local runtime port {port}") + + +def _stop_process(process: subprocess.Popen[bytes]) -> None: + if os.name == "posix": + with suppress(ProcessLookupError): + os.killpg(process.pid, signal.SIGTERM) + elif process.poll() is None: + process.terminate() + if process.poll() is None: + with suppress(subprocess.TimeoutExpired): + process.wait(timeout=5.0) + if os.name == "posix": + try: + os.killpg(process.pid, 0) + except (PermissionError, ProcessLookupError): + return + with suppress(ProcessLookupError): + os.killpg(process.pid, signal.SIGKILL) + elif process.poll() is None: + process.kill() + if process.poll() is None: + with suppress(subprocess.TimeoutExpired): + process.wait(timeout=5.0) + + +def _server_command(env_path: Path, app: str, port: int) -> list[str]: + command = [ + "python", + "-m", + "uvicorn", + app, + "--host", + "127.0.0.1", + "--port", + str(port), + "--log-level", + "warning", + ] + uv = shutil.which("uv") + if uv is not None: + return [uv, "run", "--project", str(env_path), *command] + command[0] = sys.executable + return command + + +def _contains_property(document: dict[str, Any], node: Any, name: str) -> bool: + seen_refs: set[str] = set() + + def visit(value: Any) -> bool: + if isinstance(value, dict): + reference = value.get("$ref") + if isinstance(reference, str) and reference.startswith("#/"): + if reference in seen_refs: + return False + seen_refs.add(reference) + resolved: Any = document + for part in reference[2:].split("/"): + if not isinstance(resolved, dict) or part not in resolved: + return False + resolved = resolved[part] + if visit(resolved): + return True + properties = value.get("properties") + if isinstance(properties, dict) and name in properties: + return True + return any(visit(item) for key, item in value.items() if key != "$ref") + if isinstance(value, list): + return any(visit(item) for item in value) + return False + + return visit(node) + + +def _discover_runtime_declarations( + runtime_url: str, timeout_s: float +) -> dict[str, Any]: + declarations: dict[str, Any] = {} + try: + openapi_response = requests.get( + f"{runtime_url}/openapi.json", + timeout=timeout_s, + allow_redirects=False, + ) + openapi = openapi_response.json() + except (requests.RequestException, ValueError): + openapi = None + try: + schema_response = requests.get( + f"{runtime_url}/schema", + timeout=timeout_s, + allow_redirects=False, + ) + schemas = schema_response.json() + except (requests.RequestException, ValueError): + schemas = None + + if isinstance(openapi, dict): + paths = openapi.get("paths", {}) + if isinstance(paths, dict): + task_paths = sorted( + path + for path in paths + if path == "/list_environments" + or path.endswith("/tasks") + or path.endswith("/num_tasks") + ) + if task_paths: + declarations["tasks"] = {"valid": True, "paths": task_paths} + declarations["seeds"] = { + "valid": _contains_property(openapi, paths.get("/reset"), "seed"), + "endpoint": "/reset", + } + trajectory_paths = sorted( + path + for path in paths + if "trajectory" in path.lower() or "episode" in path.lower() + ) + if trajectory_paths: + declarations["trajectories"] = { + "valid": True, + "paths": trajectory_paths, + } + + if isinstance(schemas, dict): + declarations["rewards"] = { + "valid": _contains_property( + schemas, schemas.get("observation", {}), "reward" + ), + "schema": "observation.reward", + } + + try: + tools_response = requests.post( + f"{runtime_url}/mcp", + json={"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}, + timeout=timeout_s, + allow_redirects=False, + ) + tools_payload = tools_response.json() + except (requests.RequestException, ValueError): + tools_payload = None + if isinstance(tools_payload, dict): + result = tools_payload.get("result") + tools = result.get("tools") if isinstance(result, dict) else None + if isinstance(tools, list): + names = [tool.get("name") for tool in tools if isinstance(tool, dict)] + valid = len(names) == len(tools) and all( + isinstance(name, str) and name for name in names + ) + declarations["tools"] = { + "valid": valid and len(set(names)) == len(names), + "count": len(tools), + "names": names, + } + + try: + from websockets.sync.client import connect as ws_sync_connect + + parsed = urlparse(runtime_url) + websocket_scheme = "wss" if parsed.scheme == "https" else "ws" + websocket_url = f"{websocket_scheme}://{parsed.netloc}/ws" + with ws_sync_connect(websocket_url, open_timeout=timeout_s): + pass + except Exception as exc: + declarations["websocket"] = { + "valid": False, + "error_type": type(exc).__name__, + } + else: + declarations["websocket"] = {"valid": True, "endpoint": "/ws"} + return declarations + + +@contextmanager +def _launch_environment(env_path: Path, timeout_s: float) -> Iterator[str]: + app, port = _runtime_configuration(env_path) + base_url = f"http://127.0.0.1:{port}" + _ensure_port_available(port) + with TemporaryDirectory(prefix="openenv-validation-") as temporary_home: + process_env = _runtime_environment( + env_path, temporary_home=Path(temporary_home) + ) + process = subprocess.Popen( + _server_command(env_path, app, port), + cwd=env_path, + env=process_env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + try: + deadline = time.monotonic() + timeout_s + health_url = f"{base_url}/health" + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError( + f"Local environment server exited with code {process.returncode}" + ) + try: + response = requests.get( + health_url, + timeout=min(1.0, timeout_s), + allow_redirects=False, + ) + except requests.RequestException: + time.sleep(0.1) + continue + if response.status_code == 200: + break + time.sleep(0.1) + else: + raise TimeoutError( + f"Local environment did not become ready within {timeout_s:g}s" + ) + yield base_url + finally: + _stop_process(process) + + +def _run( + target: str | Path, + *, + profile: ValidationProfile, + source_path: Path | None, + spec_load: SpecLoad, + policy: ValidationPolicy | None, + runtime_url: str | None, + runtime_error: str | None, + runtime_unavailable_reason: str | None, + timeout_s: float, +) -> ValidationReport: + discovered: dict[str, Any] = {} + available: set[ValidationCapability] = set() + if source_path is not None: + available.add(ValidationCapability.SOURCE) + if runtime_url is not None or ( + runtime_error is not None and runtime_unavailable_reason is None + ): + available.add(ValidationCapability.RUNTIME) + if runtime_url is not None: + try: + discovered["runtime_report"] = validate_running_environment( + runtime_url, timeout_s=timeout_s + ) + except Exception as exc: + discovered["runtime_probe_error"] = type(exc).__name__ + else: + try: + discovered["runtime_declarations"] = _discover_runtime_declarations( + runtime_url, timeout_s + ) + except Exception as exc: + discovered["runtime_declarations_error"] = type(exc).__name__ + elif runtime_error is not None: + discovered["runtime_probe_error"] = runtime_error + if runtime_unavailable_reason is not None: + discovered["capability_unavailable_reasons"] = { + ValidationCapability.RUNTIME.value: runtime_unavailable_reason + } + + capabilities = RunnerCapabilities( + runner="local", + available=frozenset(available), + official=False, + isolation_mode=None, + ) + plan, context = build_validation_plan( + source_path or str(target), + profile=profile, + capabilities=capabilities, + spec_load=spec_load, + policy=policy, + runtime_url=runtime_url, + discovered=discovered, + ) + return execute_validation_plan(plan, context) + + +def run_local_validation( + target: str | Path, + *, + profile: ValidationProfile | str = ValidationProfile.STATIC, + runtime_url: str | None = None, + timeout_s: float = 5.0, + spec_id: str | None = None, + spec_registry: ValidationSpecRegistry | None = None, + policy: ValidationPolicy | None = None, +) -> ValidationReport: + """Run the selected profile locally and never claim remote certification.""" + selected_profile = ( + profile + if isinstance(profile, ValidationProfile) + else ValidationProfile(profile) + ) + raw_target = str(target) + target_is_url = _looks_like_url(raw_target) + if target_is_url and selected_profile is ValidationProfile.STATIC: + raise ValueError("The static profile requires a local source directory") + source_path = None if target_is_url else Path(target) + if source_path is not None: + source_path = source_path.resolve() + if not source_path.is_dir(): + raise ValueError("Validation target must be an existing source directory") + effective_runtime_url = runtime_url or (raw_target if target_is_url else None) + if effective_runtime_url is not None: + effective_runtime_url = _normalize_runtime_url(effective_runtime_url) + + if source_path is None: + if spec_id not in {None, "openenv"} or spec_registry is not None: + raise ValueError("Runtime URLs currently support only the OpenEnv spec") + spec_load = runtime_openenv_spec_load() + else: + registry = spec_registry or DEFAULT_SPEC_REGISTRY + spec_load = registry.resolve(source_path, spec_id=spec_id) + + if selected_profile is ValidationProfile.STATIC: + return _run( + target, + profile=selected_profile, + source_path=source_path, + spec_load=spec_load, + policy=policy, + runtime_url=None, + runtime_error=None, + runtime_unavailable_reason=None, + timeout_s=timeout_s, + ) + + subject = spec_load.subject + execution_model = subject.spec.execution_model if subject is not None else None + if subject is None or execution_model is not ExecutionModel.SERVED: + reason = ( + "source_spec_unavailable" + if execution_model is None + else f"unsupported_execution_model:{execution_model.value}" + ) + return _run( + target, + profile=selected_profile, + source_path=source_path, + spec_load=spec_load, + policy=policy, + runtime_url=None, + runtime_error=None, + runtime_unavailable_reason=reason, + timeout_s=timeout_s, + ) + + if effective_runtime_url is not None: + return _run( + target, + profile=selected_profile, + source_path=source_path, + spec_load=spec_load, + policy=policy, + runtime_url=effective_runtime_url, + runtime_error=None, + runtime_unavailable_reason=None, + timeout_s=timeout_s, + ) + + if source_path is None: + return _run( + target, + profile=selected_profile, + source_path=None, + spec_load=spec_load, + policy=policy, + runtime_url=None, + runtime_error=None, + runtime_unavailable_reason="runtime_target_unavailable", + timeout_s=timeout_s, + ) + + try: + with _launch_environment(source_path, timeout_s) as launched_url: + return _run( + target, + profile=selected_profile, + source_path=source_path, + spec_load=spec_load, + policy=policy, + runtime_url=launched_url, + runtime_error=None, + runtime_unavailable_reason=None, + timeout_s=timeout_s, + ) + except Exception as exc: + return _run( + target, + profile=selected_profile, + source_path=source_path, + spec_load=spec_load, + policy=policy, + runtime_url=None, + runtime_error=type(exc).__name__, + runtime_unavailable_reason=None, + timeout_s=timeout_s, + ) + + +def format_shared_validation_report(report: ValidationReport) -> str: + """Render a concise human-readable shared report.""" + marker = "OK" if report.passed else "FAIL" + lines = [ + f"[{marker}] {report.target}: {report.profile.value} validation", + f"Policy: {report.policy_version}", + ] + identity = report.spec.spec if report.spec is not None else report.spec_identity + if identity is not None: + lines.append( + f"Spec: {identity.spec_id} {identity.spec_version or 'unknown'} " + f"({identity.execution_model.value}; " + f"adapter {identity.adapter.adapter_id}@{identity.adapter.adapter_version})" + ) + for result in report.results: + lines.append( + f" [{result.status.value.upper()}] {result.criterion_id} " + f"({result.severity.value})" + ) + if result.message: + lines.append(f" {result.message}") + lines.append("Certification: not claimed by local validation") + return "\n".join(lines) diff --git a/src/openenv/validation/runtime_probe.py b/src/openenv/validation/runtime_probe.py new file mode 100644 index 000000000..eb638a032 --- /dev/null +++ b/src/openenv/validation/runtime_probe.py @@ -0,0 +1,441 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Runtime URL normalization and legacy API conformance probing.""" + +from __future__ import annotations + +from typing import Any +from urllib.parse import urlparse + +import requests + + +def _make_criterion( + criterion_id: str, + description: str, + passed: bool, + *, + required: bool = True, + details: str | None = None, + expected: Any | None = None, + actual: Any | None = None, +) -> dict[str, Any]: + """Create a standard criterion result payload.""" + criterion: dict[str, Any] = { + "id": criterion_id, + "description": description, + "passed": passed, + "required": required, + } + if details is not None: + criterion["details"] = details + if expected is not None: + criterion["expected"] = expected + if actual is not None: + criterion["actual"] = actual + return criterion + + +def _normalize_runtime_url(base_url: str) -> str: + """Normalize and validate a runtime target URL.""" + target = base_url.strip() + if not target: + raise ValueError("Runtime URL cannot be empty") + + if "://" not in target: + target = f"http://{target}" + + try: + parsed = urlparse(target) + hostname = parsed.hostname + _ = parsed.port + except ValueError as exc: + raise ValueError(f"Invalid runtime URL: {base_url}") from exc + if ( + parsed.scheme.lower() not in {"http", "https"} + or not parsed.netloc + or not hostname + ): + raise ValueError(f"Invalid runtime URL: {base_url}") + if parsed.username is not None or parsed.password is not None: + raise ValueError("Runtime URL must not contain credentials") + if parsed.query or parsed.fragment: + raise ValueError("Runtime URL must not contain a query string or fragment") + + return target.rstrip("/") + + +def _runtime_standard_profile(api_version: str) -> str: + """Resolve the runtime standard profile for an API version.""" + if api_version.startswith("1."): + return "openenv-http/1.x" + return "openenv-http/unknown" + + +def _build_summary(criteria: list[dict[str, Any]]) -> dict[str, Any]: + """Build a compact pass/fail summary for a criteria list.""" + total_count = len(criteria) + passed_count = sum(1 for criterion in criteria if criterion.get("passed", False)) + failed_criteria = [ + criterion.get("id", "unknown") + for criterion in criteria + if not criterion.get("passed", False) + ] + required_criteria = [ + criterion for criterion in criteria if criterion.get("required", True) + ] + required_total_count = len(required_criteria) + required_passed_count = sum( + 1 for criterion in required_criteria if criterion.get("passed", False) + ) + + return { + "passed_count": passed_count, + "total_count": total_count, + "failed_criteria": failed_criteria, + "required_passed_count": required_passed_count, + "required_total_count": required_total_count, + } + + +def validate_running_environment( + base_url: str, timeout_s: float = 5.0 +) -> dict[str, Any]: + """ + Validate a running OpenEnv server against runtime API standards. + + The returned JSON report contains an overall pass/fail result and + per-criterion outcomes that can be consumed in CI. + """ + normalized_url = _normalize_runtime_url(base_url) + criteria: list[dict[str, Any]] = [] + + report: dict[str, Any] = { + "target": normalized_url, + "validation_type": "running_environment", + "standard_version": "unknown", + "standard_profile": "openenv-http/unknown", + "mode": "unknown", + "passed": False, + "summary": {}, + "criteria": criteria, + } + + openapi_paths: dict[str, Any] = {} + api_version = "unknown" + + # Criterion: OpenAPI endpoint reachable with a declared version. + try: + openapi_response = requests.get( + f"{normalized_url}/openapi.json", + timeout=timeout_s, + allow_redirects=False, + ) + except requests.RequestException as exc: + criteria.append( + _make_criterion( + "openapi_version_available", + "GET /openapi.json returns OpenAPI info.version", + False, + details=f"Request failed: {type(exc).__name__}: {exc}", + expected={"status_code": 200, "info.version": "string"}, + ) + ) + else: + try: + openapi_json = openapi_response.json() + except ValueError: + openapi_json = None + + openapi_ok = ( + openapi_response.status_code == 200 + and isinstance(openapi_json, dict) + and isinstance(openapi_json.get("info"), dict) + and isinstance(openapi_json["info"].get("version"), str) + ) + + if openapi_ok: + api_version = str(openapi_json["info"]["version"]) + openapi_paths = openapi_json.get("paths", {}) + criteria.append( + _make_criterion( + "openapi_version_available", + "GET /openapi.json returns OpenAPI info.version", + True, + expected={"status_code": 200, "info.version": "string"}, + actual={ + "status_code": openapi_response.status_code, + "info.version": api_version, + }, + ) + ) + else: + criteria.append( + _make_criterion( + "openapi_version_available", + "GET /openapi.json returns OpenAPI info.version", + False, + details="Response missing required OpenAPI info.version field", + expected={"status_code": 200, "info.version": "string"}, + actual={ + "status_code": openapi_response.status_code, + "body_type": ( + type(openapi_json).__name__ + if openapi_json is not None + else "non_json" + ), + }, + ) + ) + + report["standard_version"] = api_version + report["standard_profile"] = _runtime_standard_profile(api_version) + + # Criterion: Health endpoint. + try: + health_response = requests.get( + f"{normalized_url}/health", + timeout=timeout_s, + allow_redirects=False, + ) + except requests.RequestException as exc: + criteria.append( + _make_criterion( + "health_endpoint", + "GET /health returns healthy status", + False, + details=f"Request failed: {type(exc).__name__}: {exc}", + expected={"status_code": 200, "status": "healthy"}, + ) + ) + else: + try: + health_json = health_response.json() + except ValueError: + health_json = None + + health_ok = ( + health_response.status_code == 200 + and isinstance(health_json, dict) + and health_json.get("status") == "healthy" + ) + criteria.append( + _make_criterion( + "health_endpoint", + "GET /health returns healthy status", + health_ok, + expected={"status_code": 200, "status": "healthy"}, + actual={ + "status_code": health_response.status_code, + "status": ( + health_json.get("status") + if isinstance(health_json, dict) + else None + ), + }, + ) + ) + + # Criterion: Metadata endpoint has required fields. + try: + metadata_response = requests.get( + f"{normalized_url}/metadata", + timeout=timeout_s, + allow_redirects=False, + ) + except requests.RequestException as exc: + criteria.append( + _make_criterion( + "metadata_endpoint", + "GET /metadata returns name and description", + False, + details=f"Request failed: {type(exc).__name__}: {exc}", + expected={"status_code": 200, "fields": ["name", "description"]}, + ) + ) + else: + try: + metadata_json = metadata_response.json() + except ValueError: + metadata_json = None + + metadata_ok = ( + metadata_response.status_code == 200 + and isinstance(metadata_json, dict) + and isinstance(metadata_json.get("name"), str) + and isinstance(metadata_json.get("description"), str) + ) + criteria.append( + _make_criterion( + "metadata_endpoint", + "GET /metadata returns name and description", + metadata_ok, + expected={"status_code": 200, "fields": ["name", "description"]}, + actual={ + "status_code": metadata_response.status_code, + "name": ( + metadata_json.get("name") + if isinstance(metadata_json, dict) + else None + ), + "description": ( + metadata_json.get("description") + if isinstance(metadata_json, dict) + else None + ), + }, + ) + ) + + # Criterion: Schema endpoint returns action/observation/state. + try: + schema_response = requests.get( + f"{normalized_url}/schema", + timeout=timeout_s, + allow_redirects=False, + ) + except requests.RequestException as exc: + criteria.append( + _make_criterion( + "schema_endpoint", + "GET /schema returns action, observation, and state schemas", + False, + details=f"Request failed: {type(exc).__name__}: {exc}", + expected={ + "status_code": 200, + "fields": ["action", "observation", "state"], + }, + ) + ) + else: + try: + schema_json = schema_response.json() + except ValueError: + schema_json = None + + schema_ok = ( + schema_response.status_code == 200 + and isinstance(schema_json, dict) + and isinstance(schema_json.get("action"), dict) + and isinstance(schema_json.get("observation"), dict) + and isinstance(schema_json.get("state"), dict) + ) + criteria.append( + _make_criterion( + "schema_endpoint", + "GET /schema returns action, observation, and state schemas", + schema_ok, + expected={ + "status_code": 200, + "fields": ["action", "observation", "state"], + }, + actual={ + "status_code": schema_response.status_code, + "has_action": ( + isinstance(schema_json.get("action"), dict) + if isinstance(schema_json, dict) + else False + ), + "has_observation": ( + isinstance(schema_json.get("observation"), dict) + if isinstance(schema_json, dict) + else False + ), + "has_state": ( + isinstance(schema_json.get("state"), dict) + if isinstance(schema_json, dict) + else False + ), + }, + ) + ) + + # Criterion: MCP endpoint is reachable. + try: + mcp_response = requests.post( + f"{normalized_url}/mcp", + json={}, + timeout=timeout_s, + allow_redirects=False, + ) + except requests.RequestException as exc: + criteria.append( + _make_criterion( + "mcp_endpoint", + "POST /mcp is reachable and returns JSON-RPC payload", + False, + details=f"Request failed: {type(exc).__name__}: {exc}", + expected={"status_code": 200, "jsonrpc": "2.0"}, + ) + ) + else: + try: + mcp_json = mcp_response.json() + except ValueError: + mcp_json = None + + mcp_ok = ( + mcp_response.status_code == 200 + and isinstance(mcp_json, dict) + and mcp_json.get("jsonrpc") == "2.0" + ) + criteria.append( + _make_criterion( + "mcp_endpoint", + "POST /mcp is reachable and returns JSON-RPC payload", + mcp_ok, + expected={"status_code": 200, "jsonrpc": "2.0"}, + actual={ + "status_code": mcp_response.status_code, + "jsonrpc": ( + mcp_json.get("jsonrpc") if isinstance(mcp_json, dict) else None + ), + }, + ) + ) + + # Criterion: mode endpoint contract consistency via OpenAPI paths. + if isinstance(openapi_paths, dict) and openapi_paths: + has_reset = "/reset" in openapi_paths + has_step = "/step" in openapi_paths + has_state = "/state" in openapi_paths + + if has_reset: + report["mode"] = "simulation" + mode_ok = has_step and has_state + expected_paths = {"/reset": True, "/step": True, "/state": True} + else: + report["mode"] = "production" + mode_ok = not has_step and not has_state + expected_paths = {"/reset": False, "/step": False, "/state": False} + + criteria.append( + _make_criterion( + "mode_endpoint_consistency", + "OpenAPI endpoint set matches OpenEnv mode contract", + mode_ok, + expected=expected_paths, + actual={ + "/reset": has_reset, + "/step": has_step, + "/state": has_state, + }, + ) + ) + else: + criteria.append( + _make_criterion( + "mode_endpoint_consistency", + "OpenAPI endpoint set matches OpenEnv mode contract", + False, + details="Cannot determine mode without OpenAPI paths", + expected={"openapi.paths": "present"}, + actual={"openapi.paths": "missing"}, + ) + ) + + report["passed"] = all( + criterion["passed"] for criterion in criteria if criterion.get("required", True) + ) + report["summary"] = _build_summary(criteria) + return report diff --git a/src/openenv/validation/specs/__init__.py b/src/openenv/validation/specs/__init__.py index 98e3ac8fa..dd385ca79 100644 --- a/src/openenv/validation/specs/__init__.py +++ b/src/openenv/validation/specs/__init__.py @@ -26,7 +26,7 @@ ValidationSubject, ) from .harbor import find_harbor_verifier_script, load_harbor_requirements -from .openenv import OpenEnvSpecAdapter +from .openenv import OpenEnvSpecAdapter, runtime_openenv_spec_load DEFAULT_SPEC_REGISTRY = ValidationSpecRegistry((OpenEnvSpecAdapter(),)) @@ -57,4 +57,5 @@ "ValidationSubject", "find_harbor_verifier_script", "load_harbor_requirements", + "runtime_openenv_spec_load", ] diff --git a/src/openenv/validation/specs/openenv.py b/src/openenv/validation/specs/openenv.py index af5730c9a..f1b12a317 100644 --- a/src/openenv/validation/specs/openenv.py +++ b/src/openenv/validation/specs/openenv.py @@ -15,6 +15,8 @@ AdapterIdentity, DetectionMode, ExecutionModel, + RequirementsLoad, + RequirementsState, SpecIdentity, SpecLoad, SpecLoadState, @@ -138,3 +140,15 @@ def inspect(self, root: Path) -> SpecLoad: document_digest=f"sha256:{hashlib.sha256(document).hexdigest()}", ) return SpecLoad(state=SpecLoadState.LOADED, subject=subject) + + +def runtime_openenv_spec_load() -> SpecLoad: + """Describe an explicitly supplied OpenEnv runtime without source files.""" + adapter = OpenEnvSpecAdapter() + subject = ValidationSubject( + spec=adapter._identity(None), + signature_path=None, + detection_mode=DetectionMode.RUNTIME, + requirements=RequirementsLoad(state=RequirementsState.ABSENT), + ) + return SpecLoad(state=SpecLoadState.LOADED, subject=subject) diff --git a/tests/test_cli/test_validate.py b/tests/test_cli/test_validate.py index 6078f949f..cf710d4e1 100644 --- a/tests/test_cli/test_validate.py +++ b/tests/test_cli/test_validate.py @@ -59,7 +59,8 @@ def _write_minimal_valid_env( def test_validate_running_environment_success() -> None: """Runtime validator returns passing criteria for a conforming server.""" - def _fake_get(url: str, timeout: float) -> _MockResponse: + def _fake_get(url: str, timeout: float, *, allow_redirects: bool) -> _MockResponse: + assert allow_redirects is False if url.endswith("/openapi.json"): return _MockResponse( 200, @@ -87,7 +88,10 @@ def _fake_get(url: str, timeout: float) -> _MockResponse: ) raise AssertionError(f"Unexpected GET url: {url}") - def _fake_post(url: str, json: dict, timeout: float) -> _MockResponse: + def _fake_post( + url: str, json: dict, timeout: float, *, allow_redirects: bool + ) -> _MockResponse: + assert allow_redirects is False if url.endswith("/mcp"): return _MockResponse( 200, @@ -115,7 +119,8 @@ def _fake_post(url: str, json: dict, timeout: float) -> _MockResponse: def test_validate_running_environment_failure() -> None: """Runtime validator marks report as failed when criteria fail.""" - def _fake_get(url: str, timeout: float) -> _MockResponse: + def _fake_get(url: str, timeout: float, *, allow_redirects: bool) -> _MockResponse: + assert allow_redirects is False if url.endswith("/openapi.json"): return _MockResponse( 200, @@ -140,7 +145,10 @@ def _fake_get(url: str, timeout: float) -> _MockResponse: ) raise AssertionError(f"Unexpected GET url: {url}") - def _fake_post(url: str, json: dict, timeout: float) -> _MockResponse: + def _fake_post( + url: str, json: dict, timeout: float, *, allow_redirects: bool + ) -> _MockResponse: + assert allow_redirects is False if url.endswith("/mcp"): return _MockResponse( 200, diff --git a/tests/test_validation/test_local.py b/tests/test_validation/test_local.py new file mode 100644 index 000000000..4a9948b98 --- /dev/null +++ b/tests/test_validation/test_local.py @@ -0,0 +1,257 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Tests for local validation profiles and runtime launch helpers.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from unittest.mock import patch + +import pytest +from openenv.validation.local import ( + _runtime_environment, + _server_command, + run_local_validation, +) +from openenv.validation.models import ( + CheckOutcome, + ValidationCapability, + ValidationCheck, + ValidationPolicy, + ValidationProfile, + ValidationSeverity, +) +from openenv.validation.specs import ( + AdapterIdentity, + DetectionMode, + ExecutionModel, + RequirementsLoad, + RequirementsState, + SpecIdentity, + SpecLoad, + SpecLoadState, + ValidationRequirements, + ValidationSpecRegistry, + ValidationSubject, +) + +from ._helpers import write_valid_env + + +class _OneShotAdapter: + spec_id = "one-shot-local" + adapter_id = "one-shot-local-adapter" + adapter_version = "1" + execution_model = ExecutionModel.ONE_SHOT + signature_files = ("task.oneshot",) + + def detect(self, root: Path) -> bool: + return (root / "task.oneshot").is_file() + + def inspect(self, root: Path) -> SpecLoad: + identity = SpecIdentity( + spec_id=self.spec_id, + spec_version="1", + adapter=AdapterIdentity(self.adapter_id, self.adapter_version), + execution_model=self.execution_model, + ) + if not self.detect(root): + return SpecLoad(state=SpecLoadState.ABSENT, identity=identity) + return SpecLoad( + state=SpecLoadState.LOADED, + subject=ValidationSubject( + spec=identity, + signature_path="task.oneshot", + detection_mode=DetectionMode.AUTO, + requirements=RequirementsLoad( + state=RequirementsState.LOADED, + requirements=ValidationRequirements(), + ), + ), + ) + + +def test_static_profile_skips_absent_requirements_without_certifying( + tmp_path: Path, +) -> None: + env_dir = tmp_path / "test_env" + write_valid_env(env_dir) + + report = run_local_validation(env_dir, profile=ValidationProfile.STATIC) + payload = report.to_dict() + + requirements_result = next( + result + for result in payload["criteria"] + if result["id"] == "source.validation_requirements" + ) + assert requirements_result["status"] == "skip" + assert payload["passed"] is True + assert payload["certified"] is False + assert payload["certification_eligible"] is False + assert payload["policy_version"] + assert payload["spec"]["id"] == "openenv" + assert payload["spec"]["execution_model"] == "served" + assert "repo_sha" in payload + + +def test_present_unsupported_requirements_are_blocking(tmp_path: Path) -> None: + env_dir = tmp_path / "test_env" + write_valid_env(env_dir) + (env_dir / "task.toml").write_text('schema_version = "2.0"\n') + + report = run_local_validation(env_dir, profile=ValidationProfile.STATIC) + result = next( + result + for result in report.results + if result.criterion_id == "source.validation_requirements" + ) + + assert result.status.value == "fail" + assert result.severity is ValidationSeverity.BLOCKING + assert report.passed is False + + +def test_unsupported_spec_retains_top_level_provenance(tmp_path: Path) -> None: + env_dir = tmp_path / "test_env" + write_valid_env(env_dir) + (env_dir / "openenv.yaml").write_text("spec_version: 2\n") + + report = run_local_validation(env_dir, profile=ValidationProfile.STATIC) + payload = report.to_dict() + + assert report.spec is None + assert report.spec_identity is not None + assert payload["spec"] == { + "adapter": {"id": "openenv-yaml", "version": "1"}, + "execution_model": "served", + "id": "openenv", + "version": "2", + } + + +def test_one_shot_spec_is_not_launched_as_an_openenv_server(tmp_path: Path) -> None: + (tmp_path / "task.oneshot").write_text("signature") + runtime_check = ValidationCheck( + criterion_id="spec.one_shot.replay", + requirement="Replay the one-shot package", + capabilities=frozenset({ValidationCapability.RUNTIME}), + severity=ValidationSeverity.BLOCKING, + timeout_s=1.0, + evaluator=lambda _context: CheckOutcome.pass_(), + ) + policy = ValidationPolicy( + version="one-shot-local-v1", + supported_subjects=frozenset({("one-shot-local", ExecutionModel.ONE_SHOT)}), + checks=(runtime_check,), + ) + + with patch("openenv.validation.local._launch_environment") as launch: + report = run_local_validation( + tmp_path, + profile=ValidationProfile.RUNTIME, + spec_registry=ValidationSpecRegistry((_OneShotAdapter(),)), + policy=policy, + ) + + launch.assert_not_called() + assert report.spec is not None + assert report.spec.spec.execution_model is ExecutionModel.ONE_SHOT + assert report.results[0].status.value == "skip" + assert report.results[0].evidence["unavailable_reasons"] == { + "runtime": "unsupported_execution_model:one_shot" + } + + +def test_static_validation_never_reads_symlinked_source_files(tmp_path: Path) -> None: + env_dir = tmp_path / "test_env" + env_dir.mkdir() + outside = tmp_path / "outside.yaml" + outside.write_text( + "spec_version: 1\n" + "name: TOPSECRET_VALUE\n" + "runtime: fastapi\n" + "app: server.app:app\n" + "port: 8000\n" + ) + try: + (env_dir / "openenv.yaml").symlink_to(outside) + except OSError: + pytest.skip("symbolic links are unavailable on this platform") + + report = run_local_validation(env_dir, profile=ValidationProfile.STATIC) + serialized = json.dumps(report.to_dict()) + + assert report.passed is False + assert "TOPSECRET_VALUE" not in serialized + openenv_manifest = next( + result + for result in report.results + if result.criterion_id == "source.openenv_manifest" + ) + assert openenv_manifest.status.value == "fail" + + +def test_static_checks_reject_symlinked_project_files(tmp_path: Path) -> None: + env_dir = tmp_path / "test_env" + write_valid_env(env_dir) + outside = tmp_path / "outside" + outside.mkdir() + replacements = { + env_dir / "pyproject.toml": outside / "pyproject.toml", + env_dir / "uv.lock": outside / "uv.lock", + env_dir / "server" / "app.py": outside / "app.py", + env_dir / "server" / "Dockerfile": outside / "Dockerfile", + } + for target, source in replacements.items(): + source.write_text(target.read_text()) + target.unlink() + try: + target.symlink_to(source) + except OSError: + pytest.skip("symbolic links are unavailable on this platform") + + report = run_local_validation(env_dir, profile=ValidationProfile.STATIC) + by_id = {result.criterion_id: result for result in report.results} + + assert by_id["source.project_layout"].status.value == "fail" + assert by_id["source.dependencies"].status.value == "fail" + assert by_id["source.lockfile"].status.value == "fail" + assert by_id["source.schema_sources"].status.value == "fail" + assert by_id["source.dockerfile"].status.value == "fail" + + +def test_local_runtime_does_not_forward_ambient_credentials(tmp_path: Path) -> None: + ambient = { + "HOME": str(tmp_path), + "PATH": os.environ.get("PATH", ""), + "HF_TOKEN": "hf-secret", + "OPENAI_API_KEY": "openai-secret", + "AWS_SECRET_ACCESS_KEY": "aws-secret", + "PYTHONPATH": "extra-python-path", + } + + with patch.dict(os.environ, ambient, clear=True): + process_env = _runtime_environment(tmp_path) + + assert process_env["HOME"] == str(tmp_path) + assert process_env["OPENENV_VALIDATION"] == "1" + assert "extra-python-path" in process_env["PYTHONPATH"] + assert "HF_TOKEN" not in process_env + assert "OPENAI_API_KEY" not in process_env + assert "AWS_SECRET_ACCESS_KEY" not in process_env + + +def test_server_command_honors_manifest_port(tmp_path: Path) -> None: + with patch("openenv.validation.local.shutil.which", return_value=None): + command = _server_command(tmp_path, "server.app:app", 8123) + + assert command[1:4] == ["-m", "uvicorn", "server.app:app"] + assert command[-3:] == ["8123", "--log-level", "warning"] + + +def test_static_profile_rejects_url_in_public_api() -> None: + with pytest.raises(ValueError, match="local source"): + run_local_validation("https://example.com", profile=ValidationProfile.STATIC)