diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3e6bf15..a1ce9ca 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -111,3 +111,25 @@ jobs: - name: An out-of-tree package reaches every extension seam run: make check-extensions + + e2e: + name: Installed CLI e2e + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 + with: + version-file: "pyproject.toml" + python-version: "3.12" + enable-cache: true + - run: make install-ci + - run: make test-e2e E2E_ARGS="--artifacts test-results/e2e" + - name: E2e diagnostics on failure + if: failure() + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: e2e-failure + path: test-results/e2e/ + if-no-files-found: warn + retention-days: 14 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 56f9a64..24975f8 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -118,7 +118,8 @@ making a single model call. ### Reproducing CI and coverage -Run `make validate-ci` for the native checks and clean distribution validation. +Run `make validate-ci` for native checks, clean distribution validation, and +installed CLI e2e tests against that same validated wheel. `make validate-native` runs static checks, dependency license policy, coverage, the offline self-check, and installed extension checks without building the distribution archives. CI runs the same test suite across supported Python @@ -130,6 +131,11 @@ example `make coverage PYTEST_ARGS="--junitxml=junit.xml"`. See [Local and CI validation](docs/engineering/ci.md) for bundle membership, report artifacts, and failure diagnosis. +Run `make test-e2e` for the isolated CLI workflow alone, or +`make test-e2e WHEEL_DIR=dist` to use the single existing wheel in `dist/` +without rebuilding. See [Installed CLI e2e tests](docs/engineering/e2e.md) for +setup, offline execution, scenarios, and diagnostics. + ### Choosing models and providers Three providers are built in, each needing only a credential. Pick the shipped diff --git a/MANIFEST.in b/MANIFEST.in index dff59ce..55c873e 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -1,3 +1,4 @@ # The test suite runs from a checkout. setuptools would otherwise ship part of # it in the sdist, without the conftest.py files it needs to run. prune tests +prune tests_e2e diff --git a/Makefile b/Makefile index 3ac8263..34ef38a 100644 --- a/Makefile +++ b/Makefile @@ -26,7 +26,8 @@ UV_VERSION := $(shell sed -n 's/^required-version = "==\([^"]*\)"/\1/p' pyprojec # change would trigger a second repo-wide reformat. PY_SRC ?= src scripts templates PY_TESTS ?= tests conftest.py -PY_ALL := $(PY_SRC) $(PY_TESTS) +PY_E2E ?= tests_e2e +PY_ALL := $(PY_SRC) $(PY_TESTS) $(PY_E2E) # Worker count for the parallel suite. `auto` follows the local core count, # which suits a CPU-bound suite with every LLM call mocked (~2x on a laptop). @@ -60,7 +61,7 @@ COV_FAIL_UNDER ?= 84 .DEFAULT_GOAL := help .PHONY: help bootstrap check-tools check-lock install-dev install-ci install-kernel health-check check-notebooks-clean lint lint-fix format format-check check-all \ - check-hooks check-all-fix validate-native validate-ci test test-fast coverage smoke dist check-dist check-release-tag check-extensions check-doc-links clean-notebooks check-license-headers update-license-headers check-dependency-licenses clean + check-hooks check-all-fix validate-native validate-ci test test-fast test-e2e coverage smoke dist check-dist check-release-tag check-extensions check-doc-links clean-notebooks check-license-headers update-license-headers check-dependency-licenses clean help: ## Show the available targets @@ -128,6 +129,9 @@ check-all-fix: lint-fix format ## Fix everything fixable, then you re-review # PYTEST_ARGS is a passthrough, so a one-off run does not need a new target: # make test PYTEST_ARGS="--durations=25 -k financial" +E2E_ARGS ?= +WHEEL_DIR ?= + PYTEST_ARGS ?= test: ## Run the full test suite (parallel) @@ -142,9 +146,15 @@ coverage: ## Run tests with coverage and enforce the floor --cov-report="xml:$(COV_XML)" \ --cov-fail-under=$(COV_FAIL_UNDER) $(PYTEST_ARGS) +# Keep the installed suite separate: ordinary pytest's root conftest blocks +# loopback, and the e2e runner must never import the editable source tree. +test-e2e: check-tools ## Test an installed wheel through CLI subprocesses and a local HTTP provider + $(UV) run --frozen python scripts/run_e2e.py --uv "$(UV)" $(if $(WHEEL_DIR),--wheel-dir "$(WHEEL_DIR)") $(E2E_ARGS) + validate-native: check-all check-dependency-licenses coverage smoke check-extensions ## Check source and installed extension behavior -validate-ci: validate-native check-dist ## Run the native CI checks, including distribution validation +validate-ci: validate-native check-dist ## Validate source, then test the same clean distribution artifacts + $(MAKE) test-e2e WHEEL_DIR=dist smoke: ## Offline self-check of config, registries, and report builders $(UV) run usersim smoke diff --git a/docs/engineering/ci.md b/docs/engineering/ci.md index 75e72d0..8a0c9c7 100644 --- a/docs/engineering/ci.md +++ b/docs/engineering/ci.md @@ -2,7 +2,7 @@ CI calls repository Make targets so failures can be reproduced locally. After `make install-dev`, run `make validate-ci` to execute the native validation -bundle, including distribution checks. +bundle, including distribution checks and installed CLI tests. | Target | Checks | | --- | --- | @@ -13,13 +13,20 @@ bundle, including distribution checks. | `make check-extensions` | Out-of-tree installed extension behavior. | | `make validate-native` | The five targets above. | | `make check-dist` | Build and validate archives, then check a clean wheel installation. | -| `make validate-ci` | `validate-native` plus `check-dist`. | +| `make test-e2e` | Installed CLI workflow against a loopback provider in isolated environments. | +| `make validate-ci` | `validate-native` and `check-dist`, followed by e2e against the resulting wheel. | The distribution archives are produced by `check-dist`; extension checks build a separate temporary wheel. CI also runs the test suite across Python 3.11 through 3.14; a local bundle checks the active Python environment. Provider health checks run separately and need credentials. +The separate `Installed CLI e2e` job runs the [installed workflow](e2e.md) on +Python 3.12 through `make test-e2e`. It retains failure diagnostics under +`test-results/e2e/` as `e2e-failure` for 14 days. The local bundle passes +`WHEEL_DIR=dist` after archive validation so the e2e test installs that same +wheel without rebuilding. + ## Coverage The Python 3.12 CI matrix job runs `make coverage`; the other Python versions diff --git a/docs/engineering/e2e.md b/docs/engineering/e2e.md new file mode 100644 index 0000000..0511512 --- /dev/null +++ b/docs/engineering/e2e.md @@ -0,0 +1,67 @@ +# Installed CLI end-to-end tests + +Run `make test-e2e` to test a built UserSim wheel through its CLI and real HTTP client. The harness uses synthetic stored inputs and a deterministic OpenAI-compatible provider on loopback; it requires no provider credentials or downloaded persona datasets. + +Setup needs uv, a supported Python, and access to dependency indexes and the tokenizer data URL. The repository's frozen lock supplies runtime dependencies. The test harness installs its own pinned pytest and PyArrow into a second environment. No development dependencies or editable UserSim install enter the runtime environment. + +```bash +make test-e2e +``` + +The equivalent direct command is: + +```bash +uv run --frozen python scripts/run_e2e.py +``` + +Pass an existing wheel to test the original release artifact without rebuilding: + +```bash +uv run --frozen python scripts/run_e2e.py --wheel dist/usersim-VERSION-py3-none-any.whl +# Or require exactly one wheel in the release directory: +uv run --frozen python scripts/run_e2e.py --wheel-dir dist +``` + +`--wheel` and `--wheel-dir` are mutually exclusive and cannot be combined with `--run-only`. An empty or ambiguous wheel directory fails setup. `--uv /path/to/uv` selects the setup executable when it is not the one on `PATH`. + +## Isolation and offline execution + +The runner creates a directory outside the checkout containing `runtime/`, `harness/`, a copied test suite, a fresh home directory, and a tokenizer cache. It verifies that UserSim imports from `runtime/`. After admission, the runner writes empty `.env` and `.env.local` files in its owned work directory on every invocation, including `--run-only`. Admission checks these files and the direct preparation, cache, environment, and readiness entries before any output changes: symlinks, wrong file/directory kinds, and multiply linked writable files are rejected. Normal interpreter symlinks inside a prepared virtualenv remain supported. These stop the CLI's upward dotenv search before it reaches files outside the harness. CLI subprocesses run in fresh case directories beneath that boundary outside the source tree; the tests never import UserSim from the checkout or patch its probes, scorers, storage, or reporting. + +The runtime receives a small environment allowlist and a synthetic local API key. External sockets are rejected by a test-only `sitecustomize` guard; loopback remains available to the fake provider. Hugging Face offline settings, extension-discovery suppression, and `NEMO_TELEMETRY_ENABLED=false` and other telemetry opt-outs are explicit. A subprocess verifies that installed DataDesigner telemetry is disabled. The ordinary suite's network guard and root conftest are not loaded. + +Prepare environments while network access is available, then execute them without installation or downloads: + +```bash +uv run --frozen python scripts/run_e2e.py --work-dir /tmp/usersim-e2e --prepare-only +uv run --frozen python scripts/run_e2e.py --work-dir /tmp/usersim-e2e --run-only +``` + +`--work-dir` must resolve outside the checkout and cannot be an ancestor of it. Paths containing `..` use their canonical locations for this check; symlinked work and artifact paths are rejected. Reuse the same path for both phases; the prepared environment and copied suite are a snapshot of setup. Changes to tests or wheels require another preparation. A new preparation removes the previous `installed.json` receipt before changing the environment, so an interrupted setup cannot advertise the old wheel as a valid prepared runtime. + +## Scenarios and assertions + +| Scenario | Contract checked | +| --- | --- | +| Complete workflow | `simulate → eval → report → select` preserves identities across Parquet reload, obtains real judge scores, writes report HTML, and selects the scored rows. | +| Stored-input resume | A second simulation reuses its run and trajectory IDs, preserves `input_id`, writes no duplicate rows, and makes no provider requests. | +| Tool calling | A weather function call travels through the real HTTP client, tool execution, and persisted transcript. A transient server error is retried. | +| Non-Latin locale | A Japanese input keeps its locale and native-script user text through persistence and evaluation. | +| Guarded tool request | A three-turn health-disclosure scenario forwards explicit `tool_choice`, preserves configured `extra_body`, and merges per-call nested reasoning settings. Its shipped synthetic clinical profile emits the expected placeholder warning. | +| Malformed response | Invalid evaluator JSON produces a nonzero CLI exit and empty judge results rather than successful scores. | +| Failed request | A provider HTTP 400 causes a nonzero CLI exit without published evaluation rows. | +| Network boundary | An attempted external connection fails with the harness's explicit network error. | + +These cases verify client integration and artifact contracts. They do not measure model quality or establish acceptance by real providers. The health-disclosure case uses the shipped placeholder bank to exercise request plumbing and does not establish clinical validity. + +## Results and diagnosis + +Each execution exclusively creates `output/e2e/run-*/` and prints its exact path. Use `--artifacts ROOT` to select another output root; existing attempts and unrelated root files are left alone. A root on a bind mount works because only a new child is created, rather than renaming the mount. The runner prints its external work directory and copies scenario logs, synthetic inputs, recorded provider request bodies, trajectory/evaluation/selection Parquet, and report files into the artifact directory. `junit.xml` records the test result. The exported `installed.json` receipt retains the runtime import and asset locations and adds the selected wheel path and `wheel_sha256` captured during preparation. The digest binds prepared/run-only evidence to the wheel bytes, even if the original path later names a different file. Existing prepared environments without the digest need preparation again to provide this evidence. `installed.json` is the sole installation receipt. Captured requests contain only synthetic data and the local test model names. + +Rejected admission leaves the work and output roots unchanged. Admission checks work/output paths, owned work entries, run-only readiness, and selected wheel inputs that could be removed by work-directory preparation. Safe paths containing `..` are canonicalized after lexical symlink checks. Once admitted, a new attempt directory is created before setup or execution, so even an early failure cannot mix previous scores, JUnit, or receipts into the attempt. The output root can contain earlier results or other caller-owned files; no marker, rotation, or automatic cleanup is required. Remove result directories when you no longer need them. + +`--prepare-only` changes preparation state in the external work directory and creates no result directory. Successful preparation writes `installed.json`; an interrupted preparation removes readiness, so `--run-only` refuses it. Execution copies that receipt into its fresh result directory alongside JUnit and scenario files. CI uploads the output root recursively, including each attempt's diagnostics. + +Read a failed command's `/.log`, then compare `/requests.json` with the expected payload. The work directory retains both environments and uncopied DataDesigner diagnostics for inspection. Setup commands, CLI subprocesses, provider readiness, and the complete pytest process have timeouts. An empty test collection fails. + +The runner does not remove the external work directory. Remove it after diagnosis to reclaim the installed environments and tokenizer cache. diff --git a/scripts/run_e2e.py b/scripts/run_e2e.py new file mode 100644 index 0000000..926a676 --- /dev/null +++ b/scripts/run_e2e.py @@ -0,0 +1,281 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Prepare and run installed-wheel CLI tests outside the source checkout.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shutil +import subprocess +import sys +import tempfile +import uuid +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] + + +def command(args: list[str], *, cwd: Path, env: dict[str, str] | None = None) -> None: + print("+", " ".join(args), flush=True) + subprocess.run(args, cwd=cwd, env=env, check=True, timeout=600) + + +def environment(work: Path) -> dict[str, str]: + # Runtime configuration is deliberately independent of the invoking shell. + env = {name: os.environ[name] for name in ("PATH", "LANG", "LC_ALL", "TMPDIR") if name in os.environ} + env.update( + { + "HOME": str(work / "home"), + "TIKTOKEN_CACHE_DIR": str(work / "tokenizers"), + "USERSIM_DISABLE_EXTENSIONS": "1", + "USERSIM_MOVE_REASONING_EFFORT": "off", + "DO_NOT_TRACK": "1", + "NEMO_TELEMETRY_ENABLED": "false", + "HF_HUB_OFFLINE": "1", + "HF_DATASETS_OFFLINE": "1", + "OTEL_SDK_DISABLED": "true", + "E2E_API_KEY": "synthetic-local-key", + "PYTHONNOUSERSITE": "1", + } + ) + Path(env["HOME"]).mkdir(exist_ok=True) + return env + + +def selected_wheel(args: argparse.Namespace, parser: argparse.ArgumentParser) -> Path | None: + if args.wheel_dir: + wheels = sorted(args.wheel_dir.resolve().glob("*.whl")) + if len(wheels) != 1: + parser.error(f"--wheel-dir expects exactly one wheel, found {len(wheels)} in {args.wheel_dir}") + wheel = wheels[0].resolve(strict=True) + elif args.wheel: + wheel = args.wheel.resolve(strict=True) + else: + return None + if not wheel.is_file(): + parser.error(f"selected wheel must be a regular file: {wheel}") + return wheel + + +def archive(work: Path, args: argparse.Namespace, parser: argparse.ArgumentParser) -> Path: + wheel = selected_wheel(args, parser) + if wheel is not None: + return wheel + dist = work / "dist" + dist.mkdir(exist_ok=True) + command([args.uv, "build", "--wheel", "--out-dir", str(dist)], cwd=ROOT) + wheels = list(dist.glob("*.whl")) + if len(wheels) != 1: + parser.error(f"expected exactly one wheel in {dist}, found {len(wheels)}") + return wheels[0] + + +def record_install(work: Path, wheel: Path, env: dict[str, str]) -> None: + runtime, suite = work / "runtime", work / "suite" + python = runtime / "bin/python" + # tiktoken downloads encoding data lazily. Download once during setup, + # before the socket guard and container's network isolation take effect. + command([str(python), "-c", "import tiktoken; tiktoken.get_encoding('cl100k_base')"], cwd=work, env=env) + if suite.exists(): + shutil.rmtree(suite) + shutil.copytree(ROOT / "tests_e2e", suite) + metadata = subprocess.check_output( + [ + str(python), + "-c", + "import json, usersim; from data_designer.engine.models.telemetry import TELEMETRY_ENABLED; assert not TELEMETRY_ENABLED; from usersim.engine.core._assets import packaged_assets_dir; print(json.dumps({'module': usersim.__file__, 'assets': str(packaged_assets_dir())}))", + ], + cwd=work, + env=env, + text=True, + timeout=30, + ) + installed = json.loads(metadata.strip().splitlines()[-1]) + if not Path(installed["module"]).resolve().is_relative_to(runtime.resolve()): + raise RuntimeError(f"UserSim did not import from the runtime environment: {installed}") + with wheel.open("rb") as stream: + installed.update(wheel=str(wheel), wheel_sha256=hashlib.file_digest(stream, "sha256").hexdigest()) + (work / "installed.json").write_text(json.dumps(installed), encoding="utf-8") + + +def prepare(work: Path, args: argparse.Namespace, parser: argparse.ArgumentParser) -> None: + (work / "installed.json").unlink(missing_ok=True) + runtime, harness = work / "runtime", work / "harness" + python = runtime / "bin/python" + env = environment(work) + wheel = archive(work, args, parser) + # Export locked runtime dependencies without the editable project or dev tools. + requirements = work / "runtime-requirements.txt" + command( + [args.uv, "export", "--frozen", "--no-dev", "--no-emit-project", "--quiet", "--output-file", str(requirements)], + cwd=ROOT, + ) + for target in (runtime, harness): + if target.exists(): + shutil.rmtree(target) + command([args.uv, "venv", "--python", sys.executable, str(target)], cwd=work) + command([args.uv, "pip", "sync", "--python", str(python), str(requirements)], cwd=work) + command([args.uv, "pip", "install", "--python", str(python), "--no-deps", str(wheel)], cwd=work) + command( + [args.uv, "pip", "install", "--python", str(harness / "bin/python"), "pytest==9.0.2", "pyarrow==24.0.0"], + cwd=work, + ) + record_install(work, wheel, env) + + +def admit_paths(work: Path, artifacts: Path) -> tuple[Path, Path]: + for path in (work, artifacts): + if any(part.is_symlink() for part in (path, *path.parents)): + raise ValueError("e2e work and artifact paths must not use symlinks") + work, artifacts, source = work.resolve(), artifacts.resolve(), ROOT.resolve() + if work == source or source in work.parents or work in source.parents: + raise ValueError("e2e work directory must not overlap the checkout root or its ancestors") + if artifacts == source or artifacts in source.parents: + raise ValueError("artifact destination must not overlap the checkout root or its ancestors") + if artifacts == work or work in artifacts.parents or artifacts in work.parents: + raise ValueError("artifact destination must not overlap the e2e work directory") + directories = [ + work, + artifacts, + *(work / name for name in ("home", "tokenizers", "dist", "runtime", "harness", "suite", "cases")), + ] + files = [work / name for name in (".env", ".env.local", "runtime-requirements.txt", "installed.json")] + for path in [*directories, *files]: + if path.is_symlink(): + raise ValueError(f"owned e2e work entry must not use symlinks: {path}") + if path.exists(): + if path in directories and not path.is_dir(): + raise ValueError(f"owned e2e work entry must be a directory: {path}") + if path in files and (not path.is_file() or path.stat().st_nlink != 1): + raise ValueError(f"owned e2e work entry must be a singly linked regular file: {path}") + return work, artifacts + + +def validate_run_inputs(work: Path, args: argparse.Namespace, parser: argparse.ArgumentParser) -> Path | None: + wheel = selected_wheel(args, parser) + if wheel is not None: + targets = (work / name for name in ("runtime", "harness", "suite", "cases")) + overwritten = (work / name for name in (".env", ".env.local", "runtime-requirements.txt", "installed.json")) + if wheel in overwritten: + raise ValueError(f"selected wheel would be overwritten by e2e setup: {wheel}") + for target in targets: + if wheel.is_relative_to(target): + raise ValueError(f"selected wheel would be removed by e2e setup or execution: {wheel}") + if args.run_only: + receipt = work / "installed.json" + if not receipt.is_file(): + parser.error("--run-only requires a prepared --work-dir") + try: + installed = json.loads(receipt.read_text(encoding="utf-8")) + except (OSError, ValueError): + parser.error("--run-only requires a valid installed.json readiness receipt") + if not isinstance(installed, dict) or not isinstance(installed.get("assets"), str) or not installed["assets"]: + parser.error("--run-only readiness receipt requires an installed assets path") + return wheel + + +def attempt_directory(work: Path, root: Path) -> Path: + work, root = admit_paths(work, root) + root.mkdir(parents=True, exist_ok=True) + artifacts = Path(tempfile.mkdtemp(prefix="run-", dir=root)) + print(f"E2e attempt artifacts: {artifacts}", flush=True) + return artifacts + + +def run_suite(work: Path, artifacts: Path, env: dict[str, str]) -> int: + runtime, harness, suite = work / "runtime", work / "harness", work / "suite" + shutil.copyfile(work / "installed.json", artifacts / "installed.json") + cases = work / "cases" + if cases.exists(): + shutil.rmtree(cases) + cases.mkdir() + env.update( + { + "E2E_RUNTIME": str(runtime), + "E2E_ASSETS": json.loads((work / "installed.json").read_text())["assets"], + "E2E_ARTIFACTS": str(cases), + "PYTHONPATH": str(suite / "network_guard"), + "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1", + } + ) + try: + result = subprocess.run( + [ + str(harness / "bin/python"), + "-m", + "pytest", + "-c", + str(suite / "pytest.ini"), + str(suite), + "-v", + "--junitxml", + str(artifacts / "junit.xml"), + ], + cwd=work, + env=env, + check=False, + timeout=900, + ) + except subprocess.TimeoutExpired: + print("Installed CLI suite exceeded 900 seconds", file=sys.stderr) + returncode = 1 + else: + returncode = result.returncode + finally: + shutil.copytree(cases, artifacts, dirs_exist_ok=True) + print(f"E2e work directory: {work}\nArtifacts: {artifacts}") + return returncode + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--uv", default=os.environ.get("UV", "uv"), help="uv executable used for setup.") + wheel_source = parser.add_mutually_exclusive_group() + wheel_source.add_argument("--wheel", type=Path, help="Use an existing wheel instead of building one.") + wheel_source.add_argument( + "--wheel-dir", type=Path, help="Use the single existing wheel in this directory; never build." + ) + parser.add_argument( + "--work-dir", type=Path, help="Persistent external work directory (default: a new temporary one)." + ) + parser.add_argument( + "--artifacts", type=Path, default=ROOT / "output/e2e", help="Root for fresh per-attempt output." + ) + mode = parser.add_mutually_exclusive_group() + mode.add_argument( + "--prepare-only", action="store_true", help="Install environments and tokenizer cache; do not run tests." + ) + mode.add_argument( + "--run-only", action="store_true", help="Run prepared environments without installing or downloading." + ) + args = parser.parse_args() + if args.run_only and (args.wheel or args.wheel_dir): + parser.error("--run-only cannot be combined with --wheel or --wheel-dir; prepare that wheel first") + work = ( + args.work_dir.absolute() if args.work_dir else Path(tempfile.gettempdir()) / f"usersim-e2e-{uuid.uuid4().hex}" + ) + artifacts = args.artifacts.absolute() + work, artifacts = admit_paths(work, artifacts) + wheel = validate_run_inputs(work, args, parser) + if wheel is not None: + args.wheel, args.wheel_dir = wheel, None + if not args.prepare_only: + artifacts = attempt_directory(work, artifacts) + work.mkdir(parents=True, exist_ok=True) + # The CLI searches upward from each case cwd; stop at our owned boundary. + for filename in (".env", ".env.local"): + (work / filename).write_text("", encoding="utf-8") + if not args.run_only: + prepare(work, args, parser) + if args.prepare_only: + print(f"Prepared {work}. Run again with --work-dir {work} --run-only.") + return 0 + return run_suite(work, artifacts, environment(work)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/update_license_headers.py b/scripts/update_license_headers.py index d76b0aa..6c5c4ab 100644 --- a/scripts/update_license_headers.py +++ b/scripts/update_license_headers.py @@ -403,7 +403,7 @@ def main( total_processed = total_updated = total_skipped = 0 # This project is one distribution under src/, so the roots are flat. - for folder in [] if args.files else ["src", "tests", "scripts", "templates"]: + for folder in [] if args.files else ["src", "tests", "tests_e2e", "scripts", "templates"]: folder_path = repo_path / folder if not folder_path.exists(): continue diff --git a/tests/test_e2e_harness.py b/tests/test_e2e_harness.py new file mode 100644 index 0000000..f9721a5 --- /dev/null +++ b/tests/test_e2e_harness.py @@ -0,0 +1,467 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Installed-test setup must honor explicit archives and reject ambiguous modes.""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +import sys +from pathlib import Path + +import pytest +import yaml + +ROOT = Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts/run_e2e.py" + + +@pytest.mark.parametrize("flag", ["--wheel", "--wheel-dir"]) +def test_run_only_rejects_archive_selection(flag, tmp_path): + result = subprocess.run( + [sys.executable, str(RUNNER), "--run-only", flag, str(tmp_path / "nonexistent")], + capture_output=True, + text=True, + check=False, + timeout=10, + ) + assert result.returncode == 2 + assert "--run-only cannot be combined" in result.stderr + assert not result.stdout + + +@pytest.mark.parametrize("count", [0, 2]) +def test_archive_directory_must_have_exactly_one_wheel(count, tmp_path): + wheel_dir = tmp_path / "wheels" + wheel_dir.mkdir() + for index in range(count): + (wheel_dir / f"example-{index}.whl").touch() + result = subprocess.run( + [ + sys.executable, + str(RUNNER), + "--wheel-dir", + str(wheel_dir), + "--work-dir", + str(tmp_path / "work"), + "--artifacts", + str(tmp_path / "artifacts"), + "--uv", + "must-not-run-uv", + ], + capture_output=True, + text=True, + check=False, + timeout=10, + ) + assert result.returncode == 2 + assert f"expects exactly one wheel, found {count}" in result.stderr + assert not result.stdout, "invalid archive selection must fail before setup or build commands" + + +@pytest.mark.parametrize("status", [0, 9]) +def test_make_uses_selected_wheel_and_preserves_failure(status, tmp_path, command_test_env): + uv = tmp_path / "fake-uv" + recorded = tmp_path / "arguments.json" + version = re.search(r'required-version = "==([^"]+)"', (ROOT / "pyproject.toml").read_text()).group(1) + uv.write_text( + f"#!{sys.executable}\n" + "import json, os, pathlib, sys\n" + f"if sys.argv[1:] == ['--version']: print('uv {version}'); sys.exit(0)\n" + "pathlib.Path(os.environ['E2E_RECORDED_ARGS']).write_text(json.dumps(sys.argv[1:]))\n" + f"sys.exit({status})\n" + ) + uv.chmod(0o755) + wheels = tmp_path / "release wheels" + result = subprocess.run( + ["make", "test-e2e", f"UV={uv}", f"WHEEL_DIR={wheels}", "E2E_ARGS=--artifacts output/e2e-custom"], + cwd=ROOT, + env=command_test_env(E2E_RECORDED_ARGS=str(recorded)), + capture_output=True, + text=True, + check=False, + timeout=10, + ) + assert (result.returncode == 0) is (status == 0), result.stderr + arguments = json.loads(recorded.read_text()) + assert arguments[:4] == ["run", "--frozen", "python", "scripts/run_e2e.py"] + assert arguments[arguments.index("--wheel-dir") + 1] == str(wheels) + assert arguments[arguments.index("--uv") + 1] == str(uv) + assert arguments[-2:] == ["--artifacts", "output/e2e-custom"] + + +def test_ci_runs_installed_e2e_and_retains_diagnostics(): + workflow = yaml.safe_load((ROOT / ".github/workflows/ci.yml").read_text()) + job = workflow["jobs"]["e2e"] + assert job["timeout-minutes"] <= 30 + commands = [step["run"] for step in job["steps"] if "run" in step] + assert commands == ["make install-ci", 'make test-e2e E2E_ARGS="--artifacts test-results/e2e"'] + upload = next(step for step in job["steps"] if step.get("uses", "").startswith("actions/upload-artifact@")) + assert upload["if"] == "failure()" + assert upload["with"]["path"] == "test-results/e2e/" + assert not any(step.get("continue-on-error") for step in job["steps"]) + + +@pytest.mark.parametrize("filename,run_only", [(".env", False), (".env.local", True)]) +def test_external_work_directory_stops_ancestor_dotenv_loading(filename, run_only, tmp_path, monkeypatch): + import importlib.util + + from usersim.cli._env import load_local_env + + spec = importlib.util.spec_from_file_location("e2e_dotenv", RUNNER) + runner = importlib.util.module_from_spec(spec) + spec.loader.exec_module(runner) + ancestor = tmp_path / "external" + work = ancestor / "persistent-work" + work.mkdir(parents=True) + (ancestor / filename).write_text("USERSIM_E2E_ANCESTOR_CANARY=unexpected\n") + (work / "installed.json").write_text(json.dumps({"assets": "unused"})) + monkeypatch.delenv("USERSIM_E2E_ANCESTOR_CANARY", raising=False) + monkeypatch.delenv("E2E_ARGS", raising=False) + monkeypatch.delenv("WHEEL_DIR", raising=False) + checked = [] + + def check_cli_directories(work): + for case in ("workflow", "tool-choice"): + cwd = work / "cases" / case + cwd.mkdir(parents=True, exist_ok=True) + monkeypatch.chdir(cwd) + load_local_env() + assert "USERSIM_E2E_ANCESTOR_CANARY" not in os.environ + checked.append(case) + + def prepare(work, args, parser): + check_cli_directories(work) + + def run_suite(work, artifacts, env): + check_cli_directories(work) + return 0 + + monkeypatch.setattr(runner, "prepare", prepare) + monkeypatch.setattr(runner, "run_suite", run_suite) + argv = [str(RUNNER), "--work-dir", str(work), "--artifacts", str(tmp_path / "artifacts")] + if run_only: + argv.append("--run-only") + monkeypatch.setattr(sys, "argv", argv) + assert runner.main() == 0 + assert checked == ["workflow", "tool-choice"] * (1 if run_only else 2) + + +def test_install_receipt_records_wheel_bytes(tmp_path, monkeypatch): + import hashlib + import importlib.util + + spec = importlib.util.spec_from_file_location("e2e_receipt", RUNNER) + runner = importlib.util.module_from_spec(spec) + spec.loader.exec_module(runner) + wheel = tmp_path / "example.whl" + wheel.write_bytes(b"selected wheel bytes") + installed = {"module": str(tmp_path / "runtime/lib/usersim/__init__.py"), "assets": "packaged-assets"} + monkeypatch.setattr(runner, "command", lambda *args, **kwargs: None) + monkeypatch.setattr(runner.subprocess, "check_output", lambda *args, **kwargs: json.dumps(installed)) + runner.record_install(tmp_path, wheel, {}) + receipt = json.loads((tmp_path / "installed.json").read_text()) + assert receipt["wheel"] == str(wheel) + assert receipt["wheel_sha256"] == hashlib.sha256(wheel.read_bytes()).hexdigest() + assert receipt["module"] == installed["module"] + assert receipt["assets"] == installed["assets"] + + +def test_run_only_exports_original_install_receipt(tmp_path, monkeypatch): + import importlib.util + from types import SimpleNamespace + + spec = importlib.util.spec_from_file_location("e2e_export", RUNNER) + runner = importlib.util.module_from_spec(spec) + spec.loader.exec_module(runner) + work = tmp_path / "work" + artifacts = tmp_path / "artifacts" + work.mkdir() + artifacts.mkdir() + receipt = {"assets": "installed-assets", "wheel": "/old/build.whl", "wheel_sha256": "original-digest"} + (work / "installed.json").write_text(json.dumps(receipt)) + monkeypatch.setattr(runner.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=0)) + assert runner.run_suite(work, artifacts, {}) == 0 + assert json.loads((artifacts / "installed.json").read_text()) == receipt + + +@pytest.fixture +def e2e_runner(monkeypatch): + import importlib.util + + spec = importlib.util.spec_from_file_location("e2e_harness_runner", RUNNER) + runner = importlib.util.module_from_spec(spec) + spec.loader.exec_module(runner) + monkeypatch.delenv("E2E_ARGS", raising=False) + monkeypatch.delenv("WHEEL_DIR", raising=False) + return runner + + +@pytest.fixture +def prepared_case(tmp_path, monkeypatch, e2e_runner): + runner = e2e_runner + work, root = tmp_path / "work", tmp_path / "results" + work.mkdir() + root.mkdir() + (root / "operator-note.txt").write_text("unrelated file") + (work / "installed.json").write_text(json.dumps({"assets": "assets", "wheel_sha256": "original-digest"})) + + def invoke(*flags): + monkeypatch.setattr(sys, "argv", [str(RUNNER), "--work-dir", str(work), "--artifacts", str(root), *flags]) + return runner.main() + + return runner, work, root, invoke + + +@pytest.mark.parametrize("failure", ["nonzero", "timeout"]) +def test_fresh_attempts_isolate_failure_from_previous_success(failure, prepared_case, monkeypatch): + from types import SimpleNamespace + + runner, work, root, invoke = prepared_case + statuses = iter([0, 7]) + + def execute(args, **kwargs): + cases = Path(kwargs["env"]["E2E_ARTIFACTS"]) + code = next(statuses) + if not code: + (cases / "scores.parquet").write_bytes(b"passing scores") + Path(args[args.index("--junitxml") + 1]).write_text('') + else: + (cases / "failure.log").write_text("partial failure") + if failure == "timeout": + raise subprocess.TimeoutExpired(args, 900) + return SimpleNamespace(returncode=code) + + monkeypatch.setattr(runner.subprocess, "run", execute) + assert invoke("--run-only") == 0 + first = next(root.glob("run-*")) + (work / "installed.json").write_text(json.dumps({"assets": "assets", "wheel_sha256": "new-digest"})) + assert invoke("--run-only") == (1 if failure == "timeout" else 7) + second = next(p for p in root.glob("run-*") if p != first) + assert (first / "scores.parquet").read_bytes() == b"passing scores" + assert not (second / "scores.parquet").exists() + assert not (second / "junit.xml").exists() + assert (second / "failure.log").read_text() == "partial failure" + assert json.loads((second / "installed.json").read_text())["wheel_sha256"] == "new-digest" + assert (root / "operator-note.txt").read_text() == "unrelated file" + + +def test_preparation_state_has_no_result_directory_and_invalidates_failed_readiness(prepared_case, monkeypatch): + runner, work, root, invoke = prepared_case + with monkeypatch.context() as first: + first.setattr(runner, "prepare", lambda *args: None) + assert invoke("--prepare-only") == 0 + assert list(root.iterdir()) == [root / "operator-note.txt"] + + def interrupted_environment(work): + assert not (work / "installed.json").exists() + raise RuntimeError("interrupted preparation") + + monkeypatch.setattr(runner, "environment", interrupted_environment) + with pytest.raises(RuntimeError, match="interrupted preparation"): + invoke() + attempt = next(root.glob("run-*")) + assert list(attempt.iterdir()) == [] + with pytest.raises(SystemExit) as failure: + invoke("--run-only") + assert failure.value.code == 2 + assert list(root.glob("run-*")) == [attempt] + + +@pytest.mark.parametrize("kind", ["symlink", "dangling", "directory", "fifo", "hardlink"]) +def test_owned_file_admission_rejects_each_invalid_kind_unchanged(kind, prepared_case, tmp_path): + runner, work, root, _ = prepared_case + target, entry = tmp_path / "outside", work / ".env" + if kind != "dangling": + target.write_bytes(b"external bytes") + if kind in {"symlink", "dangling"}: + entry.symlink_to(target) + elif kind == "directory": + entry.mkdir() + elif kind == "fifo": + os.mkfifo(entry) + else: + os.link(target, entry) + before = (work / "installed.json").read_bytes() + with pytest.raises(ValueError): + runner.admit_paths(work, root) + assert (work / "installed.json").read_bytes() == before + assert not list(root.glob("run-*")) + assert target.read_bytes() == b"external bytes" if kind != "dangling" else not target.exists() + + +@pytest.mark.parametrize( + "name", + [ + ".env", + ".env.local", + "runtime-requirements.txt", + "installed.json", + "home", + "tokenizers", + "dist", + "runtime", + "harness", + "suite", + "cases", + ], +) +def test_every_owned_slot_is_protected(name, prepared_case, tmp_path): + runner, work, root, _ = prepared_case + target = tmp_path / "outside" + target.write_bytes(b"external bytes") + entry = work / name + entry.unlink(missing_ok=True) + entry.symlink_to(target) + with pytest.raises(ValueError, match="symlink"): + runner.admit_paths(work, root) + assert target.read_bytes() == b"external bytes" + assert not list(root.glob("run-*")) + + +def test_directory_kind_and_export_hardlink_are_rejected(prepared_case, tmp_path): + runner, work, root, _ = prepared_case + (work / "home").write_text("not a directory") + with pytest.raises(ValueError, match="directory"): + runner.admit_paths(work, root) + (work / "home").unlink() + outside = tmp_path / "outside" + outside.write_bytes(b"external export") + os.link(outside, work / "runtime-requirements.txt") + with pytest.raises(ValueError, match="singly linked"): + runner.admit_paths(work, root) + assert outside.read_bytes() == b"external export" + + +@pytest.mark.parametrize("mode", ["--prepare-only", "--run-only"]) +def test_main_rejects_unsafe_work_before_any_attempt(mode, prepared_case, tmp_path, monkeypatch): + runner, work, root, invoke = prepared_case + outside = tmp_path / "outside" + outside.write_bytes(b"external bytes") + (work / ".env.local").symlink_to(outside) + monkeypatch.setattr(runner, "prepare", lambda *args: pytest.fail("unsafe preparation admitted")) + with pytest.raises(ValueError, match="symlink"): + invoke(mode) + assert outside.read_bytes() == b"external bytes" + assert not list(root.glob("run-*")) + + +@pytest.mark.parametrize( + "location", + [ + "work", + "work-child", + "work-ancestor", + "source", + "source-ancestor", + "normalized-source", + "work-source", + "artifact-symlink", + "lexical-symlink", + ], +) +def test_root_admission_rejects_overlap_and_lexical_links(location, tmp_path, monkeypatch, e2e_runner): + runner = e2e_runner + source, work, root = tmp_path / "checkout", tmp_path / "work", tmp_path / "results" + source.mkdir() + work.mkdir() + (source / "tracked.txt").write_text("unchanged") + (tmp_path / "existing").mkdir() + monkeypatch.setattr(runner, "ROOT", source) + if location == "work": + root = work + elif location == "work-child": + root = work / "child" + elif location == "work-ancestor": + root = tmp_path + elif location == "source": + root = source + elif location == "source-ancestor": + work = tmp_path + elif location == "normalized-source": + work = tmp_path / "existing/../checkout/work" + elif location == "work-source": + work = source + elif location == "artifact-symlink": + root = tmp_path / "output-link" + root.symlink_to(source, target_is_directory=True) + else: + (tmp_path / "link").symlink_to(tmp_path / "existing", target_is_directory=True) + work = tmp_path / "link/../work" + before = sorted(str(p.relative_to(tmp_path)) for p in tmp_path.rglob("*")) + with pytest.raises(ValueError): + runner.admit_paths(work, root) + assert sorted(str(p.relative_to(tmp_path)) for p in tmp_path.rglob("*")) == before + assert (source / "tracked.txt").read_text() == "unchanged" + + +@pytest.mark.parametrize("receipt", [None, "not-json", "[]", "{}", '{"assets": null}']) +def test_run_only_readiness_failure_leaves_existing_results_unchanged(receipt, prepared_case): + _, work, root, invoke = prepared_case + identity = work / "installed.json" + identity.unlink() if receipt is None else identity.write_text(receipt) + with pytest.raises(SystemExit) as failure: + invoke("--run-only") + assert failure.value.code == 2 + assert list(root.iterdir()) == [root / "operator-note.txt"] + + +@pytest.mark.parametrize( + "name", ["runtime", "harness", "suite", "cases", ".env", ".env.local", "runtime-requirements.txt", "installed.json"] +) +def test_selected_input_must_survive_work_mutation(name, prepared_case): + import argparse + + runner, work, root, _ = prepared_case + if name in {"runtime", "harness", "suite", "cases"}: + directory = work / name + directory.mkdir() + wheel = directory / "selected.whl" + else: + wheel = work / name + wheel.write_bytes(b"original wheel") + args = argparse.Namespace(wheel=wheel, wheel_dir=None, run_only=False) + with pytest.raises(ValueError, match="wheel"): + runner.validate_run_inputs(work, args, argparse.ArgumentParser()) + assert wheel.read_bytes() == b"original wheel" + assert not list(root.glob("run-*")) + + +def test_existing_output_inputs_and_normal_venv_links_remain_usable(prepared_case, monkeypatch): + runner, work, root, invoke = prepared_case + (work / "runtime/bin").mkdir(parents=True) + (work / "runtime/bin/python").symlink_to(sys.executable) + wheel = root / "previous-result.whl" + wheel.write_bytes(b"original wheel") + selected = [] + monkeypatch.setattr( + runner, "prepare", lambda work, args, parser: selected.append(runner.archive(work, args, parser)) + ) + assert invoke("--prepare-only", "--wheel", str(wheel)) == 0 + assert selected == [wheel.resolve()] + assert wheel.read_bytes() == b"original wheel" + assert (work / "runtime/bin/python").is_symlink() + assert not list(root.glob("run-*")) + + +def test_safe_parent_alias_uses_canonical_installed_import(tmp_path, monkeypatch, e2e_runner): + runner = e2e_runner + (tmp_path / "existing").mkdir() + work = tmp_path / "existing/../work" + work.mkdir() + (work / "runtime").mkdir() + wheel = tmp_path / "original.whl" + wheel.write_bytes(b"original wheel") + installed = {"module": str((work / "runtime/lib/usersim/__init__.py").resolve()), "assets": "assets"} + monkeypatch.setattr(runner, "command", lambda *args, **kwargs: None) + monkeypatch.setattr(runner.subprocess, "check_output", lambda *args, **kwargs: json.dumps(installed)) + runner.record_install(work, wheel, {}) + monkeypatch.setattr(runner, "prepare", lambda work, *args: runner.record_install(work, wheel, {})) + monkeypatch.setattr( + sys, "argv", [str(RUNNER), "--prepare-only", "--work-dir", str(work), "--artifacts", str(tmp_path / "results")] + ) + assert runner.main() == 0 + assert json.loads((work / "installed.json").read_text())["module"] == installed["module"] diff --git a/tests/test_packaging_contract.py b/tests/test_packaging_contract.py index 4bcc16f..4ce7853 100644 --- a/tests/test_packaging_contract.py +++ b/tests/test_packaging_contract.py @@ -321,6 +321,7 @@ def test_tests_are_pruned_from_the_sdist(self) -> None: """The suite runs from a checkout; a partial copy without conftest.py would not run.""" directives = [line.split() for line in (_REPO_ROOT / "MANIFEST.in").read_text().splitlines()] assert ["prune", "tests"] in directives + assert ["prune", "tests_e2e"] in directives class TestManifestProvenance: diff --git a/tests_e2e/conftest.py b/tests_e2e/conftest.py new file mode 100644 index 0000000..db68c80 --- /dev/null +++ b/tests_e2e/conftest.py @@ -0,0 +1,11 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""The installed harness must collect scenarios, never silently skip all tests.""" + +import pytest + + +def pytest_collection_finish(session): + if not session.items: + raise pytest.UsageError("installed CLI e2e collected zero cases") diff --git a/tests_e2e/network_guard/sitecustomize.py b/tests_e2e/network_guard/sitecustomize.py new file mode 100644 index 0000000..bbda4a8 --- /dev/null +++ b/tests_e2e/network_guard/sitecustomize.py @@ -0,0 +1,47 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Test-only socket boundary: permit loopback, reject external connections.""" + +import ipaddress +import socket + +_connect = socket.socket.connect +_connect_ex = socket.socket.connect_ex +_getaddrinfo = socket.getaddrinfo + + +def local(host): + if isinstance(host, bytes): + host = host.decode("ascii") + if host == "localhost": + return + try: + if ipaddress.ip_address(host).is_loopback: + return + except ValueError: + pass + raise OSError(f"E2E blocks external network: {host}") + + +def connect(sock, address): + if sock.family in (socket.AF_INET, socket.AF_INET6): + local(address[0]) + return _connect(sock, address) + + +def connect_ex(sock, address): + if sock.family in (socket.AF_INET, socket.AF_INET6): + local(address[0]) + return _connect_ex(sock, address) + + +def getaddrinfo(host, *args, **kwargs): + if host is not None: + local(host) + return _getaddrinfo(host, *args, **kwargs) + + +socket.socket.connect = connect +socket.socket.connect_ex = connect_ex +socket.getaddrinfo = getaddrinfo diff --git a/tests_e2e/pytest.ini b/tests_e2e/pytest.ini new file mode 100644 index 0000000..941a3e1 --- /dev/null +++ b/tests_e2e/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +testpaths = . +addopts = --strict-markers diff --git a/tests_e2e/test_cli_workflow.py b/tests_e2e/test_cli_workflow.py new file mode 100644 index 0000000..629754f --- /dev/null +++ b/tests_e2e/test_cli_workflow.py @@ -0,0 +1,418 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Real CLI, HTTP transport, persisted rows, and downstream artifact contracts.""" + +from __future__ import annotations + +import json +import os +import subprocess +import threading +import urllib.request +from contextlib import contextmanager +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +import pyarrow.parquet as pq +import pytest + +RUNTIME = Path(os.environ["E2E_RUNTIME"]) +ARTIFACTS = Path(os.environ["E2E_ARTIFACTS"]) + + +class Provider(ThreadingHTTPServer): + daemon_threads = True + + def __init__(self): + self.requests = [] + self.mode = "ok" + self.failures = 0 + self.transient_remaining = 0 + super().__init__(("127.0.0.1", 0), Handler) + + +class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def send(self, code, body): + encoded = json.dumps(body, ensure_ascii=False).encode() + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(encoded))) + self.end_headers() + self.wfile.write(encoded) + + def do_GET(self): + self.send(200, {"object": "list", "data": []}) + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + self.server.requests.append(body) + model = body["model"] + if model == "assistant_model" and body.get("tools") and self.server.transient_remaining: + self.server.transient_remaining -= 1 + self.server.failures += 1 + self.send(500, {"error": {"message": "deliberate transient local failure", "type": "server_error"}}) + return + if model == "evaluator_model" and self.server.mode == "http-error": + self.server.failures += 1 + self.send(400, {"error": {"message": "deliberate local provider failure", "type": "invalid_request_error"}}) + return + message = {"role": "assistant", "content": "A clear answer about the requested task."} + if body.get("response_format", {}).get("type") == "json_schema": + schema = body["response_format"]["json_schema"]["schema"] + message["content"] = json.dumps( + { + key: {"score": 4, "reasoning": "The response addresses the synthetic task."} + for key in schema["properties"] + } + ) + if model == "evaluator_model" and self.server.mode == "malformed": + self.server.failures += 1 + message["content"] = "this is not JSON" + elif model == "judge_model": + message["content"] = "The request is appropriate.success" + elif model == "summary_model": + message["content"] = "no" + elif model == "api_response_model": + message["content"] = '{"status": "ok", "temperature": 22}' + elif model == "user_model" and body.get("tools"): + message.update( + { + "content": None, + "tool_calls": [ + { + "id": "call_move", + "type": "function", + "function": { + "name": "commit_move", + "arguments": json.dumps({"move": "express_distress", "topic": "", "intensity": 0.3}), + }, + } + ], + } + ) + elif model == "user_model": + prompt = json.dumps(body["messages"], ensure_ascii=False) + message["content"] = ( + "東京の天気を調べてください。今日の外出の予定に使いたいです。" + if "Japanese" in prompt + else "Please check today's weather in Austin for my afternoon walk." + ) + elif model == "assistant_model" and body.get("tools") and body.get("tool_choice") != "none": + name = body["tools"][0]["function"]["name"] + message.update( + { + "content": None, + "tool_calls": [ + { + "id": "call_e2e", + "type": "function", + "function": {"name": name, "arguments": '{"city":"Austin"}'}, + } + ], + } + ) + self.send( + 200, + { + "id": "chatcmpl-local-e2e", + "object": "chat.completion", + "created": 1, + "model": model, + "choices": [ + { + "index": 0, + "message": message, + "finish_reason": "tool_calls" if "tool_calls" in message else "stop", + } + ], + "usage": {"prompt_tokens": 20, "completion_tokens": 10, "total_tokens": 30}, + }, + ) + + +@contextmanager +def provider(work): + server = Provider() + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + with urllib.request.urlopen(f"http://127.0.0.1:{server.server_port}/health", timeout=5) as response: + assert response.status == 200 + yield server + finally: + (work / "requests.json").write_text(json.dumps(server.requests, indent=2, ensure_ascii=False), encoding="utf-8") + server.shutdown() + thread.join(timeout=5) + server.server_close() + assert not thread.is_alive() + + +def cli(cwd, name, *args, expected=0): + try: + result = subprocess.run( + [str(RUNTIME / "bin/usersim"), *map(str, args)], + cwd=cwd, + text=True, + capture_output=True, + timeout=180, + check=False, + ) + except subprocess.TimeoutExpired as error: + (cwd / f"{name}.log").write_bytes((error.stdout or b"") + (error.stderr or b"")) + raise + (cwd / f"{name}.log").write_text(result.stdout + result.stderr, encoding="utf-8") + assert result.returncode == expected, f"{name} exited {result.returncode}:\n{result.stdout}\n{result.stderr}" + return result + + +def rows(path): + files = sorted(path.rglob("part-*.parquet")) + assert files, f"No persisted Parquet under {path}" + return [ + { + **{ + part.split("=", 1)[0]: part.split("=", 1)[1] + for part in file.relative_to(path).parts[:-1] + if "=" in part + }, + **row, + } + for file in files + for row in pq.ParquetFile(file).read().to_pylist() + ] + + +def models(path, server): + aliases = ("user_model", "assistant_model", "judge_model", "summary_model", "api_response_model", "evaluator_model") + text = ( + "models = [\n" + + ",\n".join( + f'{{alias="{alias}", model="{alias}", provider="local", max_tokens=256, timeout=5, max_parallel_requests=1, extra_body={{e2e_marker="preserved", chat_template_kwargs={{enable_thinking=true, e2e_nested="preserved"}}}}}}' + for alias in aliases + ) + + f'\n]\n[[providers]]\nname="local"\nendpoint="http://127.0.0.1:{server.server_port}/v1"\nprovider_type="openai"\napi_key="E2E_API_KEY"\n' + ) + path.write_text(text, encoding="utf-8") + + +def inputs(path): + # Same stored-row/config fields used by test_cli_simulate_inputs.py. These + # synthetic personas require no downloaded population or sampler patches. + common = { + "name": "conversation_messages", + "column_type": "conversation-simulator", + "max_turns": 1, + "max_query_attempts": 1, + "tools_column": "tools", + "toolset_name_column": "toolset_name", + "assets_dir": os.environ["E2E_ASSETS"], + "random_seed": 1, + } + tool = { + "type": "function", + "function": { + "name": "weather", + "description": "Read current weather in a city.", + "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}, + }, + } + fixture = [] + for index, (locale, probe, name) in enumerate( + (("en_US", "tool_calling", "Sarah"), ("ja_JP", "general_open_ended", "花子")) + ): + fixture.append( + { + "trajectory_id": f"synthetic-input-{index}", + "persona_uuid": f"synthetic-persona-{index}", + "locale": locale, + "persona": { + "first_name": name, + "last_name": "Test", + "age": 42, + "city": "Austin", + "occupation": "teacher", + "education_level": "bachelors", + }, + "probe_type": probe, + "probe_variant": "default", + "theme": { + "type": "weather", + "description": "Plan an afternoon walk.", + "tool_expected": probe == "tool_calling", + }, + "tools": [tool], + "toolset_name": "weather", + "toolset_category": "weather", + "usersim_config": {**common, "locale": locale}, + } + ) + path.write_text("".join(json.dumps(row, ensure_ascii=False) + "\n" for row in fixture), encoding="utf-8") + return fixture + + +@pytest.fixture(scope="module") +def completed_workflow(): + work = ARTIFACTS / "workflow" + work.mkdir(parents=True, exist_ok=True) + assert not (work / "src").exists() + with provider(work) as server: + config = work / "models.toml" + models(config, server) + source = work / "inputs.jsonl" + fixture = inputs(source) + server.transient_remaining = 1 + trajectories, evaluations = work / "trajectories", work / "evaluations" + cli(work, "simulate", "simulate", "--inputs", source, "--models", config, "--out", trajectories) + assert server.failures == 1 + saved = rows(trajectories) + assert len(saved) == 2 + assert {row["input_id"] for row in saved} == {row["trajectory_id"] for row in fixture} + assert {row["locale"] for row in saved} == {"en_US", "ja_JP"} + for row in saved: + outcome = json.loads(row["simulation_outcome"]) + assert outcome["status"] == "ok", outcome + messages = json.loads(row["conversation_messages"]) + assert any(message["role"] == "assistant" for message in messages) + if row["locale"] == "ja_JP": + assert "東京" in json.dumps(messages, ensure_ascii=False) + else: + assert outcome["n_tool_calls"] >= 1 + assert any(message["role"] == "tool" for message in messages) + before = {row["trajectory_id"] for row in saved} + first_runs = sorted(p.name for p in trajectories.glob("run=*")) + assert len(first_runs) == 1 + assert before.isdisjoint({row["trajectory_id"] for row in fixture}) + calls = len(server.requests) + cli(work, "resume", "simulate", "--inputs", source, "--models", config, "--out", trajectories) + assert len(rows(trajectories)) == 2 + assert {row["trajectory_id"] for row in rows(trajectories)} == before + assert sorted(p.name for p in trajectories.glob("run=*")) == first_runs + assert len(server.requests) == calls, "resume made provider calls for already saved rows" + cli(work, "eval", "eval", "--trajectories", trajectories, "--models", config, "--out", evaluations) + assert sorted(p.name for p in evaluations.glob("run=*")) == first_runs + judged = rows(evaluations) + assert len(judged) == 2 + assert {row["trajectory_id"] for row in judged} == before + for row in judged: + cell = json.loads(row["assistant_eval"]) + assert not cell["skipped"] + assert cell["axes"] + assert all(result["evaluator_model"]["score"] == 4 for result in cell["axes"].values()) + reports, curated = work / "reports", work / "curated" + cli(work, "report", "report", "--trajectories", trajectories, "--evaluations", evaluations, "--out", reports) + html = list(reports.rglob("*.html")) + assert html and html[0].stat().st_size > 1000 + cli(work, "select", "select", "--trajectories", trajectories, "--evaluations", evaluations, "--out", curated) + selected = rows(curated) + assert len(selected) == 2 + assert {row["trajectory_id"] for row in selected} == before + assert list(curated.rglob("_selection_manifest.json")) + assistant = [request for request in server.requests if request["model"] == "assistant_model"] + assert any(request.get("tools") for request in assistant) + assert all(request.get("e2e_marker") == "preserved" for request in assistant) + assert any(request.get("response_format", {}).get("type") == "json_schema" for request in server.requests) + + return trajectories + + +def test_installed_cli_workflow(completed_workflow): + assert len(rows(completed_workflow)) == 2 + + +def test_external_network_blocked(): + telemetry = subprocess.run( + [ + str(RUNTIME / "bin/python"), + "-c", + "from data_designer.engine.models.telemetry import TELEMETRY_ENABLED; from data_designer.engine.models.facade import TELEMETRY_ENABLED as facade_enabled; assert not TELEMETRY_ENABLED and not facade_enabled", + ], + text=True, + capture_output=True, + timeout=15, + check=False, + ) + assert telemetry.returncode == 0, telemetry.stderr + + result = subprocess.run( + [str(RUNTIME / "bin/python"), "-c", "import socket; socket.create_connection(('example.com',443),timeout=1)"], + text=True, + capture_output=True, + timeout=5, + check=False, + ) + assert result.returncode != 0 + assert "E2E blocks external network" in result.stderr + + +@pytest.mark.parametrize("mode", ["malformed", "http-error"]) +def test_provider_failure_is_not_a_score(mode, completed_workflow): + trajectories = completed_workflow + work = ARTIFACTS / mode + work.mkdir(parents=True, exist_ok=True) + with provider(work) as server: + server.mode = mode + config = work / "models.toml" + models(config, server) + output = work / "evaluations" + result = cli( + work, + "eval", + "eval", + "--trajectories", + trajectories, + "--models", + config, + "--axes", + "helpfulness", + "--out", + output, + expected=1, + ) + assert server.failures >= 1 + assert ( + "no judge returned a verdict" in result.stderr + if mode == "malformed" + else "deliberate local provider failure" in result.stderr + ) + if mode == "malformed": + cells = [json.loads(row["assistant_eval"]) for row in rows(output)] + assert len(cells) == 2 + assert all(cell["axes"] == {"helpfulness": {}} for cell in cells) + else: + assert not list(output.rglob("part-*.parquet")) + + +def test_tool_choice_and_call_extra_body(): + work = ARTIFACTS / "tool-choice" + work.mkdir(parents=True, exist_ok=True) + with provider(work) as server: + config = work / "models.toml" + models(config, server) + source = work / "inputs.jsonl" + row = inputs(source)[0] + row["probe_type"] = "health_therapy_disclosure" + row["probe_variant"] = "guarded" + row["usersim_config"]["max_turns"] = 3 + source.write_text(json.dumps(row) + "\n", encoding="utf-8") + cli(work, "simulate", "simulate", "--inputs", source, "--models", config, "--out", work / "trajectories") + moves = [ + request + for request in server.requests + if request.get("tools") and request["tools"][0]["function"]["name"] == "commit_move" + ] + assert moves, "guarded follow-up did not call the real move HTTP path" + assert all(request["tool_choice"] == "auto" for request in moves) + assert all(request["e2e_marker"] == "preserved" for request in moves) + assert all( + request.get("chat_template_kwargs") == {"enable_thinking": False, "e2e_nested": "preserved"} + for request in moves + ) + outcome = json.loads(rows(work / "trajectories")[0]["simulation_outcome"]) + assert outcome["status"] == "completed_with_warnings", outcome + assert {warning["kind"] for warning in outcome["warnings"]} == {"used_placeholder_clinical_profile"} + assert outcome["failure_class"] is None + assert outcome["n_turns"] == 3