app/backend/tests/flow/test_identity.py

"""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 concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import pytest

from fluksio import cli
from fluksio.flow import identity
from fluksio.flow.dashboards import (
    DASHBOARD_DIR,
    DashboardDef,
    DashboardExists,
    DashboardNotFound,
    DashboardStore,
    default_dashboard,
    results_dashboard,
)
from fluksio.flow.identity import COMMIT_MESSAGE, IdsRefused, assign_ids
from fluksio.flow.schemas import FlowCreate, FlowDef
from fluksio.flow.store import (
    ISSUED_ID,
    FlowExists,
    FlowNotFound,
    FlowStore,
    IdentityChange,
    IdTaken,
    RecordedKey,
    ReservedKey,
    StaleVersion,
    new_id,
)

# -----------------------------------------------------------------------------
# The store
# -----------------------------------------------------------------------------


class Graves:
    """Tombstones and history, as `fluksio.flow.tombstones.Tombstones` keeps
    them in the database, which store tests have none of."""

    def __init__(self) -> None:
        self.stones: dict[str, str] = {}
        #: Ids that history rows name.
        self.history: set[str] = set()

    def bury(self, flow_id: str, name: str) -> None:
        self.stones[flow_id] = name

    def lift(self, flow_id: str) -> None:
        self.stones.pop(flow_id, None)

    def recorded(self, flow_id: str) -> str | None:
        if flow_id in self.stones:
            return self.stones[flow_id]
        return flow_id if flow_id in self.history else None


@pytest.fixture
def graves() -> Graves:
    return Graves()


@pytest.fixture
def store(tmp_path: Path, graves: Graves) -> FlowStore:
    return FlowStore(tmp_path / "flows", tombstones=graves)  # type: ignore[arg-type]


def test_a_new_flow_gets_an_issued_id_and_keeps_it(store: FlowStore):
    created = store.create(FlowDef(name="heating"))
    flow_id = created.id
    assert ISSUED_ID.fullmatch(flow_id) and flow_id.startswith("f_")
    assert created.name == "heating" and created.version == 1
    assert store.key_index() == {"heating": flow_id}

    published = store.publish_flow(flow_id, created.version)
    store.write_draft(published.model_copy(update={"title": "Warm"}), 1)
    discarded = store.discard_draft(flow_id)

    on_disk = json.loads((store.root / flow_id / "flow.json").read_text())
    assert (on_disk["id"], on_disk["name"]) == (flow_id, "heating")
    assert (discarded.id, discarded.name) == (flow_id, "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_neither_creates_nor_changes_an_id_or_a_key(store: FlowStore):
    with pytest.raises(FlowNotFound):
        store.write_draft(FlowDef(name="nobody"), 0)
    with pytest.raises(DashboardNotFound):
        DashboardStore(store).write_draft(default_dashboard("nobody"), 0)

    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."""
    first = store.create(FlowDef(name="heating"))
    assert store.key_index() == {"heating": first.id}

    with pytest.raises(FlowExists):
        store.create(FlowDef(name="heating"))
    with pytest.raises(FlowExists):
        store.create(FlowDef(name="heating"), "x_heating")
    with pytest.raises(FlowExists):
        store.write_flow(FlowDef(id="y_heating", name="heating"))

    store.delete_flow(first.id)
    second = store.create(FlowDef(name="heating"))
    assert second.id != first.id
    assert store.key_index() == {"heating": second.id}


def test_a_key_belongs_to_one_dashboard_drafts_included(store: FlowStore):
    dashboards = DashboardStore(store)
    first = dashboards.create(default_dashboard("home"))
    assert ISSUED_ID.fullmatch(first.id) and first.id.startswith("d_")
    # Its own id space: a flow may share the key.
    store.create(FlowDef(name="home"))

    with pytest.raises(DashboardExists):
        dashboards.create(default_dashboard("home"))
    with pytest.raises(DashboardExists):
        dashboards.write(DashboardDef(id="y_home", name="home"))

    dashboards.delete(first.id)
    assert dashboards.create(default_dashboard("home"), "home").id == "home"


def test_a_pin_is_an_id_nothing_holds(store: FlowStore):
    store.create(FlowDef(name="heating"), "x_heating")
    with pytest.raises(IdTaken):
        store.create(FlowDef(name="cooling"), "x_heating")
    # Another flow's key would make naming either ambiguous.
    with pytest.raises(IdTaken):
        store.create(FlowDef(name="cooling"), "heating")
    # And a directory left behind would hand the new flow what is in it.
    (store.root / "leftover" / "nodes").mkdir(parents=True)
    with pytest.raises(IdTaken):
        store.create(FlowDef(name="leftover"), "leftover")
    dashboards = DashboardStore(store)
    dashboards.create(default_dashboard("home"), "x_home")
    with pytest.raises(IdTaken):
        dashboards.create(default_dashboard("house"), "x_home")


def test_an_issued_id_avoids_whatever_history_or_a_directory_holds(
    store: FlowStore, graves: Graves, monkeypatch: pytest.MonkeyPatch
):
    candidates = iter(["f_aaaaaaaa", "f_bbbbbbbb", "f_cccccccc", "f_dddddddd"])
    monkeypatch.setattr(
        "fluksio.flow.store.new_id",
        lambda kind, taken: next(c for c in candidates if not taken(c)),
    )
    graves.bury("f_aaaaaaaa", "gone")
    graves.history.add("f_bbbbbbbb")
    (store.root / "f_cccccccc").mkdir()

    assert store.create(FlowDef(name="heating")).id == "f_dddddddd"


def test_a_deleted_flow_is_buried_before_its_files_go(
    store: FlowStore, graves: Graves, monkeypatch: pytest.MonkeyPatch
):
    flow_id = store.create(FlowDef(name="heating")).id

    def crash(path: Path) -> None:
        raise OSError("disk gone")

    monkeypatch.setattr("fluksio.flow.store.shutil.rmtree", crash)
    with pytest.raises(OSError):
        store.delete_flow(flow_id)
    # The name is safe, and the live document still wins over it.
    assert graves.stones == {flow_id: "heating"}
    assert store.read_flow(flow_id, draft=True).name == "heating"

    monkeypatch.undo()
    assert store.delete_flow(flow_id) == "heating"
    assert graves.stones == {flow_id: "heating"}
    assert not store.exists(flow_id)


def test_a_delete_that_stops_after_the_files_has_kept_the_name(
    store: FlowStore, graves: Graves, monkeypatch: pytest.MonkeyPatch
):
    flow_id = store.create(FlowDef(name="heating")).id

    def killed(message: str) -> None:
        raise OSError("killed before the commit")

    monkeypatch.setattr(store, "_commit", killed)
    with pytest.raises(OSError):
        store.delete_flow(flow_id)
    monkeypatch.undo()

    assert not store.exists(flow_id)
    assert graves.stones == {flow_id: "heating"}
    with pytest.raises(FlowNotFound):
        store.delete_flow(flow_id)


def test_a_pinned_restore_takes_up_its_history_under_its_own_key(
    store: FlowStore, graves: Graves
):
    flow_id = store.create(FlowDef(name="heating")).id
    store.delete_flow(flow_id)

    with pytest.raises(RecordedKey):
        store.create(FlowDef(name="warmth"), flow_id)
    assert graves.stones == {flow_id: "heating"}

    restored = store.create(FlowDef(name="heating"), flow_id)
    assert restored.id == flow_id and graves.stones == {}

    # History from before tombstones is filed under an id that was its key.
    graves.history.add("legacy")
    with pytest.raises(RecordedKey):
        store.create(FlowDef(name="other"), "legacy")
    assert store.create(FlowDef(name="legacy"), "legacy").id == "legacy"


def test_a_key_reused_after_a_delete_starts_a_new_history(
    store: FlowStore, graves: Graves
):
    old = store.create(FlowDef(name="heating")).id
    store.delete_flow(old)

    new = store.create(FlowDef(name="heating")).id

    assert new != old
    assert graves.stones == {old: "heating"}


def test_two_creates_of_one_key_make_one_flow(store: FlowStore):
    def create(_: int) -> str:
        try:
            return store.create(FlowDef(name="heating")).id
        except FlowExists:
            return ""

    with ThreadPoolExecutor(max_workers=8) as pool:
        made = [flow_id for flow_id in pool.map(create, range(8)) if flow_id]

    assert len(made) == 1
    assert store.list_flows() == made


def test_two_results_dashboards_of_one_flow_make_one(store: FlowStore):
    dashboards = DashboardStore(store)
    flow = FlowDef(id="x_lab", name="lab", mode="batch")

    def generate(_: int) -> str:
        try:
            return dashboards.create(results_dashboard(flow), "", True).id
        except DashboardExists:
            return ""

    with ThreadPoolExecutor(max_workers=4) as pool:
        made = [board for board in pool.map(generate, range(4)) if board]

    assert len(made) == 1
    assert dashboards.read(made[0]).name == "lab_results"


def test_a_discard_leaves_no_old_draft_version_to_match(store: FlowStore):
    flow_id = store.create(FlowDef(name="heating")).id
    published = store.publish_flow(flow_id, 1)
    held = store.write_draft(published.model_copy(update={"title": "A"}), 1)
    store.write_draft(held.model_copy(update={"title": "B"}), held.version)

    restored = store.discard_draft(flow_id)

    # Above every draft version handed out, so an editor holding one is stale.
    assert restored.version == 4 and restored.name == "heating"
    with pytest.raises(StaleVersion):
        store.write_draft(held.model_copy(update={"title": "late"}), held.version)
    assert store.write_draft(restored.model_copy(update={"title": "C"}), 4).version == 5

    dashboards = DashboardStore(store)
    board = dashboards.create(default_dashboard("home"))
    dashboards.publish(board.id, 1)
    dashboards.write_draft(dashboards.read(board.id).model_copy(), 1)
    assert dashboards.discard_draft(board.id).version == 3
    # A direct write goes above a draft too.
    dashboards.write_draft(dashboards.read(board.id).model_copy(), 3)
    assert dashboards.write(dashboards.read(board.id)).version == 5


def test_a_copy_gets_an_id_of_its_own_and_a_key_nobody_holds(store: FlowStore):
    store.write_flow(FlowDef(name="heating"))
    store.create(FlowDef(name="heating_copy"))
    dashboards = DashboardStore(store)
    dashboards.write(default_dashboard("home"))
    dashboards.create(default_dashboard("home_copy"))

    copy = store.duplicate_flow("heating", busy=lambda key: key == "heating_copy_2")
    board = dashboards.duplicate("home")

    # `_copy_2` still had work behind it, so the copy went past it.
    assert copy.name == "heating_copy_3" and ISSUED_ID.fullmatch(copy.id)
    assert board.name == "home_copy_2" and ISSUED_ID.fullmatch(board.id)


def test_a_new_key_shaped_like_an_issued_id_is_refused(store: FlowStore):
    with pytest.raises(ReservedKey):
        store.create(FlowDef(name="f_k3q7m2xa"))
    with pytest.raises(ReservedKey):
        DashboardStore(store).create(default_dashboard("d_k3q7m2xa"))
    assert store.create(FlowDef(name="f_k3q7m2xa_lab")).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


def test_what_a_create_takes_is_a_document_but_for_its_version():
    """`FlowCreate` lists the fields again; one added to `FlowDef` belongs there."""
    assert set(FlowCreate.model_fields) == set(FlowDef.model_fields) - {"version"}
    with pytest.raises(ValueError):
        FlowCreate(name="Not A Key")