feat(test-seeding): add gated POST /test/seed for e2e test resource seeding#301
feat(test-seeding): add gated POST /test/seed for e2e test resource seeding#301dm36 wants to merge 4 commits into
Conversation
…eeding E2E tests for FGAC behavior (scaleapi/agentex#380 and follow-ups) need to assert authz outcomes against real resource rows — denied-get on a real event proves the 404 is the authz collapse, not "row doesn't exist." Today there's no public POST /events; events are persisted by the worker on ACP stream-back, which makes black-box seeding awkward. This adds a test-only seeding endpoint at POST /test/seed that writes resource rows directly via the existing repository layer, bypassing the natural-flow side effects (ACP forward, etc.). Defense in depth on the gate: - `ENABLE_TEST_SEEDING` env flag (default False). - `ENVIRONMENT != production` hard-check in app.py at router-mount time. Prod never has the route at all, regardless of flag. - `X-Test-Seed-Token` shared-secret header, `hmac.compare_digest`. - All gate failures return 404 (not 401/403) so the route's existence isn't advertised. Resource types are discriminated; events ship first. The use case is structured so adding `seed_task` / `seed_api_key` / etc. is mechanical (includes a TODO for the FGAC register_resource pattern those will need). Audit: - Every seeded event has `{"seeded": true, "seeded_at": <iso8601>}` injected into its `content` payload — downstream filterable. - Structured info log per seed call with principal + resource id. Events are intentionally NOT registered as a top-level FGAC resource here — they delegate read authz to the parent agent, matching the natural flow. Verified against AgentexResourceType and routes/events.py. Tests: 9 new integration tests cover gate-flag-off, prod-env-hard-gate, wrong/missing/unconfigured token, happy path with + without content, audit-marker presence, and id override. Full integration API suite (243 tests) green — no regressions. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Snapshot the overrides dict at fixture setup, restore in finally. Covers both the fixture's own injections (TestSeedingUseCase, get_seeding_env_vars) AND any mid-test mutations from `_override_env`. Without this, entries leak into fastapi_app's global state and can bleed into other test modules that share the same app instance. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
| if ( | ||
| _test_seeding_env_vars is not None | ||
| and _test_seeding_env_vars.ENABLE_TEST_SEEDING | ||
| and _test_seeding_env_vars.ENVIRONMENT != Environment.PROD |
There was a problem hiding this comment.
from Claude, but seems legit.
The prod "hard gate" is a deny-list on an unvalidated string: ENVIRONMENT is typed str | None and populated raw from os.environ (no enum validation), so unset/None, "prod", "Production", or any typo passes != Environment.PROD. The layer described as the strongest guarantee only holds if prod sets exactly "production". Invert to an allow-list so unknown environments fail closed: if env_vars.ENVIRONMENT not in (Environment.DEV, Environment.STAGING): raise not_found - same change at the mount site in app.py.
| class TestSeedingUseCase: # noqa: PT001 — not a pytest class; "Test" prefix is the use-case domain name | ||
| __test__ = False # tell pytest not to collect this as a test class | ||
| """Test-only resource seeding. |
There was a problem hiding this comment.
In Python, only a string literal that is the first statement in the class body is treated as a docstring. Because __test__ = False appears first, this multi-line string is just a discarded expression — TestSeedingUseCase.__doc__ will be None. Swapping the two lines fixes it without any functional change.
| class TestSeedingUseCase: # noqa: PT001 — not a pytest class; "Test" prefix is the use-case domain name | |
| __test__ = False # tell pytest not to collect this as a test class | |
| """Test-only resource seeding. | |
| class TestSeedingUseCase: # noqa: PT001 — not a pytest class; "Test" prefix is the use-case domain name | |
| """Test-only resource seeding. |
…s populated Python only treats a string literal as __doc__ when it is the FIRST statement in the class body. The previous order put __test__ first, which silently turned the docstring into a discarded expression — __doc__ was None. Swap the two: docstring first, __test__ second. Verified TestSeedingUseCase.__doc__ is now populated and pytest still skips collection (__test__ = False as a class attribute is honored regardless of where it sits in the body). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
…g PROD
Address review (harvhan): ENVIRONMENT is typed `str | None` on
EnvironmentVariables and populated raw from os.environ with no enum
coercion. A deny-list against `Environment.PROD` ("production") therefore
fails OPEN on any value the gate doesn't recognize — unset, "prod",
"Production", typos, or any new environment name.
Switch both gate sites (router mount in app.py + per-request check in
the seeding route) to an allow-list against Environment.DEV ("development")
and Environment.STAGING. Anything else fails closed.
Tests: extend the gate suite with a parametrized `test_unknown_environment
_returns_404` covering None, "", "prod", "Production", "dev", "qa". 15
seeding tests now pass (was 9). Existing `test_prod_env_returns_404...`
left in place since prod IS one of the rejected cases — the new test
covers the broader allow-list contract.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
| if body.resource_type == "event": | ||
| payload = body.payload | ||
| event_entity = await use_case.seed_event( | ||
| task_id=str(payload.task_id), | ||
| agent_id=str(payload.agent_id), | ||
| content=payload.content, | ||
| id_override=str(payload.id) if payload.id is not None else None, | ||
| principal_id=principal_id, | ||
| ) | ||
| return Event.model_validate(event_entity) |
There was a problem hiding this comment.
Not sure I get the overall idea here.
I dont think seeding just event is enough because event permissions depend on the agent right? If we want to just seed the events why not just mock out the DB instead of creating a whole endpoint to create DB entries?
There was a problem hiding this comment.
Also what is the eventual intention of these tests we are writing? Do we intend them to run on each PR/each deployment?
Summary
Adds a test-only seeding endpoint at `POST /test/seed` so the e2e suite (scaleapi/agentex#378) can directly insert resource rows for FGAC assertions, bypassing the natural-flow side effects (ACP forward, worker pickup, etc.).
The endpoint writes via the existing repository layer — same path the natural flow takes — so seeded rows are indistinguishable from real ones to downstream consumers, except for an audit marker (below).
Why
E2E tests for FGAC need to assert authz outcomes (e.g. denied-get → 404) against real resource rows. A 404 on a missing row doesn't prove the same code path as a 404 on a denied row. Black-box seeding via the natural flow requires a healthy running agent for events, which is heavy/flaky for an e2e suite.
For events specifically, scaleapi/agentex#378 ships a workaround that exploits a quirk of `task_service.create_event_and_forward_to_acp` (writes row before forward). That works for events but isn't a general solution — this endpoint is.
Gate (defense in depth)
All gate failures return 404, not 401/403 — the route's existence isn't advertised.
Audit
Scope today, extension later
Ships event seeding only. The use case is structured so adding `seed_task` / `seed_api_key` / `seed_schedule` is mechanical — each new type wraps its own branch and (for FGAC-registered types) mirrors the natural flow's `register_resource` call. A `TODO` comment marks that integration point for the next contributor.
Events are intentionally NOT registered as a top-level FGAC resource here — they delegate read authz to the parent agent, matching the natural flow. Verified against `AgentexResourceType` and `routes/events.py`.
Test plan
Files
🤖 Generated with Claude Code
Greptile Summary
Adds a gated
POST /test/seedendpoint so the e2e suite can insert resource rows directly — bypassing ACP, worker pickup, and other natural-flow side effects — for FGAC assertion testing. The endpoint is protected by three independent layers: an opt-in env flag, an allow-list environment check that fails closed on prod/unknown environments, and a constant-time shared-secret header check; all failures return 404 to avoid advertising the route.src/api/routes/test_seeding.py— new router with a discriminated request schema (SeedEventRequest) and a_require_test_seeding_enableddependency that enforces all three gate layers per request.src/domain/use_cases/test_seeding_use_case.py—TestSeedingUseCase.seed_eventwrites directly via the event repository and injects a{\"seeded\": true, \"seeded_at\": ...}audit marker into every persisted row.src/api/app.py— conditionally mounts the router at startup using the same allow-list check; emits alogger.warningif it mounts so the state is visible in logs.Confidence Score: 5/5
Safe to merge. The three-layer gate (env flag + allow-list environment check + constant-time token) is well-designed and fails closed; the router is never mounted in production by construction.
The gating logic is careful and consistent between mount-time (app.py) and per-request (test_seeding.py) checks. The test suite covers all gate failure paths and the happy path with audit-marker validation. The two findings are minor edge cases that do not affect the core security properties of the implementation.
No files require special attention. The _ALLOWED_ENVS set is defined in both app.py and test_seeding.py — the in-code comment calls this out — so reviewers should ensure both are updated together if a new non-prod environment is ever added.
Important Files Changed
Sequence Diagram
sequenceDiagram participant E2E as E2E Test Client participant App as FastAPI app.py (startup) participant Gate as _require_test_seeding_enabled participant Route as POST /test/seed participant UC as TestSeedingUseCase participant Repo as EventRepository App->>App: EnvironmentVariables.refresh() alt "ENABLE_TEST_SEEDING AND ENVIRONMENT in {DEV, STAGING}" App->>App: include_router(test_seeding.router) + logger.warning else App->>App: router not mounted end E2E->>Route: POST /test/seed + X-Test-Seed-Token header Route->>Gate: Depends(_require_test_seeding_enabled) Gate->>Gate: check ENVIRONMENT in allow-list Gate->>Gate: check ENABLE_TEST_SEEDING flag Gate->>Gate: check TEST_SEED_TOKEN configured Gate->>Gate: hmac.compare_digest(header, expected) alt any check fails Gate-->>E2E: 404 Not Found else all checks pass Gate-->>Route: proceed Route->>UC: seed_event(task_id, agent_id, content, ...) UC->>UC: merge content + inject seeded/seeded_at marker UC->>Repo: create(id, task_id, agent_id, content_entity) Repo-->>UC: EventEntity UC->>UC: logger.info(test seeding wrote resource) UC-->>Route: EventEntity Route-->>E2E: 201 Created (Event schema) endPrompt To Fix All With AI
Reviews (4): Last reviewed commit: "fix(test-seeding): allow-list non-prod e..." | Re-trigger Greptile
Context used:
Learned From
scaleapi/scaleapi#126926
Learned From
scaleapi/scaleapi#127117