70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
"""A deleted flow's last key, kept for as long as history names its id."""
|
|
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from sqlmodel import Session, delete, select
|
|
|
|
from fluksio.flow import tombstones
|
|
from fluksio.flow.schemas import FlowDef
|
|
from fluksio.flow.store import FlowStore
|
|
from fluksio.flow.tombstones import Tombstones, names_of
|
|
from fluksio.models import EngineEvent, FlowTombstone, MetricBucket, Run
|
|
|
|
|
|
def test_a_tombstone_stays_while_history_names_its_id(db: Session) -> None:
|
|
graves = Tombstones()
|
|
for flow_id in ("f_gonegone", "f_trainee", "f_measured", "f_forgotten"):
|
|
graves.bury(flow_id, flow_id.removeprefix("f_"))
|
|
now = datetime.now(UTC)
|
|
# A training run is filed under the training flow, and names the node it
|
|
# trains in another, by that flow's id.
|
|
db.add(
|
|
Run(
|
|
id="tomb-run",
|
|
flow_id="wren_train",
|
|
for_node="f_trainee.lights",
|
|
created_at=now,
|
|
)
|
|
)
|
|
db.add(Run(id="tomb-own", flow_id="f_gonegone", created_at=now))
|
|
db.add(MetricBucket(flow_id="other", node="f_measured.sensor", bucket=now))
|
|
db.add(EngineEvent(ts=now, type="audit", flow_id="f_forgotten_too"))
|
|
db.commit()
|
|
|
|
assert graves.recorded("f_trainee") == "trainee"
|
|
assert graves.recorded("wren_train") == "wren_train" # history, no tombstone
|
|
assert graves.recorded("f_nothing") is None
|
|
|
|
# Committed before it is checked, so a wrong count does not leave this
|
|
# session holding the write lock for every test after it.
|
|
pruned = tombstones.prune(db)
|
|
db.commit()
|
|
assert pruned == 1
|
|
kept = {stone.id for stone in db.exec(select(FlowTombstone))}
|
|
assert kept == {"f_gonegone", "f_trainee", "f_measured"}
|
|
|
|
db.exec(delete(Run).where(Run.id.in_(["tomb-run", "tomb-own"]))) # type: ignore[attr-defined,call-overload]
|
|
db.exec(delete(MetricBucket)) # type: ignore[call-overload]
|
|
db.exec(delete(EngineEvent)) # type: ignore[call-overload]
|
|
db.commit()
|
|
pruned = tombstones.prune(db)
|
|
db.commit()
|
|
assert pruned == 3
|
|
|
|
|
|
def test_a_name_comes_from_the_live_flow_then_its_tombstone_then_the_id(
|
|
db: Session, tmp_path: Path
|
|
) -> None:
|
|
store = FlowStore(tmp_path / "flows")
|
|
store.write_flow(FlowDef(id="f_livelive", name="heating"))
|
|
Tombstones().bury("f_deadbeef", "cooling")
|
|
|
|
assert names_of(["f_livelive", "f_deadbeef", "old_key"], store) == {
|
|
"f_livelive": "heating",
|
|
"f_deadbeef": "cooling",
|
|
"old_key": "old_key",
|
|
}
|
|
Tombstones().lift("f_deadbeef")
|
|
assert db.get(FlowTombstone, "f_deadbeef") is None
|