stroblmeandClaude Opus 5
32f751e115
`receive_text` raises `WebSocketDisconnect` once the peer has gone and `RuntimeError` on every call after that. The loop swallowed both — the exception was checked only to decide whether the frame held a media subscription — and immediately made another read, which raised again. A closed tab therefore left a coroutine reading a dead socket at full tilt: around 15,000 turns a second, one orphaned `Queue.get()` per turn parked on the bus queue's waiter list. The event loop stalled behind it, `/health` answered 503 on its loop-lag check, and Docker's probe plus Traefik took the API out of the pool mid-request — which is what `make verify` was dying on, with `TypeError: Failed to fetch` from the browser. The spin ended only when an event finally reached the send path and the socket raised there, so it lasted from milliseconds to minutes depending on what the bus happened to be carrying. Four abandoned sockets logged 2,518 `Task was destroyed but it is pending!` lines and up to 2.6s of loop lag; twenty now log none and cost 1.0ms. A failed read is the disconnect, so it ends the stream. The getter created beside it is cancelled rather than abandoned: nothing reads it after the turn that made it, and left behind it would also take an event the live reader wanted. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CL9zvnnvcp1mvA8o7impxk
546 lines
18 KiB
Python
546 lines
18 KiB
Python
import asyncio
|
|
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
from fastapi import WebSocketDisconnect
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import func
|
|
from sqlmodel import Session, select
|
|
|
|
from fluksio.api.routes import flows as flows_route
|
|
from fluksio.core.config import settings
|
|
from fluksio.flow.events import event_bus
|
|
from fluksio.models import Run, RunArtifact, RunMetric, RunNode
|
|
|
|
PREFIX = f"{settings.API_V1_STR}/flows"
|
|
|
|
WORKING_NODE = """
|
|
def process():
|
|
return {"reading": 21.5}
|
|
"""
|
|
|
|
BROKEN_NODE = """
|
|
def process():
|
|
raise RuntimeError("boom")
|
|
"""
|
|
|
|
|
|
def a_flow(name: str = "demo") -> dict:
|
|
return {
|
|
"name": name,
|
|
"title": "Demo",
|
|
"nodes": [
|
|
{
|
|
"id": "sensor",
|
|
"type": "python",
|
|
"provides": [{"name": "reading", "dtype": "float"}],
|
|
},
|
|
{
|
|
"id": "logger",
|
|
"type": "python",
|
|
"requires": [{"name": "reading", "dtype": "float"}],
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
def test_flows_require_authentication(client: TestClient) -> None:
|
|
assert client.get(f"{PREFIX}/").status_code == 401
|
|
|
|
|
|
def test_save_then_read_flow(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
response = client.put(
|
|
f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow()
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
response = client.get(f"{PREFIX}/demo", headers=superuser_token_headers)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["definition"]["title"] == "Demo"
|
|
assert {node["id"] for node in body["definition"]["nodes"]} == {"sensor", "logger"}
|
|
|
|
|
|
def test_name_mismatch_is_rejected(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
response = client.put(
|
|
f"{PREFIX}/other", headers=superuser_token_headers, json=a_flow("demo")
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_broken_node_is_reported_and_siblings_stay_active(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
|
|
client.put(
|
|
f"{PREFIX}/demo/nodes/sensor/source",
|
|
headers=superuser_token_headers,
|
|
json={"code": WORKING_NODE},
|
|
)
|
|
|
|
response = client.put(
|
|
f"{PREFIX}/demo/nodes/logger/source",
|
|
headers=superuser_token_headers,
|
|
json={"code": "def process(reading:\n"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "error"
|
|
|
|
statuses = client.get(f"{PREFIX}/demo", headers=superuser_token_headers).json()
|
|
by_id = {node["id"]: node for node in statuses["nodes"]}
|
|
assert by_id["demo.sensor"]["status"] == "active"
|
|
assert by_id["demo.logger"]["status"] == "error"
|
|
|
|
|
|
def test_running_a_flow_produces_values(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
|
|
client.put(
|
|
f"{PREFIX}/demo/nodes/sensor/source",
|
|
headers=superuser_token_headers,
|
|
json={"code": WORKING_NODE},
|
|
)
|
|
client.put(
|
|
f"{PREFIX}/demo/nodes/logger/source",
|
|
headers=superuser_token_headers,
|
|
json={"code": "def process(reading):\n return {}\n"},
|
|
)
|
|
|
|
response = client.post(
|
|
f"{PREFIX}/demo/run", headers=superuser_token_headers, json={"inputs": {}}
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["values"]["demo.reading"]["value"] == 21.5
|
|
|
|
|
|
def test_editing_does_not_deploy_until_published(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
saved = client.put(
|
|
f"{PREFIX}/staged", headers=superuser_token_headers, json=a_flow("staged")
|
|
).json()
|
|
assert saved["has_draft"] is True
|
|
# Nothing is running yet, so the engine knows no nodes of this flow.
|
|
assert (
|
|
client.get(f"{PREFIX}/staged/state", headers=superuser_token_headers).json()[
|
|
"nodes"
|
|
]
|
|
== []
|
|
)
|
|
|
|
published = client.post(
|
|
f"{PREFIX}/staged/publish",
|
|
headers=superuser_token_headers,
|
|
json={"version": saved["definition"]["version"]},
|
|
)
|
|
assert published.status_code == 200
|
|
assert published.json()["has_draft"] is False
|
|
assert {
|
|
node["id"]
|
|
for node in client.get(
|
|
f"{PREFIX}/staged/state", headers=superuser_token_headers
|
|
).json()["nodes"]
|
|
} == {"staged.sensor", "staged.logger"}
|
|
|
|
client.delete(f"{PREFIX}/staged", headers=superuser_token_headers)
|
|
|
|
|
|
def test_a_stale_save_is_refused(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
saved = client.put(
|
|
f"{PREFIX}/contested", headers=superuser_token_headers, json=a_flow("contested")
|
|
).json()
|
|
stale = saved["definition"]
|
|
|
|
client.put(
|
|
f"{PREFIX}/contested",
|
|
headers=superuser_token_headers,
|
|
json={**stale, "title": "Mine"},
|
|
)
|
|
|
|
response = client.put(
|
|
f"{PREFIX}/contested",
|
|
headers=superuser_token_headers,
|
|
json={**stale, "title": "Theirs"},
|
|
)
|
|
assert response.status_code == 409
|
|
assert response.json()["detail"]["current_version"] == stale["version"] + 1
|
|
|
|
client.delete(f"{PREFIX}/contested", headers=superuser_token_headers)
|
|
|
|
|
|
def test_discarding_a_draft_restores_what_is_running(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
saved = client.put(
|
|
f"{PREFIX}/reverted", headers=superuser_token_headers, json=a_flow("reverted")
|
|
).json()
|
|
client.post(
|
|
f"{PREFIX}/reverted/publish",
|
|
headers=superuser_token_headers,
|
|
json={"version": saved["definition"]["version"]},
|
|
)
|
|
published = client.get(f"{PREFIX}/reverted", headers=superuser_token_headers).json()
|
|
|
|
client.put(
|
|
f"{PREFIX}/reverted",
|
|
headers=superuser_token_headers,
|
|
json={**published["definition"], "title": "Scratch that"},
|
|
)
|
|
|
|
response = client.post(
|
|
f"{PREFIX}/reverted/discard-draft", headers=superuser_token_headers
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["has_draft"] is False
|
|
assert response.json()["definition"]["title"] == "Demo"
|
|
|
|
client.delete(f"{PREFIX}/reverted", headers=superuser_token_headers)
|
|
|
|
|
|
def test_stopping_a_flow_takes_it_off_the_engine(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
saved = client.put(
|
|
f"{PREFIX}/halted", headers=superuser_token_headers, json=a_flow("halted")
|
|
).json()
|
|
client.post(
|
|
f"{PREFIX}/halted/publish",
|
|
headers=superuser_token_headers,
|
|
json={"version": saved["definition"]["version"]},
|
|
)
|
|
|
|
stopped = client.post(f"{PREFIX}/halted/stop", headers=superuser_token_headers)
|
|
assert stopped.status_code == 200
|
|
assert stopped.json()["enabled"] is False
|
|
# Nothing of it is loaded, so there is nothing to run.
|
|
assert (
|
|
client.post(
|
|
f"{PREFIX}/halted/run", headers=superuser_token_headers, json={"inputs": {}}
|
|
).status_code
|
|
== 409
|
|
)
|
|
|
|
started = client.post(f"{PREFIX}/halted/start", headers=superuser_token_headers)
|
|
assert started.json()["enabled"] is True
|
|
assert (
|
|
client.post(
|
|
f"{PREFIX}/halted/run", headers=superuser_token_headers, json={"inputs": {}}
|
|
).status_code
|
|
== 200
|
|
)
|
|
|
|
client.delete(f"{PREFIX}/halted", headers=superuser_token_headers)
|
|
|
|
|
|
def test_pausing_is_reported_back(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
saved = client.put(
|
|
f"{PREFIX}/held", headers=superuser_token_headers, json=a_flow("held")
|
|
).json()
|
|
client.post(
|
|
f"{PREFIX}/held/publish",
|
|
headers=superuser_token_headers,
|
|
json={"version": saved["definition"]["version"]},
|
|
)
|
|
|
|
assert (
|
|
client.post(f"{PREFIX}/held/pause", headers=superuser_token_headers).status_code
|
|
== 200
|
|
)
|
|
assert (
|
|
client.get(f"{PREFIX}/held", headers=superuser_token_headers).json()["paused"]
|
|
is True
|
|
)
|
|
|
|
client.post(f"{PREFIX}/held/resume", headers=superuser_token_headers)
|
|
assert (
|
|
client.get(f"{PREFIX}/held", headers=superuser_token_headers).json()["paused"]
|
|
is False
|
|
)
|
|
|
|
client.delete(f"{PREFIX}/held", headers=superuser_token_headers)
|
|
|
|
|
|
def test_unconnected_input_is_surfaced(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
flow = a_flow()
|
|
flow["nodes"][0]["provides"] = [] # nothing produces "reading" any more
|
|
client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=flow)
|
|
|
|
issues = client.post(
|
|
f"{PREFIX}/demo/validate", headers=superuser_token_headers
|
|
).json()["issues"]
|
|
assert any(issue["code"] == "unconnected_input" for issue in issues)
|
|
|
|
|
|
def test_delete_flow(
|
|
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
|
) -> None:
|
|
client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
|
|
# Finished: a flow with a live run refuses to be deleted, which is its own
|
|
# test below.
|
|
db.add(Run(id="run-1", flow="demo", status="ok", created_at=datetime.now(UTC)))
|
|
db.add(RunNode(run_id="run-1", node="sensor"))
|
|
db.add(RunMetric(run_id="run-1", name="loss", step=-1))
|
|
db.add(RunArtifact(run_id="run-1", name="model.pt"))
|
|
db.commit()
|
|
|
|
assert (
|
|
client.delete(f"{PREFIX}/demo", headers=superuser_token_headers).status_code
|
|
== 200
|
|
)
|
|
assert (
|
|
client.get(f"{PREFIX}/demo", headers=superuser_token_headers).status_code == 404
|
|
)
|
|
|
|
# The run outlives the flow: a run record is a record of what ran, and
|
|
# `DELETE /runs?flow=` is the only thing that clears one.
|
|
db.expire_all()
|
|
for model in (Run, RunNode, RunMetric, RunArtifact):
|
|
assert db.exec(select(func.count()).select_from(model)).one() == 1
|
|
assert (
|
|
client.get(
|
|
f"{settings.API_V1_STR}/runs/run-1", headers=superuser_token_headers
|
|
).status_code
|
|
== 200
|
|
)
|
|
|
|
|
|
def test_delete_flow_is_refused_while_a_run_is_live(
|
|
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
|
) -> None:
|
|
client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
|
|
db.add(
|
|
Run(id="run-live", flow="demo", status="running", created_at=datetime.now(UTC))
|
|
)
|
|
db.commit()
|
|
|
|
response = client.delete(f"{PREFIX}/demo", headers=superuser_token_headers)
|
|
assert response.status_code == 409
|
|
assert "run-live" in response.json()["detail"]
|
|
# Its driver is still writing node rows against that id, so nothing may go.
|
|
assert (
|
|
client.get(f"{PREFIX}/demo", headers=superuser_token_headers).status_code == 200
|
|
)
|
|
|
|
|
|
def test_node_types_are_listed(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
types = client.get(f"{PREFIX}/node-types", headers=superuser_token_headers).json()
|
|
by_type = {entry["type"]: entry for entry in types}
|
|
assert by_type["python"]["has_source"] is True
|
|
assert "properties" in by_type["mqtt"]["params_schema"]
|
|
|
|
|
|
def test_rename_flow(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
|
|
|
|
response = client.post(
|
|
f"{PREFIX}/demo/rename",
|
|
headers=superuser_token_headers,
|
|
json={"new_name": "demo_renamed"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["definition"]["name"] == "demo_renamed"
|
|
|
|
assert (
|
|
client.get(f"{PREFIX}/demo", headers=superuser_token_headers).status_code == 404
|
|
)
|
|
assert (
|
|
client.get(
|
|
f"{PREFIX}/demo_renamed", headers=superuser_token_headers
|
|
).status_code
|
|
== 200
|
|
)
|
|
|
|
# Put it back so the tests that follow find the flow they expect.
|
|
client.post(
|
|
f"{PREFIX}/demo_renamed/rename",
|
|
headers=superuser_token_headers,
|
|
json={"new_name": "demo"},
|
|
)
|
|
|
|
|
|
def test_rename_onto_a_taken_name_is_refused(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
|
|
client.put(
|
|
f"{PREFIX}/occupied", headers=superuser_token_headers, json=a_flow("occupied")
|
|
)
|
|
|
|
response = client.post(
|
|
f"{PREFIX}/demo/rename",
|
|
headers=superuser_token_headers,
|
|
json={"new_name": "occupied"},
|
|
)
|
|
assert response.status_code == 409
|
|
|
|
client.delete(f"{PREFIX}/occupied", headers=superuser_token_headers)
|
|
|
|
|
|
def test_rename_rejects_an_invalid_name(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
|
|
|
|
response = client.post(
|
|
f"{PREFIX}/demo/rename",
|
|
headers=superuser_token_headers,
|
|
json={"new_name": "Not A Flow Name"},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_a_node_that_raises_answers_with_its_error(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
"""A manual run fails the way a scheduled one does, not as a 500."""
|
|
saved = client.put(
|
|
f"{PREFIX}/failing", headers=superuser_token_headers, json=a_flow("failing")
|
|
).json()
|
|
client.put(
|
|
f"{PREFIX}/failing/nodes/sensor/source",
|
|
headers=superuser_token_headers,
|
|
json={"code": BROKEN_NODE},
|
|
)
|
|
client.post(
|
|
f"{PREFIX}/failing/publish",
|
|
headers=superuser_token_headers,
|
|
json={"version": saved["definition"]["version"]},
|
|
)
|
|
|
|
response = client.post(
|
|
f"{PREFIX}/failing/nodes/sensor/trigger",
|
|
headers=superuser_token_headers,
|
|
json={"values": {}},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "boom" in response.json()["detail"]
|
|
|
|
client.delete(f"{PREFIX}/failing", headers=superuser_token_headers)
|
|
|
|
|
|
def test_acknowledging_a_failure_is_announced_on_the_bus(
|
|
client: TestClient,
|
|
superuser_token_headers: dict[str, str],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Another browser holds the same marker; the event is what clears it."""
|
|
saved = client.put(
|
|
f"{PREFIX}/acked", headers=superuser_token_headers, json=a_flow("acked")
|
|
).json()
|
|
client.post(
|
|
f"{PREFIX}/acked/publish",
|
|
headers=superuser_token_headers,
|
|
json={"version": saved["definition"]["version"]},
|
|
)
|
|
|
|
published: list[dict] = []
|
|
monkeypatch.setattr(event_bus, "publish", published.append)
|
|
response = client.post(
|
|
f"{PREFIX}/acked/nodes/sensor/acknowledge", headers=superuser_token_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
event = next(e for e in published if e["type"] == "node_error_acknowledged")
|
|
assert event["node"] == "acked.sensor"
|
|
|
|
client.delete(f"{PREFIX}/acked", headers=superuser_token_headers)
|
|
|
|
|
|
def test_renaming_a_flow_leaves_nothing_under_the_old_name(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
saved = client.put(
|
|
f"{PREFIX}/movable", headers=superuser_token_headers, json=a_flow("movable")
|
|
).json()
|
|
client.put(
|
|
f"{PREFIX}/movable/nodes/sensor/source",
|
|
headers=superuser_token_headers,
|
|
json={"code": WORKING_NODE},
|
|
)
|
|
client.post(
|
|
f"{PREFIX}/movable/publish",
|
|
headers=superuser_token_headers,
|
|
json={"version": saved["definition"]["version"]},
|
|
)
|
|
client.post(
|
|
f"{PREFIX}/movable/run", headers=superuser_token_headers, json={"inputs": {}}
|
|
)
|
|
|
|
state = client.app.state.flow_controller.state
|
|
assert [key for key in state.keys() if "movable." in key]
|
|
|
|
client.post(
|
|
f"{PREFIX}/movable/rename",
|
|
headers=superuser_token_headers,
|
|
json={"new_name": "moved"},
|
|
)
|
|
|
|
# The values repopulate under the new name on the next run; what the old
|
|
# name left behind would sit there for good.
|
|
assert [key for key in state.keys() if "movable." in key] == []
|
|
|
|
client.delete(f"{PREFIX}/moved", headers=superuser_token_headers)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_the_live_socket_ends_when_its_client_goes(
|
|
client: TestClient, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""A departing client ends the stream instead of restarting the read.
|
|
|
|
`receive_text` raises for good once the peer has gone, so reading it again
|
|
spun this loop at full tilt: the event loop stalled, `/health` answered
|
|
503 and one queue getter was orphaned per turn.
|
|
"""
|
|
monkeypatch.setattr(flows_route, "user_from_token", lambda session, token: object())
|
|
|
|
class Peer:
|
|
"""One frame, then gone — the way a closed tab leaves."""
|
|
|
|
def __init__(self) -> None:
|
|
self.app = client.app
|
|
self.frames = ['{"type": "media", "names": []}']
|
|
self.sent: list[str] = []
|
|
|
|
async def accept(self) -> None:
|
|
return None
|
|
|
|
async def send_text(self, text: str) -> None:
|
|
self.sent.append(text)
|
|
|
|
async def receive_text(self) -> str:
|
|
if self.frames:
|
|
return self.frames.pop(0)
|
|
raise WebSocketDisconnect(1006)
|
|
|
|
peer = Peer()
|
|
subscribed = len(event_bus._subscribers)
|
|
|
|
await asyncio.wait_for(flows_route.flow_events(peer, token="not-a-jwt"), timeout=5)
|
|
|
|
# The snapshot went out, and the frame in between did not end the stream:
|
|
# reaching the disconnect at all means the second read happened.
|
|
assert peer.sent
|
|
assert peer.frames == []
|
|
assert len(event_bus._subscribers) == subscribed
|
|
# A turn of the loop for the cancellations the handler asked for.
|
|
await asyncio.sleep(0.05)
|
|
parked = [t for t in asyncio.all_tasks() if "Queue.get" in repr(t.get_coro())]
|
|
assert parked == []
|