app/backend/tests/utils/rename.py

"""What a key rename has to move, read where each kind of thing keeps names.

Only the fields classified as holding message names, never a text search: a
flow from before ids keeps its key as its id, in its directory and its history
rows; a deleted flow that had the same key keeps its history; and payloads,
logs and git history keep whatever they say.
"""

import json
from collections.abc import Iterable
from pathlib import Path
from typing import Any

from sqlmodel import Session, col, select

from fluksio.core.config import settings
from fluksio.core.db import engine
from fluksio.flow.dashboards import DASHBOARD_BINDINGS, DASHBOARD_DIR
from fluksio.flow.messages import flow_of
from fluksio.flow.namespace import FLOW_FIELDS, NAME, OUTPUTS, Rename, names, touches
from fluksio.flow.queue import WorkQueue
from fluksio.flow.rename import DEDUP, SQL_NAMES
from fluksio.flow.state import StateBackend
from fluksio.models import (
    AlertSent,
    EngineEvent,
    FlowRun,
    Run,
    RunArtifact,
    RunMetric,
    RunNode,
)

#: The history tables by name, as `SQL_NAMES` has them.
_TABLES = {model.__tablename__: model for model in (RunMetric, RunArtifact, RunNode)}


def _documents(root: Path) -> Iterable[tuple[Path, tuple[tuple[str, ...], ...]]]:
    for path in sorted(root.glob("*/flow*.json")):
        # `_devices/flow.json` is a device, not a flow.
        if not path.parent.name.startswith(("_", ".")):
            yield path, FLOW_FIELDS
    for path in sorted((root / DASHBOARD_DIR).glob("*/dashboard*.json")):
        yield path, DASHBOARD_BINDINGS


def leftovers(
    key: str,
    root: Path,
    flow_ids: Iterable[str],
    state: StateBackend | None = None,
    queue: WorkQueue | None = None,
) -> list[str]:
    """Every classified field that still names something of the flow key
    *key*: document keys and bindings under *root*, the history of the flows
    *flow_ids*, the alert ledger and channels, live state and queued work."""

    def named(name: str) -> bool:
        return "." in name and flow_of(name) == key

    found = []
    for path, paths in _documents(root):
        rel = path.relative_to(root)
        data = json.loads(path.read_text())
        if paths is FLOW_FIELDS and data.get("name") == key:
            found.append(f"{rel}: its key")
        found += [f"{rel}: {name}" for name in names(data, paths) if named(name)]

    runs = select(Run.id).where(col(Run.flow_id).in_([*flow_ids]))
    with Session(engine) as session:
        for (table, column), holds in SQL_NAMES.items():
            if holds == DEDUP:
                # The tail is a node, or a flow key on its own.
                found += [
                    f"{table}.{column}: {one}"
                    for one in session.exec(select(AlertSent.key)).all()
                    if flow_of(one.partition(":")[2]) == key
                ]
                continue
            model = _TABLES[table]
            values = session.exec(
                select(getattr(model, column)).where(col(model.run_id).in_(runs))
            ).all()
            if holds == NAME:
                found += [f"{table}.{column}: {v}" for v in values if named(v)]
            elif holds == OUTPUTS:
                found += [
                    f"{table}.{column}: {name}"
                    for text in values
                    for name in json.loads(text or "{}")
                    if named(name)
                ]

    try:
        alerts = json.loads(settings.ALERTS_FILE.read_text())
    except FileNotFoundError:
        alerts = {}
    found += [
        f"alerts.json: {channel['config']['message']}"
        for channel in alerts.get("channels", [])
        if channel["kind"] == "dashboard" and named(channel["config"]["message"])
    ]

    if state is not None:
        found += [one for one in state.keys() if touches(key, one)]
        found += [f"names: {one}" for one in state.message_names() if named(one)]
    if queue is not None:
        probe = Rename(key, f"{key}_probe")
        found += [
            f"queued: {one.node}" for one in queue.live_items() if probe.item(one)
        ]
        markers = queue.count(probe)["done_markers"]
        found += [f"done markers: {markers}"] if markers else []
    return found


#: Columns that hold ids, which a rename never changes.
_IDENTITIES: tuple[Any, ...] = (
    Run.id,
    Run.flow_id,
    Run.for_node,
    RunNode.node,
    RunMetric.node,
    RunArtifact.node,
    FlowRun.id,
    FlowRun.flow_id,
)


def stability(root: Path) -> dict[str, Any]:
    """What a rename leaves exactly as it was: the store's directories,
    every file in it but the documents, and the id columns of the history
    (`EngineEvent` without the audit rows a rename adds)."""
    documents = {path for path, _ in _documents(root)}
    files = {
        str(path.relative_to(root)): path.read_bytes()
        for path in sorted(root.rglob("*"))
        if path.is_file() and ".git" not in path.parts and path not in documents
    }
    directories = sorted(
        str(path.relative_to(root))
        for path in root.rglob("*")
        if path.is_dir() and ".git" not in path.parts
    )
    with Session(engine) as session:
        ids = {
            str(column): sorted(map(str, session.exec(select(column)).all()))
            for column in _IDENTITIES
        }
        ids["engine_event"] = sorted(
            map(
                str,
                session.exec(
                    select(EngineEvent.flow_id, EngineEvent.node).where(
                        col(EngineEvent.type) != "audit"
                    )
                ).all(),
            )
        )
    return {"files": files, "directories": directories, "ids": ids}