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

Flows and dashboards get ids: the store is addressed by id with a key index, the first start gives every document its key as id, fluksio store migrate --dry-run and ls; the rename routes are gone

Browse Source
This commit is contained in:
stroblme
2026-11-02 16:54:11 +01:00
parent de13c48c90
commit fae85f5ecb
20 changed files with 1287 additions and 420 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/api/routes/dashboards.py
+8 -19
View File
@@ -26,10 +26,6 @@ router = APIRouter(
)
class RenameRequest(BaseModel):
name: str
class PublishRequest(BaseModel):
version: int
@@ -101,7 +97,10 @@ async def generate_results_dashboard(
if await run_in_threadpool(store.exists, name):
raise HTTPException(status_code=409, detail=f"'{name}' already exists")
written = await run_in_threadpool(store.write, results_dashboard(definition))
try:
written = await run_in_threadpool(store.write, results_dashboard(definition))
except DashboardExists as exc:
raise HTTPException(status_code=409, detail=str(exc))
await run_in_threadpool(_apply_history_limits, store, controller)
event_bus.publish(
{"type": "dashboard_changed", "dashboard": name, "ts": time.time()}
@@ -123,6 +122,8 @@ async def save_dashboard(
"""
if body.name != name:
raise HTTPException(status_code=422, detail="The name in the body must match")
if body.id and body.id != name:
raise HTTPException(status_code=422, detail="The id in the body must match")
try:
saved = await run_in_threadpool(store.write_draft, body, body.version)
except StaleVersion as exc:
@@ -133,6 +134,8 @@ async def save_dashboard(
"current_version": exc.current,
},
)
except DashboardExists as exc:
raise HTTPException(status_code=409, detail=str(exc))
return saved
@@ -211,17 +214,3 @@ async def duplicate_dashboard(name: str, store: DashboardStoreDep) -> Any:
return await run_in_threadpool(store.duplicate, name)
except DashboardNotFound:
raise HTTPException(status_code=404, detail=f"No dashboard named '{name}'")
@router.post("/{name}/rename", response_model=DashboardDef)
async def rename_dashboard(
name: str, body: RenameRequest, store: DashboardStoreDep
) -> Any:
try:
return await run_in_threadpool(store.rename, name, body.name)
except DashboardNotFound:
raise HTTPException(status_code=404, detail=f"No dashboard named '{name}'")
except DashboardExists:
raise HTTPException(status_code=409, detail=f"'{body.name}' already exists")
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc))
backend/fluksio/api/routes/flows.py
+7 -35
View File
@@ -140,10 +140,6 @@ class ValidationResult(BaseModel):
issues: list[ValidationIssue] = []
class RenameRequest(BaseModel):
new_name: str
class PublishRequest(BaseModel):
version: int
@@ -338,6 +334,7 @@ def read_flows(controller: FlowControllerDep) -> Any:
statuses = controller.node_statuses(name)
summaries.append(
FlowSummary(
id=definition.id,
name=definition.name,
title=definition.title,
mode=definition.mode,
@@ -539,6 +536,10 @@ async def save_flow(
raise HTTPException(
status_code=400, detail="The flow name in the body must match the URL"
)
if definition.id and definition.id != name:
raise HTTPException(
status_code=400, detail="The flow id in the body must match the URL"
)
duplicates = {n.id for n in definition.nodes}
if len(duplicates) != len(definition.nodes):
@@ -559,6 +560,8 @@ async def save_flow(
status_code=409,
detail={"message": str(exc), "current_version": exc.current},
)
except FlowExists as exc:
raise HTTPException(status_code=409, detail=str(exc))
return await run_in_threadpool(_detail, controller, stored)
@@ -641,37 +644,6 @@ async def delete_flow(
return Message(message=f"Deleted flow '{name}'")
@router.post("/{name}/rename", response_model=FlowDetail)
async def rename_flow(
name: str,
body: RenameRequest,
controller: FlowControllerDep,
) -> Any:
"""Rename a flow, along with every reference to its messages."""
if not NAME_PATTERN.match(body.new_name):
raise HTTPException(
status_code=400,
detail=(
"Use lowercase letters, digits and underscores, starting with a letter"
),
)
try:
renamed = await run_in_threadpool(
controller.store.rename_flow, name, body.new_name
)
except FlowNotFound:
raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
except FlowExists as exc:
raise HTTPException(status_code=409, detail=str(exc))
# The old name is nobody's namespace now; its values would sit there under a
# flow that no longer exists. They repopulate under the new name on the next
# run, so this is cleanup rather than a migration.
await run_in_threadpool(controller.forget_flow, name)
await controller.reload()
return await run_in_threadpool(_detail, controller, renamed)
@router.post("/{name}/duplicate", response_model=FlowDetail)
async def duplicate_flow(
name: str, controller: FlowControllerDep, user: CurrentUser
backend/fluksio/cli.py
+81 -1
View File
@@ -1,4 +1,4 @@
"""`fluksio serve`, `fluksio enroll`, `fluksio worker`, `sync`, `run`, `runs`.
"""`fluksio serve`, `enroll`, `worker`, `store`, `sync`, `run`, `runs`.
The point of this module is a machine nobody can route to: a node on a cluster
where ports cannot be opened, or a laptop with no Docker. `fluksio serve`
@@ -686,6 +686,71 @@ def cmd_enroll(args: argparse.Namespace) -> int:
return result
def _store_dir(raw: str) -> Path | None:
path = Path(raw).expanduser()
if path.is_dir():
return path
print(f"{path} is not a directory", file=sys.stderr, flush=True)
return None
def cmd_store_migrate(args: argparse.Namespace) -> int:
"""Say what the engine's next start would do to a store's ids.
The same check the start makes, reading and nothing else: no store is
opened, so no repository is created and no other migration runs. Point it
at a copy of the store; the engine does the migration itself, with the
store to itself.
"""
from fluksio.flow.identity import plan_ids
root = _store_dir(args.dir)
if root is None:
return 2
plan = plan_ids(root)
flows = sum(1 for kind, _, _ in plan.objects if kind == "flow")
_say(f"{root}: {flows} flows, {len(plan.objects) - flows} dashboards")
if plan.problems:
_say("The engine would refuse to start, and change nothing, because:")
for problem in plan.problems:
_say(f" {problem}")
return 1
if plan.pending:
_say("An interrupted start gave these their id, and would commit them:")
for rel in sorted(plan.pending):
_say(f" {rel}")
if plan.stamps:
_say("The engine would give these their key as id, in one commit:")
for rel, version in sorted(plan.stamps.items()):
_say(f" {rel} (version {version})")
if not plan.pending and not plan.stamps:
_say("Nothing to do: every flow and dashboard has its id.")
return 0
def cmd_store_ls(args: argparse.Namespace) -> int:
"""Every flow and dashboard in a store, by id, key and kind."""
from fluksio.flow.identity import plan_ids
root = _store_dir(args.dir)
if root is None:
return 2
plan = plan_ids(root)
width = max([len("ID"), *(len(obj_id) for _, obj_id, _ in plan.objects)])
keys = max([len("KEY"), *(len(key) for _, _, key in plan.objects)])
_say(f"{'ID':<{width}} {'KEY':<{keys}} KIND")
for kind, obj_id, key in sorted(plan.objects):
_say(f"{obj_id:<{width}} {key:<{keys}} {kind}")
if plan.problems:
print(
f"{len(plan.problems)} problem(s): `fluksio store migrate --dry-run "
f"{root}` lists them",
file=sys.stderr,
flush=True,
)
return 0
def cmd_worker(rest: list[str]) -> int:
from fluksio_worker.agent import main as worker_main
@@ -804,6 +869,21 @@ def _parser() -> argparse.ArgumentParser:
add_help=False,
)
store = subparsers.add_parser("store", help="look into a flow store directory")
store_commands = store.add_subparsers(dest="store_command", required=True)
migrate = store_commands.add_parser(
"migrate", help="check what the next engine start migrates"
)
# Required: the engine migrates on its start, with the store to itself.
migrate.add_argument(
"--dry-run", action="store_true", required=True, help="change nothing"
)
migrate.add_argument("dir", help="the store, e.g. a copy of <data dir>/flows")
migrate.set_defaults(func=cmd_store_migrate)
ls = store_commands.add_parser("ls", help="list flows and dashboards by id")
ls.add_argument("dir", help="the store, e.g. <data dir>/flows")
ls.set_defaults(func=cmd_store_ls)
# The client half: talking to an engine rather than being one.
from fluksio.sdk.cli import add_parsers
backend/fluksio/flow/dashboards.py
+67 -32
View File
@@ -25,13 +25,15 @@ from typing import Any, Literal
from pydantic import BaseModel, Field, field_validator, model_validator
from fluksio.flow.messages import DType, qualify
from fluksio.flow.schemas import FlowDef, _validate_name
from fluksio.flow.schemas import FlowDef, _validate_id, _validate_name
from fluksio.flow.store import (
FlowStore,
StaleVersion,
_updated_at,
checked,
claim,
copy_name,
read_keys,
)
#: Sibling of the shared-node library, and likewise not a flow.
@@ -500,6 +502,11 @@ def _flatten_pages(pages: list[Any]) -> list[dict[str, Any]]:
class DashboardDef(BaseModel):
"""A dashboard as stored, and as the API hands it over."""
#: Which dashboard this is, like a flow's (``FlowDef.id``), in an id space
#: of its own: the directory, never changed once given, the key for one
#: from before ids. Optional on the way in, always set on the way out.
id: str = ""
#: The key, unique among dashboards. Nothing binds to it, so it is a label.
name: str
title: str = ""
#: How many columns the grid is cut into, so a dashboard can be matched to
@@ -550,6 +557,11 @@ class DashboardDef(BaseModel):
data["widgets"] = _flatten_pages(pages)
return data
@field_validator("id")
@classmethod
def _check_id(cls, value: str) -> str:
return _validate_id(value)
@field_validator("name")
@classmethod
def _check_name(cls, value: str) -> str:
@@ -610,6 +622,7 @@ class DashboardDef(BaseModel):
class DashboardSummary(BaseModel):
"""A dashboard in a list, without its contents."""
id: str = ""
name: str
title: str = ""
#: The glyph this dashboard draws on a panel's rail, so a list can show it
@@ -648,6 +661,9 @@ class DashboardExists(ValueError):
super().__init__(name)
self.name = name
def __str__(self) -> str:
return f"There is already a dashboard named '{self.name}'"
class DashboardStore:
"""Dashboards in the flow store's repository, invisible to the flow listing.
@@ -665,6 +681,8 @@ class DashboardStore:
self._published: dict[str, DashboardDef | None] = {}
self._published_at = -1
self._published_lock = threading.Lock()
#: `key_index` and the flow store revision it was read at.
self._keys: tuple[int, dict[str, str]] = (-1, {})
def _file(self, name: str) -> Path:
return self.root / checked(name) / "dashboard.json"
@@ -689,6 +707,7 @@ class DashboardStore:
continue
summaries.append(
DashboardSummary(
id=defn.id,
name=defn.name,
title=defn.title,
icon=defn.icon,
@@ -725,10 +744,45 @@ class DashboardStore:
path = self._file(name)
if not path.exists():
raise DashboardNotFound(name)
return DashboardDef.model_validate_json(path.read_text()).model_copy(
update={"has_draft": self.has_draft(name)}
defn = DashboardDef.model_validate_json(path.read_text())
# One from before ids is known by its directory, which is its key.
return defn.model_copy(
update={"id": defn.id or name, "has_draft": self.has_draft(name)}
)
def key_index(self) -> dict[str, str]:
"""Every dashboard's key, to its id, whether published or only a draft.
Read again after any commit of the store, as `FlowStore.key_index`.
Shared: read it, never change it.
"""
at, keys = self._keys
if at == self.flows.revision:
return keys
at = self.flows.revision
found = {p.parent for p in self.root.glob("*/dashboard.json")}
found |= {p.parent for p in self.root.glob("*/dashboard.draft.json")}
keys = read_keys(sorted(found), "dashboard")
self._keys = (at, keys)
return keys
def _claim(self, defn: DashboardDef) -> DashboardDef:
"""*defn* with its id, once it is certain to be a save the store allows.
Under the flow store's write lock; see `claim`. An id left out is the
key, which is how every dashboard is created for now.
"""
dash_id = checked(defn.id or defn.name)
current = self.read(dash_id, draft=True) if self.exists(dash_id) else None
claim(
dash_id,
defn.name,
(current.id, current.name) if current else None,
self.key_index(),
DashboardExists,
)
return defn.model_copy(update={"id": dash_id})
def published(self, name: str) -> DashboardDef | None:
"""The published document, or None, as of the store's last commit.
@@ -756,7 +810,8 @@ class DashboardStore:
caller that already has the finished document — a test, or a seed.
"""
with self._lock, self.flows._write_lock:
path = self._file(defn.name)
defn = self._claim(defn)
path = self._file(defn.id)
current = 0
if path.exists():
current = DashboardDef.model_validate_json(path.read_text()).version
@@ -779,14 +834,15 @@ class DashboardStore:
does not exist yet: creating one is its first draft.
"""
with self._lock, self.flows._write_lock:
defn = self._claim(defn)
current = 0
if self.exists(defn.name):
current = self.read(defn.name, draft=True).version
if self.exists(defn.id):
current = self.read(defn.id, draft=True).version
if base_version is not None and base_version != current:
raise StaleVersion(defn.name, current)
saved = defn.model_copy(update={"version": current + 1, "has_draft": True})
path = self._draft_file(defn.name)
path = self._draft_file(defn.id)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(self._dump(saved))
self.flows._commit(f"Update draft of dashboard '{defn.name}'")
@@ -828,30 +884,6 @@ class DashboardStore:
pass
self.flows._commit(f"Delete dashboard '{name}'")
def rename(self, name: str, new_name: str) -> DashboardDef:
defn = self.read(name, draft=True)
if self.exists(new_name):
raise DashboardExists(new_name)
with self.flows._write_lock:
renamed = defn.model_copy(update={"name": new_name})
self._file(new_name).parent.mkdir(parents=True, exist_ok=True)
# Whichever files the dashboard has move; one nobody published yet
# has only the draft, and renaming it must not publish it.
if self.is_published(name):
published = self.read(name).model_copy(update={"name": new_name})
self._file(new_name).write_text(self._dump(published))
self._file(name).unlink()
# An unpublished edit belongs to the dashboard, so it moves too.
if self.has_draft(name):
self._draft_file(new_name).write_text(self._dump(renamed))
self._draft_file(name).unlink()
try:
self._file(name).parent.rmdir()
except OSError:
pass
self.flows._commit(f"Rename dashboard '{name}' to '{new_name}'")
return renamed
def duplicate(self, name: str) -> DashboardDef:
"""Copy the working copy into a new dashboard, as a draft only.
@@ -860,9 +892,12 @@ class DashboardStore:
"""
with self._lock, self.flows._write_lock:
source = self.read(name, draft=True)
new_name = copy_name(name, self.exists)
# The copy's id is its key, so it has to be free as both.
keys = self.key_index()
new_name = copy_name(source.name, lambda n: self.exists(n) or n in keys)
copy = source.model_copy(
update={
"id": new_name,
"name": new_name,
"title": f"{source.title} (copy)" if source.title else "",
"version": 1,
backend/fluksio/flow/identity.py
+280
View File
@@ -0,0 +1,280 @@
"""Giving every flow and dashboard stored before ids existed its id, once.
Before ids, a flow was its key: ``name`` was its directory, its URL, its hook
and its rows in the database. The id now does those jobs and the key keeps its
own, the namespace of the flow's messages (see `fluksio.flow.store`). A store
from before is moved over by writing each document's key into it as its id.
Nothing else changes, since every directory, URL, hook, device and row already
holds that string.
`assign_ids` does it when the engine starts, and refuses to touch a store whose
documents do not say clearly who they are. `plan_ids` is the same reading
without the writing, which is what ``fluksio store migrate --dry-run`` prints
and ``fluksio store ls`` lists.
"""
from __future__ import annotations
import json
import logging
import os
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from fluksio.flow.dashboards import DASHBOARD_DIR
logger = logging.getLogger(__name__)
#: The one commit the documents given an id land in.
COMMIT_MESSAGE = "Assign ids to flows and dashboards"
#: Each kind's documents, published first.
_DOCUMENTS = {
"flow": ("flow.json", "flow.draft.json"),
"dashboard": ("dashboard.json", "dashboard.draft.json"),
}
Documents = dict[tuple[str, str], dict[str, dict[str, Any]]]
class IdsRefused(RuntimeError):
"""A store whose documents do not say clearly enough who they are."""
def __init__(self, root: Path, problems: list[str]) -> None:
super().__init__(root, problems)
self.root = root
self.problems = problems
def __str__(self) -> str:
lines = "\n".join(f" {problem}" for problem in self.problems)
return (
f"Could not give the flows and dashboards in {self.root} their ids, "
f"so nothing was changed:\n{lines}\n"
"Fix these by hand and start again. `fluksio store migrate "
f"--dry-run {self.root}` checks a store without starting the engine."
)
@dataclass
class IdPlan:
"""What `assign_ids` would do to a store, read without changing it."""
#: Every object found, as ``(kind, id, key)``. The id is the directory,
#: which is also the one a document without an id would be given.
objects: list[tuple[str, str, str]] = field(default_factory=list)
#: Documents to give their id, relative to the store, with the version
#: each moves to.
stamps: dict[str, int] = field(default_factory=dict)
#: Documents an interrupted run already gave theirs, not yet committed.
pending: dict[str, int] = field(default_factory=dict)
#: Why it refuses, a line each. Nothing is written while there is one.
problems: list[str] = field(default_factory=list)
def _marker(root: Path) -> Path:
"""Where a run writes down what it is about to do, before doing any of it.
Inside ``.git``, where no commit picks it up, or in the store itself when
there is no repository, and so no commit either.
"""
git = root / ".git"
return (git if git.is_dir() else root) / ".assign-ids.json"
def _scan(root: Path) -> tuple[Documents, list[str]]:
"""Every document as plain JSON, by ``(kind, directory)``, and the unreadable."""
found: Documents = {}
problems = []
for kind, base in (("flow", root), ("dashboard", root / DASHBOARD_DIR)):
for name in _DOCUMENTS[kind]:
for path in sorted(base.glob(f"*/{name}")):
directory = path.parent.name
# `_devices/flow.json` is a device named "flow", and no other
# `_` directory is a flow either.
if kind == "flow" and directory.startswith(("_", ".")):
continue
rel = path.relative_to(root).as_posix()
try:
data = json.loads(path.read_text())
except (OSError, ValueError) as exc:
problems.append(f"{rel}: not readable as JSON ({exc})")
continue
if not isinstance(data, dict) or not isinstance(data.get("name"), str):
problems.append(f"{rel}: has no name")
continue
found.setdefault((kind, directory), {})[rel] = data
return found, problems
def plan_ids(root: Path) -> IdPlan:
"""What `assign_ids` would do to the store at *root*, changing nothing.
A document without an id is to get its directory as one, which has to be
its ``name`` as well: the engine runs a flow under the name in its
document, so when the two differ which one the flow is cannot be told, and
that is for a person to settle. Refused likewise: a document that is not
JSON, an id that is not its directory, a published document and draft
with different keys, and two objects of one kind sharing a key.
"""
plan = IdPlan()
found, plan.problems = _scan(root)
try:
recorded = json.loads(_marker(root).read_text())
except FileNotFoundError:
recorded = {}
owners: dict[tuple[str, str], list[str]] = {}
for (kind, directory), docs in sorted(found.items()):
before = len(plan.problems)
keys = sorted({data["name"] for data in docs.values()})
for key in keys:
owners.setdefault((kind, key), []).append(directory)
if len(keys) > 1:
plan.problems.append(
f"{kind} '{directory}': its published document and its draft "
f"have different keys, '{keys[0]}' and '{keys[1]}'"
)
# The working copy's key: the draft's, when there is one.
plan.objects.append((kind, directory, list(docs.values())[-1]["name"]))
unstamped = []
for rel, data in docs.items():
given, name = data.get("id"), data["name"]
if given and given != directory:
plan.problems.append(f"{rel}: its id '{given}' is not its directory")
elif not given and name != directory:
plan.problems.append(
f"{rel}: its name '{name}' is not its directory '{directory}'"
)
elif not isinstance(data.get("version", 1), int):
plan.problems.append(f"{rel}: its version is not a number")
elif not given:
unstamped.append(rel)
if not unstamped or len(plan.problems) > before:
continue
# Past both, so neither an editor on the draft nor one on what a
# discard goes back to can save over it.
target = max(data.get("version", 1) for data in docs.values()) + 1
for rel in unstamped:
# An interrupted run's number, unless the document moved past it.
planned = recorded.get(rel, 0)
own = docs[rel].get("version", 1)
plan.stamps[rel] = planned if planned > own else target
for (kind, key), directories in sorted(owners.items()):
if len(directories) > 1:
plan.problems.append(
f"the {kind}s {' and '.join(map(repr, directories))} share "
f"the key '{key}'"
)
plan.pending = {
rel: version
for rel, version in recorded.items()
if rel not in plan.stamps and (root / rel).exists()
}
return plan
def assign_ids(root: Path) -> list[str]:
"""Give every flow and dashboard at *root* that has no id its key as one.
For the engine's startup, with the store to itself and before anything
reads a definition. Raises `IdsRefused`, having written nothing, on a
problem `plan_ids` finds. Otherwise each document gets its id and moves
past the higher version of itself and its sibling, so an editor holding
the old one gets a conflict, after a discard too. All of it is one commit
of only those files, whatever else in the store is uncommitted.
The versions are written down before the first document is, and the note
goes once the commit is made, so a run interrupted anywhere in between is
finished by the next, with the same numbers. A store that has been through
it changes nothing and says nothing. Returns the documents changed.
"""
plan = plan_ids(root)
if plan.problems:
raise IdsRefused(root, plan.problems)
changed = {**plan.pending, **plan.stamps}
if not changed:
return []
marker = _marker(root)
if plan.stamps:
_replace(marker, json.dumps(changed, indent=2) + "\n")
for rel, version in sorted(plan.stamps.items()):
path = root / rel
data = json.loads(path.read_text())
data.pop("id", None)
stamped = {"id": path.parent.name, **data, "version": version}
_replace(path, json.dumps(stamped, indent=2, ensure_ascii=False) + "\n")
paths = sorted(changed)
committed = _commit(root, paths)
logger.warning(
"Gave each flow and dashboard without an id its key as one, in %s",
", ".join(paths),
)
if committed:
marker.unlink(missing_ok=True)
return paths
def _replace(path: Path, text: str) -> None:
"""Write *path* whole or not at all: a crash leaves the old file or the new."""
temp = path.with_name(f".{path.name}.tmp")
with temp.open("w") as handle:
handle.write(text)
handle.flush()
os.fsync(handle.fileno())
os.replace(temp, path)
def _commit(root: Path, paths: list[str]) -> bool:
"""Commit *paths* alone, leaving whatever else is uncommitted as it was.
True once nothing is left to commit: done now, done by the run this one
finishes, or no git to do it with, in which case the store is just files,
as `FlowStore` takes it too.
"""
if not (root / ".git").exists():
return True
def git(*args: str) -> subprocess.CompletedProcess[str]:
# `gc.auto=0` for the reason `FlowStore._git` gives.
return subprocess.run(
["git", "-C", str(root), "-c", "gc.auto=0", *args],
capture_output=True,
text=True,
check=False,
)
try:
git("add", "--", *paths)
if git("diff", "--cached", "--quiet", "HEAD", "--", *paths).returncode == 0:
return True
result = git(
"-c",
"user.name=fluksio",
"-c",
"user.email=fluksio@localhost",
"commit",
"-q",
"--only",
"-m",
COMMIT_MESSAGE,
"--",
*paths,
)
except FileNotFoundError:
return True
if result.returncode != 0:
logger.warning(
"Could not commit the ids, so the next start tries again: %s",
(result.stderr or result.stdout).strip(),
)
return False
return True
__all__ = ["COMMIT_MESSAGE", "IdPlan", "IdsRefused", "assign_ids", "plan_ids"]
backend/fluksio/flow/schemas.py
+16
View File
@@ -32,6 +32,11 @@ def _validate_name(value: str) -> str:
return value
def _validate_id(value: str) -> str:
"""An id is a name too, so it stays one safe path segment. Empty is unset."""
return value and _validate_name(value)
def _megabytes(value: Any) -> Any:
"""``2G`` and ``512M`` and ``512`` all mean a number of megabytes."""
if not isinstance(value, str):
@@ -302,6 +307,11 @@ class FlowInput(BaseModel):
class FlowDef(BaseModel):
"""One atomic flow."""
#: Which flow this is: the directory it is stored in, never changed once
#: given. A flow from before ids existed has its key as its id. Optional
#: on the way in, since the store fills it, and always set on the way out.
id: str = ""
#: The key: unique among flows, and the namespace of the flow's messages.
name: str
title: str = ""
nodes: list[NodeDef] = Field(default_factory=list)
@@ -331,6 +341,11 @@ class FlowDef(BaseModel):
),
)
@field_validator("id")
@classmethod
def _check_id(cls, value: str) -> str:
return _validate_id(value)
@field_validator("name")
@classmethod
def _check_name(cls, value: str) -> str:
@@ -402,6 +417,7 @@ class MessageHistory(BaseModel):
class FlowSummary(BaseModel):
id: str = ""
name: str
title: str = ""
#: Whether running it means one finite execution or leaving it running.
backend/fluksio/flow/store.py
+168 -60
View File
@@ -10,6 +10,12 @@ Editing is separated from running. The editor writes ``flow.draft.json`` and
``nodes/*.py``. Publishing promotes the draft over the published files and
removes it, so a flow directory without draft files is simply one with nothing
unpublished — which is what every flow written before this existed looks like.
A flow is two names. Its id is which flow it is: the directory, the argument
every method here takes, and never changed once given. Its key, ``name``, is
the namespace of its messages and unique among flows. A flow from before ids
existed has its key as its id (`fluksio.flow.identity`), and so does every new
one for now. Published and draft documents always share both.
"""
from __future__ import annotations
@@ -17,12 +23,15 @@ from __future__ import annotations
import json
import logging
import re
import secrets
import shutil
import string
import subprocess
import threading
from collections.abc import Callable, Iterable
from datetime import UTC, datetime
from pathlib import Path
from typing import Literal
from fluksio.flow.schemas import FlowDef, _validate_name
@@ -66,6 +75,91 @@ def checked(name: str) -> str:
raise InvalidName(name) from None
#: The shape of an id the server issues: a kind, then 40 random bits.
ISSUED_ID = re.compile(r"[fd]_[a-z2-7]{8}")
class ReservedKey(InvalidName):
"""A new key shaped like an issued id, which it could one day equal."""
def __str__(self) -> str:
return (
f"'{self.name}' has the shape of a generated id, f_ or d_ and eight "
"letters or digits, which a new key may not have"
)
class IdentityChange(ValueError):
"""A save that would change which object a document is, or its key.
Ids never change, and a key changes only by renaming, which moves its
messages along with it. A save that did either would bypass that.
"""
def claim(
obj_id: str,
key: str,
current: tuple[str, str] | None,
keys: dict[str, str],
taken: type[ValueError],
) -> None:
"""Refuse a write that changes who an object is, or takes another's key.
*current* is the stored ``(id, name)`` at *obj_id*, None for a new object,
and *keys* the store's key index. Called under the store's write lock, so
two creates of one key cannot both pass. *taken* is what a key already in
use raises: `FlowExists`, or the dashboards' own.
"""
if current is not None:
stored_id, stored_key = current
if stored_id != obj_id:
raise IdentityChange(f"'{obj_id}' holds a document of '{stored_id}'")
if stored_key != key:
raise IdentityChange(
f"'{obj_id}' has the key '{stored_key}', which a save cannot "
f"change to '{key}'"
)
return
# New keys only: one this shape that exists already stays saveable.
if ISSUED_ID.fullmatch(key):
raise ReservedKey(key)
if keys.get(key, obj_id) != obj_id:
raise taken(key)
def read_keys(directories: Iterable[Path], document: str) -> dict[str, str]:
"""The key each directory's object holds, to the directory: key to id.
Read from the working copy, ``<document>.draft.json`` when there is one, so
an object that is only a draft holds its key as much as a published one.
"""
keys: dict[str, str] = {}
for directory in directories:
draft = directory / f"{document}.draft.json"
path = draft if draft.exists() else directory / f"{document}.json"
try:
keys.setdefault(json.loads(path.read_text())["name"], directory.name)
except (OSError, ValueError, KeyError, TypeError):
# Unreadable to the engine too, which reports it when it loads.
continue
return keys
def new_id(kind: Literal["f", "d"], taken: Callable[[str], bool]) -> str:
"""A fresh id for a flow (``f``) or a dashboard (``d``), e.g. ``f_k3q7m2xa``.
Random is not unique, so *taken* decides, under the store's write lock: it
has to know every live id, every directory left behind and every key,
since a key equal to an id makes an argument naming either ambiguous.
"""
alphabet = string.ascii_lowercase + "234567"
while True:
candidate = f"{kind}_" + "".join(secrets.choice(alphabet) for _ in range(8))
if not taken(candidate):
return candidate
class FlowNotFound(KeyError):
def __init__(self, name: str) -> None:
super().__init__(name)
@@ -174,6 +268,8 @@ class FlowStore:
self._read_all: list[FlowDef] = []
self._read_all_at = -1
self._read_all_lock = threading.Lock()
#: `key_index` and the revision it was read at.
self._keys: tuple[int, dict[str, str]] = (-1, {})
if not (self.root / ".git").exists():
self._git("init", "-q")
self._commit("Initialise flow store", allow_empty=True)
@@ -441,7 +537,7 @@ class FlowStore:
):
if not path.exists():
continue
definition = FlowDef.model_validate_json(path.read_text())
definition = self._load(path)
users = [n for n in definition.nodes if n.source_ref == name]
if not users:
continue
@@ -549,16 +645,56 @@ class FlowStore:
+ list(self._draft_nodes_dir(name).glob("*.py"))
)
@staticmethod
def _load(path: Path) -> FlowDef:
"""A stored document, with the id one from before ids is known by."""
flow = FlowDef.model_validate_json(path.read_text())
if not flow.id:
flow.id = path.parent.name
return flow
def read_flow(self, name: str, draft: bool = False) -> FlowDef:
"""The published flow, or with ``draft`` the working copy."""
if draft:
path = self._draft_file(name)
if path.exists():
return FlowDef.model_validate_json(path.read_text())
return self._load(path)
path = self._flow_file(name)
if not path.exists():
raise FlowNotFound(name)
return FlowDef.model_validate_json(path.read_text())
return self._load(path)
def key_index(self) -> dict[str, str]:
"""Every flow's key, to its id, whether published or only a draft.
Read again after any commit, which every write makes — a draft-only
create or a delete as much as a save. Shared: read it, never change it.
"""
at, keys = self._keys
if at == self.revision:
return keys
# Taken before reading, so a commit landing meanwhile reads it again.
at = self.revision
keys = read_keys((self.root / n for n in self.list_flows()), "flow")
self._keys = (at, keys)
return keys
def _claim(self, flow: FlowDef) -> FlowDef:
"""*flow* with its id, once it is certain to be a save the store allows.
Under `_write_lock`. An id left out is the key, which is how every
flow is created for now.
"""
flow_id = checked(flow.id or flow.name)
current = self.read_flow(flow_id, draft=True) if self.exists(flow_id) else None
claim(
flow_id,
flow.name,
(current.id, current.name) if current else None,
self.key_index(),
FlowExists,
)
return flow.model_copy(update={"id": flow_id})
def read_all(self) -> list[FlowDef]:
"""Every published flow — what the engine runs.
@@ -583,15 +719,17 @@ class FlowStore:
def write_flow(self, flow: FlowDef) -> bool:
"""Publish a flow directly. Returns False when nothing actually changed."""
path = self._flow_file(flow.name)
content = flow.model_dump_json(indent=2) + "\n"
if path.exists() and path.read_text() == content:
return False
with self._write_lock:
flow = self._claim(flow)
path = self._flow_file(flow.id)
content = flow.model_dump_json(indent=2) + "\n"
if path.exists() and path.read_text() == content:
return False
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content)
self._commit(f"Update flow '{flow.name}'")
return True
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content)
self._commit(f"Update flow '{flow.name}'")
return True
def write_draft(self, flow: FlowDef, base_version: int) -> FlowDef:
"""Save unpublished changes, refusing to overwrite someone else's.
@@ -601,9 +739,10 @@ class FlowStore:
write would silently discard their work.
"""
with self._write_lock:
flow = self._claim(flow)
current: FlowDef | None = None
if self.exists(flow.name):
current = self.read_flow(flow.name, draft=True)
if self.exists(flow.id):
current = self.read_flow(flow.id, draft=True)
if base_version != current.version:
raise StaleVersion(flow.name, current.version)
if _same_content(current, flow):
@@ -612,7 +751,7 @@ class FlowStore:
stored = flow.model_copy(
update={"version": (current.version if current else 0) + 1}
)
path = self._draft_file(flow.name)
path = self._draft_file(flow.id)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(stored.model_dump_json(indent=2) + "\n")
self._commit(f"Update draft of flow '{flow.name}'")
@@ -656,48 +795,12 @@ class FlowStore:
return self.read_flow(name)
def delete_flow(self, name: str) -> None:
directory = self._flow_dir(name)
if not directory.exists():
raise FlowNotFound(name)
shutil.rmtree(directory)
self._commit(f"Delete flow '{name}'")
def rename_flow(self, name: str, new_name: str) -> FlowDef:
"""Rename a flow, carrying its nodes and any references to it.
A flow's name is the namespace of its messages, so other flows reading
``old.temperature`` are rewritten to read ``new.temperature`` — leaving
them pointing at a flow that no longer exists would break them silently.
"""
if not self.exists(name):
raise FlowNotFound(name)
if self.exists(new_name):
raise FlowExists(new_name)
self._flow_dir(name).rename(self._flow_dir(new_name))
# Published and draft alike carry the name, and both have to move.
for path in (self._flow_file(new_name), self._draft_file(new_name)):
if not path.exists():
continue
flow = FlowDef.model_validate_json(path.read_text())
path.write_text(
flow.model_copy(update={"name": new_name}).model_dump_json(indent=2)
+ "\n"
)
for other in self.list_flows():
if other == new_name:
continue
for path in (self._flow_file(other), self._draft_file(other)):
if not path.exists():
continue
flow = FlowDef.model_validate_json(path.read_text())
if self._retarget(flow, f"{name}.", f"{new_name}."):
path.write_text(flow.model_dump_json(indent=2) + "\n")
self._commit(f"Rename flow '{name}' to '{new_name}'")
return self.read_flow(new_name, draft=True)
with self._write_lock:
directory = self._flow_dir(name)
if not directory.exists():
raise FlowNotFound(name)
shutil.rmtree(directory)
self._commit(f"Delete flow '{name}'")
def duplicate_flow(self, name: str) -> FlowDef:
"""Copy a flow's working copy into a new flow that is only a draft.
@@ -710,19 +813,24 @@ class FlowStore:
with self._write_lock:
source = self.read_flow(name, draft=True)
# The directory rather than `exists`: anything left in it would
# otherwise end up in the copy.
new_name = copy_name(name, lambda n: self._flow_dir(n).exists())
# otherwise end up in the copy. The copy's id is its key, so that
# has to be free among the keys too.
keys = self.key_index()
new_name = copy_name(
source.name, lambda n: self._flow_dir(n).exists() or n in keys
)
copy = source.model_copy(
update={
"id": new_name,
"name": new_name,
"title": f"{source.title} (copy)" if source.title else "",
"version": 1,
"origin": None,
}
)
# Its own messages named with its own name are the copy's now.
self._retarget(copy, f"{name}.", f"{new_name}.")
# Its own messages named with its own key are the copy's now.
self._retarget(copy, f"{source.name}.", f"{new_name}.")
path = self._draft_file(new_name)
path.parent.mkdir(parents=True)
path.write_text(copy.model_dump_json(indent=2) + "\n")
backend/fluksio/main.py
+12 -1
View File
@@ -40,6 +40,7 @@ from fluksio.flow.demand import demand
from fluksio.flow.devices import DeviceStore, DeviceUnavailable
from fluksio.flow.events import event_bus
from fluksio.flow.executor import ExecutionService
from fluksio.flow.identity import assign_ids
from fluksio.flow.metrics import MetricsCollector
from fluksio.flow.nodes.http import close_shared_client
from fluksio.flow.pipeline import ValueSource
@@ -58,7 +59,7 @@ from fluksio.flow.resources import (
from fluksio.flow.runs import RUN_STATE_TTL, RunService, sweep_artifacts
from fluksio.flow.secrets import init_secrets
from fluksio.flow.state import MemoryState, RedisState, StateBackend
from fluksio.flow.store import FlowStore, InvalidName
from fluksio.flow.store import FlowStore, IdentityChange, InvalidName
from fluksio.flow.watchdog import LoopWatchdog
from fluksio.flow.workers import PythonWorkerPool
@@ -301,6 +302,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
events=event_bus,
max_cascades=settings.FLOW_MAX_CASCADES,
)
# Every flow and dashboard has an id before anything reads one: the
# store's own `_migrate`, the device migrations below and every cache
# after them write documents back with whatever id they read.
await run_in_threadpool(assign_ids, settings.FLOWS_DIR)
store = FlowStore(settings.FLOWS_DIR)
# The packages node code imports. In the background, because a first
# install of anything substantial takes minutes and `uv` is given five
@@ -577,6 +582,12 @@ async def invalid_name(request: Request, exc: InvalidName) -> JSONResponse: # n
return JSONResponse(status_code=400, content={"detail": str(exc)})
@app.exception_handler(IdentityChange)
async def identity_change(request: Request, exc: IdentityChange) -> JSONResponse: # noqa: ARG001
"""A save that would change an object's id or key, which no save may."""
return JSONResponse(status_code=409, content={"detail": str(exc)})
@app.exception_handler(DeviceUnavailable)
async def device_unavailable(request: Request, exc: DeviceUnavailable) -> JSONResponse: # noqa: ARG001
"""A device credential that cannot be checked now is not a refused one."""
backend/tests/api/routes/test_dashboards.py
+26
View File
@@ -86,3 +86,29 @@ def test_a_save_based_on_a_version_someone_moved_past_is_refused(
assert stale.status_code == 409
assert stale.json()["detail"]["current_version"] == 1
def test_a_dashboard_is_known_by_an_id_no_save_changes(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
saved = client.put(
f"{PREFIX}/put_identified",
headers=superuser_token_headers,
json=a_dashboard("put_identified"),
).json()
assert saved["id"] == "put_identified"
elsewhere = client.put(
f"{PREFIX}/put_identified",
headers=superuser_token_headers,
json={**saved, "id": "elsewhere"},
)
assert elsewhere.status_code == 422
renamed = client.post(
f"{PREFIX}/put_identified/rename",
headers=superuser_token_headers,
json={"name": "renamed"},
)
assert renamed.status_code in (404, 405)
client.delete(f"{PREFIX}/put_identified", headers=superuser_token_headers)
backend/tests/api/routes/test_flows.py
+39 -98
View File
@@ -77,6 +77,45 @@ def test_name_mismatch_is_rejected(
assert response.status_code == 400
def test_a_flow_is_known_by_an_id_no_save_changes(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
"""For now the id of a new flow is its key, and the URL names both."""
saved = client.put(
f"{PREFIX}/identified",
headers=superuser_token_headers,
json=a_flow("identified"),
).json()["definition"]
assert saved["id"] == "identified"
listed = client.get(f"{PREFIX}/", headers=superuser_token_headers).json()["data"]
assert {"id": "identified", "name": "identified"}.items() <= next(
f for f in listed if f["name"] == "identified"
).items()
elsewhere = client.put(
f"{PREFIX}/identified",
headers=superuser_token_headers,
json={**saved, "id": "elsewhere"},
)
assert elsewhere.status_code == 400
# A key changes by a rename, and that route is not there yet.
renamed = client.post(
f"{PREFIX}/identified/rename",
headers=superuser_token_headers,
json={"new_name": "renamed"},
)
assert renamed.status_code in (404, 405)
# Shaped like an id the server issues, which a new key may not be.
issued = client.put(
f"{PREFIX}/f_k3q7m2xa",
headers=superuser_token_headers,
json=a_flow("f_k3q7m2xa"),
)
assert issued.status_code == 400
client.delete(f"{PREFIX}/identified", headers=superuser_token_headers)
def test_broken_node_is_reported_and_siblings_stay_active(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
@@ -347,68 +386,6 @@ def test_node_types_are_listed(
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:
@@ -467,42 +444,6 @@ def test_acknowledging_a_failure_is_announced_on_the_bus(
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
backend/tests/api/routes/test_path_names.py
-1
View File
@@ -58,7 +58,6 @@ def test_the_stores_refuse_them_themselves(tmp_path: Path) -> None:
for call in (
lambda: flows.delete_flow(".."),
lambda: flows.read_flow("_lib"),
lambda: flows.rename_flow("..", "elsewhere"),
lambda: flows.write_node_source("demo", "../x", "code"),
lambda: flows.write_lib_source("..", "code"),
lambda: flows.exists("demo\n"),
backend/tests/flow/test_dashboards.py
+5 -15
View File
@@ -110,17 +110,11 @@ def test_a_new_dashboard_is_a_draft_until_it_is_published(store: DashboardStore)
assert store.read("house").title == created.title
def test_an_unpublished_dashboard_can_be_renamed_and_deleted(store: DashboardStore):
"""And renaming it does not put it on a wall either."""
def test_an_unpublished_dashboard_can_be_deleted(store: DashboardStore):
store.write_draft(default_dashboard("house"), 0)
store.rename("house", "home")
store.delete("house")
assert not store.exists("house")
assert store.has_draft("home") and not store.is_published("home")
store.delete("home")
assert not store.exists("home")
def test_a_duplicate_is_an_unpublished_copy_of_the_working_copy(
@@ -139,16 +133,12 @@ def test_a_duplicate_is_an_unpublished_copy_of_the_working_copy(
assert store.duplicate("house").name == "house_copy_2"
def test_deleting_and_renaming(store: DashboardStore):
def test_deleting(store: DashboardStore):
store.write(default_dashboard("house"))
renamed = store.rename("house", "home")
assert renamed.name == "home"
assert not store.exists("house")
store.delete("home")
store.delete("house")
with pytest.raises(DashboardNotFound):
store.read("home")
store.read("house")
def test_the_deepest_chart_decides_how_much_past_is_kept(store: DashboardStore):
backend/tests/flow/test_drafts.py
-19
View File
@@ -164,25 +164,6 @@ def test_resaving_the_published_source_creates_no_draft(store: FlowStore):
assert not store.has_draft("heating")
def test_renaming_carries_the_draft_and_repoints_other_drafts(store: FlowStore):
store.write_draft(a_flow(), 0)
store.write_draft(
FlowDef(
name="display",
nodes=[NodeDef(id="gauge", requires=[MessageSpec(name="heating.temp")])],
),
0,
)
renamed = store.rename_flow("heating", "warmth")
assert renamed.name == "warmth"
assert store.has_draft("warmth")
assert store.read_flow("display", draft=True).nodes[0].requires[0].name == (
"warmth.temp"
)
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.nodes.append(NodeDef(id="reader", requires=[MessageSpec(name="heating.temp")]))
backend/tests/flow/test_identity.py
+517
View File
@@ -0,0 +1,517 @@
"""Ids: who a flow or dashboard is, apart from what it is called.
The store keeps them, and the startup migration gives every document from
before them its key as its id. A mistake in either loses data silently, which
is why the migration is driven through crashes, refusals and stores without git.
"""
import json
import logging
import subprocess
from pathlib import Path
import pytest
from fluksio import cli
from fluksio.flow import identity
from fluksio.flow.dashboards import (
DASHBOARD_DIR,
DashboardDef,
DashboardExists,
DashboardStore,
default_dashboard,
)
from fluksio.flow.identity import COMMIT_MESSAGE, IdsRefused, assign_ids
from fluksio.flow.schemas import FlowDef
from fluksio.flow.store import (
ISSUED_ID,
FlowExists,
FlowStore,
IdentityChange,
ReservedKey,
new_id,
)
# -----------------------------------------------------------------------------
# The store
# -----------------------------------------------------------------------------
@pytest.fixture
def store(tmp_path: Path) -> FlowStore:
return FlowStore(tmp_path / "flows")
def test_a_flow_is_given_its_key_as_id_and_keeps_both(store: FlowStore):
created = store.write_draft(FlowDef(name="heating"), 0)
assert (created.id, created.name) == ("heating", "heating")
published = store.publish_flow("heating", created.version)
store.write_draft(published.model_copy(update={"title": "Warm"}), 1)
discarded = store.discard_draft("heating")
assert json.loads((store.root / "heating" / "flow.json").read_text())["id"] == (
"heating"
)
assert (discarded.id, discarded.name) == ("heating", "heating")
def test_a_document_from_before_ids_reads_with_its_directory_as_id(store: FlowStore):
(store.root / "heating").mkdir()
(store.root / "heating" / "flow.json").write_text('{"name": "heating"}')
(store.root / DASHBOARD_DIR / "home").mkdir(parents=True)
(store.root / DASHBOARD_DIR / "home" / "dashboard.json").write_text(
'{"name": "home"}'
)
assert store.read_flow("heating").id == "heating"
assert DashboardStore(store).read("home").id == "home"
# And a save of it, without an id, is the same flow and not a new one.
assert store.write_draft(FlowDef(name="heating", title="Warm"), 1).version == 2
def test_a_save_cannot_change_an_id_or_a_key(store: FlowStore):
store.write_flow(FlowDef(id="x_heating", name="heating"))
with pytest.raises(IdentityChange):
store.write_draft(FlowDef(id="x_heating", name="warmth"), 1)
with pytest.raises(IdentityChange):
store.write_flow(FlowDef(id="x_heating", name="warmth"))
dashboards = DashboardStore(store)
dashboards.write(DashboardDef(id="x_home", name="home"))
with pytest.raises(IdentityChange):
dashboards.write_draft(DashboardDef(id="x_home", name="house"), 1)
assert dashboards.read("x_home").name == "home"
def test_a_key_belongs_to_one_flow_drafts_included(store: FlowStore):
"""A draft nobody published holds its key, and a delete lets it go."""
store.write_draft(FlowDef(id="x_heating", name="heating"), 0)
assert store.key_index() == {"heating": "x_heating"}
with pytest.raises(FlowExists):
store.write_draft(FlowDef(name="heating"), 0)
with pytest.raises(FlowExists):
store.write_flow(FlowDef(id="y_heating", name="heating"))
store.delete_flow("x_heating")
assert store.write_draft(FlowDef(name="heating"), 0).id == "heating"
assert store.key_index() == {"heating": "heating"}
def test_a_key_belongs_to_one_dashboard_drafts_included(store: FlowStore):
dashboards = DashboardStore(store)
dashboards.write_draft(DashboardDef(id="x_home", name="home"), 0)
# Its own id space: a flow may share the key.
store.write_draft(FlowDef(name="home"), 0)
with pytest.raises(DashboardExists):
dashboards.write_draft(default_dashboard("home"), 0)
with pytest.raises(DashboardExists):
dashboards.write(DashboardDef(id="y_home", name="home"))
dashboards.delete("x_home")
assert dashboards.write(default_dashboard("home")).id == "home"
def test_a_copy_takes_a_key_nobody_holds(store: FlowStore):
store.write_flow(FlowDef(name="heating"))
store.write_draft(FlowDef(id="x_other", name="heating_copy"), 0)
dashboards = DashboardStore(store)
dashboards.write(default_dashboard("home"))
dashboards.write_draft(DashboardDef(id="x_other", name="home_copy"), 0)
copy = store.duplicate_flow("heating")
board = dashboards.duplicate("home")
assert (copy.id, copy.name) == ("heating_copy_2", "heating_copy_2")
assert (board.id, board.name) == ("home_copy_2", "home_copy_2")
def test_a_new_key_shaped_like_an_issued_id_is_refused(store: FlowStore):
with pytest.raises(ReservedKey):
store.write_draft(FlowDef(name="f_k3q7m2xa"), 0)
with pytest.raises(ReservedKey):
DashboardStore(store).write_draft(default_dashboard("d_k3q7m2xa"), 0)
assert store.write_draft(FlowDef(name="f_k3q7m2xa_lab"), 0).name == (
"f_k3q7m2xa_lab"
)
def test_an_existing_key_shaped_like_an_issued_id_stays_saveable(store: FlowStore):
(store.root / "f_k3q7m2xa").mkdir()
(store.root / "f_k3q7m2xa" / "flow.json").write_text('{"name": "f_k3q7m2xa"}')
saved = store.write_draft(FlowDef(name="f_k3q7m2xa", title="Old"), 1)
assert (saved.id, saved.title) == ("f_k3q7m2xa", "Old")
def test_a_new_id_is_random_and_avoids_whatever_is_taken():
asked: list[str] = []
def taken(candidate: str) -> bool:
asked.append(candidate)
return len(asked) < 4
issued = new_id("d", taken)
assert issued == asked[-1] and len(asked) == 4
assert all(ISSUED_ID.fullmatch(c) and c.startswith("d_") for c in asked)
fresh = {new_id("f", lambda _: False) for _ in range(50)}
assert len(fresh) == 50
assert all(ISSUED_ID.fullmatch(c) and c.startswith("f_") for c in fresh)
# -----------------------------------------------------------------------------
# The migration
# -----------------------------------------------------------------------------
def put(root: Path, rel: str, data: dict | str) -> None:
path = root / rel
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(data if isinstance(data, str) else json.dumps(data))
def git(root: Path, *args: str) -> str:
return subprocess.run(
["git", "-C", str(root), *args], capture_output=True, text=True, check=True
).stdout
#: Every document a store from before ids holds, by path. The id each gets.
GOLDEN = {
# Published and a draft, and a field this version does not model.
"heating/flow.json": {
"name": "heating",
"version": 3,
"later": {"kept": True},
"nodes": [{"id": "sensor", "provides": [{"name": "temp"}]}],
},
"heating/flow.draft.json": {
"name": "heating",
"version": 5,
"nodes": [{"id": "reader", "requires": [{"name": "solar.watts"}]}],
},
# Only a draft.
"lab/flow.draft.json": {"name": "lab", "version": 1},
# Stopped, which is a file beside it that has no id to get.
"stopped/flow.json": {"name": "stopped"},
# A dashboard and a flow share a key: separate id spaces.
"home/flow.json": {"name": "home", "version": 2},
f"{DASHBOARD_DIR}/home/dashboard.json": {"name": "home", "version": 4},
f"{DASHBOARD_DIR}/home/dashboard.draft.json": {
"name": "home",
"version": 4,
"widgets": [],
},
# Pages and sections, as a dashboard was written before it was one grid.
f"{DASHBOARD_DIR}/legacy/dashboard.json": {
"name": "legacy",
"version": 2,
"pages": [
{
"sections": [
{"widgets": [{"id": "t", "type": "stat", "config": {}}]},
]
}
],
},
f"{DASHBOARD_DIR}/draft_only/dashboard.draft.json": {"name": "draft_only"},
}
#: What is not a flow or a dashboard, and so is never touched.
BYSTANDERS = {
"stopped/runtime.json": '{"enabled": false}',
# A device called "flow", which `*/flow.json` would otherwise find.
"_devices/flow.json": '{"id": "flow", "title": "Hall"}',
"_lib/shared.py": "def process():\n return {}\n",
"heating/nodes/sensor.py": "def process():\n return {'temp': 1}\n",
}
#: The versions the golden store moves to: past published and draft alike.
VERSIONS = {
"heating/flow.json": 6,
"heating/flow.draft.json": 6,
"lab/flow.draft.json": 2,
"stopped/flow.json": 2,
"home/flow.json": 3,
f"{DASHBOARD_DIR}/home/dashboard.json": 5,
f"{DASHBOARD_DIR}/home/dashboard.draft.json": 5,
f"{DASHBOARD_DIR}/legacy/dashboard.json": 3,
f"{DASHBOARD_DIR}/draft_only/dashboard.draft.json": 2,
}
def golden(root: Path, extra: dict | None = None) -> Path:
"""A store in the shape it had before ids, committed."""
FlowStore(root)
for rel, data in {**GOLDEN, **BYSTANDERS, **(extra or {})}.items():
put(root, rel, data)
git(root, "add", "-A")
git(root, "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-qm", "Old")
return root
def snapshot(root: Path) -> dict[str, bytes]:
"""Every file, git's own included, so a write anywhere shows."""
return {
str(p.relative_to(root)): p.read_bytes()
for p in sorted(root.rglob("*"))
if p.is_file()
}
def documents(root: Path) -> dict[str, dict]:
return {rel: json.loads((root / rel).read_text()) for rel in GOLDEN}
def assert_migrated(root: Path) -> None:
for rel, data in documents(root).items():
before = GOLDEN[rel]
assert data["id"] == Path(rel).parent.name, rel
assert data["version"] == VERSIONS[rel], rel
# Nothing else about it changed, the fields nobody models included.
assert {k: v for k, v in data.items() if k not in ("id", "version")} == {
k: v for k, v in before.items() if k != "version"
}, rel
for rel, text in BYSTANDERS.items():
assert (root / rel).read_text() == text, rel
assert not (root / ".git" / ".assign-ids.json").exists()
def test_a_store_from_before_ids_gets_each_key_as_id(tmp_path: Path, caplog):
root = golden(tmp_path / "flows")
commits = git(root, "rev-list", "--count", "HEAD")
with caplog.at_level(logging.WARNING, logger=identity.__name__):
changed = assign_ids(root)
assert changed == sorted(GOLDEN)
assert_migrated(root)
assert "heating/flow.json" in caplog.text
# One commit of those documents and nothing else.
assert git(root, "log", "-1", "--format=%s").strip() == COMMIT_MESSAGE
assert sorted(git(root, "show", "--name-only", "--format=", "HEAD").split()) == (
sorted(GOLDEN)
)
assert int(git(root, "rev-list", "--count", "HEAD")) == int(commits) + 1
# What the stores read is what they read before, but for the version.
flows, dashboards = FlowStore(root), DashboardStore(FlowStore(root))
assert flows.read_flow("heating", draft=True).nodes[0].id == "reader"
assert flows.key_index() == {n: n for n in ("heating", "home", "lab", "stopped")}
assert [w.id for w in dashboards.read("legacy").widgets] == ["t"]
assert dashboards.key_index() == {n: n for n in ("draft_only", "home", "legacy")}
# Once is all: a second start changes nothing and says nothing.
before = snapshot(root)
caplog.clear()
with caplog.at_level(logging.WARNING, logger=identity.__name__):
assert assign_ids(root) == []
assert snapshot(root) == before
assert caplog.text == ""
@pytest.mark.parametrize("survives", range(len(GOLDEN) + 1))
def test_a_run_interrupted_part_way_is_finished_by_the_next(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, survives: int
):
"""The note of the versions is the first write, so it is always there."""
root = golden(tmp_path / "flows")
real = identity._replace
writes: list[Path] = []
def crash(path: Path, text: str) -> None:
if len(writes) == survives:
# Half of the temporary file, and never the replace.
path.with_name(f".{path.name}.tmp").write_text(text[:20])
raise OSError("power cut")
writes.append(path)
real(path, text)
monkeypatch.setattr(identity, "_replace", crash)
with pytest.raises(OSError):
assign_ids(root)
monkeypatch.setattr(identity, "_replace", real)
assert assign_ids(root) == sorted(GOLDEN)
assert_migrated(root)
assert git(root, "log", "-1", "--format=%s").strip() == COMMIT_MESSAGE
assert git(root, "status", "--porcelain") == ""
assert not list(root.rglob("*.tmp"))
def test_a_run_interrupted_before_its_commit_is_committed_by_the_next(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
):
"""Every document has its id by then, and the note is what is left."""
root = golden(tmp_path / "flows")
def crash(*_: object) -> bool:
raise OSError("power cut")
monkeypatch.setattr(identity, "_commit", crash)
with pytest.raises(OSError):
assign_ids(root)
monkeypatch.undo()
assert git(root, "log", "-1", "--format=%s").strip() == "Old"
assert assign_ids(root) == sorted(GOLDEN)
assert_migrated(root)
assert git(root, "log", "-1", "--format=%s").strip() == COMMIT_MESSAGE
assert assign_ids(root) == []
def test_documents_that_have_their_id_keep_their_version(tmp_path: Path):
"""A pair half given its id moves the other half past both."""
root = golden(
tmp_path / "flows",
{
"heating/flow.json": {"id": "heating", "name": "heating", "version": 3},
# An older engine saved this after the published one got its id.
"heating/flow.draft.json": {"name": "heating", "version": 7},
"lab/flow.draft.json": {"id": "lab", "name": "lab", "version": 1},
# Moved apart by a rename, with its id: nothing to do with it.
"warmth/flow.json": {"id": "warmth", "name": "radiators"},
},
)
changed = assign_ids(root)
assert "heating/flow.json" not in changed and "lab/flow.draft.json" not in changed
assert "warmth/flow.json" not in changed
assert json.loads((root / "heating/flow.json").read_text())["version"] == 3
assert json.loads((root / "heating/flow.draft.json").read_text())["version"] == 8
assert json.loads((root / "lab/flow.draft.json").read_text())["version"] == 1
@pytest.mark.parametrize(
("extra", "named"),
[
# Which one the flow is cannot be told: the engine runs it as `warmth`.
({"heating/flow.json": {"name": "warmth"}}, "heating/flow.json"),
({"heating/flow.json": {"id": "warmth", "name": "heating"}}, "'warmth'"),
(
{
"pair/flow.json": {"id": "pair", "name": "one"},
"pair/flow.draft.json": {"id": "pair", "name": "two"},
},
"'one' and 'two'",
),
(
{
"a/flow.json": {"id": "a", "name": "same"},
"c/flow.draft.json": {"id": "c", "name": "same"},
},
"'a' and 'c' share the key 'same'",
),
({"lab/flow.draft.json": "{not json"}, "lab/flow.draft.json"),
({f"{DASHBOARD_DIR}/home/dashboard.json": {"name": "hall"}}, "'hall'"),
],
)
def test_a_store_in_doubt_is_refused_and_left_as_it_was(
tmp_path: Path, extra: dict, named: str
):
root = golden(tmp_path / "flows", extra)
before = snapshot(root)
with pytest.raises(IdsRefused) as refused:
assign_ids(root)
assert named in str(refused.value)
assert f"fluksio store migrate --dry-run {root}" in str(refused.value)
assert snapshot(root) == before
def test_uncommitted_changes_beside_it_stay_uncommitted(tmp_path: Path):
root = golden(tmp_path / "flows")
(root / "_lib" / "shared.py").write_text("edited behind the API\n")
(root / "notes.txt").write_text("someone's notes\n")
git(root, "add", "notes.txt")
assign_ids(root)
assert sorted(git(root, "show", "--name-only", "--format=", "HEAD").split()) == (
sorted(GOLDEN)
)
assert git(root, "status", "--porcelain").splitlines() == [
" M _lib/shared.py",
"A notes.txt",
]
def test_a_store_without_a_repository(tmp_path: Path):
root = tmp_path / "flows"
for rel, data in {**GOLDEN, **BYSTANDERS}.items():
put(root, rel, data)
assert assign_ids(root) == sorted(GOLDEN)
assert_migrated(root)
assert not (root / ".assign-ids.json").exists()
assert not (root / ".git").exists()
def test_a_machine_without_git(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
root = golden(tmp_path / "flows")
real = subprocess.run
def no_git(cmd, *args, **kwargs):
if cmd and cmd[0] == "git":
raise FileNotFoundError(2, "No such file or directory", "git")
return real(cmd, *args, **kwargs)
monkeypatch.setattr(identity.subprocess, "run", no_git)
assert assign_ids(root) == sorted(GOLDEN)
assert_migrated(root)
# -----------------------------------------------------------------------------
# `fluksio store`
# -----------------------------------------------------------------------------
def test_the_dry_run_changes_no_bytes(tmp_path: Path, capsys):
root = golden(tmp_path / "flows")
before = snapshot(root)
assert cli.main(["store", "migrate", "--dry-run", str(root)]) == 0
out = capsys.readouterr().out
assert "4 flows, 3 dashboards" in out
assert "heating/flow.draft.json (version 6)" in out
assert snapshot(root) == before
def test_the_dry_run_creates_no_repository_and_names_what_it_refuses(
tmp_path: Path, capsys
):
root = tmp_path / "flows"
for rel, data in {**GOLDEN, "odd/flow.json": {"name": "even"}}.items():
put(root, rel, data)
before = snapshot(root)
assert cli.main(["store", "migrate", "--dry-run", str(root)]) == 1
assert "odd/flow.json: its name 'even' is not its directory 'odd'" in (
capsys.readouterr().out
)
assert snapshot(root) == before
assert not (root / ".git").exists()
def test_ls_lists_ids_keys_and_kinds(tmp_path: Path, capsys):
root = golden(
tmp_path / "flows", {"warmth/flow.json": {"id": "warmth", "name": "radiators"}}
)
assert cli.main(["store", "ls", str(root)]) == 0
rows = [line.split() for line in capsys.readouterr().out.splitlines()]
assert rows[0] == ["ID", "KEY", "KIND"]
assert ["home", "home", "dashboard"] in rows and ["home", "home", "flow"] in rows
assert ["warmth", "radiators", "flow"] in rows
backend/tests/flow/test_store.py
+1 -43
View File
@@ -8,7 +8,7 @@ import pytest
from fluksio.flow.messages import MessageSpec
from fluksio.flow.schemas import FlowDef, NodeDef
from fluksio.flow.store import FlowExists, FlowNotFound, FlowStore
from fluksio.flow.store import FlowNotFound, FlowStore
@pytest.fixture
@@ -92,48 +92,6 @@ def test_deleting_removes_flow_and_its_nodes(store: FlowStore):
assert not (store.root / "heating").exists()
def test_renaming_a_flow_carries_its_nodes(store: FlowStore):
store.write_flow(a_flow())
store.write_node_source("heating", "sensor", "def process():\n return {}\n")
renamed = store.rename_flow("heating", "warmth")
assert renamed.name == "warmth"
assert store.list_flows() == ["warmth"]
assert "def process" in store.read_node_source("warmth", "sensor")
def test_renaming_a_flow_repoints_the_flows_reading_it(store: FlowStore):
store.write_flow(a_flow())
store.write_flow(
FlowDef(
name="display",
nodes=[
NodeDef(
id="gauge",
# Reads across the flow boundary, so the name must follow.
requires=[MessageSpec(name="heating.temp")],
)
],
)
)
store.rename_flow("heating", "warmth")
display = store.read_flow("display")
assert display.nodes[0].requires[0].name == "warmth.temp"
def test_renaming_onto_an_existing_name_is_refused(store: FlowStore):
store.write_flow(a_flow())
store.write_flow(FlowDef(name="warmth"))
with pytest.raises(FlowExists):
store.rename_flow("heating", "warmth")
assert store.list_flows() == ["heating", "warmth"]
def test_opening_the_store_renames_trigger_nodes_to_debounce(store: FlowStore):
"""The old name has no alias, so what is stored under it is rewritten once."""
store.write_flow(