stroblmeandClaude Opus 5
d375ea97cc
/observability/timeseries and /flows read every metric_minute row in the window
and folded them in Python, so the 7d preset pulled a week of rows on each 30 s
poll. date_bin() does the binning now — the row count drops to the slices asked
for, and to flows × 60 for the sparklines. A window of zero hours used to divide
by nothing and answer 500; windows are clamped to an hour at the low end and to
the retention period at the high end, past which there is nothing to find.
/observability/runs returns {data, count} rather than a bare list, so a minute
busier than the 200-row cap says so instead of quietly showing its newest 200.
The count is only queried when the page comes back full, which keeps the poll
from handing back what the fold just saved.
failures_24h leaves the summary — the Home tile counts errors over the selected
window from the rollups, and nothing had read the field since.
Deleting a flow now takes its Run rows and their nodes, metrics and artifacts
with it. This lives in the route rather than in forget_flow because renaming a
flow calls that too, and a rename must keep its history. The observability
rollups stay: they are the record of what ran, and retention already prunes them.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Uq8mtNb97A7praJLyeEYgs
439 lines
14 KiB
Python
439 lines
14 KiB
Python
from datetime import UTC, datetime
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import func
|
|
from sqlmodel import Session, select
|
|
|
|
from app.core.config import settings
|
|
from app.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())
|
|
db.add(Run(id="run-1", flow="demo", 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
|
|
)
|
|
|
|
# Deleting the flow takes its runs with it, so a reseeded demo starts clean.
|
|
db.expire_all()
|
|
for model in (Run, RunNode, RunMetric, RunArtifact):
|
|
assert db.exec(select(func.count()).select_from(model)).one() == 0
|
|
|
|
|
|
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_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)
|