Superuser route to flush the metrics collector now

POST /observability/flush handles what the collector has queued and writes
it, so the soak harness no longer sleeps FLUSH_INTERVAL_S before clearing
its own rows.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx
This commit is contained in:
2026-10-15 09:53:19 +02:00
co-authored by Claude Opus 5.5
parent efe80cb3c2
commit 0d382edcbf
8 changed files with 106 additions and 11 deletions
+31 -1
View File
@@ -1,9 +1,11 @@
import time
from datetime import UTC, datetime, timedelta
from fastapi.testclient import TestClient
from sqlmodel import Session
from sqlmodel import Session, select
from fluksio.core.config import settings
from fluksio.flow.events import event_bus
from fluksio.models import EngineEvent, FlowRun, MetricBucket
PREFIX = f"{settings.API_V1_STR}/observability"
@@ -482,3 +484,31 @@ def test_a_node_failing_on_every_call_makes_the_summary_degraded(
assert body["nodes"]["failing"] == 1
(problem,) = [p for p in body["problems"] if "failing" in p]
assert "house.calc" in problem and "house.old" not in problem
def test_a_superuser_can_flush_the_collector_now(
client: TestClient,
superuser_token_headers: dict[str, str],
normal_user_token_headers: dict[str, str],
db: Session,
) -> None:
"""Whoever reads or clears the tables need not wait out the interval."""
event_bus.publish(
{
"type": "audit",
"action": "flushed",
"flow": FLOW,
"user": "a@example.com",
"ts": time.time(),
}
)
url = f"{PREFIX}/flush"
assert client.post(url, headers=normal_user_token_headers).status_code == 403
assert client.post(url, headers=superuser_token_headers).status_code == 200
written = db.exec(
select(EngineEvent).where(
EngineEvent.flow == FLOW, EngineEvent.detail == "flushed"
)
).all()
assert len(written) == 1