423 lines
15 KiB
Python
423 lines
15 KiB
Python
"""Renaming a flow's or a dashboard's key over HTTP (sections 5.2 and 5.3 of
|
|
docs/private/id_refac_rename.md): who may, what a request is answered with,
|
|
what waits for it, and what keeps working by id afterwards."""
|
|
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from fluksio.api.routes import oauth
|
|
from fluksio.core.config import settings
|
|
from fluksio.flow import rename as renaming
|
|
from fluksio.main import app
|
|
|
|
API = settings.API_V1_STR
|
|
FLOWS = f"{API}/flows"
|
|
DASHBOARDS = f"{API}/dashboards"
|
|
DEVICES = f"{API}/devices"
|
|
MESSAGES = f"{API}/messages"
|
|
|
|
TWICE = "def process(tick):\n return {'twice': tick * 2}\n"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _journal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""A journal of the test's own, and pairing unlimited, as in the device
|
|
tests."""
|
|
monkeypatch.setattr(settings, "RENAME_JOURNAL_FILE", tmp_path / "rename.json")
|
|
for bucket in [b for b in oauth._hits if b.startswith("pair:")]:
|
|
del oauth._hits[bucket]
|
|
|
|
|
|
def _key() -> str:
|
|
return f"rn_{uuid.uuid4().hex[:8]}"
|
|
|
|
|
|
def _flow(
|
|
client: TestClient,
|
|
headers: dict[str, str],
|
|
key: str,
|
|
codes: dict[str, str] | None = None,
|
|
**document: Any,
|
|
) -> str:
|
|
"""A flow under an issued id, its python nodes' *codes* written, and
|
|
published; its id."""
|
|
made = client.post(f"{FLOWS}/", headers=headers, json={"name": key, **document})
|
|
assert made.status_code == 201, made.text
|
|
definition = made.json()["definition"]
|
|
for node_id, code in (codes or {}).items():
|
|
saved = client.put(
|
|
f"{FLOWS}/{definition['id']}/nodes/{node_id}/source",
|
|
headers=headers,
|
|
json={"code": code},
|
|
)
|
|
assert saved.status_code == 200, saved.text
|
|
version = client.get(f"{FLOWS}/{definition['id']}", headers=headers).json()[
|
|
"definition"
|
|
]["version"]
|
|
published = client.post(
|
|
f"{FLOWS}/{definition['id']}/publish",
|
|
headers=headers,
|
|
json={"version": version},
|
|
)
|
|
assert published.status_code == 200, published.text
|
|
return str(definition["id"])
|
|
|
|
|
|
def _hooked(code_node: dict | None = None) -> dict[str, Any]:
|
|
"""A webhook putting ``tick`` in, and what reads it."""
|
|
nodes: list[dict[str, Any]] = [
|
|
{
|
|
"id": "hook",
|
|
"type": "http",
|
|
"params": {"url": "/tick", "method": "POST"},
|
|
"provides": [{"name": "tick", "dtype": "float"}],
|
|
}
|
|
]
|
|
if code_node is not None:
|
|
nodes.append(code_node)
|
|
return {"nodes": nodes}
|
|
|
|
|
|
def _request(
|
|
client: TestClient,
|
|
headers: dict[str, str],
|
|
obj_id: str,
|
|
new: str,
|
|
kind: str = FLOWS,
|
|
**override: Any,
|
|
) -> dict[str, Any]:
|
|
plan = client.post(
|
|
f"{kind}/{obj_id}/rename/plan", headers=headers, json={"new_name": new}
|
|
)
|
|
assert plan.status_code == 200, plan.text
|
|
body = plan.json()
|
|
return {
|
|
"request_id": uuid.uuid4().hex,
|
|
"expected_name": body["name"],
|
|
"expected_version": body["version"],
|
|
"new_name": new,
|
|
"fingerprint": body["fingerprint"],
|
|
**override,
|
|
}
|
|
|
|
|
|
def _in_thread(call) -> tuple[threading.Thread, list[httpx.Response]]:
|
|
answered: list[httpx.Response] = []
|
|
thread = threading.Thread(target=lambda: answered.append(call()), daemon=True)
|
|
thread.start()
|
|
return thread, answered
|
|
|
|
|
|
def _until(check, timeout: float = 20.0) -> None:
|
|
deadline = time.monotonic() + timeout
|
|
while not check():
|
|
assert time.monotonic() < deadline, "timed out"
|
|
time.sleep(0.05)
|
|
|
|
|
|
def test_a_rename_is_a_superusers_and_answered_once(
|
|
client: TestClient,
|
|
superuser_token_headers: dict[str, str],
|
|
normal_user_token_headers: dict[str, str],
|
|
) -> None:
|
|
"""Every refusal by its code, then the rename, the same answer for the
|
|
same request, and its webhook where it was, by id."""
|
|
su = superuser_token_headers
|
|
key, new = _key(), _key()
|
|
flow_id = _flow(client, su, key, **_hooked())
|
|
try:
|
|
body = _request(client, su, flow_id, new)
|
|
url = f"{FLOWS}/{flow_id}/rename"
|
|
assert (
|
|
client.post(url, headers=normal_user_token_headers, json=body).status_code
|
|
== 403
|
|
)
|
|
assert (
|
|
client.post(f"{FLOWS}/f_nothere0/rename", headers=su, json=body).status_code
|
|
== 404
|
|
)
|
|
for override, status, code in (
|
|
({"new_name": "Not A Key"}, 422, "invalid_name"),
|
|
({"expected_version": body["expected_version"] - 1}, 409, "stale"),
|
|
({"fingerprint": "sha256:moved"}, 409, "plan_changed"),
|
|
):
|
|
refused = client.post(url, headers=su, json={**body, **override})
|
|
assert (refused.status_code, refused.json()["code"]) == (status, code)
|
|
assert "refusals" in refused.json()["plan"]
|
|
|
|
renamed = client.post(url, headers=su, json=body)
|
|
assert renamed.status_code == 200, renamed.text
|
|
result = renamed.json()
|
|
assert (
|
|
result["kind"],
|
|
result["id"],
|
|
result["old_name"],
|
|
result["new_name"],
|
|
) == (
|
|
"flow",
|
|
flow_id,
|
|
key,
|
|
new,
|
|
)
|
|
assert result["op"] == body["request_id"]
|
|
assert client.post(url, headers=su, json=body).json() == result
|
|
again = client.post(url, headers=su, json={**body, "request_id": "other"})
|
|
assert (again.status_code, again.json()["code"]) == (409, "stale")
|
|
|
|
detail = client.get(f"{FLOWS}/{flow_id}", headers=su).json()
|
|
assert detail["definition"]["name"] == new
|
|
# The webhook is under the id, which the rename left alone.
|
|
hooked = client.post(f"/hooks/{flow_id}/tick", json={"tick": 7.0})
|
|
assert hooked.status_code == 200, hooked.text
|
|
|
|
def values() -> dict[str, Any]:
|
|
found = client.get(f"{FLOWS}/{flow_id}/state", headers=su).json()
|
|
return dict(found["values"])
|
|
|
|
# Once the cascade it queued has run.
|
|
_until(lambda: f"{new}.tick" in values())
|
|
assert values()[f"{new}.tick"]["value"] == 7.0
|
|
assert f"{key}.tick" not in values()
|
|
summary = client.get(f"{API}/observability/summary", headers=su).json()
|
|
assert summary["rename"] is None
|
|
finally:
|
|
client.delete(f"{FLOWS}/{flow_id}", headers=su)
|
|
|
|
|
|
def test_a_devices_credential_follows_the_renamed_flow(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
"""A message it may send moves with the flow, and renaming its dashboard
|
|
leaves the device's list as it was."""
|
|
su = superuser_token_headers
|
|
key, new, board, device = _key(), _key(), _key(), _key()
|
|
flow_id = _flow(
|
|
client, su, key, inputs=[{"spec": {"name": "button", "dtype": "float"}}]
|
|
)
|
|
made = client.post(f"{DASHBOARDS}/", headers=su, json={"name": board}).json()
|
|
dash_id = made["id"]
|
|
made["widgets"] = [
|
|
{"id": "s", "type": "slider", "config": {"target": f"{key}.button"}}
|
|
]
|
|
saved = client.put(f"{DASHBOARDS}/{dash_id}", headers=su, json=made)
|
|
assert saved.status_code == 200, saved.text
|
|
client.post(
|
|
f"{DASHBOARDS}/{dash_id}/publish",
|
|
headers=su,
|
|
json={"version": saved.json()["version"]},
|
|
)
|
|
created = client.post(f"{DEVICES}/{device}", headers=su).json()
|
|
client.put(
|
|
f"{DEVICES}/{device}",
|
|
headers=su,
|
|
json={**created, "dashboards": [dash_id]},
|
|
)
|
|
started = client.post(f"{DEVICES}/pair").json()
|
|
client.post(f"{DEVICES}/{device}/pair", headers=su, json={"code": started["code"]})
|
|
token = client.get(
|
|
f"{DEVICES}/pair/{started['code']}", params={"secret": started["secret"]}
|
|
).json()["access_token"]
|
|
panel = {"Authorization": f"Bearer {token}"}
|
|
|
|
def send(name: str) -> int:
|
|
return client.post(
|
|
f"{MESSAGES}/{name}", headers=panel, json={"value": 1.0}
|
|
).status_code
|
|
|
|
try:
|
|
assert send(f"{key}.button") == 200
|
|
renamed = client.post(
|
|
f"{FLOWS}/{flow_id}/rename",
|
|
headers=su,
|
|
json=_request(client, su, flow_id, new),
|
|
)
|
|
assert renamed.status_code == 200, renamed.text
|
|
assert renamed.json()["documents"][-1]["id"] == dash_id
|
|
assert send(f"{key}.button") == 403
|
|
assert send(f"{new}.button") == 200
|
|
|
|
renamed = client.post(
|
|
f"{DASHBOARDS}/{dash_id}/rename",
|
|
headers=su,
|
|
json=_request(client, su, dash_id, f"{board}_b", kind=DASHBOARDS),
|
|
)
|
|
assert renamed.status_code == 200, renamed.text
|
|
assert client.get(f"{DASHBOARDS}/{dash_id}", headers=su).json()["name"] == (
|
|
f"{board}_b"
|
|
)
|
|
listed = client.get(f"{DEVICES}/{device}", headers=su).json()
|
|
assert listed["dashboards"] == [dash_id]
|
|
assert send(f"{new}.button") == 200
|
|
finally:
|
|
client.delete(f"{DEVICES}/{device}", headers=su)
|
|
client.delete(f"{DASHBOARDS}/{dash_id}", headers=su)
|
|
client.delete(f"{FLOWS}/{flow_id}", headers=su)
|
|
|
|
|
|
def test_what_arrives_during_a_rename_waits_or_is_turned_away(
|
|
client: TestClient,
|
|
superuser_token_headers: dict[str, str],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Saves, publishes, creates, deletes, runs, imports, sync and messages
|
|
get 503 while the rename holds the engine, its own flow's webhook too.
|
|
Another flow's webhook is taken, and what it sets off runs once the
|
|
rename lets go. Reads carry on."""
|
|
su = superuser_token_headers
|
|
key, new, other_key = _key(), _key(), _key()
|
|
flow_id = _flow(client, su, key, **_hooked())
|
|
other = _flow(
|
|
client,
|
|
su,
|
|
other_key,
|
|
{"double": TWICE},
|
|
**_hooked(
|
|
{
|
|
"id": "double",
|
|
"type": "python",
|
|
"requires": [{"name": "tick", "dtype": "float"}],
|
|
"provides": [{"name": "twice", "dtype": "float"}],
|
|
}
|
|
),
|
|
)
|
|
stale = client.get(f"{FLOWS}/{flow_id}", headers=su).json()["definition"]
|
|
holding, go = threading.Event(), threading.Event()
|
|
begin = renaming.begin
|
|
|
|
def held_begin(*args: Any, **kwargs: Any) -> Any:
|
|
holding.set()
|
|
assert go.wait(20)
|
|
return begin(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(renaming, "begin", held_begin)
|
|
body = _request(client, su, flow_id, new)
|
|
thread, answered = _in_thread(
|
|
lambda: client.post(f"{FLOWS}/{flow_id}/rename", headers=su, json=body)
|
|
)
|
|
try:
|
|
assert holding.wait(20)
|
|
for method, path, payload in (
|
|
("put", f"{FLOWS}/{flow_id}", stale),
|
|
("post", f"{FLOWS}/{flow_id}/publish", {"version": stale["version"]}),
|
|
("post", f"{FLOWS}/", {"name": _key()}),
|
|
("delete", f"{FLOWS}/{other}", None),
|
|
("post", f"{API}/runs/flows/{other}", {"inputs": {}}),
|
|
("post", f"{API}/runs/import", []),
|
|
("post", f"{MESSAGES}/{other_key}.tick", {"value": 1.0}),
|
|
):
|
|
refused = client.request(method, path, headers=su, json=payload)
|
|
assert refused.status_code == 503, (path, refused.text)
|
|
assert refused.json()["code"] == "renaming"
|
|
assert refused.headers["retry-after"] == "5"
|
|
assert (
|
|
client.post(f"/hooks/{flow_id}/tick", json={"tick": 1.0}).status_code == 503
|
|
)
|
|
assert (
|
|
client.post(f"/hooks/{other}/tick", json={"tick": 3.0}).status_code == 200
|
|
)
|
|
assert client.get(f"{FLOWS}/{flow_id}", headers=su).status_code == 200
|
|
summary = client.get(f"{API}/observability/summary", headers=su).json()
|
|
assert summary["rename"]["phase"] == "prepared"
|
|
|
|
def doubled() -> Any:
|
|
values = client.get(f"{FLOWS}/{other}/state", headers=su).json()["values"]
|
|
return values.get(f"{other_key}.twice", {}).get("value")
|
|
|
|
time.sleep(0.3)
|
|
assert doubled() is None
|
|
finally:
|
|
go.set()
|
|
thread.join(30)
|
|
assert answered[0].status_code == 200, answered[0].text
|
|
try:
|
|
_until(lambda: doubled() == 6.0)
|
|
# An editor still holding the old document is refused, not saved back.
|
|
assert (
|
|
client.put(f"{FLOWS}/{flow_id}", headers=su, json=stale).status_code == 409
|
|
)
|
|
finally:
|
|
client.delete(f"{FLOWS}/{other}", headers=su)
|
|
client.delete(f"{FLOWS}/{flow_id}", headers=su)
|
|
|
|
|
|
def test_a_rename_waits_for_running_work_and_can_be_called_off(
|
|
client: TestClient, superuser_token_headers: dict[str, str], tmp_path: Path
|
|
) -> None:
|
|
"""It says what it waits for, and called off it changes nothing."""
|
|
su = superuser_token_headers
|
|
key, new = _key(), _key()
|
|
go = tmp_path / "go"
|
|
slow = (
|
|
"import os, time\n"
|
|
"def process():\n"
|
|
f" while not os.path.exists({str(go)!r}):\n"
|
|
" time.sleep(0.05)\n"
|
|
" return {'reading': 1.0}\n"
|
|
)
|
|
flow_id = _flow(
|
|
client,
|
|
su,
|
|
key,
|
|
{"sensor": slow},
|
|
nodes=[
|
|
{
|
|
"id": "sensor",
|
|
"type": "python",
|
|
"provides": [{"name": "reading", "dtype": "float"}],
|
|
}
|
|
],
|
|
)
|
|
trigger = f"{FLOWS}/{flow_id}/nodes/sensor/trigger"
|
|
triggering, triggered = _in_thread(
|
|
lambda: client.post(trigger, headers=su, json={"values": {}})
|
|
)
|
|
body = _request(client, su, flow_id, new)
|
|
try:
|
|
# In, and counted, before the gate closes.
|
|
_until(lambda: f"POST {trigger}" in app.state.rename_gate._admitted)
|
|
thread, answered = _in_thread(
|
|
lambda: client.post(f"{FLOWS}/{flow_id}/rename", headers=su, json=body)
|
|
)
|
|
|
|
def waiting() -> Any:
|
|
summary = client.get(f"{API}/observability/summary", headers=su).json()
|
|
return summary["rename"]
|
|
|
|
_until(
|
|
lambda: waiting() is not None and waiting()["busy"] == [f"POST {trigger}"]
|
|
)
|
|
assert waiting()["phase"] == "draining"
|
|
assert client.put(f"{FLOWS}/{flow_id}", headers=su, json={}).status_code == 503
|
|
|
|
cancel = f"{FLOWS}/{flow_id}/rename/{body['request_id']}"
|
|
assert (
|
|
client.delete(f"{FLOWS}/{flow_id}/rename/other", headers=su).status_code
|
|
== 404
|
|
)
|
|
assert client.delete(cancel, headers=su).status_code == 200
|
|
thread.join(30)
|
|
assert (answered[0].status_code, answered[0].json()["code"]) == (409, "aborted")
|
|
assert waiting() is None
|
|
finally:
|
|
go.touch()
|
|
triggering.join(60)
|
|
assert triggered[0].status_code == 200, triggered[0].text
|
|
try:
|
|
definition = client.get(f"{FLOWS}/{flow_id}", headers=su).json()["definition"]
|
|
assert (definition["name"], definition["version"]) == (
|
|
key,
|
|
body["expected_version"],
|
|
)
|
|
# The gate is open again, and the same rename now goes through.
|
|
renamed = client.post(f"{FLOWS}/{flow_id}/rename", headers=su, json=body)
|
|
assert renamed.status_code == 200, renamed.text
|
|
finally:
|
|
client.delete(f"{FLOWS}/{flow_id}", headers=su)
|