diff --git a/backend/tests/functional/webhook/conftest.py b/backend/tests/functional/webhook/conftest.py index c0940a99e24..584ec3e4416 100644 --- a/backend/tests/functional/webhook/conftest.py +++ b/backend/tests/functional/webhook/conftest.py @@ -12,6 +12,9 @@ from infrahub.core.node import Node from infrahub.events.models import EventBranchContext, EventContext from infrahub.task_manager.flow_run.prefect_client import PrefectClientAdapter +from infrahub.trigger.constants import NAME_SEPARATOR +from infrahub.trigger.models import TriggerType +from infrahub.trigger.setup import gather_all_automations from infrahub.webhook.tasks import process from infrahub.workflows.catalogue import ( WEBHOOK_CONFIGURE, @@ -101,6 +104,23 @@ async def prefect_client(prefect_test_fixture: None) -> AsyncGenerator[PrefectCl yield client +@pytest.fixture(scope="class", autouse=True) +async def delete_webhook_automations(prefect_client: PrefectClient) -> AsyncGenerator[None, None]: + """Delete the webhook automations a test class registered, once the class is done with them. + + The Prefect test server is session-scoped, so an automation outlives the class that created it + while the webhook node behind it is dropped with the class database. A surviving all-branches + automation then turns every event any later test emits in this session into a scheduled + webhook-process run that no worker ever executes, filling the server's database and pushing the + run count past the API's 200-row page size. + """ + yield + webhook_prefix = f"{TriggerType.WEBHOOK.value}{NAME_SEPARATOR}" + for automation in await gather_all_automations(client=prefect_client): + if automation.id and automation.name.startswith(webhook_prefix): + await prefect_client.delete_automation(automation_id=automation.id) + + @pytest.fixture(scope="class") def flow_run_querier(prefect_client: PrefectClient) -> FlowRunQuerying: """A read-only view of the Prefect client, exposing only flow-run querying to tests.""" diff --git a/backend/tests/functional/webhook/test_render.py b/backend/tests/functional/webhook/test_render.py index 9216e9e17d9..11627ee8cf2 100644 --- a/backend/tests/functional/webhook/test_render.py +++ b/backend/tests/functional/webhook/test_render.py @@ -6,6 +6,7 @@ from uuid import uuid4 from prefect.client.schemas.filters import DeploymentFilter, DeploymentFilterId +from prefect.client.schemas.sorting import FlowRunSort from prefect.events.schemas.events import Event, Resource from prefect.types import DateTime @@ -19,6 +20,7 @@ if TYPE_CHECKING: from infrahub_sdk import InfrahubClient from prefect.client.orchestration import PrefectClient + from prefect.client.schemas.objects import FlowRun from infrahub.database import InfrahubDatabase @@ -63,7 +65,22 @@ async def test_branchless_event_triggers_webhook_process( deployment = await prefect_client.read_deployment_by_name(f"{WEBHOOK_PROCESS.name}/{WEBHOOK_PROCESS.name}") deployment_filter = DeploymentFilter(id=DeploymentFilterId(any_=[deployment.id])) - runs_before = len(await prefect_client.read_flow_runs(deployment_filter=deployment_filter)) + + async def read_process_runs() -> list[FlowRun]: + # Earlier tests leave webhook-process runs behind and a read is capped at the server's + # 200-row page size, so run counts saturate and only a run's identity is a usable signal. + return await prefect_client.read_flow_runs( + deployment_filter=deployment_filter, sort=FlowRunSort.EXPECTED_START_TIME_DESC + ) + + runs_before = {run.id for run in await read_process_runs()} + + async def read_new_runs() -> list[FlowRun]: + return [ + run + for run in await read_process_runs() + if run.id not in runs_before and run.parameters.get("webhook_id") == webhook.id + ] # A branch-less event: the resource carries no infrahub.branch.name, the id is a UUID and the # occurred time a datetime -- all values the action parameters must render as plain strings. @@ -76,10 +93,20 @@ async def test_branchless_event_triggers_webhook_process( ) await prefect_client._client.post("/events", json=[event.model_dump(mode="json")]) - runs_after = runs_before + new_runs: list[FlowRun] = [] for _ in range(PREFECT_EVENT_WAIT_SECONDS): - runs_after = len(await prefect_client.read_flow_runs(deployment_filter=deployment_filter)) - if runs_after > runs_before: + new_runs = await read_new_runs() + if new_runs: break await asyncio.sleep(1) - assert runs_after > runs_before, "webhook-process deployment was not run; server-side parameter render failed" + assert new_runs, "webhook-process deployment was not run; server-side parameter render failed" + + # Every value the deployment receives has to be a plain string, an absent branch included. + parameters = new_runs[0].parameters + assert parameters["event_id"] == str(event.id) + assert parameters["event_type"] == "infrahub.node.created" + assert parameters["event_occured_at"] == "2026-01-01 00:00:00+00:00" + branch_name = parameters["branch_name"] + assert isinstance(branch_name, str) + assert not branch_name + assert parameters["event_payload"] == {"data": {"node_id": "abc"}, "context": {}} diff --git a/dev/knowledge/backend/testing.md b/dev/knowledge/backend/testing.md index b5d0bae4624..046ec5ebfd7 100644 --- a/dev/knowledge/backend/testing.md +++ b/dev/knowledge/backend/testing.md @@ -418,6 +418,20 @@ async def test_logs_warning(caplog: pytest.LogCaptureFixture) -> None: This matches the pattern used in `test_webhook_header.py` and `test_models.py`. +### Prefect Server State Outlives the Test Class + +The Prefect test server is session-scoped — one per xdist worker — while the database and the +fixtures that populate it are class-scoped, so whatever a class registers on that server survives +it. Two rules follow: + +- Delete the automations a class created at its teardown. A surviving all-branches webhook + automation turns every event any later test emits into a scheduled flow run — no worker runs in + the functional suite, so nothing executes them — filling the server's SQLite database. +- Never assert on a flow-run count. `read_flow_runs()` returns at most `PREFECT_API_DEFAULT_LIMIT` + (200) rows and the API rejects a larger `limit`, so once that page is full a before/after + comparison saturates and can never be true again. Read newest-first + (`FlowRunSort.EXPECTED_START_TIME_DESC`) and identify the run by its id or parameters instead. + ### Functional Tests with `TestInfrahubApp` `TestInfrahubApp` provides a `memory_cache` fixture (class-scoped) that injects a `MemoryCache` via `dependency_provider.scope(build_cache, ...)`. Use it in functional tests to pre-fill and assert on cache state: