diff --git a/.github/workflows/ci-test.yaml b/.github/workflows/ci-test.yaml index acb7a466ff..b0fe661ec1 100644 --- a/.github/workflows/ci-test.yaml +++ b/.github/workflows/ci-test.yaml @@ -22,12 +22,19 @@ on: jobs: # Path filtering at job level, not trigger paths-ignore: an untriggered # workflow reports no check run, so a required check waits forever; a job - # skipped via `if:` reports conclusion `skipped`, which passes. Gates only - # unit/integration — e2e runs regardless (docker/** changes affect it). + # skipped via `if:` reports conclusion `skipped`, which passes. Two scopes: + # `relevant` gates unit/integration; `e2e_relevant` gates e2e and skips only + # docs — the e2e image build is the sole CI check that compiles the frontend, + # and docker/** changes must be e2e-tested. changes: runs-on: ubuntu-latest outputs: - relevant: ${{ steps.filter.outputs.relevant }} + # Filtering is a PR-time saving only. On workflow_dispatch there is no + # base to diff, so paths-filter resolves against the default branch and a + # manual run on main diffs main against itself — every filter false, every + # tier skipped, gate green with nothing run. Force both true off-PR. + relevant: ${{ github.event_name != 'pull_request' || steps.filter.outputs.relevant == 'true' }} + e2e_relevant: ${{ github.event_name != 'pull_request' || steps.filter.outputs.e2e_relevant == 'true' }} steps: - name: Checkout repository uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 @@ -35,6 +42,7 @@ jobs: persist-credentials: false - name: Check for changes outside ignored paths + if: github.event_name == 'pull_request' uses: dorny/paths-filter@7b450fff21473bca461d4b92ce414b9d0420d706 # v4.0.2 id: filter with: @@ -46,6 +54,8 @@ jobs: - "!docker/**" - "!frontend/**" - "!docs/**" + e2e_relevant: + - "!docs/**" # Tiers share no state (separate runners, disjoint groups), so they run as # parallel matrix legs; cross-tier reconciliation happens in `report`. @@ -114,11 +124,12 @@ jobs: if-no-files-found: ignore retention-days: 14 - # Own runner: builds images then runs e2e against the full compose platform. - # Not gated by `changes` — docker/** edits don't count as "relevant" yet must - # be e2e-tested. + # Builds images then runs e2e against the full compose platform. Gated by + # `e2e_relevant`, which unlike `relevant` keeps docker/** and frontend/** in + # scope — both are built here — and skips only docs-only changes. e2e: - if: github.event.pull_request.draft == false + needs: changes + if: needs.changes.outputs.e2e_relevant == 'true' && github.event.pull_request.draft == false runs-on: ubuntu-latest timeout-minutes: 45 env: @@ -168,8 +179,39 @@ jobs: - name: Set up env files run: ./run-platform.sh -e + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4 + + # Bake instead of `compose build` so each ephemeral runner restores layers + # from the GHA cache the release workflow also writes (matching per-service + # scopes), instead of building cold every run. `load` exports the images + # into the daemon so the compose stack can boot them. - name: Build service images - run: docker compose -f docker/docker-compose.build.yaml build + uses: docker/bake-action@d3418bd7d0e9324001bca92fa8ba175ea7e6dc9b # v7 + with: + files: ./docker/docker-compose.build.yaml + load: true + set: | + *.context=. + *.args.VERSION=${{ github.sha }} + frontend.cache-from=type=gha,scope=frontend + frontend.cache-to=type=gha,mode=max,scope=frontend + backend.cache-from=type=gha,scope=backend + backend.cache-to=type=gha,mode=max,scope=backend + runner.cache-from=type=gha,scope=runner + runner.cache-to=type=gha,mode=max,scope=runner + tool-sidecar.cache-from=type=gha,scope=tool-sidecar + tool-sidecar.cache-to=type=gha,mode=max,scope=tool-sidecar + platform-service.cache-from=type=gha,scope=platform-service + platform-service.cache-to=type=gha,mode=max,scope=platform-service + x2text-service.cache-from=type=gha,scope=x2text-service + x2text-service.cache-to=type=gha,mode=max,scope=x2text-service + tool-classifier.cache-from=type=gha,scope=tool-classifier + tool-classifier.cache-to=type=gha,mode=max,scope=tool-classifier + tool-text_extractor.cache-from=type=gha,scope=tool-text_extractor + tool-text_extractor.cache-to=type=gha,mode=max,scope=tool-text_extractor + worker-unified.cache-from=type=gha,scope=worker-unified + worker-unified.cache-to=type=gha,mode=max,scope=worker-unified - name: Run e2e tier (rig boots the platform via compose) env: @@ -184,15 +226,6 @@ jobs: mv reports/.coverage "reports/.coverage.tier-e2e" fi - - name: Capture docker compose logs on failure - if: failure() - run: | - mkdir -p reports - docker compose -p unstract-test \ - -f docker/docker-compose.yaml \ - -f tests/compose/docker-compose.test.yaml \ - logs --no-color > reports/docker-compose-logs.txt || true - - name: Upload e2e reports artifact if: always() uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 @@ -204,9 +237,9 @@ jobs: report: needs: [changes, test, e2e] - # `always()` so a red tier still posts a report. e2e always runs on - # non-draft, so gate only on draft; the fail step below distinguishes a - # legitimate skip (irrelevant paths) from a real tier failure. + # `always()` so a red tier still posts a report. Any tier may skip on + # irrelevant paths; the fail step below distinguishes a legitimate skip + # from a real tier failure. if: always() && github.event.pull_request.draft == false runs-on: ubuntu-latest steps: @@ -295,12 +328,15 @@ jobs: retention-days: 14 - name: Fail if any tier failed - # Single branch-protection check. A skipped `test` means irrelevant - # paths (path-filter skip == pass); e2e always runs, so it must be - # green. Runs after the comment so a red tier still reports. + # Single branch-protection check. A skipped tier means irrelevant paths + # (path-filter skip == pass); only an actual failure fails the gate. + # Runs after the comment so a red tier still reports. if: always() run: | + changes_result="${{ needs.changes.result }}" test_result="${{ needs.test.result }}" e2e_result="${{ needs.e2e.result }}" + # A broken filter job skips both tiers, which would read as pass. + [ "$changes_result" = "success" ] || exit 1 [ "$test_result" = "success" ] || [ "$test_result" = "skipped" ] || exit 1 - [ "$e2e_result" = "success" ] || exit 1 + [ "$e2e_result" = "success" ] || [ "$e2e_result" = "skipped" ] || exit 1 diff --git a/tests/README.md b/tests/README.md index 40abddd1bf..982b12f553 100644 --- a/tests/README.md +++ b/tests/README.md @@ -24,11 +24,12 @@ tests/ │ ├── coverage.py # Per-group coverage files + combine │ └── critical_paths.py # Gap + regression detection ├── e2e/ -│ ├── conftest.py # Session-scoped `platform` fixture +│ ├── conftest.py # `platform` fixture + `provisioned_workflow` chain │ ├── smoke/ # Login → /health smoke -│ ├── workflows/ # (future) workflow execution e2e -│ ├── api_deployment/ # (future) API deployment e2e -│ ├── prompt_studio/ # (future) Prompt Studio e2e +│ ├── workflows/ # Workflow execution e2e (mocked LLM) +│ ├── api_deployment/ # API deployment e2e: run, callback delivery, fan-out (all async) +│ ├── etl/ # ETL pipeline e2e (MinIO source + destination) +│ ├── prompt_studio/ # Prompt Studio fetch-response e2e │ └── hurl/ # (future) hurl-based HTTP suites ├── integration/ # Cross-service tests needing infra but not full platform ├── fixtures/ # Sample PDFs, JSON, adapter configs @@ -153,6 +154,34 @@ The rig brings the platform up **once** per `run` invocation (if any selected gr The `platform` pytest fixture in `tests/e2e/conftest.py` reads those env vars; e2e tests run elsewhere (without the rig) just skip with a clear message. +### Hermetic LLM (`UNSTRACT_LLM_MOCK_RESPONSE`) + +Execute-path e2e tests must not call a real provider, so the rig sets `UNSTRACT_LLM_MOCK_RESPONSE` (default `MOCK_LLM_OK`) before boot, for any runtime and treating an exported empty string as unset. The test overlay forwards it into the workers, and `unstract.sdk1.llm` passes it to litellm as `mock_response`: for the non-streaming completion path litellm returns the string verbatim with fixed usage (10 prompt / 20 completion / 30 total), so both the answer and the token counts are exact-assertable. Streaming (`stream_complete`) goes through a different litellm path whose usage differs, so don't assume 10/20/30 there. Sentinels like `litellm.RateLimitError` force error paths. Unset (the production default) the hook is a no-op. + +Mocking needs two conditions, not one: `ENVIRONMENT` must also be `test` or `development`. Production sets neither variable, so a stray mock var alone cannot fake completions and their billing — and a refusal is logged rather than silent, since setting the var at all means someone expected mocking. `development` is allowed because that is what base compose sets on the workers that run the injection; the test overlay pins `ENVIRONMENT=test` on those same two workers explicitly, so the tier can't lose its mock to a base-compose edit. + +A CI/dev override wins (the rig only fills an unset value). Running these tests under the rig **fails** if the var is missing; running **without** the rig just skips the execute-path tests — export it on both sides (your shell and the workers) if you boot the stack yourself. + +While the mock is active platform-wide, the LLM adapter's test-connection cannot pass: it regex-matches the completion for a specific city, which `MOCK_LLM_OK` won't satisfy — so `adapter-register-llm` stays an e2e gap for now. + +Only `LLM` completions are mocked, not embeddings: `provisioned_workflow` pins `chunk_size=0` so indexing never invokes `litellm.embedding`. A test that needs chunking will need the embedding path mocked too. + +### Fan-out (`MAX_PARALLEL_FILE_BATCHES`) + +Defaults to `1`, meaning every file of a multi-file run lands in one batch and is processed serially — so a fan-out test would pass without any fan-out happening. The overlay defaults it to `3` on `backend`, which is what normally takes effect: workers ask the backend for this value and fall back to their own env only when they can't reach it or it carries no value (the overlay sets it on `worker-api-deployment-v2` too, to keep the fallback in step). Batches are `min(MAX_PARALLEL_FILE_BATCHES, num_files)`, so N files with the same N gives one batch each. + +**The fan-out half is an open gap** (`workflow-execution-fan-out` in `critical_paths.yaml`), and closing it needs a product change, not a cleverer test. Nothing persists a batch or task id: the batch index is a discarded loop local, and the celery task id only ever reaches worker stdout. That leaves per-file timing as the only proxy, and timing does not work here — measured on CI, three files genuinely fanned out finished *further* apart (durations `6.6 / 18.1 / 22.9`) than the ~2s steps seen when they share a batch, because three concurrent tool containers on a loaded runner cost more in contention than they gain in overlap. No threshold separates those two distributions, and no delay value fixes it: the contention scales with the parallelism being measured. + +Worth stating plainly because the design is tempting: overlap of the row windows doesn't discriminate either, since a serialised batch pre-creates all its rows in one call and they overlap trivially. + +What would close it is a batch or task identifier on `WorkflowFileExecution` — the test then asserts N distinct ids for N files, with no timing and no stall. That is also ordinary execution observability, which is the argument for doing it in the product rather than the test. + +`e2e-api-deployment` still asserts the rejoin: one result per file, all files counted into `successful_files`, one row per file. + +### ETL (MinIO) + +`tests/e2e/etl` runs a pipeline from a source connector to a destination connector. MinIO is the only storage connector the compose stack both boots and registers — the local-filesystem one would need no infra but is never registered (`local_storage/` has no `__init__.py`, so `register_connectors` skips it), which is why the mounted `./workflow_data:/data` volume can't be used as an ETL endpoint. The test seeds and reads its objects over the published port (`UNSTRACT_MINIO_ENDPOINT`, default `localhost:9000`) while the workers reach the same store over the compose network (`UNSTRACT_MINIO_INTERNAL_URL`, default `http://unstract-minio:9000`). It skips when no MinIO answers, so it does not fail a runtime that publishes none. + --- ## Reports diff --git a/tests/compose/docker-compose.test.yaml b/tests/compose/docker-compose.test.yaml index a5cbbee0c3..feb46d6eda 100644 --- a/tests/compose/docker-compose.test.yaml +++ b/tests/compose/docker-compose.test.yaml @@ -1,21 +1,16 @@ -# E2E test overlay for docker/docker-compose.yaml. -# -# Used by tests/rig/runtime.py's ComposeRuntime: -# docker compose -f docker/docker-compose.yaml \ -# -f tests/compose/docker-compose.test.yaml up -d --wait -# -# Keep this file minimal so e2e tests run against images as close to production -# as possible. Override only what the test stack actually needs (test creds, -# pinned image tags, ENVIRONMENT=test sentinel). +# E2E overlay for docker/docker-compose.yaml — override only what the test +# stack needs, so e2e runs against production-like images. services: backend: - # Defaults to `latest` so contributors can run without setting - # UNSTRACT_TEST_VERSION. CI pins it to the commit SHA of the images built - # in the same run (see the e2e job in .github/workflows/ci-test.yaml). + # CI pins the tag to the SHA built in the same run; `latest` lets a + # contributor run without setting it. image: unstract/backend:${UNSTRACT_TEST_VERSION:-latest} environment: - ENVIRONMENT=test + # Workers ask the backend for this, so this is the value that normally + # takes effect for fan-out. + - MAX_PARALLEL_FILE_BATCHES=${MAX_PARALLEL_FILE_BATCHES:-3} platform-service: image: unstract/platform-service:${UNSTRACT_TEST_VERSION:-latest} @@ -26,3 +21,23 @@ services: image: unstract/runner:${UNSTRACT_TEST_VERSION:-latest} environment: - ENVIRONMENT=test + + # Execute-path e2e must never reach a real provider. The delay gives each + # mocked completion a known cost so per-file durations stay comparable, and + # ENVIRONMENT is the mock's second condition — set here rather than inherited + # so the tier can't lose its mock to a base-compose edit. + worker-executor-v2: + environment: + - ENVIRONMENT=test + - UNSTRACT_LLM_MOCK_RESPONSE=${UNSTRACT_LLM_MOCK_RESPONSE:-} + + worker-file-processing-v2: + environment: + - ENVIRONMENT=test + - UNSTRACT_LLM_MOCK_RESPONSE=${UNSTRACT_LLM_MOCK_RESPONSE:-} + + # Only the worker's fallback if it can't reach the backend; kept in step so it + # can't quietly serialise the fan-out test. + worker-api-deployment-v2: + environment: + - MAX_PARALLEL_FILE_BATCHES=${MAX_PARALLEL_FILE_BATCHES:-3} diff --git a/tests/critical_paths.yaml b/tests/critical_paths.yaml index 6fc07e721e..00429280e6 100644 --- a/tests/critical_paths.yaml +++ b/tests/critical_paths.yaml @@ -49,6 +49,7 @@ paths: - id: workflow-create-execute description: "Create a workflow, configure source+destination, execute, poll, fetch result." covered_by: [e2e-workflow] + proof: marker - id: api-deployment-provision description: "Deploying a workflow as an API mints a usable key and a resolvable endpoint." @@ -63,6 +64,7 @@ paths: - id: api-deployment-run description: "Deploy a workflow as an API, POST a document, receive structured JSON." covered_by: [e2e-api-deployment] + proof: marker - id: prompt-studio-author description: "Create a Prompt Studio project and add a prompt to it." @@ -70,8 +72,9 @@ paths: proof: marker - id: prompt-studio-fetch-response - description: "Prompt Studio: create project, add prompt, run single-pass, get response." + description: "Prompt Studio: create project, add prompt, run a prompt, get response." covered_by: [e2e-prompt-studio] + proof: marker - id: connector-register-test description: "Connector credentials are validated against the live system and stored encrypted." @@ -80,7 +83,8 @@ paths: - id: pipeline-etl-execute description: "Run an ETL pipeline from source connector to destination." - covered_by: [] # gap + covered_by: [e2e-etl] + proof: marker - id: usage-aggregate-read description: "Per-run token usage aggregates correctly and stays scoped to its organization." @@ -89,12 +93,18 @@ paths: - id: usage-token-tracking description: "Per-execution token usage is recorded and retrievable." - covered_by: [] # gap + covered_by: [e2e-api-deployment] + proof: marker + # Deliberate gap. The rejoin half is covered by e2e-api-deployment, but the + # fan-out half has no observable: no batch or task id is persisted, and timing + # can't stand in for one (see the test's module docstring). Closing it needs + # batch identity on WorkflowFileExecution, not another test. - id: workflow-execution-fan-out description: "Multi-file workflow execution fans out to file-processing workers and rejoins." - covered_by: [] # gap + covered_by: [] - id: callback-result-delivery description: "Async results are posted back via the callback worker." - covered_by: [] # gap + covered_by: [e2e-api-deployment] + proof: marker diff --git a/tests/e2e/api_deployment/__init__.py b/tests/e2e/api_deployment/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/e2e/api_deployment/conftest.py b/tests/e2e/api_deployment/conftest.py new file mode 100644 index 0000000000..2261c1e806 --- /dev/null +++ b/tests/e2e/api_deployment/conftest.py @@ -0,0 +1,154 @@ +"""Fixtures shared by the API-deployment e2e tests.""" + +from __future__ import annotations + +import time +import uuid +from dataclasses import dataclass +from urllib.parse import parse_qs, urlparse + +import pytest +import requests + +from tests.e2e.conftest import ProvisionedWorkflow + +# A file-bearing execution reaches COMPLETED only once the chord callback +# rejoins, which the synchronous POST cannot wait out reliably under load. Every +# api-deployment test therefore dispatches async (timeout=0) and polls the +# status endpoint — the one path the codebase itself proves works. +# Per-test budget. Three tests in this group share one 600s group timeout, so +# this stays well under a third of it to leave room for provisioning. +_POLL_TIMEOUT_SECONDS = 150 +# A slow poll response is not a verdict on the execution — keep waiting rather +# than failing the test on a transient blip. +_TRANSIENT = (requests.exceptions.Timeout, requests.exceptions.ConnectionError) + + +@dataclass(frozen=True) +class ApiDeployment: + """A deployed API endpoint and the key that opens it.""" + + session: requests.Session + base: str # backend root; status_api is returned rooted at it + prefix: str + exec_url: str + api_key: str + + @property + def auth(self) -> dict[str, str]: + return {"Authorization": f"Bearer {self.api_key}"} + + +def dispatch_async( + deployment: ApiDeployment, + files: list | dict, + **form: object, +) -> tuple[str, str]: + """POST an async execution (timeout=0) and return (execution_id, status_url). + + Leaves the result for the callback to deliver; asserts the immediate PENDING + handshake so a caller only ever polls a genuinely dispatched execution. + """ + resp = deployment.session.post( + deployment.exec_url, + headers=deployment.auth, + data={"timeout": 0, **form}, + files=files, + timeout=60, + ) + assert resp.status_code == 200, f"dispatch: HTTP {resp.status_code}: {resp.text}" + message = resp.json()["message"] + assert message["execution_status"] == "PENDING", message + assert message["result"] is None, message + # The async PENDING body carries the execution id only inside status_api's + # query string, not as a top-level field. + status_api = message["status_api"] + assert status_api, message + status_url = f"{deployment.base}/{status_api.lstrip('/')}" + execution_id = parse_qs(urlparse(status_api).query).get("execution_id", [None])[0] + assert execution_id, message + return execution_id, status_url + + +def poll_delivered( + deployment: ApiDeployment, status_url: str, *, include_metadata: bool = False +) -> dict: + """Poll until the result is delivered (HTTP 200), tolerating not-ready blips. + + The status endpoint answers 422 both while running and after a failure, so + the body is read to tell them apart rather than spinning until the timeout on + an execution that already gave up. The read is destructive: the first 200 + acknowledges the result, so call this once per execution. + """ + params = {"include_metadata": str(include_metadata).lower()} + deadline = time.monotonic() + _POLL_TIMEOUT_SECONDS + last = "" + while time.monotonic() < deadline: + try: + resp = deployment.session.get( + status_url, headers=deployment.auth, params=params, timeout=30 + ) + except _TRANSIENT as exc: + last = repr(exc) + time.sleep(2) + continue + last = resp.text + if resp.status_code == 200: + return resp.json() + # 406 means the result was already acknowledged and cleared — a prior + # 200 was lost in transit, so the body is gone and cannot be recovered. + if resp.status_code == 406: + pytest.fail(f"result already acknowledged (200 lost in transit): {last}") + # 422 is the only "still running / failed" answer; anything else is a + # real error the view surfaced (400/500/…), not a poll-protocol hiccup. + if resp.status_code != 422: + pytest.fail(f"unexpected poll status HTTP {resp.status_code}: {last}") + status = _status_of(resp) + assert status, f"422 body carried no status (contract change?): {last}" + assert status not in ("ERROR", "STOPPED"), f"execution failed: {last}" + time.sleep(2) + pytest.fail(f"result never delivered within {_POLL_TIMEOUT_SECONDS}s; last: {last}") + + +def _status_of(resp: requests.Response) -> str: + try: + return str(resp.json().get("status", "")) + except ValueError: # a non-JSON body has no status; the caller asserts on it + return "" + + +@pytest.fixture(scope="session") +def api_deployment(provisioned_workflow: ProvisionedWorkflow) -> ApiDeployment: + """Deploy the provisioned workflow as an API. + + Session-scoped, so it runs once per xdist worker (not once per group) when + the group is parallelised. + """ + pw = provisioned_workflow + api_name = f"e2edep{uuid.uuid4().hex[:8]}" + resp = pw.session.post( + f"{pw.prefix}/api/deployment/", + headers={"X-CSRFToken": pw.session.cookies.get("csrftoken", "")}, + json={ + "workflow": pw.workflow_id, + "display_name": f"e2e {api_name}", + "description": "e2e api deployment", + "api_name": api_name, + "is_active": True, + }, + timeout=30, + ) + assert resp.status_code == 201, f"deploy: {resp.text}" + body = resp.json() + endpoint = body["api_endpoint"] + return ApiDeployment( + session=pw.session, + base=pw.base, + prefix=pw.prefix, + exec_url=( + endpoint + if endpoint.startswith("http") + else f"{pw.base}/{endpoint.lstrip('/')}" + ), + api_key=body["api_key"], + ) diff --git a/tests/e2e/api_deployment/test_api_deployment_async.py b/tests/e2e/api_deployment/test_api_deployment_async.py new file mode 100644 index 0000000000..21fb867300 --- /dev/null +++ b/tests/e2e/api_deployment/test_api_deployment_async.py @@ -0,0 +1,46 @@ +"""E2E: async API execution, whose result only the callback worker can deliver. + +With a file to process, COMPLETED is written by exactly one code path — the +chord callback running in the callback worker. The api-deployment worker itself +only ever writes EXECUTING, ERROR, or (with zero files) COMPLETED. So a polled +COMPLETED carrying the per-file result is proof the callback ran and rejoined. +""" + +from __future__ import annotations + +import io + +import pytest + +from tests.e2e.api_deployment.conftest import ( + ApiDeployment, + dispatch_async, + poll_delivered, +) + +pytestmark = [pytest.mark.e2e, pytest.mark.critical] + + +@pytest.mark.critical_path("callback-result-delivery") +def test_async_execution_result_delivered_by_callback( + api_deployment: ApiDeployment, llm_mock_response: str +) -> None: + document = io.BytesIO(b"Async probe. This document is about async widgets.") + _, status_url = dispatch_async( + api_deployment, {"files": ("async-probe.txt", document, "text/plain")} + ) + + body = poll_delivered(api_deployment, status_url) + assert body["status"] == "COMPLETED", body + + file_result = body["message"][0] + assert file_result["status"] == "Success", file_result + assert file_result["result"]["output"]["answer"] == llm_mock_response + + # The read is destructive: fetching a delivered result acknowledges it, and + # acknowledged results are gone. Pinning it stops a retry loop from being + # added later that would silently swallow the only copy. + again = api_deployment.session.get( + status_url, headers=api_deployment.auth, timeout=30 + ) + assert again.status_code == 406, f"expected acknowledged, got {again.text}" diff --git a/tests/e2e/api_deployment/test_api_deployment_fan_out.py b/tests/e2e/api_deployment/test_api_deployment_fan_out.py new file mode 100644 index 0000000000..1a225d5d72 --- /dev/null +++ b/tests/e2e/api_deployment/test_api_deployment_fan_out.py @@ -0,0 +1,82 @@ +"""E2E: a multi-file execution rejoins after being dispatched to workers. + +The rejoin is the chord callback: one result per file, and per-file rows counted +back up into successful_files. That half is directly observable. + +The fan-out half is NOT asserted here, and the critical path is recorded as a +gap. Nothing persists a batch or task id, so the only available proxy was +per-file timing, and timing cannot decide it: on a loaded CI runner three files +genuinely running in parallel finish further apart than three run serially, +because the contention they create outweighs the overlap they gain. Closing the +gap needs batch identity on the row, not a better statistic. +""" + +from __future__ import annotations + +import io + +import pytest + +from tests.e2e.api_deployment.conftest import ( + ApiDeployment, + dispatch_async, + poll_delivered, +) + +pytestmark = [pytest.mark.e2e, pytest.mark.critical] + +# One file per batch, matching the overlay's MAX_PARALLEL_FILE_BATCHES. +_DOCUMENTS = { + "fan-alpha.txt": b"Alpha document. This one is about alpha widgets and invoices.", + "fan-beta.txt": b"Beta document. A different text about beta gadgets entirely.", + "fan-gamma.txt": b"Gamma document. Yet another distinct body concerning gamma.", +} + + +def test_multi_file_execution_rejoins( + api_deployment: ApiDeployment, llm_mock_response: str +) -> None: + # Bodies must differ: identical files are deduplicated by hash on ingest and + # would come back as fewer results, which reads as a lost file. + files = [ + ("files", (name, io.BytesIO(body), "text/plain")) + for name, body in _DOCUMENTS.items() + ] + execution_id, status_url = dispatch_async(api_deployment, files) + body = poll_delivered(api_deployment, status_url) + assert body["status"] == "COMPLETED", body + + # Keyed by name, not position: batches rejoin in completion order. + by_name = {entry["file"]: entry for entry in body["message"]} + assert sorted(by_name) == sorted(_DOCUMENTS), body["message"] + for name, entry in by_name.items(): + assert entry["status"] == "Success", (name, entry) + assert entry["result"]["output"]["answer"] == llm_mock_response, (name, entry) + + _assert_totals_rejoined(api_deployment, execution_id) + rows = _fetch_file_rows(api_deployment, execution_id) + assert len(rows) == len(_DOCUMENTS), rows + + +def _assert_totals_rejoined(deployment: ApiDeployment, execution_id: str) -> None: + """Every file is counted back into the execution's totals.""" + resp = deployment.session.get( + f"{deployment.prefix}/execution/{execution_id}/", timeout=30 + ) + resp.raise_for_status() + execution = resp.json() + assert execution["total_files"] == len(_DOCUMENTS), execution + # Counted from the per-file rows the workers wrote, so this is the rejoin + # itself rather than a status the dispatcher could set alone. + assert execution["successful_files"] == len(_DOCUMENTS), execution + assert execution["failed_files"] == 0, execution + + +def _fetch_file_rows(deployment: ApiDeployment, execution_id: str) -> list[dict]: + """The per-file rows, which exist only because the workers wrote them.""" + resp = deployment.session.get( + f"{deployment.prefix}/execution/{execution_id}/files/", timeout=30 + ) + resp.raise_for_status() + body = resp.json() + return body if isinstance(body, list) else body.get("results", []) diff --git a/tests/e2e/api_deployment/test_api_deployment_run.py b/tests/e2e/api_deployment/test_api_deployment_run.py new file mode 100644 index 0000000000..6f87ca1aa7 --- /dev/null +++ b/tests/e2e/api_deployment/test_api_deployment_run.py @@ -0,0 +1,47 @@ +"""E2E: deploy a workflow as an API, POST a document, get structured JSON back. + +Dispatches async and polls the status endpoint for the result: a file-bearing +execution only reaches COMPLETED once the callback rejoins, which the sync POST +cannot reliably wait out under CI load. +""" + +from __future__ import annotations + +import io + +import pytest + +from tests.e2e.api_deployment.conftest import ( + ApiDeployment, + dispatch_async, + poll_delivered, +) + +pytestmark = [pytest.mark.e2e, pytest.mark.critical] + + +@pytest.mark.critical_path("api-deployment-run") +@pytest.mark.critical_path("usage-token-tracking") +def test_api_deployment_returns_mocked_answer( + api_deployment: ApiDeployment, llm_mock_response: str +) -> None: + document = io.BytesIO(b"Hello invoice 123. This is a test document about widgets.") + _, status_url = dispatch_async( + api_deployment, {"files": ("probe.txt", document, "text/plain")} + ) + body = poll_delivered(api_deployment, status_url, include_metadata=True) + assert body["status"] == "COMPLETED", body + + file_result = body["message"][0] + assert file_result["status"] == "Success", file_result + assert file_result["result"]["output"]["answer"] == llm_mock_response + + # 10/20/30 is litellm's fixed usage for one mocked completion. Counts are + # summed per reason, so a multiple of it means the prompt made extra calls + # rather than that tracking is broken. + usage = file_result["result"]["metadata"]["extraction_llm"][0] + counts = (usage["input_tokens"], usage["output_tokens"], usage["total_tokens"]) + assert counts == (10, 20, 30), ( + f"expected exactly one mocked extraction call, got {counts} — a multiple " + f"means the prompt issued more completions than this test assumes: {usage}" + ) diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index d4ba147507..7ebfb13cd6 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -9,11 +9,48 @@ from __future__ import annotations import os +import time +import uuid +from dataclasses import dataclass import pytest import requests -from tests.rig.runtime import PlatformEndpoints +from tests.rig.runtime import ( + LLM_MOCK_RESPONSE_ENV, + PlatformEndpoints, +) + +# Mirrors unstract.core.data_models.ExecutionStatus.terminal_statuses(); kept a +# local literal because the e2e venv carries only pytest/requests/minio, and +# these are the JSON status strings the HTTP API returns anyway. +TERMINAL_EXECUTION_STATUSES = {"COMPLETED", "ERROR", "STOPPED"} + +class CsrfSession(requests.Session): + """Stamps X-CSRFToken from its own cookies on unsafe methods. + + Centralises what a handful of per-file helpers each re-implemented (and had + already drifted on — some merged caller headers, some dropped them). + """ + + _UNSAFE = frozenset({"POST", "PUT", "PATCH", "DELETE"}) + + def request(self, method: str, url: str, **kwargs: object) -> requests.Response: + if method.upper() in self._UNSAFE: + headers = dict(kwargs.get("headers") or {}) + headers.setdefault("X-CSRFToken", self.cookies.get("csrftoken", "")) + kwargs["headers"] = headers + return super().request(method, url, **kwargs) + + +# Adapter registry ids from unstract.sdk1, chosen so no real provider is called: +# the vectordb/x2text NoOp rows return canned output and the LLM is mocked. The +# embedding adapter is real but never hit — chunk_size=0 keeps indexing off the +# path. Fake creds persist because adapter create does not validate connectivity. +_LLM_ADAPTER = "openai|502ecf49-e47c-445c-9907-6d4b90c5cd17" +_EMBED_ADAPTER = "openai|717a0b0e-3bbc-41dc-9f0c-5689437a1151" +_VECTORDB_ADAPTER = "noOpVectorDb|ca4d6056-4971-4bc8-97e3-9e36290b5bc0" +_X2TEXT_ADAPTER = "noOpX2text|mp66d1op-7100-d340-9101-846fc7115676" @pytest.fixture(scope="session") @@ -39,7 +76,7 @@ def authed_session(platform: PlatformEndpoints) -> requests.Session: should build their own session instead. """ base = platform.backend_url.rstrip("/") - session = requests.Session() + session = CsrfSession() resp = session.post( f"{base}/api/v1/login", @@ -69,3 +106,236 @@ def authed_session(platform: PlatformEndpoints) -> requests.Session: ) resp.raise_for_status() return session + + +@pytest.fixture(scope="session") +def llm_mock_response() -> str: + """The exact string the workers were told to return for every completion. + + Skips rather than guesses when unset: there is no deterministic answer to + assert unless the workers got the same value. + """ + value = os.environ.get(LLM_MOCK_RESPONSE_ENV) + if not value: + # Under a rig run the value is mandatory; skipping green would silently + # drop the whole execute path. Outside the rig, skip is still right. + if os.environ.get("UNSTRACT_RIG_SESSION_ID"): + pytest.fail( + f"{LLM_MOCK_RESPONSE_ENV} unset under a rig run — the rig must set " + "it before booting the platform." + ) + pytest.skip( + f"{LLM_MOCK_RESPONSE_ENV} not set — execute-path e2e needs the LLM mock." + ) + return value + + +@dataclass(frozen=True) +class ProvisionedWorkflow: + """Handles for a hermetic, execute-ready workflow (one Prompt Studio tool).""" + + session: requests.Session + base: str # backend root, e.g. http://localhost:8000 + prefix: str # tenant-scoped API root: {base}/api/v1/unstract/{org_id} + org_id: str + workflow_id: str + tool_id: str + prompt_registry_id: str + profile_id: str + prompt_id: str + prompt_key: str + + +def _org_id(session: requests.Session, base: str) -> str: + orgs = session.get(f"{base}/api/v1/organization", timeout=10) + orgs.raise_for_status() + return orgs.json()["organizations"][0]["id"] + + +def _post(session: requests.Session, url: str, **kw: object) -> requests.Response: + return session.post(url, timeout=60, **kw) # CsrfSession stamps the header + + +def _patch(session: requests.Session, url: str, **kw: object) -> requests.Response: + return session.patch(url, timeout=60, **kw) + + +def wait_for_execution( + session: requests.Session, + prefix: str, + execution_id: str, + *, + timeout: float = 300.0, +) -> dict: + """Poll an execution until terminal, failing loudly on timeout. + + A timed-out poll fails with the last-seen body so a slow worker reads + differently from a genuine terminal failure. + """ + deadline = time.monotonic() + timeout + execution: dict = {} + while time.monotonic() < deadline: + resp = session.get(f"{prefix}/execution/{execution_id}/", timeout=30) + resp.raise_for_status() + execution = resp.json() + if execution.get("status") in TERMINAL_EXECUTION_STATUSES: + return execution + time.sleep(2) + pytest.fail(f"execution not terminal within {timeout}s: {execution}") + + +@pytest.fixture(scope="session") +def provisioned_workflow( + platform: PlatformEndpoints, authed_session: requests.Session +) -> ProvisionedWorkflow: + """Stand up an API workflow backed by a single Prompt Studio tool. + + Provisioned over HTTP rather than via the ORM so setup exercises the same + surface as the app. Executes hermetically: the LLM is mocked, and + chunk_size=0 keeps embedding and vectordb out of the path. + """ + s = authed_session + base = platform.backend_url.rstrip("/") + org_id = _org_id(s, base) + prefix = f"{base}/api/v1/unstract/{org_id}" + sfx = uuid.uuid4().hex[:8] # adapter/tool names are unique per org + + def create_adapter(adapter_id: str, adapter_type: str, metadata: dict) -> str: + name = f"e2e-{adapter_type.lower()}-{sfx}" + resp = _post( + s, + f"{prefix}/adapter/", + json={ + "adapter_id": adapter_id, + "adapter_name": name, + "adapter_type": adapter_type, + "adapter_metadata": {"adapter_name": name, **metadata}, + }, + ) + assert resp.status_code == 201, f"adapter {adapter_type}: {resp.text}" + return resp.json()["id"] + + # api_base is required by the openai adapter's JSON schema, so adapter + # creation rejects the row without it even though the completion is mocked. + llm_id = create_adapter( + _LLM_ADAPTER, + "LLM", + { + "api_key": "sk-test", + "model": "gpt-4o", + "api_base": "https://api.openai.com/v1", + }, + ) + embed_id = create_adapter( + _EMBED_ADAPTER, + "EMBEDDING", + {"api_key": "sk-test", "model": "text-embedding-3-small"}, + ) + vdb_id = create_adapter(_VECTORDB_ADAPTER, "VECTOR_DB", {"wait_time": 0}) + x2t_id = create_adapter(_X2TEXT_ADAPTER, "X2TEXT", {"wait_time": 0}) + + resp = _post( + s, + f"{prefix}/prompt-studio/", + json={"tool_name": f"e2e-{sfx}", "description": "e2e execute", "author": "e2e"}, + ) + assert resp.status_code == 201, f"prompt-studio project: {resp.text}" + tool_id = resp.json()["tool_id"] + + # Patch the auto-created default profile; a second one would break export. + resp = s.get(f"{prefix}/prompt-studio/prompt-studio-profile/{tool_id}/", timeout=30) + resp.raise_for_status() + profile_id = next(p["profile_id"] for p in resp.json() if p.get("is_default")) + resp = _patch( + s, + f"{prefix}/prompt-studio/profile-manager/{profile_id}/", + json={ + "llm": llm_id, + "embedding_model": embed_id, + "vector_store": vdb_id, + "x2text": x2t_id, + "chunk_size": 0, + "chunk_overlap": 0, + }, + ) + assert resp.status_code == 200, f"profile patch: {resp.text}" + + prompt_key = "answer" + resp = _post( + s, + f"{prefix}/prompt-studio/prompt-studio-prompt/{tool_id}/", + json={ + "tool_id": tool_id, + "prompt_key": prompt_key, + "prompt": "What is this document about?", + "prompt_type": "PROMPT", + # text keeps the answer the raw completion; the structured types are + # re-serialised on the way out and would not match the mock verbatim. + "enforce_type": "text", + "sequence_number": 1, + "active": True, + "profile_manager": profile_id, + }, + ) + assert resp.status_code == 201, f"add prompt: {resp.text}" + prompt_id = resp.json()["prompt_id"] + + resp = _post( + s, + f"{prefix}/prompt-studio/export/{tool_id}", + json={"force_export": True, "is_shared_with_org": True}, + ) + assert resp.status_code == 200, f"export: {resp.text}" + # The filter arg is mandatory: without one the view returns no queryset. + resp = s.get( + f"{prefix}/prompt-studio/registry/", params={"custom_tool": tool_id}, timeout=30 + ) + resp.raise_for_status() + regs = resp.json() + reg_list = regs if isinstance(regs, list) else regs.get("results", []) + assert reg_list, f"registry empty for tool {tool_id}" + prompt_registry_id = reg_list[0]["prompt_registry_id"] + + resp = _post(s, f"{prefix}/workflow/", json={"workflow_name": f"e2e-wf-{sfx}"}) + assert resp.status_code == 201, f"create workflow: {resp.text}" + workflow_id = resp.json()["id"] + + resp = s.get( + f"{prefix}/workflow/endpoint/", params={"workflow": workflow_id}, timeout=30 + ) + resp.raise_for_status() + eps = resp.json() + eps = eps if isinstance(eps, list) else eps.get("results", []) + patched = 0 + for endpoint in eps: + if endpoint.get("workflow") == workflow_id: + resp = _patch( + s, + f"{prefix}/workflow/endpoint/{endpoint['id']}/", + json={"connection_type": "API"}, + ) + assert resp.status_code == 200, f"patch endpoint: {resp.text}" + patched += 1 + # A workflow always has exactly SOURCE+DESTINATION; a half-configured one + # here would only surface downstream in an execute test. + assert patched == 2, f"expected SOURCE+DESTINATION, patched {patched}: {eps}" + + resp = _post( + s, + f"{prefix}/tool_instance/", + json={"workflow_id": workflow_id, "tool_id": prompt_registry_id}, + ) + assert resp.status_code == 201, f"attach tool: {resp.text}" + + return ProvisionedWorkflow( + session=s, + base=base, + prefix=prefix, + org_id=org_id, + workflow_id=workflow_id, + tool_id=tool_id, + prompt_registry_id=prompt_registry_id, + profile_id=profile_id, + prompt_id=prompt_id, + prompt_key=prompt_key, + ) diff --git a/tests/e2e/etl/__init__.py b/tests/e2e/etl/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/e2e/etl/conftest.py b/tests/e2e/etl/conftest.py new file mode 100644 index 0000000000..7cff0c6920 --- /dev/null +++ b/tests/e2e/etl/conftest.py @@ -0,0 +1,200 @@ +"""Fixtures for the ETL e2e tests: an object store and a filesystem workflow. + +MinIO is the only storage connector the compose stack both boots and registers, +so it stands in for "a connector" here. The local-filesystem connector would +need no infra at all but is never registered, so it cannot be selected. +""" + +from __future__ import annotations + +import os +import uuid +from dataclasses import dataclass +from typing import TYPE_CHECKING + +import pytest +import requests + +from tests.e2e.conftest import ProvisionedWorkflow + +if TYPE_CHECKING: + from minio import Minio + +# Registry id from unstract.connectors (filesystems/minio). +_MINIO_CONNECTOR = "minio|c799f6e3-2b57-434e-aaac-b5daa415da19" + +# Created by the stack's minio-bootstrap; the ETL run only adds prefixes to it. +_BUCKET = "unstract" + + +@dataclass(frozen=True) +class MinioFixture: + """Where the test and the workers each reach the object store.""" + + client: Minio + bucket: str + # The workers resolve MinIO over the compose network, the test over a + # published port, so the two see different endpoints for the same store. + internal_url: str + access_key: str + secret_key: str + + +@dataclass(frozen=True) +class EtlWorkflow: + """A workflow whose endpoints read and write an object store.""" + + session: requests.Session + prefix: str + workflow_id: str + input_prefix: str + output_prefix: str + + +@pytest.fixture(scope="session") +def minio_store() -> MinioFixture: + """The stack's MinIO, or a skip when this runtime doesn't publish one.""" + rig_session = os.environ.get("UNSTRACT_RIG_SESSION_ID") + try: + import minio + except ImportError as exc: + # The rig injects minio>=7.2.0, so a missing client under it is a + # provisioning fault, not a legitimate "not installed" skip. + if rig_session: + pytest.fail(f"rig run but minio client not importable: {exc}") + pytest.skip("minio client not installed") + endpoint = os.environ.get("UNSTRACT_MINIO_ENDPOINT", "localhost:9000") + access_key = os.environ.get("UNSTRACT_MINIO_ACCESS_KEY", "minio") + secret_key = os.environ.get("UNSTRACT_MINIO_SECRET_KEY", "minio123") + # One scheme drives both halves so they can't disagree: the workers' URL and + # the test client's TLS flag. The compose stack serves MinIO plain; a + # TLS-fronted deployment sets UNSTRACT_MINIO_SCHEME=https. + scheme = os.environ.get("UNSTRACT_MINIO_SCHEME", "http") + internal_url = os.environ.get( + "UNSTRACT_MINIO_INTERNAL_URL", f"{scheme}://unstract-minio:9000" + ) + client = minio.Minio( + endpoint, access_key=access_key, secret_key=secret_key, secure=scheme == "https" + ) + try: + exists = client.bucket_exists(_BUCKET) + except Exception as exc: # noqa: BLE001 - narrow verdict below + # Under the rig the stack is meant to be up, so an unusable MinIO is a + # real failure to surface, not a store that legitimately isn't there. + if rig_session: + pytest.fail(f"rig provisioned the stack but MinIO is unusable: {exc}") + pytest.skip(f"MinIO unreachable at {endpoint}: {exc}") + if not exists: + pytest.skip(f"MinIO bucket {_BUCKET!r} missing at {endpoint}") + return MinioFixture( + client=client, + bucket=_BUCKET, + internal_url=internal_url, + access_key=access_key, + secret_key=secret_key, + ) + + +def _post(session: requests.Session, url: str, **kwargs: object) -> requests.Response: + return session.post(url, timeout=60, **kwargs) # CsrfSession stamps the header + + +def _patch(session: requests.Session, url: str, **kwargs: object) -> requests.Response: + return session.patch(url, timeout=60, **kwargs) + + +@pytest.fixture(scope="session") +def etl_workflow( + provisioned_workflow: ProvisionedWorkflow, minio_store: MinioFixture +) -> EtlWorkflow: + """A second workflow, reading and writing MinIO instead of the API. + + Reuses the exported tool but not the workflow: endpoints are per-workflow + and the API one is already committed to serving the deployment tests. + """ + pw = provisioned_workflow + s = pw.session + sfx = uuid.uuid4().hex[:8] + input_prefix = f"e2e-in-{sfx}" + output_prefix = f"e2e-out-{sfx}" + + def create_connector(connector_type: str) -> str: + name = f"e2e-{connector_type.lower()}-{sfx}" + resp = _post( + s, + f"{pw.prefix}/connector/", + json={ + "connector_id": _MINIO_CONNECTOR, + "connector_name": name, + "connector_type": connector_type, + "connector_metadata": { + "connectorName": name, + "key": minio_store.access_key, + "secret": minio_store.secret_key, + "endpoint_url": minio_store.internal_url, + "region_name": "us-east-1", + }, + }, + ) + assert resp.status_code == 201, f"connector {connector_type}: {resp.text}" + return resp.json()["id"] + + source_id = create_connector("INPUT") + destination_id = create_connector("OUTPUT") + + resp = _post(s, f"{pw.prefix}/workflow/", json={"workflow_name": f"e2e-etl-{sfx}"}) + assert resp.status_code == 201, f"create workflow: {resp.text}" + workflow_id = resp.json()["id"] + + resp = s.get( + f"{pw.prefix}/workflow/endpoint/", params={"workflow": workflow_id}, timeout=30 + ) + resp.raise_for_status() + body = resp.json() + endpoints = body if isinstance(body, list) else body.get("results", []) + by_type = { + e["endpoint_type"]: e for e in endpoints if e.get("workflow") == workflow_id + } + assert {"SOURCE", "DESTINATION"} <= set(by_type), endpoints + + resp = _patch( + s, + f"{pw.prefix}/workflow/endpoint/{by_type['SOURCE']['id']}/", + json={ + "connection_type": "FILESYSTEM", + "connector_instance_id": source_id, + "configuration": { + "folders": [f"/{minio_store.bucket}/{input_prefix}"], + "processSubDirectories": False, + "maxFiles": 1, + "fileProcessingOrder": "unordered", + }, + }, + ) + assert resp.status_code == 200, f"source endpoint: {resp.text}" + + resp = _patch( + s, + f"{pw.prefix}/workflow/endpoint/{by_type['DESTINATION']['id']}/", + json={ + "connection_type": "FILESYSTEM", + "connector_instance_id": destination_id, + "configuration": {"outputFolder": f"{minio_store.bucket}/{output_prefix}"}, + }, + ) + assert resp.status_code == 200, f"destination endpoint: {resp.text}" + + resp = _post( + s, + f"{pw.prefix}/tool_instance/", + json={"workflow_id": workflow_id, "tool_id": pw.prompt_registry_id}, + ) + assert resp.status_code == 201, f"attach tool: {resp.text}" + + return EtlWorkflow( + session=s, + prefix=pw.prefix, + workflow_id=workflow_id, + input_prefix=input_prefix, + output_prefix=output_prefix, + ) diff --git a/tests/e2e/etl/test_pipeline_etl_execute.py b/tests/e2e/etl/test_pipeline_etl_execute.py new file mode 100644 index 0000000000..6b68ab14df --- /dev/null +++ b/tests/e2e/etl/test_pipeline_etl_execute.py @@ -0,0 +1,94 @@ +"""E2E: run an ETL pipeline from a source connector to a destination connector. + +Distinct from the API paths: nothing is uploaded over HTTP and no result is +returned to the caller. The source connector discovers the file, and the proof +the pipeline ran end to end is the answer landing in the destination store. +""" + +from __future__ import annotations + +import io +import uuid + +import pytest + +from tests.e2e.conftest import wait_for_execution +from tests.e2e.etl.conftest import EtlWorkflow, MinioFixture + +pytestmark = [pytest.mark.e2e, pytest.mark.critical] + +_EXECUTION_TIMEOUT_SECONDS = 300 +_DOCUMENT = b"ETL probe. This document is about pipeline widgets and invoices." + + +@pytest.mark.critical_path("pipeline-etl-execute") +def test_etl_pipeline_writes_answer_to_destination( + etl_workflow: EtlWorkflow, minio_store: MinioFixture, llm_mock_response: str +) -> None: + _seed_source_document(etl_workflow, minio_store) + + pipeline_id = _create_pipeline(etl_workflow) + resp = etl_workflow.session.post( + f"{etl_workflow.prefix}/pipeline/execute/", + headers={"X-CSRFToken": etl_workflow.session.cookies.get("csrftoken", "")}, + json={"pipeline_id": pipeline_id}, + timeout=60, + ) + assert resp.status_code == 200, f"execute pipeline: {resp.text}" + execution_id = resp.json()["execution"]["execution_id"] + + execution = wait_for_execution( + etl_workflow.session, + etl_workflow.prefix, + execution_id, + timeout=_EXECUTION_TIMEOUT_SECONDS, + ) + assert execution.get("status") == "COMPLETED", execution + assert execution.get("successful_files") == 1, execution + + written = _read_destination_output(minio_store, etl_workflow.output_prefix) + assert llm_mock_response in written, written[:500] + + +def _seed_source_document(workflow: EtlWorkflow, store: MinioFixture) -> None: + """Put the document where the source connector will discover it.""" + store.client.put_object( + store.bucket, + f"{workflow.input_prefix}/probe.txt", + io.BytesIO(_DOCUMENT), + length=len(_DOCUMENT), + content_type="text/plain", + ) + + +def _create_pipeline(workflow: EtlWorkflow) -> str: + # No cron_string: that would schedule it instead of leaving it on demand. + resp = workflow.session.post( + f"{workflow.prefix}/pipeline/", + headers={"X-CSRFToken": workflow.session.cookies.get("csrftoken", "")}, + # Name is capped at 32 characters. + json={ + "pipeline_name": f"e2e-etl-{uuid.uuid4().hex[:8]}", + "workflow": workflow.workflow_id, + "pipeline_type": "ETL", + }, + timeout=60, + ) + assert resp.status_code == 201, f"create pipeline: {resp.text}" + return resp.json()["id"] + + +def _read_destination_output(store: MinioFixture, output_prefix: str) -> str: + """Return the destination object's body, whatever the connector named it.""" + objects = list( + store.client.list_objects( + store.bucket, prefix=f"{output_prefix}/", recursive=True + ) + ) + assert objects, f"nothing written under {output_prefix}/" + response = store.client.get_object(store.bucket, objects[0].object_name) + try: + return response.read().decode("utf-8", errors="replace") + finally: + response.close() + response.release_conn() diff --git a/tests/e2e/prompt_studio/__init__.py b/tests/e2e/prompt_studio/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/e2e/prompt_studio/test_prompt_studio_fetch_response.py b/tests/e2e/prompt_studio/test_prompt_studio_fetch_response.py new file mode 100644 index 0000000000..bb8177f12f --- /dev/null +++ b/tests/e2e/prompt_studio/test_prompt_studio_fetch_response.py @@ -0,0 +1,103 @@ +"""E2E: run a single prompt inside Prompt Studio and read its response back. + +This is the authoring surface, not workflow execution: the answer is produced by +the executor worker (so the LLM mock applies) and written back asynchronously by +the IDE callback worker, which is why the output is polled rather than returned. +""" + +from __future__ import annotations + +import io +import time + +import pytest +import requests + +from tests.e2e.conftest import ProvisionedWorkflow + +pytestmark = [pytest.mark.e2e, pytest.mark.critical] + +_OUTPUT_TIMEOUT_SECONDS = 240 + +# Upload sniffs content rather than trusting the declared type, and OSS accepts +# only PDF — so this has to be a real one, however empty. +_MINIMAL_PDF = ( + b"%PDF-1.4\n" + b"1 0 obj\n<< /Type /Catalog /Pages 2 0 R >>\nendobj\n" + b"2 0 obj\n<< /Type /Pages /Kids [3 0 R] /Count 1 >>\nendobj\n" + b"3 0 obj\n<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] >>\nendobj\n" + b"trailer\n<< /Size 4 /Root 1 0 R >>\n" + b"%%EOF\n" +) + + +@pytest.mark.critical_path("prompt-studio-fetch-response") +def test_fetch_response_returns_mocked_answer( + provisioned_workflow: ProvisionedWorkflow, llm_mock_response: str +) -> None: + pw = provisioned_workflow + document_id = _upload_document(pw) + + resp = _post( + pw, + f"{pw.prefix}/prompt-studio/fetch_response/{pw.tool_id}", + json={ + "id": pw.prompt_id, + "document_id": document_id, + "profile_manager": pw.profile_id, + }, + ) + # Accepted, not answered: the run is dispatched and written back later. + assert resp.status_code == 202, f"fetch_response: {resp.text}" + + output = _poll_for_output(pw, document_id) + assert output == llm_mock_response, output + + +def _upload_document(pw: ProvisionedWorkflow) -> str: + resp = pw.session.post( + f"{pw.prefix}/prompt-studio/file/{pw.tool_id}", + headers={"X-CSRFToken": pw.session.cookies.get("csrftoken", "")}, + files={"file": ("probe.pdf", io.BytesIO(_MINIMAL_PDF), "application/pdf")}, + timeout=60, + ) + assert resp.status_code == 200, f"upload: {resp.text}" + return resp.json()["data"][0]["document_id"] + + +def _poll_for_output(pw: ProvisionedWorkflow, document_id: str) -> str | None: + """Wait for the answer to be written back, then return it. + + Polls the output itself rather than the task status: the task completing and + the callback having stored its result are different moments, and the status + reports only the first. + """ + deadline = time.monotonic() + _OUTPUT_TIMEOUT_SECONDS + while time.monotonic() < deadline: + rows = _prompt_outputs(pw, document_id) + for row in rows: + if row.get("output") is not None: + return row["output"] + time.sleep(2) + pytest.fail(f"no prompt output within {_OUTPUT_TIMEOUT_SECONDS}s") + + +def _prompt_outputs(pw: ProvisionedWorkflow, document_id: str) -> list[dict]: + resp = pw.session.get( + f"{pw.prefix}/prompt-studio/prompt-output/", + params={ + "tool_id": pw.tool_id, + "document_manager": document_id, + "prompt_id": pw.prompt_id, + "is_single_pass_extract": "false", + }, + timeout=30, + ) + resp.raise_for_status() + body = resp.json() + return body if isinstance(body, list) else body.get("results", []) + + +def _post(pw: ProvisionedWorkflow, url: str, **kwargs: object) -> requests.Response: + headers = {"X-CSRFToken": pw.session.cookies.get("csrftoken", "")} + return pw.session.post(url, headers=headers, timeout=60, **kwargs) diff --git a/tests/e2e/workflows/__init__.py b/tests/e2e/workflows/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/e2e/workflows/test_workflow_execute.py b/tests/e2e/workflows/test_workflow_execute.py new file mode 100644 index 0000000000..c09a3c5583 --- /dev/null +++ b/tests/e2e/workflows/test_workflow_execute.py @@ -0,0 +1,60 @@ +"""E2E: create a workflow, execute a document through it, poll to COMPLETED. + +Covers the app-facing ``/workflow/execute/`` path, distinct from the public +API-deployment endpoint. + +Asserts execution status rather than the answer, which a manual execute never +exposes over HTTP; the API-deployment test covers the answer itself. A succeeding +file is proof enough that the mock held: a real completion would fail without a +key. +""" + +from __future__ import annotations + +import io + +import pytest + +from tests.e2e.conftest import ProvisionedWorkflow, wait_for_execution + +pytestmark = [pytest.mark.e2e, pytest.mark.critical] + +_EXECUTION_TIMEOUT_SECONDS = 180 + + +@pytest.mark.critical_path("workflow-create-execute") +def test_workflow_execute_completes( + provisioned_workflow: ProvisionedWorkflow, llm_mock_response: str +) -> None: + pw = provisioned_workflow + session = pw.session + csrf = {"X-CSRFToken": session.cookies.get("csrftoken", "")} + + # Two calls by contract: the first creates a PENDING execution, the second + # uploads and dispatches it. + resp = session.post( + f"{pw.prefix}/workflow/execute/", + headers=csrf, + data={"workflow_id": pw.workflow_id}, + timeout=60, + ) + assert resp.status_code == 200, f"create execution: {resp.text}" + execution_id = resp.json()["execution_id"] + + document = io.BytesIO(b"Hello invoice 123. This is a test document about widgets.") + resp = session.post( + f"{pw.prefix}/workflow/execute/", + headers=csrf, + data={"workflow_id": pw.workflow_id, "execution_id": execution_id}, + files={"files": ("probe.txt", document, "text/plain")}, + timeout=120, + ) + assert resp.status_code == 200, f"dispatch execution: {resp.text}" + + # Uses /execution//, not /workflow/execution// which filters by + # workflow_id and 404s. + final = wait_for_execution( + session, pw.prefix, execution_id, timeout=_EXECUTION_TIMEOUT_SECONDS + ) + assert final.get("status") == "COMPLETED", final + assert final.get("successful_files") == 1, final diff --git a/tests/groups.yaml b/tests/groups.yaml index 125baa2915..30f88c9a8c 100644 --- a/tests/groups.yaml +++ b/tests/groups.yaml @@ -167,7 +167,6 @@ groups: requires_platform: true critical: true depends_on: [e2e-smoke] - optional: true e2e-api-deployment: tier: e2e @@ -175,7 +174,6 @@ groups: requires_platform: true critical: true depends_on: [e2e-smoke] - optional: true e2e-prompt-studio: tier: e2e @@ -183,7 +181,13 @@ groups: requires_platform: true critical: true depends_on: [e2e-smoke] - optional: true + + e2e-etl: + tier: e2e + paths: [tests/e2e/etl] + requires_platform: true + critical: true + depends_on: [e2e-smoke] e2e-hurl: tier: e2e diff --git a/tests/rig/cli.py b/tests/rig/cli.py index 4bbc2e682e..64e6b78bcb 100644 --- a/tests/rig/cli.py +++ b/tests/rig/cli.py @@ -33,12 +33,15 @@ load_groups, ) from tests.rig.reporting import ( + STATUS_PASS, GroupResult, parse_junit, passed_critical_path_ids, write_summary, ) from tests.rig.runtime import ( + DEFAULT_LLM_MOCK_RESPONSE, + LLM_MOCK_RESPONSE_ENV, PlatformEndpoints, PlatformRuntime, TestcontainersRuntime, @@ -55,6 +58,9 @@ # @pytest.mark.critical_path marker args land in junit properties. _PYTEST_PLUGIN_DIR = REPO_ROOT / "tests" / "rig" / "pytest_plugin" +# saxutils.escape leaves quotes alone, which is wrong inside an attribute. +_XML_ATTR_ESCAPES = {'"': """} + @lru_cache(maxsize=1) def _rig_session_id() -> str: @@ -408,6 +414,11 @@ def cmd_run(args: argparse.Namespace) -> int: reports_dir.mkdir(parents=True, exist_ok=True) needs_platform = any(manifest.get(n).requires_platform for n in runnable) + # Every runtime (not just compose) needs the mock value present for the + # execute-path e2e, and an exported empty string counts as unset — else the + # workers skip injection and the tests skip green with zero coverage. + if needs_platform and not os.environ.get(LLM_MOCK_RESPONSE_ENV): + os.environ[LLM_MOCK_RESPONSE_ENV] = DEFAULT_LLM_MOCK_RESPONSE # A group can declare `requires_services` (stateful infra like Postgres/ # Redis) without needing the whole platform — provision just that infra via # testcontainers instead of standing up every compose service. Platform wins @@ -436,18 +447,36 @@ def cmd_run(args: argparse.Namespace) -> int: ) endpoints = runtime.up() - # TODO(runtime-gate-skip): groups run unconditionally in topo order; - # there is no skip-if-a-dependency-failed logic yet. The dep edge to - # the platform gate (e2e-smoke) is enforced structurally at load time - # (_validate_platform_groups_depend_on_gate), but at runtime a red gate - # does not stop its dependents from running against a half-up stack. - # Moot today since every platform dependent is `optional: true` - # (placeholder). When promoting them to active, track failed groups - # here and skip any group whose (transitive) deps include a failure, - # writing a synthetic "skipped (dependency failed)" result. Decide then: - # does an optional/non-failing-exit gate cascade, and how far? + # Groups run in topo order, so a dependency's result is always known by + # the time its dependents come up. A red dep means the precondition it + # gates (e.g. e2e-smoke: is the platform actually up?) does not hold, so + # running its dependents only yields noise against a half-up stack. + # + # `optional` cascades like any other dep: it governs whether a failure + # gates CI, not whether the stack can be trusted. The skip itself never + # sets overall_exit — the failing dep already did if it was non-optional, + # and a blocked group attests no coverage, so --fail-on-critical-gap + # still catches a critical path that silently stopped being covered. + # Groups something else depends on: an empty run (exit 5) of one of + # these leaves its precondition unverified, so it must block dependents + # rather than reading as a clean pass. + depended_on: set[str] = set() + for n in runnable: + depended_on |= manifest.transitive_deps(n) + failed_groups: set[str] = set() for name in runnable: group = manifest.get(name) + blocked_by = tuple(sorted(manifest.transitive_deps(name) & failed_groups)) + if blocked_by: + print( + f"\n[rig] SKIP {name} " + f"(dependency failed or was blocked: {', '.join(blocked_by)})" + ) + group_results.append(_blocked_result(group, blocked_by)) + failed_groups.add(name) + if not args.dry_run: + _persist_blocked_group(reports_dir, group, blocked_by) + continue print( f"\n[rig] running group: {name} " f"(tier={group.tier}, runner={group.runner})" @@ -467,6 +496,16 @@ def cmd_run(args: argparse.Namespace) -> int: ) if result is not None: group_results.append(result) + # Tracked regardless of `optional` — see the cascade note above. + # An empty run (exit 5) blocks dependents only when something depends + # on this group: a gate that collected nothing verified nothing, but + # a leaf placeholder collecting nothing gates no one. + if ( + exit_code not in _NON_FAILING_PYTEST_EXIT_CODES + or (result is not None and (result.errors or result.failed)) + or (exit_code == 5 and name in depended_on) + ): + failed_groups.add(name) # `optional: true` groups run and surface their result in the # summary, but never gate the overall exit. This honors the # developer intent for groups that need infra we don't provision in @@ -625,6 +664,21 @@ def cmd_run(args: argparse.Namespace) -> int: # ── execution helpers ───────────────────────────────────────────────────────── +def _blocked_result(group: GroupDefinition, blocked_by: tuple[str, ...]) -> GroupResult: + """A group that never ran because a dependency failed or was itself blocked.""" + return GroupResult( + name=group.name, + tier=group.tier, + exit_code=0, + passed=0, + failed=0, + errors=0, + skipped=0, + duration_seconds=0.0, + blocked_by=blocked_by, + ) + + def _db_env_from_postgres_url(url: str) -> dict[str, str]: """Translate a provisioned Postgres URL into the discrete ``DB_*`` vars Django reads (``backend/settings/base.py``). @@ -679,7 +733,7 @@ def _coverage_attesting_groups(results: list[GroupResult]) -> set[str]: # "empty" (exit 5) stays non-failing for the build, but a covering group # that collected zero tests must not attest critical-path coverage — a # broken marker expression would otherwise report ✅ with zero tests run. - return {r.name for r in results if r.status == "pass"} + return {r.name for r in results if r.status == STATUS_PASS} def _marker_proven_paths( @@ -804,6 +858,7 @@ def _execute_group( "pytest-cov>=4.1.0", "pytest-mock>=3.12.0", "requests>=2.31.0", + "minio>=7.2.0", ) @@ -936,7 +991,7 @@ def _write_synthetic_junit(path: Path, group_name: str, exit_code: int) -> None: is_failure = exit_code != 0 and exit_code != 5 failures = 1 if is_failure else 0 failure_tag = f'' if is_failure else "" - name = saxutils.escape(group_name, {'"': """}) + name = saxutils.escape(group_name, _XML_ATTR_ESCAPES) path.write_text( '\n' f' None: ) +def _write_blocked_junit( + path: Path, group_name: str, blocked_by: tuple[str, ...] +) -> None: + """Synthesise a junit for a group that never ran because a dependency did + not pass. + + Carries the blocking group names as a property (zero test counters) so CI's + `rig report`, which rebuilds results purely from junit, renders the ⏭️ + blocked row instead of dropping the group. + """ + name = saxutils.escape(group_name, _XML_ATTR_ESCAPES) + value = saxutils.escape(",".join(blocked_by), _XML_ATTR_ESCAPES) + path.write_text( + '\n' + f'\n' + " \n" + f' \n' + " \n" + "\n" + ) + + +def _persist_blocked_group( + reports_dir: Path, group: GroupDefinition, blocked_by: tuple[str, ...] +) -> None: + """Write a blocked group's junit + exit.txt so it survives the artifact + round-trip into CI's `rig report`. Best-effort: a read-only reports dir + shouldn't abort the run. + """ + group_reports = reports_dir / group.name + try: + group_reports.mkdir(parents=True, exist_ok=True) + _write_blocked_junit(group_reports / "junit.xml", group.name, blocked_by) + (group_reports / "exit.txt").write_text("0") + except OSError as err: + print( + f"[rig] could not persist blocked group {group.name}: {err}", + file=sys.stderr, + ) + + def _spawn(cmd: list[str], *, env: dict[str, str], cwd: Path, timeout: int) -> int: start = time.monotonic() try: diff --git a/tests/rig/groups.py b/tests/rig/groups.py index 05fee9030c..07b70283ce 100644 --- a/tests/rig/groups.py +++ b/tests/rig/groups.py @@ -81,6 +81,18 @@ def names(self) -> list[str]: def names_by_tier(self, tier: Tier) -> list[str]: return sorted(n for n, g in self.groups.items() if g.tier == tier) + def transitive_deps(self, name: str) -> set[str]: + """Return every group ``name`` depends on, directly or otherwise.""" + self.get(name) # raises on unknown + deps: set[str] = set() + frontier = [name] + while frontier: + for dep in self.get(frontier.pop()).depends_on: + if dep not in deps: + deps.add(dep) + frontier.append(dep) + return deps + def expand(self, selected: list[str]) -> list[str]: """Return ``selected`` plus the transitive closure of their ``depends_on``, in topological order (dependencies before dependents). @@ -207,13 +219,11 @@ def _validate_platform_groups_depend_on_gate( graph: it guarantees the manifest can never ship a platform test that bypasses the smoke gate, and that the gate runs first in topological order. - Note: runtime skip-on-gate-failure (not running dependents when the gate - reports red) is NOT implemented yet — ``cmd_run`` executes every runnable - group unconditionally. Today this is moot because every dependent is - ``optional: true`` (a placeholder), so a smoke failure doesn't gate CI and - the dependents collect nothing. When those groups are promoted to active, - add dep-failure tracking in ``cmd_run`` so dependents skip cleanly rather - than running against a half-up stack — see the TODO there. + Runtime skip-on-gate-failure is enforced separately: ``cmd_run`` tracks + failed groups and skips (blocks) any dependent whose transitive deps + include one, so a red gate stops its dependents from running against a + half-up stack. This structural check guarantees the graph such tracking + relies on is actually wired. If the manifest declares ``requires_platform`` groups but doesn't actually define the gate, that's a manifest error — silently disabling the check diff --git a/tests/rig/reporting.py b/tests/rig/reporting.py index 21ef3f6fa9..4c8ed8eda4 100644 --- a/tests/rig/reporting.py +++ b/tests/rig/reporting.py @@ -24,12 +24,18 @@ log = logging.getLogger(__name__) -ResultStatus = Literal["pass", "empty", "fail"] +ResultStatus = Literal["pass", "empty", "fail", "blocked"] + +STATUS_PASS: ResultStatus = "pass" +STATUS_EMPTY: ResultStatus = "empty" +STATUS_FAIL: ResultStatus = "fail" +STATUS_BLOCKED: ResultStatus = "blocked" _STATUS_ICONS: dict[ResultStatus, str] = { - "pass": "✅", - "empty": "⚪", - "fail": "❌", + STATUS_PASS: "✅", + STATUS_EMPTY: "⚪", + STATUS_FAIL: "❌", + STATUS_BLOCKED: "⏭️", } @@ -43,14 +49,28 @@ class GroupResult: errors: int skipped: int duration_seconds: float + # Names of failed dependencies that stopped this group from running. Never + # green, so it attests no coverage and the path reports as a gap. + blocked_by: tuple[str, ...] = () + + def __post_init__(self) -> None: + # A blocked group never ran, so its counters must be zero; otherwise + # `status` would report "blocked" while the markdown row prints stray + # failures and coverage attestation drops it. + if self.blocked_by and ( + self.exit_code or self.passed or self.failed or self.errors or self.skipped + ): + raise ValueError("a blocked group never ran; its counters must be zero") @property def status(self) -> ResultStatus: + if self.blocked_by: + return STATUS_BLOCKED if self.exit_code == 5: # pytest "no tests collected" - return "empty" + return STATUS_EMPTY if self.exit_code == 0 and self.failed == 0 and self.errors == 0: - return "pass" - return "fail" + return STATUS_PASS + return STATUS_FAIL @property def status_icon(self) -> str: @@ -118,6 +138,23 @@ def parse_junit(group_name: str, tier: str, reports_dir: Path) -> GroupResult | skipped += sk passed += max(total - f - e - sk, 0) + blocked_by = _read_blocked_by(root) + if blocked_by: + # A blocked group's synthetic junit carries zero counters; return it as + # blocked so CI's `rig report` (which rebuilds from junit) renders the + # ⏭️ row instead of dropping the group entirely. + return GroupResult( + name=group_name, + tier=tier, + exit_code=0, + passed=0, + failed=0, + errors=0, + skipped=0, + duration_seconds=duration, + blocked_by=blocked_by, + ) + if not saw_attributes: # Junit that parses but has no counters anywhere is almost certainly a # truncated write. Don't count it as green. @@ -139,6 +176,17 @@ def parse_junit(group_name: str, tier: str, reports_dir: Path) -> GroupResult | ) +def _read_blocked_by(root: ET.Element) -> tuple[str, ...]: + """Read the ``rig-blocked-by`` property a blocked group's synthetic junit + carries, if any. + """ + for prop in root.iter("property"): + if prop.get("name") == "rig-blocked-by": + value = prop.get("value") or "" + return tuple(d for d in value.split(",") if d) + return () + + def passed_critical_path_ids(group_name: str, reports_dir: Path) -> set[str]: """Extract critical-path ids attested by *passing* tests in a group's junit. @@ -199,7 +247,9 @@ def write_summary( summary_json.write_text( json.dumps( { - "groups": [asdict(r) for r in group_results], + # `status` is a property, so it isn't in asdict; emit it + # explicitly or a consumer can't tell blocked from empty. + "groups": [{**asdict(r), "status": r.status} for r in group_results], "critical_paths": [ { "id": s.path.id, @@ -245,9 +295,16 @@ def _render_markdown( ) lines.append("|---|---|---|---:|---:|---:|---:|---:|") for r in group_results: + # Without the reason a blocked row is all zeros, indistinguishable + # from a group that simply had nothing to run. + note = ( + f" — blocked by {', '.join(f'`{d}`' for d in r.blocked_by)}" + if r.blocked_by + else "" + ) lines.append( - f"| {r.status_icon} | `{r.name}` | {r.tier} | {r.passed} | {r.failed} " - f"| {r.errors} | {r.skipped} | {r.duration_seconds:.1f} |" + f"| {r.status_icon} | `{r.name}`{note} | {r.tier} | {r.passed} " + f"| {r.failed} | {r.errors} | {r.skipped} | {r.duration_seconds:.1f} |" ) totals = _totals(group_results) lines.append( diff --git a/tests/rig/runtime.py b/tests/rig/runtime.py index 2643d9da99..aaabb1df94 100644 --- a/tests/rig/runtime.py +++ b/tests/rig/runtime.py @@ -33,6 +33,10 @@ COMPOSE_OVERLAY = REPO_ROOT / "tests" / "compose" / "docker-compose.test.yaml" BASE_COMPOSE = REPO_ROOT / "docker" / "docker-compose.yaml" +# Shared by the workers and the tests, so the exact completion is assertable. +LLM_MOCK_RESPONSE_ENV = "UNSTRACT_LLM_MOCK_RESPONSE" +DEFAULT_LLM_MOCK_RESPONSE = "MOCK_LLM_OK" + @dataclass(frozen=True) class InfraEndpoints: @@ -144,6 +148,7 @@ def down(self) -> None: files = ["-f", str(BASE_COMPOSE)] if COMPOSE_OVERLAY.exists(): files += ["-f", str(COMPOSE_OVERLAY)] + self._dump_logs(files) _run( [ "docker", @@ -158,6 +163,25 @@ def down(self) -> None: check=False, ) + def _dump_logs(self, files: list[str]) -> None: + """Capture service logs while the containers still exist. + + A failing e2e is usually explained by a worker log, and `down -v` is the + last thing that can read them -- a CI step afterwards gets an empty file. + """ + target = REPO_ROOT / "reports" / "docker-compose-logs.txt" + try: + target.parent.mkdir(parents=True, exist_ok=True) + completed = subprocess.run( # noqa: S603 + ["docker", "compose", "-p", self.project_name, *files, "logs", "--no-color"], + capture_output=True, + text=True, + check=False, + ) + target.write_text(completed.stdout) + except OSError as exc: + log.warning("Could not capture compose logs: %s", exc) + class TestcontainersRuntime: """Spin up stateful infra via testcontainers; services run locally. diff --git a/tests/rig/tests/test_cli.py b/tests/rig/tests/test_cli.py index 31d4b1f0ab..020ebd493a 100644 --- a/tests/rig/tests/test_cli.py +++ b/tests/rig/tests/test_cli.py @@ -690,3 +690,98 @@ def test_inject_infra_env_wires_provisioned_redis() -> None: assert env["REDIS_HOST"] == "redis.internal" assert env["REDIS_PORT"] == "49999" assert env["CELERY_BROKER_BASE_URL"] == "redis://redis.internal:49999" + + +def test_failed_gate_blocks_dependents_and_cascades(tmp_path: Path, monkeypatch) -> None: + """A red gate skips its dependents (and their dependents), records them as + blocked with the right ``blocked_by``, and — all groups optional — leaves + the overall exit at 0. Pins the cascade `cmd_run` wires together. + """ + import json + + from tests.rig.reporting import GroupResult + + test_dir = Path(__file__).parent + manifest_yaml = ( + "version: 1\n" + "groups:\n" + " e2e-smoke:\n" + " tier: e2e\n" + f" workdir: {test_dir}\n" + " paths: [.]\n" + " optional: true\n" + " e2e-mid:\n" + " tier: e2e\n" + f" workdir: {test_dir}\n" + " paths: [.]\n" + " optional: true\n" + " depends_on: [e2e-smoke]\n" + " e2e-leaf:\n" + " tier: e2e\n" + f" workdir: {test_dir}\n" + " paths: [.]\n" + " optional: true\n" + " depends_on: [e2e-mid]\n" + ) + (tmp_path / "groups.yaml").write_text(manifest_yaml) + (tmp_path / "critical_paths.yaml").write_text("version: 1\npaths: []\n") + + import tests.rig.cli as cli_mod + import tests.rig.critical_paths as cp_mod + import tests.rig.groups as groups_mod + + monkeypatch.setattr(groups_mod, "DEFAULT_MANIFEST", tmp_path / "groups.yaml") + monkeypatch.setattr(cp_mod, "DEFAULT_REGISTRY", tmp_path / "critical_paths.yaml") + + executed: list[str] = [] + + def fake_execute_group(group, **kwargs): + executed.append(group.name) + result = GroupResult( + name=group.name, + tier=group.tier, + exit_code=1, + passed=0, + failed=1, + errors=0, + skipped=0, + duration_seconds=0.01, + ) + return result, 1 + + monkeypatch.setattr(cli_mod, "_execute_group", fake_execute_group) + + reports_dir = tmp_path / "reports" + args = cli_mod._build_parser().parse_args( + [ + "run", + "e2e-smoke", + "e2e-mid", + "e2e-leaf", + "--no-coverage", + "--no-parallel", + "--reports-dir", + str(reports_dir), + "--baseline", + str(reports_dir / "previous-summary.json"), + ] + ) + exit_code = cli_mod.cmd_run(args) + + # Only the gate ran; its dependents were blocked before execution. + assert executed == ["e2e-smoke"], executed + # All optional, so a red gate never gates the overall exit. + assert exit_code == 0, exit_code + + summary = json.loads((reports_dir / "summary.json").read_text()) + by_name = {g["name"]: g for g in summary["groups"]} + assert by_name["e2e-mid"]["status"] == "blocked" + assert by_name["e2e-mid"]["blocked_by"] == ["e2e-smoke"] + # The intermediate block cascades: leaf names both, proving a blocked group + # is itself added to the failed set. + assert by_name["e2e-leaf"]["status"] == "blocked" + assert by_name["e2e-leaf"]["blocked_by"] == ["e2e-mid", "e2e-smoke"] + + # The blocked groups persisted a junit so CI's `rig report` can render them. + assert (reports_dir / "e2e-mid" / "junit.xml").exists() + assert (reports_dir / "e2e-leaf" / "junit.xml").exists() diff --git a/tests/rig/tests/test_groups.py b/tests/rig/tests/test_groups.py index 62b3d20513..eabd5b7f05 100644 --- a/tests/rig/tests/test_groups.py +++ b/tests/rig/tests/test_groups.py @@ -80,6 +80,34 @@ def test_expand_topological_order(tmp_path: Path) -> None: assert expanded == ["leaf", "mid", "root"] +def test_transitive_deps_reaches_indirect_dependencies(tmp_path: Path) -> None: + manifest = _write_manifest( + tmp_path, + """ + version: 1 + groups: + leaf: + tier: unit + paths: [x] + optional: true + mid: + tier: unit + paths: [x] + depends_on: [leaf] + optional: true + root: + tier: e2e + paths: [x] + depends_on: [mid] + optional: true + """, + ) + loaded = load_groups(manifest) + # Indirect deps count: a red `leaf` must be able to block `root`. + assert loaded.transitive_deps("root") == {"mid", "leaf"} + assert loaded.transitive_deps("leaf") == set() + + def test_invalid_tier_rejected(tmp_path: Path) -> None: manifest = _write_manifest( tmp_path, diff --git a/tests/rig/tests/test_reporting.py b/tests/rig/tests/test_reporting.py index 4913100dc0..499ce58ba8 100644 --- a/tests/rig/tests/test_reporting.py +++ b/tests/rig/tests/test_reporting.py @@ -106,6 +106,37 @@ def test_status_icon_round_trips() -> None: assert empty_result.status_icon == "⚪" +def test_blocked_result_is_never_green() -> None: + # A blocked group ran nothing, so its zero counters otherwise read as a + # pass and would attest coverage its tests never proved. + blocked = GroupResult("g", "e2e", 0, 0, 0, 0, 0, 0.0, blocked_by=("e2e-smoke",)) + assert blocked.status == "blocked" + assert blocked.status_icon == "⏭️" + + +def test_blocked_result_rejects_nonzero_counters() -> None: + import pytest + + with pytest.raises(ValueError, match="never ran"): + GroupResult("g", "e2e", 0, 0, 1, 0, 0, 0.0, blocked_by=("e2e-smoke",)) + + +def test_blocked_junit_round_trips_through_parse(tmp_path: Path) -> None: + # The blocked-by property a synthetic junit carries must survive `rig + # report`, which rebuilds results purely from junit on the CI runner. + from tests.rig.cli import _write_blocked_junit + + (tmp_path / "g").mkdir() + _write_blocked_junit( + tmp_path / "g" / "junit.xml", "g", ("e2e-smoke", "e2e-mid") + ) + (tmp_path / "g" / "exit.txt").write_text("0") + result = parse_junit("g", "e2e", tmp_path) + assert result is not None + assert result.status == "blocked" + assert result.blocked_by == ("e2e-smoke", "e2e-mid") + + def test_passed_critical_path_ids_collects_only_passing_marked_tests( tmp_path: Path, ) -> None: diff --git a/unstract/core/src/unstract/core/data_models.py b/unstract/core/src/unstract/core/data_models.py index b56d0487ff..c8e26f4d6a 100644 --- a/unstract/core/src/unstract/core/data_models.py +++ b/unstract/core/src/unstract/core/data_models.py @@ -407,6 +407,15 @@ def _is_failure(cls, status: str) -> bool: ExecutionStatus.is_failure = classmethod(_is_failure) +# Statuses a run cannot move out of. `is_completed` is the boolean form. +def _terminal_statuses(cls) -> frozenset["ExecutionStatus"]: + """Return the set of statuses that represent a finished run.""" + return frozenset({cls.COMPLETED, cls.ERROR, cls.STOPPED}) + + +ExecutionStatus.terminal_statuses = classmethod(_terminal_statuses) + + def is_failure_run(execution_status: str | None, failed_files: int | None) -> bool: """Canonical "did this run fail?" rule, shared across every dispatch path. diff --git a/unstract/sdk1/src/unstract/sdk1/llm.py b/unstract/sdk1/src/unstract/sdk1/llm.py index b1180a03da..1befc96eb3 100644 --- a/unstract/sdk1/src/unstract/sdk1/llm.py +++ b/unstract/sdk1/src/unstract/sdk1/llm.py @@ -4,6 +4,7 @@ from collections.abc import Callable, Generator, Mapping, Sequence from dataclasses import dataclass, field from enum import Enum +from functools import lru_cache from typing import Any, NoReturn, cast import litellm @@ -31,6 +32,51 @@ logger = logging.getLogger(__name__) +# Lets tests force a deterministic completion without a provider or a secret. +# Unset in production, where this is a no-op. +_MOCK_RESPONSE_ENV = "UNSTRACT_LLM_MOCK_RESPONSE" +# Second condition on the hatch, so a stray mock var alone can't fake +# completions and their billing. Deployments that set neither fail closed. +_ENVIRONMENT_ENV = "ENVIRONMENT" +_MOCK_ALLOWED_ENVIRONMENTS = frozenset({"test", "development"}) + + +@lru_cache(maxsize=1) +def _warn_mock_active() -> None: + # Once per process: the hatch is silent otherwise, and a stray env var in + # production would fake every completion and its billing. + logger.warning( + "%s is set — returning canned completions instead of calling the " + "provider, with synthetic token usage. Unset it outside tests.", + _MOCK_RESPONSE_ENV, + ) + + +@lru_cache(maxsize=1) +def _warn_mock_refused(environment: str) -> None: + # Loud rather than silent: the var being set at all means someone expected + # mocking, and they need to know why the bill is real. + logger.warning( + "%s is set but %s=%r is not one of %s — calling the provider for real.", + _MOCK_RESPONSE_ENV, + _ENVIRONMENT_ENV, + environment, + sorted(_MOCK_ALLOWED_ENVIRONMENTS), + ) + + +def _inject_mock_response(completion_kwargs: dict[str, object]) -> None: + mock = os.getenv(_MOCK_RESPONSE_ENV) + if not mock or "mock_response" in completion_kwargs: + return + environment = os.getenv(_ENVIRONMENT_ENV, "").strip().lower() + if environment not in _MOCK_ALLOWED_ENVIRONMENTS: + _warn_mock_refused(environment) + return + _warn_mock_active() + completion_kwargs["mock_response"] = mock + + # Drop unsupported params rather than raising errors. # Set once at module level instead of per-call to avoid repeated # global mutation in concurrent environments. @@ -328,6 +374,7 @@ def complete(self, prompt: str, **kwargs: object) -> dict[str, object]: ) completion_kwargs = self.adapter.validate({**self.kwargs, **kwargs}) + _inject_mock_response(completion_kwargs) completion_kwargs.pop("cost_model", None) completion_kwargs.pop("context_window", None) @@ -451,6 +498,7 @@ def complete_vision( ) completion_kwargs = self.adapter.validate({**self.kwargs, **kwargs}) + _inject_mock_response(completion_kwargs) completion_kwargs.pop("cost_model", None) completion_kwargs.pop("context_window", None) @@ -519,6 +567,7 @@ def stream_complete( ) completion_kwargs = self.adapter.validate({**self.kwargs, **kwargs}) + _inject_mock_response(completion_kwargs) completion_kwargs.pop("cost_model", None) completion_kwargs.pop("context_window", None) @@ -591,6 +640,7 @@ async def acomplete(self, prompt: str, **kwargs: object) -> dict[str, object]: ) completion_kwargs = self.adapter.validate({**self.kwargs, **kwargs}) + _inject_mock_response(completion_kwargs) completion_kwargs.pop("cost_model", None) completion_kwargs.pop("context_window", None) diff --git a/unstract/sdk1/tests/test_mock_response.py b/unstract/sdk1/tests/test_mock_response.py new file mode 100644 index 0000000000..6a4e0b1a59 --- /dev/null +++ b/unstract/sdk1/tests/test_mock_response.py @@ -0,0 +1,169 @@ +"""The UNSTRACT_LLM_MOCK_RESPONSE escape hatch and the litellm mock contract. + +Hermetic execute-path coverage rests on both, so pin them here: a litellm bump +that broke either would otherwise surface far from its cause. +""" + +from __future__ import annotations + +from functools import lru_cache +from importlib import import_module + +import litellm +import pytest + + +@lru_cache(maxsize=1) +def _load_llm_module() -> object: + import sys + from types import ModuleType + + # Stub python-magic for the import only, so we neither depend on libmagic + # nor leave a stub shadowing a real `magic` for the rest of the process. + inserted = "magic" not in sys.modules + if inserted: + sys.modules["magic"] = ModuleType("magic") + try: + return import_module("unstract.sdk1.llm") + finally: + if inserted: + del sys.modules["magic"] + + +def _inject(kwargs: dict[str, object]) -> dict[str, object]: + _load_llm_module()._inject_mock_response(kwargs) + return kwargs + + +@pytest.fixture(autouse=True) +def _reset_warn_cache() -> None: + _load_llm_module()._warn_mock_active.cache_clear() + _load_llm_module()._warn_mock_refused.cache_clear() + + +@pytest.fixture(autouse=True) +def _allowed_environment(monkeypatch: pytest.MonkeyPatch) -> None: + """The hatch needs a permitted ENVIRONMENT; the guard itself is tested below.""" + monkeypatch.setenv("ENVIRONMENT", "test") + + +def test_inject_is_noop_when_env_unset(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("UNSTRACT_LLM_MOCK_RESPONSE", raising=False) + assert _inject({"model": "gpt-4o"}) == {"model": "gpt-4o"} + + +def test_inject_is_noop_when_env_empty(monkeypatch: pytest.MonkeyPatch) -> None: + # The overlay exports `UNSTRACT_LLM_MOCK_RESPONSE=` (empty) into workers when + # the var is unset upstream; that empty string must stay a no-op, or every + # local stack run would silently mock completions. + monkeypatch.setenv("UNSTRACT_LLM_MOCK_RESPONSE", "") + assert _inject({"model": "gpt-4o"}) == {"model": "gpt-4o"} + + +def test_inject_sets_mock_response_when_env_set( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("UNSTRACT_LLM_MOCK_RESPONSE", "canned answer") + assert _inject({"model": "gpt-4o"})["mock_response"] == "canned answer" + + +@pytest.mark.parametrize("environment", ["production", "staging", "", " "]) +def test_mock_refused_outside_allowed_environments( + monkeypatch: pytest.MonkeyPatch, environment: str +) -> None: + # The whole point of the guard: a stray mock var in a real deployment must + # not fake completions and their billing. + monkeypatch.setenv("UNSTRACT_LLM_MOCK_RESPONSE", "canned") + monkeypatch.setenv("ENVIRONMENT", environment) + assert _inject({"model": "gpt-4o"}) == {"model": "gpt-4o"} + + +def test_mock_refused_when_environment_unset(monkeypatch: pytest.MonkeyPatch) -> None: + # Production k8s sets no ENVIRONMENT at all, so unset must fail closed. + monkeypatch.setenv("UNSTRACT_LLM_MOCK_RESPONSE", "canned") + monkeypatch.delenv("ENVIRONMENT", raising=False) + assert _inject({"model": "gpt-4o"}) == {"model": "gpt-4o"} + + +def test_refusal_is_warned_so_a_real_bill_is_never_silent( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + monkeypatch.setenv("UNSTRACT_LLM_MOCK_RESPONSE", "canned") + monkeypatch.setenv("ENVIRONMENT", "production") + with caplog.at_level("WARNING", logger="unstract.sdk1.llm"): + _inject({"model": "gpt-4o"}) + assert any("not one of" in r.message for r in caplog.records), caplog.text + + +@pytest.mark.parametrize("environment", ["test", "development", "TEST", " Development "]) +def test_mock_applies_in_allowed_environments( + monkeypatch: pytest.MonkeyPatch, environment: str +) -> None: + # `development` is what base compose sets on the workers that run the + # injection; a guard that only accepted `test` would kill the local stack. + monkeypatch.setenv("UNSTRACT_LLM_MOCK_RESPONSE", "canned") + monkeypatch.setenv("ENVIRONMENT", environment) + assert _inject({"model": "gpt-4o"})["mock_response"] == "canned" + + +def test_inject_does_not_clobber_explicit_mock_response( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("UNSTRACT_LLM_MOCK_RESPONSE", "from-env") + assert _inject({"mock_response": "explicit"})["mock_response"] == "explicit" + + +def test_inject_warns_once_while_active( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + monkeypatch.setenv("UNSTRACT_LLM_MOCK_RESPONSE", "from-env") + with caplog.at_level("WARNING", logger="unstract.sdk1.llm"): + _inject({"model": "gpt-4o"}) + _inject({"model": "gpt-4o"}) + warnings = [r for r in caplog.records if "UNSTRACT_LLM_MOCK_RESPONSE" in r.message] + # Once per process, not once per completion: a worker would flood otherwise. + assert len(warnings) == 1, caplog.text + + +def test_inject_does_not_warn_when_env_unset( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + monkeypatch.delenv("UNSTRACT_LLM_MOCK_RESPONSE", raising=False) + with caplog.at_level("WARNING", logger="unstract.sdk1.llm"): + _inject({"model": "gpt-4o"}) + assert not caplog.records, caplog.text + + +def test_every_completion_path_injects_the_mock() -> None: + # The hook is worthless if a completion method doesn't call it; deleting any + # call site would otherwise leave this suite green and only surface as an + # opaque "no API key" worker failure in the e2e tier. + import inspect + + llm = _load_llm_module() + for name in ("complete", "complete_vision", "stream_complete", "acomplete"): + src = inspect.getsource(getattr(llm.LLM, name)) + assert "_inject_mock_response(completion_kwargs)" in src, name + + +def test_litellm_mock_contract_returns_string_and_fixed_usage() -> None: + # 10/20/30 are litellm's defaults, asserted verbatim by the e2e tests. + resp = litellm.completion( + model="gpt-4o", + messages=[{"role": "user", "content": "anything"}], + mock_response="canned answer", + ) + assert resp["choices"][0]["message"]["content"] == "canned answer" + assert resp["usage"]["prompt_tokens"] == 10 + assert resp["usage"]["completion_tokens"] == 20 + assert resp["usage"]["total_tokens"] == 30 + + +def test_litellm_mock_error_sentinel_raises() -> None: + # Error paths need the sentinel to raise rather than complete normally. + with pytest.raises(litellm.RateLimitError): + litellm.completion( + model="gpt-4o", + messages=[{"role": "user", "content": "anything"}], + mock_response="litellm.RateLimitError", + )