test(webhook): identify the render test's own flow run instead of counting

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."""

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": {}}

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: