Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rename planner and document rewrite: rewrite_documents moves qualified names of the renamed flow in every flow (ports, inputs, outputs) and dashboard binding (one key table, WIDGET_BINDINGS, shared with the WidgetDef accessors, legacy pages included), published and draft, as plain JSON with both variants above either version; plan_rename reports documents, counts, fingerprint, warnings and refusals; superuser plan routes for flows and dashboards; key residue classifies state as the mapper does; duplicate retargets inputs and outputs too; queue reports its Redis version and other consumers

Browse Source
This commit is contained in:
stroblme
2026-11-02 21:06:39 +01:00
parent 81bd7c2724
commit fa2a213631
14 changed files with 1922 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/api/routes/dashboards.py
+34
View File
@@ -16,8 +16,10 @@ from fluksio.api.deps import (
DashboardStoreDep,
FlowControllerDep,
check_pin,
get_current_active_superuser,
get_current_user,
)
from fluksio.api.routes.alerts import read_config as read_alerts_config
from fluksio.flow.dashboards import (
DashboardCreate,
DashboardDef,
@@ -28,6 +30,7 @@ from fluksio.flow.dashboards import (
results_dashboard,
)
from fluksio.flow.events import event_bus
from fluksio.flow.rename import RenamePlan, RenamePlanRequest, plan_rename
from fluksio.flow.store import FlowNotFound, IdTaken, StaleVersion
from fluksio.models import Message
@@ -245,3 +248,34 @@ async def duplicate_dashboard(dashboard_id: str, store: DashboardStoreDep) -> An
return await run_in_threadpool(store.duplicate, dashboard_id)
except DashboardNotFound:
raise _missing(dashboard_id)
@router.post(
"/{dashboard_id}/rename/plan",
response_model=RenamePlan,
dependencies=[Depends(get_current_active_superuser)],
)
async def plan_dashboard_rename(
dashboard_id: str,
body: RenamePlanRequest,
store: DashboardStoreDep,
controller: FlowControllerDep,
) -> Any:
"""What renaming this dashboard's key would change, and what refuses it.
Changes nothing. Nothing binds to a dashboard's key, so the plan is the
dashboard alone; a superuser's all the same, as a flow's is.
"""
try:
return await run_in_threadpool(
lambda: plan_rename(
controller,
store,
"dashboard",
dashboard_id,
body.new_name,
read_alerts_config(),
)
)
except DashboardNotFound:
raise _missing(dashboard_id)
backend/fluksio/api/routes/flows.py
+36
View File
@@ -32,6 +32,7 @@ from fluksio.api.deps import (
CAMERA_IS_SUPERUSERS,
CODE_IS_SUPERUSERS,
CurrentUser,
DashboardStoreDep,
FlowControllerDep,
HiddenDep,
SessionDep,
@@ -39,10 +40,12 @@ from fluksio.api.deps import (
check_pin,
decode_token,
get_code_author,
get_current_active_superuser,
get_current_user,
hidden_from,
user_from_token,
)
from fluksio.api.routes.alerts import read_config as read_alerts_config
from fluksio.api.routes.messages import port_source
from fluksio.api.socket_auth import TICKET_TTL_S, issue_ticket, socket_token
from fluksio.core.config import settings
@@ -66,6 +69,7 @@ from fluksio.flow.events import event_bus
from fluksio.flow.messages import MEDIA_FAMILIES, DType, flow_of, qualify
from fluksio.flow.nodes.wren import TRAIN_FLOW, WrenNode, pack_meta
from fluksio.flow.pipeline import EngineHeld, ValidationIssue
from fluksio.flow.rename import RenamePlan, RenamePlanRequest, plan_rename
from fluksio.flow.runs import RunRejected
from fluksio.flow.schemas import (
NAME_PATTERN,
@@ -740,6 +744,38 @@ async def duplicate_flow(
return await run_in_threadpool(_detail, controller, copy)
@router.post(
"/{flow_id}/rename/plan",
response_model=RenamePlan,
dependencies=[Depends(get_current_active_superuser)],
)
async def plan_flow_rename(
flow_id: str,
body: RenamePlanRequest,
controller: FlowControllerDep,
dashboards: DashboardStoreDep,
) -> Any:
"""What renaming this flow's key would change, and what refuses it.
Changes nothing. A superuser's, as the rename is: the key names the
flow's messages, and renaming it rewrites every flow and dashboard that
names one, its history and its live state.
"""
try:
return await run_in_threadpool(
lambda: plan_rename(
controller,
dashboards,
"flow",
flow_id,
body.new_name,
read_alerts_config(),
)
)
except FlowNotFound:
raise _missing(flow_id)
# -----------------------------------------------------------------------------
# Node source
# -----------------------------------------------------------------------------
backend/fluksio/flow/controller.py
+10 -16
View File
@@ -39,6 +39,7 @@ from fluksio.flow.demand import demand
from fluksio.flow.events import EventBus
from fluksio.flow.executor import ExecutionService
from fluksio.flow.messages import MessageSpec, flow_of, qualify
from fluksio.flow.namespace import touches
from fluksio.flow.nodes import (
RESERVED_SETTINGS,
AssistantNode,
@@ -1894,20 +1895,21 @@ class FlowController:
if self.execution is not None:
self.execution.queue.clear_flow(flow)
with self.state.lock():
for key in [k for k in self.state.keys() if _in_namespace(k, flow)]:
for key in [k for k in self.state.keys() if touches(flow, k)]:
self.state.delete(key)
def key_residue(self, flow: str) -> list[str]:
"""What still lives under the flow key *flow*. Blocking.
State and live queue work in its namespace: left by a flow that held
the key before, or put there by anything still writing to it. A new
flow taking the key would inherit it, so creating one is refused while
this is not empty, and so is renaming a flow onto it. Work that was
acknowledged, dead-lettered or marked done is history and not counted.
Names what it found, state keys first.
State and live queue work in its namespace, judged as a rename maps
it (`namespace.touches`): left by a flow that held the key before, or
put there by anything still writing to it. A new flow taking the key
would inherit it, so creating one is refused while this is not empty,
and so is renaming a flow onto it. Work that was acknowledged,
dead-lettered or marked done is history and not counted. Names what it
found, state keys first.
"""
found = [key for key in self.state.keys() if _in_namespace(key, flow)]
found = [key for key in self.state.keys() if touches(flow, key)]
if self.execution is not None:
found += [
f"queued work of '{item.node}'"
@@ -2010,14 +2012,6 @@ class FlowController:
self.events.publish(event)
def _in_namespace(state_key: str, flow: str) -> bool:
"""Whether a state key is about the flow key *flow*: one of its messages,
or the engine's own bookkeeping on one of its messages or nodes, which is
keyed by those names too."""
prefix = f"{flow}."
return state_key.startswith(prefix) or f":{prefix}" in state_key
def _touches(item: WorkItem, flow: str) -> bool:
"""Whether a queued item names anything of the flow key *flow*: as its
flow, its node or one of the messages it carries."""
backend/fluksio/flow/dashboards.py
+30
View File
@@ -219,6 +219,36 @@ class WidgetFootprint(Placement):
type: WidgetType
#: Where a widget's config names a message, as paths into it
#: (`namespace.rebind`), each with the accessor of `WidgetDef` that reads it. A
#: rename rewrites exactly these, and leaves titles, labels and every other
#: config string alone, so a binding the accessors read and this does not list
#: would keep its old name. A registry test holds the two to each other.
WIDGET_BINDINGS: tuple[tuple[str, ...], ...] = (
("message",), # `messages`, and a bar's first row (`bar_rows`)
("target",), # `target`
("request",), # `target`, of a querying chart
("series", "*", "message"), # `messages`, of a chart
("rows", "*", "message"), # `bar_rows`
("inner",), # `inner_bindings`, as one name
("inner", "*", "message"), # `inner_bindings`, as a list
("reading", "message"), # `reading`
("runs", "metric"), # a chart of runs: a recorded series, named as a message
)
#: The same for a whole stored document: the bound settings
#: (`DashboardDef.setting_messages`), the widgets, and the widgets of a
#: document still written as pages and sections (`_flatten_pages`).
DASHBOARD_BINDINGS: tuple[tuple[str, ...], ...] = (
("settings", "*", "message"),
*(("widgets", "*", "config", *path) for path in WIDGET_BINDINGS),
*(
("pages", "*", "sections", "*", "widgets", "*", "config", *path)
for path in WIDGET_BINDINGS
),
)
class WidgetDef(BaseModel):
"""One tile: what it shows or does, and where it sits.
backend/fluksio/flow/namespace.py
+67 -1
View File
@@ -9,12 +9,17 @@ the new name, so every family is registered here, beside the builders that
take their prefix from it, and a key of a family that is not is refused rather
than left behind.
Documents name messages too: a flow its ports, inputs and outputs, a dashboard
its bindings. Where they do is registered beside the paths into them
(`FLOW_FIELDS`, `dashboards.DASHBOARD_BINDINGS`), and `rebind` rewrites them.
Nothing here does I/O. It maps names; moving the keys and items is the state
and queue backends' work.
and queue backends' work, and writing documents the rename's.
"""
from __future__ import annotations
from collections.abc import Callable, Iterable
from dataclasses import dataclass, replace
from typing import TYPE_CHECKING, Any
@@ -87,6 +92,62 @@ ITEM_FIELDS: dict[str, str | None] = {
"parent": None,
}
#: Where a flow document names messages, as paths into its JSON (`rebind`).
#: A bare name is the flow's own and follows its key by itself, so a rename
#: moves only the qualified ones: those the flow names in another's
#: namespace, or explicitly in its own.
FLOW_FIELDS: tuple[tuple[str, ...], ...] = (
("nodes", "*", "requires", "*", "name"),
("nodes", "*", "provides", "*", "name"),
("inputs", "*", "spec", "name"),
("outputs", "*"),
)
def rebind(
document: Any, paths: Iterable[tuple[str, ...]], rename: Callable[[str], str]
) -> int:
"""Pass every string *paths* lead to in *document* through *rename*, in
place, and count those that changed.
A path is a key per level, or ``*`` for every item of a list or value of
an object. One that leads nowhere, or to anything but a string, is passed
over, so a document missing a field or holding an older shape of it is
read as far as it goes.
"""
return sum(_rebind(document, path, rename) for path in paths)
def names(document: Any, paths: Iterable[tuple[str, ...]]) -> list[str]:
"""Every string *paths* lead to in *document* (`rebind`)."""
found: list[str] = []
def keep(name: str) -> str:
found.append(name)
return name
rebind(document, paths, keep)
return found
def _rebind(node: Any, path: tuple[str, ...], rename: Callable[[str], str]) -> int:
head, rest = path[0], path[1:]
keys: Iterable[Any] = ()
if head == "*" and isinstance(node, list):
keys = range(len(node))
elif head == "*" and isinstance(node, dict):
keys = list(node)
elif isinstance(node, dict) and head in node:
keys = (head,)
changed = 0
for key in keys:
if rest:
changed += _rebind(node[key], rest, rename)
elif isinstance(node[key], str) and (moved := rename(node[key])) != node[key]:
node[key] = moved
changed += 1
return changed
def _parts(key: str) -> tuple[str, list[tuple[str, str]]]:
"""A state key's family prefix and its parts, each with what it holds.
@@ -129,6 +190,11 @@ class Rename:
def flow(self, key: str) -> str:
return self.new if key == self.old else key
def binding(self, name: str) -> str:
"""A message name as a document holds it. A bare one is local to the
flow the document is, so only a qualified one moves."""
return self.name(name) if "." in name else name
def state_key(self, key: str) -> str | None:
"""Where the state key *key* moves to, or None if it stays.
backend/fluksio/flow/queue.py
+32
View File
@@ -457,6 +457,17 @@ class WorkQueue(ABC):
"""Release a held list left by a process that stopped mid-rename. For
startup, before anything claims. Returns how many items moved."""
def server_version(self) -> tuple[int, int] | None:
"""The major and minor version of the server the queue lives on,
None for one in this process. A rename rebuilds streams with commands
Redis has had since 7.0."""
return None
def other_consumers(self, within_ms: int) -> list[str]:
"""Consumers but this one that read the queue in the last *within_ms*:
another engine on the same server, which a rename cannot hold."""
return []
@abstractmethod
def rename(
self, rename: Rename, progress: Callable[[str], None] | None = None
@@ -1132,6 +1143,27 @@ class RedisWorkQueue(WorkQueue):
"durable": True,
}
def server_version(self) -> tuple[int, int] | None:
info = cast(dict[str, Any], self._redis.info("server"))
major, minor, *_ = str(info["redis_version"]).split(".")
return int(major), int(minor)
def other_consumers(self, within_ms: int) -> list[str]:
found: set[str] = set()
for stream in (self._stream, self._due_stream):
try:
rows = cast(
list[dict[str, Any]], self._redis.xinfo_consumers(stream, GROUP)
)
except redis.ResponseError:
continue
found |= {
str(row["name"])
for row in rows
if row["name"] != self._consumer and int(row["idle"]) < within_ms
}
return sorted(found)
def dead_letters(self, count: int = 50) -> list[dict[str, Any]]:
"""What was set aside, newest first. The entry id carries the time."""
entries = cast(
backend/fluksio/flow/rename.py
+574 -5
View File
@@ -1,23 +1,50 @@
"""Renaming a flow's key: the door that keeps the API out while it runs.
"""Renaming a flow's or a dashboard's key.
A rename moves everything named under a flow's key, in documents, history,
live state and queued work, so nothing may write through the API meanwhile.
`RenameGate` is that door. ``docs/private/id_refac_rename.md`` has the whole
protocol.
`RenameGate` is the door that keeps it out. `plan_rename` says beforehand what
a rename would change and what refuses it, and `rewrite_documents` is its
document part. ``docs/private/id_refac_rename.md`` has the whole protocol.
"""
from __future__ import annotations
import hashlib
import json
import re
import threading
from collections import Counter
from collections.abc import Callable, Iterable
from collections.abc import Callable, Iterable, Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Literal, cast
import orjson
from pydantic import BaseModel
from sqlalchemy import String
from sqlmodel import Session, col, func, select
from starlette.types import Scope
from fluksio.core.config import settings
from fluksio.flow.controller import HOOK_PREFIX
from fluksio.core.db import engine as db_engine
from fluksio.flow.alerts import AlertsConfig
from fluksio.flow.controller import HOOK_PREFIX, FlowController
from fluksio.flow.dashboards import (
DASHBOARD_BINDINGS,
DASHBOARD_DIR,
DashboardDef,
DashboardStore,
results_name,
)
from fluksio.flow.executor import wait_out
from fluksio.flow.messages import flow_of
from fluksio.flow.namespace import FLOW_FIELDS, NAME, OUTPUTS, Rename, names, rebind
from fluksio.flow.schemas import FlowDef, _validate_name
from fluksio.flow.state import MemoryState
from fluksio.flow.store import ISSUED_ID, FlowStore, InvalidName, ReservedKey
from fluksio.models import AlertSent, Run, RunArtifact, RunMetric, RunNode
Kind = Literal["flow", "dashboard"]
#: What reading looks like. Everything else under the API changes something.
_READS = frozenset({"GET", "HEAD", "OPTIONS"})
@@ -120,3 +147,545 @@ class RenameGate:
self._closed = False
self._affected = frozenset()
self.op = ""
# -----------------------------------------------------------------------------
# Documents
# -----------------------------------------------------------------------------
@dataclass
class Rewrite:
"""A stored document a rename changes, as texts before and after."""
kind: Kind
id: str
name: str
published_version: int | None
draft_version: int | None
#: Bindings moved, in whichever variant moves more of them.
fields: int
#: Bindings moved in the draft, which stays unpublished.
draft_fields: int
#: Declared in code (`FlowDef.origin`), so a sync puts the old names back.
code_defined: bool
#: ``(path, before, after)`` per variant, the path relative to the store.
files: list[tuple[str, str, str]]
_VARIANTS = (("published", ".json"), ("draft", ".draft.json"))
def _documents(
store: FlowStore, dashboards: DashboardStore
) -> Iterator[tuple[Kind, str, str, tuple[tuple[str, ...], ...]]]:
"""Every stored flow and dashboard: kind, id, path without its variant's
suffix, and where it names messages."""
for flow_id in store.list_flows():
yield "flow", flow_id, f"{flow_id}/flow", FLOW_FIELDS
found = {path.parent.name for path in dashboards.root.glob("*/dashboard*.json")}
for dash_id in sorted(found):
yield (
"dashboard",
dash_id,
f"{DASHBOARD_DIR}/{dash_id}/dashboard",
DASHBOARD_BINDINGS,
)
def _variants(root: Path, stem: str) -> dict[str, tuple[str, dict[str, Any]]]:
"""A document's published and draft variants, as text and as JSON.
Plain JSON rather than the model, so a field this version does not model,
or a dashboard still written as pages, comes through as it was. Empty for
one that does not parse, which the engine cannot read either.
"""
found: dict[str, tuple[str, dict[str, Any]]] = {}
for variant, suffix in _VARIANTS:
path = root / f"{stem}{suffix}"
if not path.exists():
continue
text = path.read_text()
try:
data = orjson.loads(text)
except orjson.JSONDecodeError:
return {}
if not isinstance(data, dict):
return {}
found[variant] = (text, data)
return found
def rewrite_documents(
store: FlowStore,
dashboards: DashboardStore,
kind: Kind,
obj_id: str,
rename: Rename,
) -> list[Rewrite]:
"""Every document the rename of *obj_id* changes, published and draft.
Reads only; writing the texts is the rename's.
The object itself takes the new key. Renaming a flow also moves every
qualified name of its namespace that a flow names (`FLOW_FIELDS`, its own
included) or a dashboard binds (`DASHBOARD_BINDINGS`). Bare names are a
flow's own and follow its key by themselves, and no other field changes.
Both variants of a document that changes go to one version above either,
so an editor holding the old text is refused rather than saving it back,
and a draft discarded later cannot bring an older version back.
"""
rewrites = []
for doc_kind, doc_id, stem, paths in _documents(store, dashboards):
own = doc_kind == kind and doc_id == obj_id
if kind == "dashboard" and not own:
continue
variants = _variants(store.root, stem)
if not variants:
continue
moved = {
variant: rebind(data, paths, rename.binding) if kind == "flow" else 0
for variant, (_, data) in variants.items()
}
if not own and not any(moved.values()):
continue
versions = {
variant: int(data.get("version", 1))
for variant, (_, data) in variants.items()
}
working = variants.get("draft") or variants["published"]
rewrite = Rewrite(
kind=doc_kind,
id=doc_id,
name=str(working[1].get("name", doc_id)),
published_version=versions.get("published"),
draft_version=versions.get("draft"),
fields=max(moved.values()),
draft_fields=moved.get("draft", 0),
code_defined=any(data.get("origin") for _, data in variants.values()),
files=[],
)
top = max(versions.values()) + 1
for variant, suffix in _VARIANTS:
if variant not in variants:
continue
text, data = variants[variant]
if own:
data["name"] = rename.new
data["version"] = top
# The stores write what pydantic writes, which this matches.
after = orjson.dumps(data, option=orjson.OPT_INDENT_2).decode()
if text.endswith("\n"):
after += "\n"
rewrite.files.append((f"{stem}{suffix}", text, after))
rewrites.append(rewrite)
return rewrites
def references(store: FlowStore, dashboards: DashboardStore, key: str) -> list[str]:
"""The flows and dashboards that name a message of the flow key *key*,
published or draft, by id.
A rename onto *key* would hand those bindings to the renamed flow, and a
rename back would take them along. Creating a flow under *key* is another
matter: that is how bindings declared ahead of their flow are met.
"""
found = []
for _, doc_id, stem, paths in _documents(store, dashboards):
bound = (
name
for _, data in _variants(store.root, stem).values()
for name in names(data, paths)
)
if any("." in name and flow_of(name) == key for name in bound):
found.append(doc_id)
return found
# -----------------------------------------------------------------------------
# Plan
# -----------------------------------------------------------------------------
#: An alert's dedup key (`alerts.dedup_key`), whose tail is a node or a flow.
DEDUP = "dedup"
#: Where the history names messages, by table and column: a qualified name
#: (`namespace.NAME`), a JSON object keyed by them (`namespace.OUTPUTS`) or an
#: alert's dedup key. A rename rewrites exactly these. Every other column
#: holds ids, which never change, or payload and free text, which are history,
#: and a registry test fails when a column is added without saying which.
SQL_NAMES: dict[tuple[str, str], str] = {
("run_metric", "name"): NAME,
("run_artifact", "name"): NAME,
("run_node", "outputs"): OUTPUTS,
("alert_sent", "key"): DEDUP,
}
#: How recently another consumer must have read the queue to count as an
#: engine still running: it blocks a second at a time.
OTHER_ENGINE_MS = 10_000
#: Run states a rename refuses: the run would read names mid-move.
_ACTIVE_RUNS = ("queued", "running")
class RenamePlanRequest(BaseModel):
"""The key a flow or dashboard would be renamed to."""
new_name: str
class RenameRefusal(BaseModel):
"""Why a rename cannot go ahead, with the ids of what is in the way."""
code: str
detail: str
ids: list[str] = []
class RenameWarning(BaseModel):
"""Something a rename changes, or leaves, that its caller should know."""
code: str
detail: str
class PlannedDocument(BaseModel):
"""A document a rename rewrites, and the versions it was planned against."""
kind: Kind
id: str
name: str
published_version: int | None
draft_version: int | None
#: Bindings moved, in whichever variant moves more of them.
fields: int
class AlertCounts(BaseModel):
channels: int
ledger_rows: int
class HistoryCounts(BaseModel):
runs: int
metric_rows: int
artifact_rows: int
output_rows: int
class RuntimeCounts(BaseModel):
#: False with memory backends, whose state and queue a restart loses anyway.
durable: bool
state_keys: int
queue_items: int
delayed: int
parked: int
done_markers: int
class RenamePlan(BaseModel):
"""What renaming a flow's or dashboard's key would change, and whether it
may. Read-only; the rename request repeats ``version`` and
``fingerprint``, and is refused when either moved since."""
kind: Kind
id: str
name: str
new_name: str
#: The working copy's version, the draft's when there is one.
version: int
#: Over the kind, id and versions of every listed document and both
#: keys, so a document changing between plan and rename is noticed.
fingerprint: str = ""
allowed: bool = False
refusals: list[RenameRefusal] = []
documents: list[PlannedDocument] = []
#: Dry-run counts, a flow's only, which inform and are not checked again.
alerts: AlertCounts | None = None
history: HistoryCounts | None = None
runtime: RuntimeCounts | None = None
warnings: list[RenameWarning] = []
def plan_rename(
controller: FlowController,
dashboards: DashboardStore,
kind: Kind,
obj_id: str,
new_name: str,
alerts: AlertsConfig,
) -> RenamePlan:
"""What renaming the flow or dashboard *obj_id* to *new_name* would change,
and what refuses it. Blocking, and changes nothing.
:raises FlowNotFound: For a flow that does not exist.
:raises DashboardNotFound: For a dashboard that does not exist.
"""
store = controller.store
current: FlowDef | DashboardDef
if kind == "flow":
current = store.read_flow(obj_id, draft=True)
keys, ids = store.key_index(), set(store.list_flows())
else:
current = dashboards.read(obj_id, draft=True)
keys = dashboards.key_index()
ids = {path.parent.name for path in dashboards.root.glob("*/dashboard*.json")}
old = current.name
plan = RenamePlan(
kind=kind, id=obj_id, name=old, new_name=new_name, version=current.version
)
def refuse(code: str, detail: str, refused: Iterable[str] = ()) -> None:
plan.refusals.append(RenameRefusal(code=code, detail=detail, ids=[*refused]))
def warn(code: str, detail: str) -> None:
plan.warnings.append(RenameWarning(code=code, detail=detail))
try:
_validate_name(new_name)
except ValueError:
refuse("invalid_name", str(InvalidName(new_name)))
return plan
if ISSUED_ID.fullmatch(new_name):
refuse("reserved_shape", str(ReservedKey(new_name)))
return plan
if new_name == old:
refuse("same_name", f"The {kind} is already called '{old}'")
return plan
rename = Rename(old, new_name)
rewrites = rewrite_documents(store, dashboards, kind, obj_id, rename)
plan.documents = [
PlannedDocument(
kind=r.kind,
id=r.id,
name=r.name,
published_version=r.published_version,
draft_version=r.draft_version,
fields=r.fields,
)
for r in rewrites
]
plan.fingerprint = _fingerprint(plan)
if new_name in keys:
refuse("name_taken", f"Another {kind} has the key '{new_name}'")
if new_name in ids - {obj_id}:
warn(
"new_name_is_id",
f"'{new_name}' is also the id of another {kind}, so a client naming "
"it by either is asked which one it means",
)
if kind == "flow":
_plan_flow(plan, controller, dashboards, rename, rewrites, alerts, refuse, warn)
plan.allowed = not plan.refusals
return plan
def _plan_flow(
plan: RenamePlan,
controller: FlowController,
dashboards: DashboardStore,
rename: Rename,
rewrites: list[Rewrite],
alerts: AlertsConfig,
refuse: Callable[..., None],
warn: Callable[[str, str], None],
) -> None:
"""The part of a plan only a flow has: what holds its namespace, and
what moves outside the documents."""
old, new = rename.old, rename.new
own = next(r for r in rewrites if r.kind == "flow" and r.id == plan.id)
if own.code_defined:
refuse(
"code_defined",
f"'{old}' is declared in code: rename it there and sync it again",
)
coded = [r.id for r in rewrites if r is not own and r.code_defined]
if coded:
refuse(
"code_defined_dependent",
f"{len(coded)} flows declared in code name messages of '{old}', and "
"a sync would put the old names back: rename them in code first",
coded,
)
flow_ids = [r.id for r in rewrites if r.kind == "flow"]
with Session(db_engine) as session:
active = session.exec(
select(Run.id, Run.flow_id, Run.for_node).where(
col(Run.status).in_(_ACTIVE_RUNS)
)
).all()
running = [
run_id
for run_id, flow_id, for_node in active
if flow_id in flow_ids or flow_of(for_node or "") in flow_ids
]
if running:
refuse(
"run_active",
f"{len(running)} runs of affected flows are queued or running",
running,
)
history, collisions, too_long = _history(session, rename, flow_ids)
ledger = [
rename.dedup_key(key)
for key in session.exec(select(AlertSent.key)).all()
if rename.dedup_key(key) != key
]
if collisions:
refuse(
"history_collision",
f"{collisions} history rows of the affected flows are already "
f"named '{new}.*'",
)
too_long += [key for key in ledger if len(key) > _length(col(AlertSent.key))]
if too_long:
refuse(
"name_too_long",
f"'{too_long[0]}' would be longer than its column holds",
)
residue = controller.key_residue(new)
referencing = references(controller.store, dashboards, new)
if residue or referencing:
named = [*residue, *(f"bindings in '{one}'" for one in referencing)]
refuse(
"target_in_use",
f"'{new}' still has {', '.join(named[:3])} behind it",
referencing,
)
state = controller.state
queue = controller.execution.queue if controller.execution else None
counts = queue.count(rename) if queue else {}
plan.runtime = RuntimeCounts(
durable=not isinstance(state, MemoryState),
state_keys=state.count(rename),
queue_items=counts.get("queue_items", 0),
delayed=counts.get("delayed", 0),
parked=counts.get("parked", 0),
done_markers=counts.get("done_markers", 0),
)
if queue is not None:
others = queue.other_consumers(OTHER_ENGINE_MS)
if others:
refuse(
"other_engine",
f"Another engine reads the queue ({', '.join(others)}), which a "
"rename cannot hold",
)
version = queue.server_version()
if version is not None and version < (7, 0):
refuse(
"redis_too_old",
f"Redis {version[0]}.{version[1]} is older than 7.0, which "
"a rename needs",
)
plan.history = history
plan.alerts = AlertCounts(
channels=sum(
channel.kind == "dashboard"
and rename.binding(str(channel.config.get("message") or ""))
!= channel.config.get("message")
for channel in alerts.channels
),
ledger_rows=len(ledger),
)
warn(
"external_references",
f"REST /messages/{old}.*, scripts, CLI, SDK and MCP key arguments and "
"hand-written bindings keep the old name",
)
if results_name(old) in dashboards.key_index():
warn(
"results_dashboard",
f"The dashboard '{results_name(old)}' keeps its name",
)
drafts = [r.id for r in rewrites if r.draft_fields]
if drafts:
warn(
"drafts_rewritten",
f"{len(drafts)} unpublished drafts change and stay unpublished",
)
pipeline = controller.pipeline
if (pipeline is not None and pipeline.is_paused(old)) or plan.runtime.parked:
warn("paused", f"'{old}' is paused, and its parked work moves with it")
if not plan.runtime.durable:
warn(
"not_durable",
"State and queued work are kept in memory, so a restart during the "
"rename loses them, as any restart does",
)
rows = history.metric_rows + history.artifact_rows + history.output_rows
warn(
"engine_pause",
f"Automations wait while {rows} history rows and "
f"{plan.runtime.state_keys} state keys move",
)
def _history(
session: Session, rename: Rename, flow_ids: list[str]
) -> tuple[HistoryCounts, int, list[str]]:
"""What moves in the history of the flows *flow_ids*, how many of its
names are in the new namespace already, and the names that would no
longer fit their column."""
old, new = f"{rename.old}.", f"{rename.new}."
grow = len(rename.new) - len(rename.old)
runs = select(Run.id).where(col(Run.flow_id).in_(flow_ids))
counts = {}
collisions = 0
too_long = []
for model in (RunMetric, RunArtifact):
name = col(model.name)
scope = col(model.run_id).in_(runs)
moving = func.substr(name, 1, len(old)) == old
counts[model] = session.exec(
select(func.count()).select_from(model).where(scope, moving)
).one()
collisions += session.exec(
select(func.count())
.select_from(model)
.where(scope, func.substr(name, 1, len(new)) == new)
).one()
longest = session.exec(
select(name).where(scope, moving).order_by(func.length(name).desc())
).first()
if longest is not None and len(longest) + grow > _length(name):
too_long.append(rename.name(longest))
outputs = session.exec(
select(RunNode.outputs).where(
col(RunNode.run_id).in_(runs),
col(RunNode.outputs).contains(f'"{old}', autoescape=True),
)
).all()
moved_outputs = sum(
any(rename.binding(key) != key for key in json.loads(text or "null") or {})
for text in outputs
)
total = session.exec(
select(func.count()).select_from(Run).where(col(Run.flow_id).in_(flow_ids))
).one()
history = HistoryCounts(
runs=total,
metric_rows=counts[RunMetric],
artifact_rows=counts[RunArtifact],
output_rows=moved_outputs,
)
return history, collisions, too_long
def _length(column: Any) -> int:
"""How many characters a text column holds."""
return cast(int, cast(String, column.type).length)
def _fingerprint(plan: RenamePlan) -> str:
seen = sorted(
(d.kind, d.id, d.published_version or 0, d.draft_version or 0)
for d in plan.documents
)
body = json.dumps([seen, plan.name, plan.new_name]).encode()
return f"sha256:{hashlib.sha256(body).hexdigest()}"
backend/fluksio/flow/store.py
+4 -16
View File
@@ -34,6 +34,7 @@ from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING, Literal
from fluksio.flow.namespace import FLOW_FIELDS, Rename, rebind
from fluksio.flow.schemas import FlowDef, NodeDef, _validate_name
if TYPE_CHECKING:
@@ -971,7 +972,9 @@ class FlowStore:
)
copy = self._claim(copy, create=True)
# Its own messages named with its own key are the copy's now.
self._retarget(copy, f"{source.name}.", f"{new_name}.")
document = copy.model_dump(mode="json")
rebind(document, FLOW_FIELDS, Rename(source.name, new_name).binding)
copy = FlowDef.model_validate(document)
path = self._draft_file(copy.id)
path.parent.mkdir(parents=True)
path.write_text(copy.model_dump_json(indent=2) + "\n")
@@ -993,21 +996,6 @@ class FlowStore:
)
return copy
@staticmethod
def _retarget(flow: FlowDef, old_prefix: str, new_prefix: str) -> bool:
"""Point this flow's cross-flow message names at a renamed flow."""
changed = False
for node in flow.nodes:
for specs in (node.requires, node.provides):
for position, spec in enumerate(specs):
if spec.name.startswith(old_prefix):
tail = spec.name[len(old_prefix) :]
specs[position] = spec.model_copy(
update={"name": new_prefix + tail}
)
changed = True
return changed
# -------------------------------------------------------------------------
# Node source
# -------------------------------------------------------------------------
backend/tests/flow/test_drafts.py
+10 -2
View File
@@ -7,7 +7,7 @@ import pytest
from fluksio.flow.controller import FlowController
from fluksio.flow.messages import MessageSpec
from fluksio.flow.schemas import FlowDef, FlowOrigin, NodeDef
from fluksio.flow.schemas import FlowDef, FlowInput, FlowOrigin, NodeDef
from fluksio.flow.store import ISSUED_ID, FlowStore, StaleVersion
from tests.utils.stores import created
@@ -166,7 +166,13 @@ def test_resaving_the_published_source_creates_no_draft(store: FlowStore):
def test_a_duplicate_is_a_local_draft_of_what_the_editor_shows(store: FlowStore):
flow = a_flow(title="Warm").model_copy(update={"origin": FlowOrigin()})
flow = a_flow(title="Warm").model_copy(
update={
"origin": FlowOrigin(),
"inputs": [FlowInput(spec=MessageSpec(name="heating.target"))],
"outputs": ["heating.temp", "temp"],
}
)
flow.nodes.append(NodeDef(id="reader", requires=[MessageSpec(name="heating.temp")]))
store.write_flow(flow)
store.write_node_source("heating", "sensor", SOURCE)
@@ -180,6 +186,8 @@ def test_a_duplicate_is_a_local_draft_of_what_the_editor_shows(store: FlowStore)
assert ISSUED_ID.fullmatch(copy.id)
assert copy.origin is None
assert copy.nodes[1].requires[0].name == "heating_copy.temp"
assert copy.inputs[0].spec.name == "heating_copy.target"
assert copy.outputs == ["heating_copy.temp", "temp"]
assert store.has_draft(copy.id) and not store.is_published(copy.id)
assert store.read_node_source(copy.id, "sensor", draft=True) == EDITED
assert not (store.root / copy.id / "runtime.json").exists()
backend/tests/flow/test_rename_runtime.py
+17
View File
@@ -4,6 +4,7 @@ FLUKSIO_TEST_REDIS_HOST (and optional PORT)."""
import os
import threading
import time
import uuid
import pytest
@@ -487,3 +488,19 @@ def test_state_is_never_moved_onto_a_key_that_exists(state):
with pytest.raises(RuntimeError, match="exists"):
state.rename_keys(HEAT)
assert state.get("warm.temp") == 2
def test_another_engine_reading_the_queue_is_seen(redis_queue):
assert redis_queue.server_version() >= (7, 0)
redis_queue.claim(1, 10)
assert redis_queue.other_consumers(10_000) == []
other = _redis_queue(redis_queue._ns, "engine-b")
try:
other.claim(1, 10)
assert redis_queue.other_consumers(10_000) == ["engine-b"]
# Gone quiet for longer than asked about: a stopped engine.
time.sleep(0.05)
assert redis_queue.other_consumers(10) == []
finally:
other.close()
backend/tests/test_rename_plan.py
+687
View File
@@ -0,0 +1,687 @@
"""Planning a rename and its document part (sections 1.5 and 5.1 of
docs/private/id_refac_rename.md, "The key rename operation" in id_refac.md)."""
import ast
import inspect
import json
import time
from collections.abc import Iterator
from datetime import UTC, datetime
from pathlib import Path
from types import SimpleNamespace
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import JSON, String
from sqlmodel import Session, SQLModel, delete
from sqlmodel.sql.sqltypes import AutoString
from fluksio.core.config import settings
from fluksio.core.db import engine
from fluksio.flow.alerts import AlertsConfig, Channel
from fluksio.flow.controller import FlowController
from fluksio.flow.dashboards import (
DASHBOARD_BINDINGS,
WIDGET_BINDINGS,
DashboardDef,
DashboardStore,
SettingDef,
WidgetDef,
)
from fluksio.flow.executor import ExecutionService
from fluksio.flow.messages import MessageSpec
from fluksio.flow.namespace import FLOW_FIELDS, Rename, names
from fluksio.flow.queue import MemoryWorkQueue, WorkItem
from fluksio.flow.rename import SQL_NAMES, RenamePlan, plan_rename, rewrite_documents
from fluksio.flow.schemas import FlowDef, FlowInput, FlowOrigin, NodeDef
from fluksio.flow.state import MemoryState
from fluksio.flow.store import FlowStore
from fluksio.models import AlertSent, Run, RunArtifact, RunMetric, RunNode
from tests.utils.stores import created
NOW = datetime.now(UTC)
def _flow(name: str, *nodes: NodeDef, **fields) -> FlowDef:
return FlowDef(name=name, nodes=list(nodes), **fields)
def _reads(node_id: str, *messages: str) -> NodeDef:
return NodeDef(id=node_id, requires=[MessageSpec(name=m) for m in messages])
def _writes(node_id: str, *messages: str) -> NodeDef:
return NodeDef(id=node_id, provides=[MessageSpec(name=m) for m in messages])
@pytest.fixture
def world(tmp_path: Path) -> Iterator[SimpleNamespace]:
"""`heating`, published with a draft over it; `lights` reading it, published;
`notes` reading it, only a draft; `heating_extra`, which only looks alike;
and `home`, a dashboard bound to it, published with a draft over it."""
store = FlowStore(tmp_path / "flows")
dashboards = DashboardStore(store)
queue = MemoryWorkQueue()
controller = FlowController(
store, state=MemoryState(), execution=ExecutionService(queue)
)
heating = _flow(
"heating",
_writes("sensor", "temp", "heating.setpoint"),
inputs=[FlowInput(spec=MessageSpec(name="heating.target"))],
outputs=["temp", "heating.setpoint"],
)
store.write_flow(heating)
draft = store.read_flow("heating", draft=True)
draft.nodes.append(_reads("valve", "heating.temp"))
store.write_draft(draft, draft.version)
store.write_flow(_flow("lights", _reads("lamp", "heating.temp", "lights.level")))
created(store, _flow("notes", _reads("note", "heating.mode")))
store.write_flow(_flow("heating_extra", _writes("x", "heating_extra.x")))
home = DashboardDef(
name="home",
settings={"theme": SettingDef(message="heating.mode", dtype="str")},
widgets=[
WidgetDef(
id="t",
type="stat",
title="heating.temp",
config={"message": "heating.temp"},
),
WidgetDef(
id="s",
type="slider",
config={
"target": "heating.setpoint",
"reading": {"message": "heating.temp", "label": "heating.temp"},
},
),
WidgetDef(id="n", type="markdown", config={"text": "heating.temp"}),
],
)
dashboards.write(home)
current = dashboards.read("home", draft=True)
dashboards.write_draft(
current.model_copy(update={"title": "Home"}), current.version
)
dashboards.write(DashboardDef(name="away"))
yield SimpleNamespace(
store=store, dashboards=dashboards, controller=controller, queue=queue
)
def _plan(
world, new: str = "warmth", kind: str = "flow", obj: str = "heating", alerts=None
) -> RenamePlan:
return plan_rename(
world.controller, world.dashboards, kind, obj, new, alerts or AlertsConfig()
)
def _codes(found) -> set[str]:
return {one.code for one in found}
def _files(root: Path) -> dict[str, bytes]:
return {
str(path.relative_to(root)): path.read_bytes()
for path in sorted(root.rglob("*"))
if path.is_file() and ".git" not in path.parts
}
def _apply(world, rewrites) -> None:
for rewrite in rewrites:
for path, _, after in rewrite.files:
(world.store.root / path).write_text(after)
def test_a_plan_lists_the_flow_what_names_it_and_what_moves(world):
world.controller.state.set("heating.temp", 20.5)
world.controller.state.set("__version__:heating.temp", 3)
world.queue.add(
WorkItem(
kind="cascade",
node="lights.lamp",
flow="lights",
outputs={"heating.temp": 1},
)
)
alerts = AlertsConfig(
channels=[
Channel(
name="board", kind="dashboard", config={"message": "heating.alerts"}
),
Channel(name="mail", kind="smtp", config={"to": "heating.x@example.com"}),
]
)
with Session(engine) as session:
session.add(Run(id="r-plan", created_at=NOW, flow_id="heating", status="ok"))
session.add(RunMetric(run_id="r-plan", name="heating.loss", step=0))
session.add(
RunArtifact(
run_id="r-plan",
name="heating.plot",
digest="d",
size=1,
media_type="image/png",
)
)
session.add(
RunNode(
run_id="r-plan", node="heating.sensor", outputs='{"heating.temp": 1}'
)
)
session.add(AlertSent(key="node_failed:heating.sensor", ts=time.time()))
session.add(AlertSent(key="node_failed:heating_extra.x", ts=time.time()))
session.commit()
try:
plan = _plan(world, alerts=alerts)
finally:
with Session(engine) as session:
for model in (RunMetric, RunArtifact, RunNode, Run, AlertSent):
session.execute(delete(model))
session.commit()
assert plan.allowed, plan.refusals
assert (plan.name, plan.new_name, plan.version) == ("heating", "warmth", 2)
listed = {
(d.kind, d.id): (d.published_version, d.draft_version, d.fields)
for d in plan.documents
}
assert listed == {
# Its own qualified names: three in the published variant, four in
# the draft.
("flow", "heating"): (1, 2, 4),
("flow", "lights"): (1, None, 1),
("flow", "notes"): (None, 1, 1),
("dashboard", "home"): (1, 2, 4),
}
assert plan.fingerprint.startswith("sha256:")
assert plan.history is not None and plan.history.model_dump() == {
"runs": 1,
"metric_rows": 1,
"artifact_rows": 1,
"output_rows": 1,
}
assert plan.alerts is not None and plan.alerts.model_dump() == {
"channels": 1,
"ledger_rows": 1,
}
assert plan.runtime is not None
assert (
plan.runtime.durable,
plan.runtime.state_keys,
plan.runtime.queue_items,
) == (False, 2, 1)
assert _codes(plan.warnings) == {
"external_references",
"drafts_rewritten",
"not_durable",
"engine_pause",
}
def test_a_dry_run_changes_no_bytes(world):
before = _files(world.store.root)
head = world.store._git("rev-parse", "HEAD").stdout
_plan(world)
_plan(world, "house", "dashboard", "home")
assert _files(world.store.root) == before
assert world.store._git("rev-parse", "HEAD").stdout == head
def test_the_fingerprint_follows_every_listed_documents_version(world):
first = _plan(world)
assert _plan(world).fingerprint == first.fingerprint
lights = world.store.read_flow("lights")
world.store.write_draft(
lights.model_copy(update={"title": "Lights"}), lights.version
)
assert _plan(world).fingerprint != first.fingerprint
def test_a_rename_rewrites_bindings_and_nothing_else(world):
legacy = world.dashboards.root / "legacy" / "dashboard.json"
legacy.parent.mkdir()
legacy.write_text(
json.dumps(
{
"name": "legacy",
"version": 4,
"pages": [
{
"sections": [
{
"widgets": [
{
"id": "w",
"type": "stat",
"config": {"message": "heating.temp"},
}
]
}
]
}
],
},
indent=2,
)
)
device = world.store.root / "_devices" / "panel" / "device.json"
device.parent.mkdir(parents=True)
device.write_text('{"name": "panel", "dashboards": ["home", "legacy"]}\n')
before = _files(world.store.root)
rewrites = rewrite_documents(
world.store, world.dashboards, "flow", "heating", Rename("heating", "warmth")
)
_apply(world, rewrites)
changed = {path for r in rewrites for path, _, _ in r.files}
after = _files(world.store.root)
# The stability set: no file appears or goes, and nothing else changes.
assert after.keys() == before.keys()
assert {path for path in after if after[path] != before[path]} == changed
assert "_devices/panel/device.json" not in changed
for kind, stem, paths in (
("flow", "*/flow*.json", FLOW_FIELDS),
("dashboard", "_dashboards/*/dashboard*.json", DASHBOARD_BINDINGS),
):
for path in world.store.root.glob(stem):
data = json.loads(path.read_text())
assert data.get("id", path.parent.name) == path.parent.name
bound = names(data, paths)
assert not [n for n in bound if n.startswith("heating.")], (kind, path)
assert (
world.store.read_flow("heating_extra").nodes[0].provides[0].name
== "heating_extra.x"
)
renamed = world.store.read_flow("heating", draft=True)
assert renamed.name == "warmth" and renamed.version == 3
assert [s.name for s in renamed.nodes[0].provides] == ["temp", "warmth.setpoint"]
assert renamed.inputs[0].spec.name == "warmth.target"
assert renamed.outputs == ["temp", "warmth.setpoint"]
assert world.store.read_flow("heating").version == 3
assert (
world.store.read_flow("notes", draft=True).nodes[0].requires[0].name
== "warmth.mode"
)
home = world.dashboards.read("home", draft=True)
assert (home.version, home.title) == (3, "Home")
assert home.settings["theme"].message == "warmth.mode"
assert home.widgets[0].messages == ["warmth.temp"]
# Labels and free text are the person's.
assert home.widgets[0].title == "heating.temp"
assert home.widgets[1].target == "warmth.setpoint"
assert home.widgets[1].reading == {
"message": "warmth.temp",
"label": "heating.temp",
}
assert home.widgets[2].config == {"text": "heating.temp"}
assert world.dashboards.read("home").version == 3
assert json.loads(legacy.read_text())["pages"][0]["sections"][0]["widgets"][0][
"config"
] == {"message": "warmth.temp"}
def test_renaming_back_restores_the_documents_but_for_versions(world):
def unversioned() -> dict:
return {
path: {k: v for k, v in json.loads(text).items() if k != "version"}
for path, text in _files(world.store.root).items()
if path.endswith(".json") and not path.endswith("runtime.json")
}
before = unversioned()
_apply(
world,
rewrite_documents(
world.store,
world.dashboards,
"flow",
"heating",
Rename("heating", "warmth"),
),
)
assert unversioned() != before
_apply(
world,
rewrite_documents(
world.store,
world.dashboards,
"flow",
"heating",
Rename("warmth", "heating"),
),
)
assert unversioned() == before
def test_a_dashboard_rename_changes_its_key_and_versions_only(world):
plan = _plan(world, "house", "dashboard", "home")
assert plan.allowed and plan.version == 2
assert [
(d.kind, d.id, d.published_version, d.draft_version, d.fields)
for d in plan.documents
] == [("dashboard", "home", 1, 2, 0)]
assert (plan.alerts, plan.history, plan.runtime, plan.warnings) == (
None,
None,
None,
[],
)
assert _codes(_plan(world, "away", "dashboard", "home").refusals) == {"name_taken"}
(rewrite,) = rewrite_documents(
world.store, world.dashboards, "dashboard", "home", Rename("home", "house")
)
_apply(world, [rewrite])
published, draft = (
world.dashboards.read("home"),
world.dashboards.read("home", draft=True),
)
assert (published.name, published.version, draft.name, draft.version) == (
"house",
3,
"house",
3,
)
assert draft.widgets[0].messages == ["heating.temp"]
@pytest.mark.parametrize(
("new", "code"),
[
("Warmth", "invalid_name"),
("f_abcdefgh", "reserved_shape"),
("heating", "same_name"),
("lights", "name_taken"),
("notes", "name_taken"),
],
)
def test_a_key_that_cannot_be_taken_is_refused(world, new, code):
plan = _plan(world, new)
assert not plan.allowed
assert code in _codes(plan.refusals)
def test_code_defined_flows_are_refused(world):
lights = world.store.read_flow("lights")
world.store.write_flow(lights.model_copy(update={"origin": FlowOrigin()}))
plan = _plan(world)
assert [(r.code, r.ids) for r in plan.refusals] == [
("code_defined_dependent", ["lights"])
]
heating = world.store.read_flow("heating")
world.store.write_flow(heating.model_copy(update={"origin": FlowOrigin()}))
assert _codes(_plan(world).refusals) == {"code_defined", "code_defined_dependent"}
def test_an_active_run_of_an_affected_flow_is_refused(world):
with Session(engine) as session:
session.add(
Run(id="r-busy", created_at=NOW, flow_id="lights", status="running")
)
session.add(
Run(
id="r-train",
created_at=NOW,
flow_id="trainer",
for_node="heating.model",
status="queued",
)
)
session.add(
Run(id="r-other", created_at=NOW, flow_id="heating_extra", status="running")
)
session.commit()
try:
plan = _plan(world)
finally:
with Session(engine) as session:
session.execute(delete(Run))
session.commit()
assert [(r.code, sorted(r.ids)) for r in plan.refusals] == [
("run_active", ["r-busy", "r-train"])
]
def test_what_still_lives_under_the_new_key_is_refused(world):
state = world.controller.state
# A node's own key is its data, even when it reads like a name.
state.set("__node__:lights.lamp:last:warmth.temp", 1)
assert _plan(world).allowed
state.set("__last_seen__:lights.lamp:warmth.temp", 1)
world.queue.add(WorkItem(kind="cascade", node="warmth.sensor", flow="warmth"))
plan = _plan(world)
assert [r.code for r in plan.refusals] == ["target_in_use"]
assert "__last_seen__:lights.lamp:warmth.temp" in plan.refusals[0].detail
assert world.controller.key_residue("warmth") == [
"__last_seen__:lights.lamp:warmth.temp",
"queued work of 'warmth.sensor'",
]
state.delete("__last_seen__:lights.lamp:warmth.temp")
world.queue.clear_flow("warmth")
away = world.dashboards.read("away")
world.dashboards.write(
away.model_copy(
update={
"widgets": [
WidgetDef(id="w", type="stat", config={"message": "warmth.temp"})
]
}
)
)
plan = _plan(world)
assert [(r.code, r.ids) for r in plan.refusals] == [("target_in_use", ["away"])]
def test_history_that_would_collide_or_overflow_is_refused(world):
long = "heating." + "x" * (128 - len("heating."))
with Session(engine) as session:
session.add(Run(id="r-hist", created_at=NOW, flow_id="heating", status="ok"))
session.add(RunMetric(run_id="r-hist", name="heating_v2.loss", step=0))
session.add(RunMetric(run_id="r-hist", name=long, step=0))
session.commit()
try:
assert _codes(_plan(world, "heating_v2").refusals) == {
"history_collision",
"name_too_long",
}
# A key no longer than the old one fits.
assert _codes(_plan(world, "warmth").refusals) == set()
finally:
with Session(engine) as session:
for model in (RunMetric, Run):
session.execute(delete(model))
session.commit()
def test_another_engine_or_an_old_redis_is_refused(world, monkeypatch):
monkeypatch.setattr(
MemoryWorkQueue, "other_consumers", lambda self, ms: ["engine-b"]
)
monkeypatch.setattr(MemoryWorkQueue, "server_version", lambda self: (6, 2))
assert _codes(_plan(world).refusals) == {"other_engine", "redis_too_old"}
def test_what_a_rename_leaves_behind_is_said(world):
# A flow whose id is the new key, under a key of its own.
world.store.create(_flow("elsewhere"), "warmth")
world.dashboards.write(DashboardDef(name="heating_results"))
world.queue.park(
"heating", WorkItem(kind="cascade", node="heating.sensor", flow="heating")
)
plan = _plan(world)
assert plan.allowed
assert {"new_name_is_id", "results_dashboard", "paused"} <= _codes(plan.warnings)
def test_every_binding_a_widget_reads_is_in_the_key_table():
"""A new binding in an accessor fails here until the table knows it, so a
rename cannot leave it behind. Keys that hold no message name are listed
as such."""
plain = {
"source",
"dtype",
"inner_dtype",
"request_dtype",
"history",
"points",
"format",
"counters",
"ids",
"group",
"flow",
"latest",
}
tree = ast.parse(inspect.getsource(WidgetDef))
read = {
call.args[0].value
for call in ast.walk(tree)
if isinstance(call, ast.Call)
and isinstance(call.func, ast.Attribute)
and call.func.attr == "get"
and call.args
and isinstance(call.args[0], ast.Constant)
}
bound = {part for path in WIDGET_BINDINGS for part in path if part != "*"}
assert read - plain == bound
assert set(SettingDef.model_fields) - {"value", "dtype"} == {"message"}
assert ("settings", "*", "message") in DASHBOARD_BINDINGS
def test_every_history_column_is_classified():
"""A column added to the history fails here until it says whether it
names a message, which a rename then rewrites (`SQL_NAMES`), or not."""
unnamed = {
# Ids, which never change: flows and nodes are filed by id.
"run": {
"id",
"flow",
"commit",
"origin_commit",
"code_digest",
"params",
"passed",
"declared",
"params_digest",
"group_id",
"parent_id",
"idempotency_key",
"external_id",
"cause",
"for_node",
"status",
"status_reason",
"labels",
"needs",
"result",
"engine",
"actor",
},
"run_node": {
"run_id",
"node",
"status",
"worker",
"error",
"logs",
"cache_key",
"cached_from",
},
"run_metric": {"run_id", "node"},
"run_artifact": {"run_id", "digest", "media_type", "filename", "node"},
"flow_run": {"id", "flow", "source", "parent", "status"},
"metric_minute": {"flow", "node"},
"engine_event": {"type", "flow", "node", "detail", "actor", "run"},
"flow_tombstone": {"id", "name"},
}
# Tables that hold no flow, node or message at all.
unrelated = {
"user",
"oauth_client",
"oauth_authorization_code",
"oauth_refresh_token",
"flavor",
"worker_token",
"connector_state",
"alembic_version",
}
found: dict[str, set[str]] = {}
for table in SQLModel.metadata.sorted_tables:
if table.name in unrelated:
continue
found[table.name] = {
column.name
for column in table.columns
if isinstance(column.type, String | AutoString | JSON)
}
classified = dict(unnamed)
for table, column in SQL_NAMES:
classified.setdefault(table, set()).add(column)
assert found == classified
def test_planning_is_a_superusers(
client: TestClient, superuser_token_headers, normal_user_token_headers
):
flows = f"{settings.API_V1_STR}/flows"
made = client.post(
f"{flows}/", headers=superuser_token_headers, json={"name": "plan_me"}
)
assert made.status_code == 201, made.text
flow_id = made.json()["definition"]["id"]
try:
body = {"new_name": "plan_me_too"}
assert (
client.post(
f"{flows}/{flow_id}/rename/plan",
headers=normal_user_token_headers,
json=body,
).status_code
== 403
)
assert (
client.post(
f"{flows}/f_nothere0/rename/plan",
headers=superuser_token_headers,
json=body,
).status_code
== 404
)
plan = client.post(
f"{flows}/{flow_id}/rename/plan", headers=superuser_token_headers, json=body
)
assert plan.status_code == 200, plan.text
assert (plan.json()["name"], plan.json()["allowed"]) == ("plan_me", True)
boards = f"{settings.API_V1_STR}/dashboards"
board = client.post(
f"{boards}/", headers=superuser_token_headers, json={"name": "plan_board"}
).json()
try:
plan = client.post(
f"{boards}/{board['id']}/rename/plan",
headers=superuser_token_headers,
json=body,
)
assert plan.status_code == 200 and plan.json()["kind"] == "dashboard"
finally:
client.delete(f"{boards}/{board['id']}", headers=superuser_token_headers)
finally:
client.delete(f"{flows}/{flow_id}", headers=superuser_token_headers)