"""API ↔ CLI parity tests for the v49 unified stack (Phase 9 / Task 9.1).

For each (HTTP endpoint, ``agnes …`` CLI command) pair listed in Section 6
of the unified-stack design doc, fire BOTH paths against the same in-memory
DB and verify they produce the **identical** DB state delta. Catches drift
where a CLI command and the corresponding endpoint quietly start writing
different rows, dropping audit lines, or fanning out one but not the other.

Bridging the CLI to the FastAPI ``TestClient``
----------------------------------------------
The CLI's ``api_get/post/put/delete`` helpers do real HTTP via ``httpx``.
For parity we patch them inside each command module (where Typer captured
them at import time) to redirect through the shared ``TestClient`` with an
admin Bearer token injected. The CLI code path is otherwise untouched —
argument parsing, error handling, slug→id resolution, audit emission all
exercise their real implementations.

Delta shape
-----------
Each test seeds the DB to a known starting state, snapshots the relevant
tables, fires the API path, snapshots again → ``delta_api``. Reset to the
same starting state, fire the CLI path, snapshot → ``delta_cli``. Assert
``delta_api == delta_cli``.

We deliberately exclude ``audit_log.created_at`` / ``data_packages.created_at``
from comparison — wall-clock columns flicker between two adjacent writes. We
DO compare row counts, schema, and content columns. Audit rows are matched
by ``(action, target)`` pairs since exact wall-clock differs.
"""

from __future__ import annotations

import uuid
from typing import Any, Callable, Dict, List, Optional, Tuple

import pytest
from typer.testing import CliRunner

from cli.main import app as cli_app
from src.db import get_system_db


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _auth(token: str) -> Dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def _patch_cli_to_testclient(monkeypatch, modules: List[str], client, token: str):
    """Redirect CLI's ``api_*`` helpers to the in-memory TestClient.

    Each CLI command module imported its api helpers at module import time
    (``from cli.client import api_get, api_post, …``). Typer captures those
    bound names — so we patch them on each module rather than on
    ``cli.client`` itself.

    Auth header is injected automatically using the admin token so admin
    endpoints accept the request.
    """
    auth = _auth(token)

    def _normalize(headers):
        out = dict(auth)
        if headers:
            out.update(headers)
        return out

    def _get(path: str, *, timeout: float = 30.0, **kwargs):
        kwargs.setdefault("headers", {})
        kwargs["headers"] = _normalize(kwargs["headers"])
        # TestClient accepts ``params`` directly like httpx.
        return client.get(path, **kwargs)

    def _post(path: str, *, timeout: float = 30.0, **kwargs):
        kwargs.setdefault("headers", {})
        kwargs["headers"] = _normalize(kwargs["headers"])
        return client.post(path, **kwargs)

    def _put(path: str, *, timeout: float = 30.0, **kwargs):
        kwargs.setdefault("headers", {})
        kwargs["headers"] = _normalize(kwargs["headers"])
        return client.put(path, **kwargs)

    def _delete(path: str, *, timeout: float = 30.0, **kwargs):
        kwargs.setdefault("headers", {})
        kwargs["headers"] = _normalize(kwargs["headers"])
        return client.delete(path, **kwargs)

    for mod_name in modules:
        for name, repl in (
            ("api_get", _get),
            ("api_post", _post),
            ("api_put", _put),
            ("api_delete", _delete),
        ):
            try:
                monkeypatch.setattr(f"{mod_name}.{name}", repl)
            except AttributeError:
                # Module didn't import this helper — skip silently.
                pass


# ---------------------------------------------------------------------------
# Snapshot helpers
# ---------------------------------------------------------------------------


def _snapshot_table(conn, sql: str, params=None) -> List[tuple]:
    rows = conn.execute(sql, params or []).fetchall()
    # Sort to make comparison order-independent.
    return sorted(tuple(r) for r in rows)


def _snapshot_audit_actions(conn, prefix: str = "") -> List[str]:
    """Audit actions are compared by (action, resource-shape) without
    wall-clock or random uuid suffixes so parity holds across two adjacent
    writes against different random ids.

    Newly-created rows pick up a fresh uuid in ``resource`` (e.g.
    ``data_package:pkg_0f7a4af03713``) so we mask the id to ``<id>`` and
    keep only the resource type prefix. The remaining shape is what we
    actually care about: same action verb + same resource type.
    """
    import re

    if prefix:
        rows = conn.execute(
            "SELECT action, resource FROM audit_log WHERE action LIKE ? ORDER BY id",
            [f"{prefix}%"],
        ).fetchall()
    else:
        rows = conn.execute(
            "SELECT action, resource FROM audit_log ORDER BY id"
        ).fetchall()
    # ``resource`` is canonically ``<type>:<id>``; normalize the id half so
    # uuid-based ids don't break equality on two adjacent runs.
    masked = []
    for action, resource in rows:
        if resource and ":" in resource:
            rtype, _ = resource.split(":", 1)
            masked.append(f"{action}|{rtype}:<id>")
        else:
            masked.append(f"{action}|{resource}")
    return masked


def _reset_audit_log(conn) -> None:
    conn.execute("DELETE FROM audit_log")


# ---------------------------------------------------------------------------
# DB seeding helpers
# ---------------------------------------------------------------------------


def _seed_group_with_user(conn, *, name: str, user_id: str) -> str:
    """Create a fresh group + add user_id to it. Returns the new group id."""
    from src.repositories.user_groups import UserGroupsRepository
    from src.repositories.user_group_members import UserGroupMembersRepository

    g = UserGroupsRepository(conn).create(name=name, description="", created_by="test")
    gid = g["id"] if isinstance(g, dict) else g
    UserGroupMembersRepository(conn).add_member(user_id, gid, source="test")
    return gid


def _seed_data_package(conn, *, slug: str, name: str = "P") -> str:
    from src.repositories.data_packages import DataPackagesRepository

    return DataPackagesRepository(conn).create(
        name=name, slug=slug, description=None,
        icon=None, color=None, created_by="test",
    )


def _seed_memory_domain(conn, *, slug: str, name: str = "D") -> str:
    from src.repositories.memory_domains import MemoryDomainsRepository

    return MemoryDomainsRepository(conn).create(
        name=name, slug=slug, description=None,
        icon=None, color=None, created_by="test",
    )


def _seed_grant_for(
    conn, group_id: str, resource_type: str, resource_id: str,
    requirement: str = "available",
) -> str:
    grant_id = str(uuid.uuid4())
    conn.execute(
        "INSERT INTO resource_grants(id, group_id, resource_type, resource_id, "
        "requirement, assigned_at, assigned_by) "
        "VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP, 'test')",
        [grant_id, group_id, resource_type, resource_id, requirement],
    )
    return grant_id


def _seed_table_registry(conn, *, name: str) -> str:
    """Insert a minimal table_registry row and return its id.

    The repo's ``register()`` requires an explicit id and returns None; we
    just pick a stable id mirroring the name so subsequent lookups don't
    need a second roundtrip.
    """
    from src.repositories.table_registry import TableRegistryRepository

    tid = f"tbl_{name}"
    TableRegistryRepository(conn).register(
        id=tid,
        name=name,
        source_type="keboola",
        bucket="b",
        source_table=name,
        query_mode="local",
    )
    return tid


def _seed_knowledge_item(conn, *, item_id: str, title: str = "T") -> None:
    conn.execute(
        "INSERT INTO knowledge_items(id, title, status) VALUES (?, ?, 'approved')",
        [item_id, title],
    )


def _purge_user_state(conn) -> None:
    """Wipe everything that the parity tests dirty. Called between API/CLI runs.

    Order matters — junction tables first to avoid FK / orphan complaints.
    """
    for sql in (
        "DELETE FROM data_package_tables",
        "DELETE FROM data_packages",
        "DELETE FROM knowledge_item_domains",
        "DELETE FROM memory_domains",
        "DELETE FROM user_stack_subscriptions",
        "DELETE FROM resource_grants",
        "DELETE FROM user_group_members WHERE source = 'test'",
        "DELETE FROM user_groups WHERE name LIKE 'parity_%'",
        "DELETE FROM knowledge_items",
        "DELETE FROM table_registry",
        "DELETE FROM audit_log",
    ):
        try:
            conn.execute(sql)
        except Exception:
            # DuckDB raises on unknown tables in some legacy schemas. The
            # parity suite only runs against v49+; the empty try keeps
            # cleanup robust if a table was renamed in a parallel branch.
            pass


# ---------------------------------------------------------------------------
# Fixture
# ---------------------------------------------------------------------------


@pytest.fixture
def parity_env(seeded_app, monkeypatch):
    """Patch CLI api_* helpers to the seeded TestClient + return runners.

    Returns a dict with the test client, admin token, the Typer CliRunner,
    and ready-made ``run_cli`` / ``run_api`` callables.
    """
    # Suppress the auto-update probe — it hits /cli/latest, prints a stderr
    # warning that CliRunner merges into ``result.output``, and breaks
    # ``json.loads(result.output)`` parsing.
    monkeypatch.setenv("AGNES_NO_UPDATE_CHECK", "1")

    client = seeded_app["client"]
    admin_token = seeded_app["admin_token"]
    analyst_token = seeded_app["analyst_token"]

    _patch_cli_to_testclient(
        monkeypatch,
        modules=[
            "cli.commands.admin_data_package",
            "cli.commands.admin_memory_domain",
            "cli.commands.stack",
            "cli.commands.admin",
        ],
        client=client,
        token=admin_token,
    )
    runner = CliRunner()

    def run_cli(args: List[str], *, expect_success: bool = True, input: str = ""):
        result = runner.invoke(cli_app, args, input=input)
        if expect_success:
            assert result.exit_code == 0, (
                f"CLI {' '.join(args)} failed exit={result.exit_code}\n"
                f"stdout={result.output}\n"
                f"stderr={getattr(result, 'stderr', '<merged>')}"
            )
        return result

    return {
        "client": client,
        "admin_token": admin_token,
        "analyst_token": analyst_token,
        "run_cli": run_cli,
    }


# ---------------------------------------------------------------------------
# Stack subscribe / unsubscribe / list
# ---------------------------------------------------------------------------


class TestStackSubscribeParity:
    """``POST /api/stack/subscribe`` ↔ ``agnes stack add <type> <id>``."""

    def _setup(self):
        conn = get_system_db()
        _purge_user_state(conn)
        gid = _seed_group_with_user(conn, name="parity_subs", user_id="analyst1")
        pkg_id = _seed_data_package(conn, slug="parity-sub-pkg")
        _seed_grant_for(conn, gid, "data_package", pkg_id, "available")
        conn.close()
        return pkg_id

    def test_subscribe_parity(self, parity_env):
        pkg_id = self._setup()

        # API path
        r = parity_env["client"].post(
            "/api/stack/subscribe",
            json={"resource_type": "data_package", "resource_id": pkg_id},
            headers=_auth(parity_env["analyst_token"]),
        )
        assert r.status_code == 200
        conn = get_system_db()
        delta_api = _snapshot_table(
            conn,
            "SELECT user_id, resource_type, resource_id "
            "FROM user_stack_subscriptions WHERE resource_id = ?",
            [pkg_id],
        )
        conn.close()

        # Reset subscriptions and re-fire via CLI. CLI uses admin token by
        # default (the global patch wires admin). To target analyst, we
        # re-patch with the analyst token for this test only.
        conn = get_system_db()
        conn.execute("DELETE FROM user_stack_subscriptions")
        conn.close()

        # Re-issue CLI under analyst auth.
        with _admin_auth_swap(parity_env, parity_env["analyst_token"]):
            parity_env["run_cli"](["stack", "add", "data_package", pkg_id])

        conn = get_system_db()
        delta_cli = _snapshot_table(
            conn,
            "SELECT user_id, resource_type, resource_id "
            "FROM user_stack_subscriptions WHERE resource_id = ?",
            [pkg_id],
        )
        conn.close()

        assert delta_api == delta_cli
        # Both paths produced exactly one (analyst1, data_package, pkg) row.
        assert delta_api == [("analyst1", "data_package", pkg_id)]


class TestStackUnsubscribeParity:
    """``DELETE /api/stack/subscription/{type}/{id}`` ↔ ``agnes stack remove …``."""

    def _setup_with_sub(self):
        conn = get_system_db()
        _purge_user_state(conn)
        gid = _seed_group_with_user(conn, name="parity_unsubs", user_id="analyst1")
        pkg_id = _seed_data_package(conn, slug="parity-unsub-pkg")
        _seed_grant_for(conn, gid, "data_package", pkg_id, "available")
        conn.execute(
            "INSERT INTO user_stack_subscriptions(user_id, resource_type, resource_id) "
            "VALUES ('analyst1', 'data_package', ?)",
            [pkg_id],
        )
        conn.close()
        return pkg_id

    def test_unsubscribe_parity(self, parity_env):
        pkg_id = self._setup_with_sub()

        # API
        r = parity_env["client"].delete(
            f"/api/stack/subscription/data_package/{pkg_id}",
            headers=_auth(parity_env["analyst_token"]),
        )
        # 0.54.26 design-rules pass moved this endpoint to 204.
        assert r.status_code == 204
        conn = get_system_db()
        delta_api = _snapshot_table(
            conn,
            "SELECT user_id, resource_type, resource_id "
            "FROM user_stack_subscriptions",
        )
        conn.close()

        # Reinstate the same subscription against the same package id (don't
        # rebuild the whole setup; pkg_id is random and would drift).
        conn = get_system_db()
        conn.execute(
            "INSERT INTO user_stack_subscriptions(user_id, resource_type, resource_id) "
            "VALUES ('analyst1', 'data_package', ?)",
            [pkg_id],
        )
        conn.close()
        with _admin_auth_swap(parity_env, parity_env["analyst_token"]):
            parity_env["run_cli"](["stack", "remove", "data_package", pkg_id])

        conn = get_system_db()
        delta_cli = _snapshot_table(
            conn,
            "SELECT user_id, resource_type, resource_id "
            "FROM user_stack_subscriptions",
        )
        conn.close()

        assert delta_api == delta_cli == []


class TestStackListParity:
    """``GET /api/stack?type=`` ↔ ``agnes stack list --type ...``."""

    def test_list_parity(self, parity_env):
        conn = get_system_db()
        _purge_user_state(conn)
        gid = _seed_group_with_user(conn, name="parity_list", user_id="analyst1")
        pkg_id = _seed_data_package(conn, slug="parity-list-pkg", name="ListPkg")
        _seed_grant_for(conn, gid, "data_package", pkg_id, "required")
        conn.close()

        # API
        r_api = parity_env["client"].get(
            "/api/stack?type=data_package",
            headers=_auth(parity_env["analyst_token"]),
        )
        assert r_api.status_code == 200
        api_ids = sorted(it["id"] for it in r_api.json()["items"])

        # CLI — JSON output for stable comparison
        with _admin_auth_swap(parity_env, parity_env["analyst_token"]):
            result = parity_env["run_cli"](
                ["stack", "list", "--type", "data_package", "--json"]
            )
        import json
        cli_ids = sorted(it["id"] for it in json.loads(result.output))

        assert api_ids == cli_ids
        assert pkg_id in api_ids


# ---------------------------------------------------------------------------
# Data Package admin CRUD
# ---------------------------------------------------------------------------


class TestDataPackageCreateParity:
    """``POST /api/admin/data-packages`` ↔ ``agnes admin data-package create``."""

    def _snapshot(self, conn) -> tuple:
        pkg_rows = _snapshot_table(
            conn, "SELECT slug, name, description, icon, color FROM data_packages"
        )
        audit_rows = _snapshot_audit_actions(conn, prefix="data_package.create")
        return (pkg_rows, audit_rows)

    def test_create_parity(self, parity_env):
        conn = get_system_db()
        _purge_user_state(conn)
        conn.close()

        # API
        r = parity_env["client"].post(
            "/api/admin/data-packages",
            json={"name": "API Pkg", "slug": "api-pkg",
                  "description": "via api", "icon": None, "color": None},
            headers=_auth(parity_env["admin_token"]),
        )
        assert r.status_code == 201
        conn = get_system_db()
        delta_api = self._snapshot(conn)
        conn.close()

        # Reset
        conn = get_system_db()
        _purge_user_state(conn)
        conn.close()

        # CLI
        parity_env["run_cli"](
            ["admin", "data-package", "create",
             "--name", "API Pkg", "--slug", "api-pkg",
             "--description", "via api"]
        )
        conn = get_system_db()
        delta_cli = self._snapshot(conn)
        conn.close()

        assert delta_api == delta_cli
        # Both wrote one package + one audit row.
        assert len(delta_api[0]) == 1
        assert len(delta_api[1]) == 1


class TestDataPackageEditParity:
    """``PUT /api/admin/data-packages/{id}`` ↔ ``agnes admin data-package edit``."""

    def _setup(self):
        conn = get_system_db()
        _purge_user_state(conn)
        pkg_id = _seed_data_package(conn, slug="parity-edit", name="OldName")
        conn.close()
        return pkg_id

    def test_edit_parity(self, parity_env):
        pkg_id = self._setup()

        r = parity_env["client"].put(
            f"/api/admin/data-packages/{pkg_id}",
            json={"name": "NewName"},
            headers=_auth(parity_env["admin_token"]),
        )
        assert r.status_code == 200
        conn = get_system_db()
        name_api = conn.execute(
            "SELECT name FROM data_packages WHERE id = ?", [pkg_id]
        ).fetchone()[0]
        audit_api = _snapshot_audit_actions(conn, prefix="data_package.update")
        conn.close()

        # Reset to OldName + audit
        conn = get_system_db()
        conn.execute(
            "UPDATE data_packages SET name = 'OldName' WHERE id = ?", [pkg_id]
        )
        _reset_audit_log(conn)
        conn.close()

        parity_env["run_cli"](
            ["admin", "data-package", "edit", pkg_id, "--name", "NewName"]
        )
        conn = get_system_db()
        name_cli = conn.execute(
            "SELECT name FROM data_packages WHERE id = ?", [pkg_id]
        ).fetchone()[0]
        audit_cli = _snapshot_audit_actions(conn, prefix="data_package.update")
        conn.close()

        assert name_api == name_cli == "NewName"
        assert audit_api == audit_cli
        assert len(audit_api) == 1


class TestDataPackageDeleteParity:
    """``DELETE /api/admin/data-packages/{id}`` ↔ ``agnes admin data-package delete``."""

    def _setup(self):
        conn = get_system_db()
        _purge_user_state(conn)
        pkg_id = _seed_data_package(conn, slug="parity-del")
        conn.close()
        return pkg_id

    def test_delete_parity(self, parity_env):
        pkg_id = self._setup()

        r = parity_env["client"].delete(
            f"/api/admin/data-packages/{pkg_id}",
            headers=_auth(parity_env["admin_token"]),
        )
        assert r.status_code == 204
        # v54: delete() is a soft delete — the row still exists but
        # carries ``deleted_at IS NOT NULL``. Parity asserts both
        # paths leave the row in the same "live=0, soft-deleted=1"
        # shape.
        conn = get_system_db()
        api_live = conn.execute(
            "SELECT COUNT(*) FROM data_packages "
            "WHERE id = ? AND deleted_at IS NULL", [pkg_id]
        ).fetchone()[0]
        api_soft = conn.execute(
            "SELECT COUNT(*) FROM data_packages "
            "WHERE id = ? AND deleted_at IS NOT NULL", [pkg_id]
        ).fetchone()[0]
        api_audit = _snapshot_audit_actions(conn, prefix="data_package.delete")
        conn.close()

        pkg_id_2 = self._setup()
        parity_env["run_cli"](
            ["admin", "data-package", "delete", pkg_id_2, "--yes"]
        )
        conn = get_system_db()
        cli_live = conn.execute(
            "SELECT COUNT(*) FROM data_packages "
            "WHERE id = ? AND deleted_at IS NULL", [pkg_id_2]
        ).fetchone()[0]
        cli_soft = conn.execute(
            "SELECT COUNT(*) FROM data_packages "
            "WHERE id = ? AND deleted_at IS NOT NULL", [pkg_id_2]
        ).fetchone()[0]
        cli_audit = _snapshot_audit_actions(conn, prefix="data_package.delete")
        conn.close()

        assert api_live == cli_live == 0
        assert api_soft == cli_soft == 1
        # Both paths emit exactly one delete audit row
        assert len(api_audit) == len(cli_audit) == 1


class TestDataPackageAddRemoveTableParity:
    """``POST /api/admin/data-packages/{id}/tables`` ↔ ``agnes admin data-package add-table``;
    plus the matching remove-table pair."""

    def _setup(self):
        conn = get_system_db()
        _purge_user_state(conn)
        pkg_id = _seed_data_package(conn, slug="parity-tbl-pkg")
        tbl_id = _seed_table_registry(conn, name="parity_tbl")
        conn.close()
        return pkg_id, tbl_id

    def test_add_table_parity(self, parity_env):
        pkg_id, tbl_id = self._setup()

        r = parity_env["client"].post(
            f"/api/admin/data-packages/{pkg_id}/tables",
            json={"table_id": tbl_id},
            headers=_auth(parity_env["admin_token"]),
        )
        assert r.status_code == 200
        conn = get_system_db()
        api_state = _snapshot_table(
            conn, "SELECT package_id, table_id FROM data_package_tables"
        )
        api_audit = _snapshot_audit_actions(conn, prefix="data_package.add_table")
        conn.close()

        # Reset junction + audit
        conn = get_system_db()
        conn.execute("DELETE FROM data_package_tables")
        _reset_audit_log(conn)
        conn.close()

        parity_env["run_cli"](
            ["admin", "data-package", "add-table", pkg_id, tbl_id]
        )
        conn = get_system_db()
        cli_state = _snapshot_table(
            conn, "SELECT package_id, table_id FROM data_package_tables"
        )
        cli_audit = _snapshot_audit_actions(conn, prefix="data_package.add_table")
        conn.close()

        assert api_state == cli_state
        assert api_audit == cli_audit
        assert api_state == [(pkg_id, tbl_id)]

    def test_remove_table_parity(self, parity_env):
        pkg_id, tbl_id = self._setup()
        # Pre-link
        conn = get_system_db()
        conn.execute(
            "INSERT INTO data_package_tables(package_id, table_id, added_by) "
            "VALUES (?, ?, 'test')",
            [pkg_id, tbl_id],
        )
        conn.close()

        r = parity_env["client"].delete(
            f"/api/admin/data-packages/{pkg_id}/tables/{tbl_id}",
            headers=_auth(parity_env["admin_token"]),
        )
        assert r.status_code == 204
        conn = get_system_db()
        api_state = _snapshot_table(
            conn, "SELECT package_id, table_id FROM data_package_tables"
        )
        api_audit = _snapshot_audit_actions(conn, prefix="data_package.remove_table")
        conn.close()

        # Re-link
        conn = get_system_db()
        conn.execute(
            "INSERT INTO data_package_tables(package_id, table_id, added_by) "
            "VALUES (?, ?, 'test')",
            [pkg_id, tbl_id],
        )
        _reset_audit_log(conn)
        conn.close()

        parity_env["run_cli"](
            ["admin", "data-package", "remove-table", pkg_id, tbl_id, "--yes"]
        )
        conn = get_system_db()
        cli_state = _snapshot_table(
            conn, "SELECT package_id, table_id FROM data_package_tables"
        )
        cli_audit = _snapshot_audit_actions(conn, prefix="data_package.remove_table")
        conn.close()

        assert api_state == cli_state == []
        assert api_audit == cli_audit


# ---------------------------------------------------------------------------
# Memory Domain admin CRUD
# ---------------------------------------------------------------------------


class TestMemoryDomainCreateParity:
    def _snapshot(self, conn) -> tuple:
        return (
            _snapshot_table(
                conn, "SELECT slug, name, description FROM memory_domains"
            ),
            _snapshot_audit_actions(conn, prefix="memory_domain.create"),
        )

    def test_create_parity(self, parity_env):
        conn = get_system_db()
        _purge_user_state(conn)
        conn.close()

        r = parity_env["client"].post(
            "/api/admin/memory-domains",
            json={"name": "Finance", "slug": "parity-finance",
                  "description": "$$$"},
            headers=_auth(parity_env["admin_token"]),
        )
        assert r.status_code == 201
        conn = get_system_db()
        api_delta = self._snapshot(conn)
        conn.close()

        conn = get_system_db()
        _purge_user_state(conn)
        conn.close()

        parity_env["run_cli"](
            ["admin", "memory-domain", "create",
             "--name", "Finance", "--slug", "parity-finance",
             "--description", "$$$"]
        )
        conn = get_system_db()
        cli_delta = self._snapshot(conn)
        conn.close()

        assert api_delta == cli_delta
        assert len(api_delta[0]) == 1


class TestMemoryDomainEditParity:
    def _setup(self):
        conn = get_system_db()
        _purge_user_state(conn)
        dom_id = _seed_memory_domain(conn, slug="parity-edit-dom", name="Old")
        conn.close()
        return dom_id

    def test_edit_parity(self, parity_env):
        dom_id = self._setup()

        r = parity_env["client"].put(
            f"/api/admin/memory-domains/{dom_id}",
            json={"name": "New"},
            headers=_auth(parity_env["admin_token"]),
        )
        assert r.status_code == 200
        conn = get_system_db()
        api_name = conn.execute(
            "SELECT name FROM memory_domains WHERE id = ?", [dom_id]
        ).fetchone()[0]
        api_audit = _snapshot_audit_actions(conn, prefix="memory_domain.update")
        conn.close()

        conn = get_system_db()
        conn.execute(
            "UPDATE memory_domains SET name = 'Old' WHERE id = ?", [dom_id]
        )
        _reset_audit_log(conn)
        conn.close()

        parity_env["run_cli"](
            ["admin", "memory-domain", "edit", dom_id, "--name", "New"]
        )
        conn = get_system_db()
        cli_name = conn.execute(
            "SELECT name FROM memory_domains WHERE id = ?", [dom_id]
        ).fetchone()[0]
        cli_audit = _snapshot_audit_actions(conn, prefix="memory_domain.update")
        conn.close()

        assert api_name == cli_name == "New"
        assert api_audit == cli_audit


class TestMemoryDomainDeleteParity:
    def _setup(self):
        conn = get_system_db()
        _purge_user_state(conn)
        dom_id = _seed_memory_domain(conn, slug="parity-del-dom")
        conn.close()
        return dom_id

    def test_delete_parity(self, parity_env):
        dom_id = self._setup()

        r = parity_env["client"].delete(
            f"/api/admin/memory-domains/{dom_id}",
            headers=_auth(parity_env["admin_token"]),
        )
        assert r.status_code == 204
        # v54: soft delete (see TestDataPackageDeleteParity above).
        conn = get_system_db()
        api_live = conn.execute(
            "SELECT COUNT(*) FROM memory_domains "
            "WHERE id = ? AND deleted_at IS NULL", [dom_id]
        ).fetchone()[0]
        api_soft = conn.execute(
            "SELECT COUNT(*) FROM memory_domains "
            "WHERE id = ? AND deleted_at IS NOT NULL", [dom_id]
        ).fetchone()[0]
        api_audit = _snapshot_audit_actions(conn, prefix="memory_domain.delete")
        conn.close()

        dom_id_2 = self._setup()
        parity_env["run_cli"](
            ["admin", "memory-domain", "delete", dom_id_2, "--yes"]
        )
        conn = get_system_db()
        cli_live = conn.execute(
            "SELECT COUNT(*) FROM memory_domains "
            "WHERE id = ? AND deleted_at IS NULL", [dom_id_2]
        ).fetchone()[0]
        cli_soft = conn.execute(
            "SELECT COUNT(*) FROM memory_domains "
            "WHERE id = ? AND deleted_at IS NOT NULL", [dom_id_2]
        ).fetchone()[0]
        cli_audit = _snapshot_audit_actions(conn, prefix="memory_domain.delete")
        conn.close()

        assert api_live == cli_live == 0
        assert api_soft == cli_soft == 1
        assert len(api_audit) == len(cli_audit) == 1


class TestMemoryDomainAddRemoveItemParity:
    def _setup(self):
        conn = get_system_db()
        _purge_user_state(conn)
        dom_id = _seed_memory_domain(conn, slug="parity-item-dom")
        _seed_knowledge_item(conn, item_id="parity_item_1", title="T")
        conn.close()
        return dom_id, "parity_item_1"

    def test_add_item_parity(self, parity_env):
        dom_id, item_id = self._setup()

        r = parity_env["client"].post(
            f"/api/admin/memory-domains/{dom_id}/items",
            json={"item_id": item_id},
            headers=_auth(parity_env["admin_token"]),
        )
        assert r.status_code == 200
        conn = get_system_db()
        api_state = _snapshot_table(
            conn,
            "SELECT domain_id, item_id FROM knowledge_item_domains",
        )
        api_audit = _snapshot_audit_actions(conn, prefix="memory_domain.add_item")
        conn.close()

        conn = get_system_db()
        conn.execute("DELETE FROM knowledge_item_domains")
        _reset_audit_log(conn)
        conn.close()

        parity_env["run_cli"](
            ["admin", "memory-domain", "add-item", dom_id, item_id]
        )
        conn = get_system_db()
        cli_state = _snapshot_table(
            conn,
            "SELECT domain_id, item_id FROM knowledge_item_domains",
        )
        cli_audit = _snapshot_audit_actions(conn, prefix="memory_domain.add_item")
        conn.close()

        assert api_state == cli_state
        assert api_audit == cli_audit
        assert api_state == [(dom_id, item_id)]

    def test_remove_item_parity(self, parity_env):
        dom_id, item_id = self._setup()
        conn = get_system_db()
        conn.execute(
            "INSERT INTO knowledge_item_domains(item_id, domain_id, added_by) "
            "VALUES (?, ?, 'test')",
            [item_id, dom_id],
        )
        conn.close()

        r = parity_env["client"].delete(
            f"/api/admin/memory-domains/{dom_id}/items/{item_id}",
            headers=_auth(parity_env["admin_token"]),
        )
        assert r.status_code == 204
        conn = get_system_db()
        api_state = _snapshot_table(
            conn,
            "SELECT domain_id, item_id FROM knowledge_item_domains",
        )
        api_audit = _snapshot_audit_actions(conn, prefix="memory_domain.remove_item")
        conn.close()

        # Re-link
        conn = get_system_db()
        conn.execute(
            "INSERT INTO knowledge_item_domains(item_id, domain_id, added_by) "
            "VALUES (?, ?, 'test')",
            [item_id, dom_id],
        )
        _reset_audit_log(conn)
        conn.close()

        parity_env["run_cli"](
            ["admin", "memory-domain", "remove-item", dom_id, item_id, "--yes"]
        )
        conn = get_system_db()
        cli_state = _snapshot_table(
            conn,
            "SELECT domain_id, item_id FROM knowledge_item_domains",
        )
        cli_audit = _snapshot_audit_actions(conn, prefix="memory_domain.remove_item")
        conn.close()

        assert api_state == cli_state == []
        assert api_audit == cli_audit


# ---------------------------------------------------------------------------
# Grants (requirement) — POST + PUT pair
# ---------------------------------------------------------------------------


class TestGrantCreateRequirementParity:
    """``POST /api/admin/grants`` + PUT requirement ↔
    ``agnes admin grant create … --requirement required``.

    CLI route: POST → if 201, PUT requirement when caller asked for
    required. We assert that both code paths end with the same single
    grant row at the requested requirement.
    """

    def _setup_group_and_package(self):
        conn = get_system_db()
        _purge_user_state(conn)
        from src.repositories.user_groups import UserGroupsRepository
        from src.repositories.user_group_members import UserGroupMembersRepository

        g = UserGroupsRepository(conn).create(
            name="parity_grant_g", description="", created_by="test",
        )
        gid = g["id"] if isinstance(g, dict) else g
        UserGroupMembersRepository(conn).add_member("analyst1", gid, source="test")
        pkg_id = _seed_data_package(conn, slug="parity-grant-pkg")
        conn.close()
        return gid, pkg_id

    def _snapshot_grant(self, conn, gid, pkg_id):
        return _snapshot_table(
            conn,
            "SELECT group_id, resource_type, resource_id, requirement "
            "FROM resource_grants WHERE group_id = ? AND resource_id = ?",
            [gid, pkg_id],
        )

    def test_grant_required_parity(self, parity_env):
        gid, pkg_id = self._setup_group_and_package()

        # API path: POST (creates available) + PUT (flips to required).
        r1 = parity_env["client"].post(
            "/api/admin/grants",
            json={"group_id": gid, "resource_type": "data_package",
                  "resource_id": pkg_id},
            headers=_auth(parity_env["admin_token"]),
        )
        assert r1.status_code == 201, r1.text
        grant_id = r1.json()["id"]
        r2 = parity_env["client"].put(
            f"/api/admin/grants/{grant_id}",
            json={"requirement": "required"},
            headers=_auth(parity_env["admin_token"]),
        )
        assert r2.status_code == 200, r2.text

        conn = get_system_db()
        api_state = self._snapshot_grant(conn, gid, pkg_id)
        conn.close()

        # Reset grants, fire CLI which internally does the same POST+PUT.
        conn = get_system_db()
        conn.execute(
            "DELETE FROM resource_grants WHERE group_id = ? AND resource_id = ?",
            [gid, pkg_id],
        )
        conn.close()

        parity_env["run_cli"](
            ["admin", "grant", "create", "parity_grant_g",
             "data_package", pkg_id, "--requirement", "required"]
        )
        conn = get_system_db()
        cli_state = self._snapshot_grant(conn, gid, pkg_id)
        conn.close()

        assert api_state == cli_state
        # Single required grant row.
        assert len(api_state) == 1
        assert api_state[0][3] == "required"


class TestGrantUpdateRequirementParity:
    """``PUT /api/admin/grants/{id}`` (requirement update) ↔ CLI grant create
    when an existing grant is detected (409 → list → PUT).
    """

    def _setup_with_grant(self, requirement: str):
        conn = get_system_db()
        _purge_user_state(conn)
        from src.repositories.user_groups import UserGroupsRepository
        from src.repositories.user_group_members import UserGroupMembersRepository

        g = UserGroupsRepository(conn).create(
            name="parity_put_g", description="", created_by="test",
        )
        gid = g["id"] if isinstance(g, dict) else g
        UserGroupMembersRepository(conn).add_member("analyst1", gid, source="test")
        pkg_id = _seed_data_package(conn, slug="parity-put-pkg")
        grant_id = _seed_grant_for(conn, gid, "data_package", pkg_id, requirement)
        conn.close()
        return gid, pkg_id, grant_id

    def test_downgrade_parity_materializes_subscriptions(self, parity_env):
        gid, pkg_id, grant_id = self._setup_with_grant("required")

        # API path
        r = parity_env["client"].put(
            f"/api/admin/grants/{grant_id}",
            json={"requirement": "available"},
            headers=_auth(parity_env["admin_token"]),
        )
        assert r.status_code == 200, r.text
        conn = get_system_db()
        api_grants = _snapshot_table(
            conn,
            "SELECT group_id, resource_type, resource_id, requirement "
            "FROM resource_grants WHERE id = ?",
            [grant_id],
        )
        api_subs = _snapshot_table(
            conn,
            "SELECT user_id, resource_type, resource_id "
            "FROM user_stack_subscriptions WHERE resource_id = ?",
            [pkg_id],
        )
        conn.close()

        # Reset to required, drop the eager subs, re-run CLI
        conn = get_system_db()
        conn.execute(
            "UPDATE resource_grants SET requirement = 'required' WHERE id = ?",
            [grant_id],
        )
        conn.execute(
            "DELETE FROM user_stack_subscriptions WHERE resource_id = ?",
            [pkg_id],
        )
        conn.close()

        # CLI: same group/resource → POST returns 409 → CLI lists → finds
        # existing grant → PUTs requirement update.
        parity_env["run_cli"](
            ["admin", "grant", "create", "parity_put_g",
             "data_package", pkg_id, "--requirement", "available"]
        )
        conn = get_system_db()
        cli_grants = _snapshot_table(
            conn,
            "SELECT group_id, resource_type, resource_id, requirement "
            "FROM resource_grants WHERE id = ?",
            [grant_id],
        )
        cli_subs = _snapshot_table(
            conn,
            "SELECT user_id, resource_type, resource_id "
            "FROM user_stack_subscriptions WHERE resource_id = ?",
            [pkg_id],
        )
        conn.close()

        assert api_grants == cli_grants
        # Soft-downgrade materialized subscription for analyst1 on BOTH paths.
        assert api_subs == cli_subs
        assert ("analyst1", "data_package", pkg_id) in api_subs


# ---------------------------------------------------------------------------
# Per-test auth swap helper (re-patches the CLI helpers to a non-admin token)
# ---------------------------------------------------------------------------


import contextlib


@contextlib.contextmanager
def _admin_auth_swap(parity_env, token: str):
    """Temporarily re-patch ``cli.commands.stack`` to use a different token.

    The fixture-level patch hard-wires the admin token. Tests that need
    analyst auth (subscribe/unsubscribe/list) flip it just for the CLI
    invocation, then restore.
    """
    from cli.commands import stack as _stack_mod

    client = parity_env["client"]

    def _normalize(headers, t):
        out = {"Authorization": f"Bearer {t}"}
        if headers:
            out.update(headers)
        return out

    orig = {
        "api_get": _stack_mod.api_get,
        "api_post": _stack_mod.api_post,
        "api_delete": _stack_mod.api_delete,
    }

    def _get(path: str, *, timeout: float = 30.0, **kwargs):
        kwargs["headers"] = _normalize(kwargs.get("headers"), token)
        return client.get(path, **kwargs)

    def _post(path: str, *, timeout: float = 30.0, **kwargs):
        kwargs["headers"] = _normalize(kwargs.get("headers"), token)
        return client.post(path, **kwargs)

    def _delete(path: str, *, timeout: float = 30.0, **kwargs):
        kwargs["headers"] = _normalize(kwargs.get("headers"), token)
        return client.delete(path, **kwargs)

    _stack_mod.api_get = _get
    _stack_mod.api_post = _post
    _stack_mod.api_delete = _delete
    try:
        yield
    finally:
        for k, v in orig.items():
            setattr(_stack_mod, k, v)