import json
from sqlite3 import Connection
from typing import cast

import pytest
from fastapi.testclient import TestClient

from jarvis_finance.api.dependencies import get_db
from jarvis_finance.api.main import create_app
from jarvis_finance.services.portfolio_policy import confirm_policy, preview_policy
from jarvis_finance.storage.database import connect_memory
from jarvis_finance.storage.migrations import apply_migrations


def db():
    conn = connect_memory()
    apply_migrations(conn)
    return conn


def payload(**changes):
    base = {
        "base_currency": "CHF",
        "effective_from": "2026-07-23",
        "horizon": "long",
        "objective": "Long term",
        "liquidity_reserve": None,
        "monthly_contribution": "100.00",
        "max_crypto_pct": "20",
        "allocations": [
            {
                "asset_class": "cash",
                "target_pct": "60",
                "lower_pct": "50",
                "upper_pct": "70",
            },
            {
                "asset_class": "equity",
                "target_pct": "40",
                "lower_pct": "30",
                "upper_pct": "50",
            },
        ],
    }
    base.update(changes)
    return base


def client_for(conn):
    app = create_app(write_mode="test")
    app.dependency_overrides[get_db] = lambda: conn
    return TestClient(app)


def preview(client, data):
    response = client.post("/api/portfolio/policy/preview", json=data)
    assert response.status_code == 200
    assert response.json()["valid"] is True
    return response.json()


def confirm(client, data, *, identifiers=None):
    identifiers = identifiers or preview(client, data)
    return client.post(
        "/api/portfolio/policy/confirm",
        json={
            **data,
            "preview_id": identifiers["preview_id"],
            "confirmation_id": identifiers["confirmation_id"],
            "confirm": True,
        },
    )


def test_preview_is_read_only_and_same_confirmation_is_idempotent():
    conn = db()
    client = client_for(conn)
    data = payload()
    before = conn.total_changes
    identifiers = preview(client, data)
    assert identifiers["preview_id"].startswith("policy-preview-")
    assert identifiers["confirmation_id"].startswith("policy-confirm-")
    assert conn.total_changes == before

    first = confirm(client, data, identifiers=identifiers)
    second = confirm(client, data, identifiers=identifiers)
    assert first.status_code == 200
    assert second.status_code == 200
    assert second.json() == {**first.json(), "idempotent": True}
    assert conn.execute("SELECT COUNT(*) FROM portfolio_policies").fetchone()[0] == 1


def test_confirmation_id_is_bound_to_preview_and_payload_fail_closed():
    conn = db()
    client = client_for(conn)
    original = payload()
    identifiers = preview(client, original)
    assert confirm(client, original, identifiers=identifiers).status_code == 200

    wrong_preview = confirm(
        client,
        original,
        identifiers={**identifiers, "preview_id": "policy-preview-wrong"},
    )
    assert wrong_preview.status_code == 400
    assert "Vorschau" in wrong_preview.json()["detail"]

    changed = payload(objective="Changed")
    changed_preview = preview(client, changed)
    conflict = confirm(
        client,
        changed,
        identifiers={
            "preview_id": changed_preview["preview_id"],
            "confirmation_id": identifiers["confirmation_id"],
        },
    )
    assert conflict.status_code == 400
    assert "verändertem Inhalt" in conflict.json()["detail"]
    assert conn.execute("SELECT COUNT(*) FROM portfolio_policies").fetchone()[0] == 1


class _EmptyCursor:
    def fetchone(self):
        return None


class _RaceConnection:
    """Hide the first lookup to force the post-IntegrityError race path."""

    def __init__(self, conn):
        self.conn = conn
        self.hide_confirmation_lookup = True

    def execute(self, sql, parameters=()):
        if (
            self.hide_confirmation_lookup
            and "FROM portfolio_policies WHERE confirmation_id=?" in sql
        ):
            self.hide_confirmation_lookup = False
            return _EmptyCursor()
        return self.conn.execute(sql, parameters)

    def commit(self):
        return self.conn.commit()

    def rollback(self):
        return self.conn.rollback()


def test_integrity_error_race_path_rechecks_request_identity():
    conn = db()
    client = client_for(conn)
    original = payload()
    identifiers = preview(client, original)
    first = confirm(client, original, identifiers=identifiers).json()
    request = {**original, **identifiers, "confirm": True}

    retried = confirm_policy(cast(Connection, _RaceConnection(conn)), request)
    assert retried == {**first, "idempotent": True}
    assert conn.execute("SELECT COUNT(*) FROM portfolio_policies").fetchone()[0] == 1
    assert conn.execute("SELECT COUNT(*) FROM audit_log").fetchone()[0] == 1

    changed = payload(objective="Changed during race")
    changed_preview = preview(client, changed)
    conflicting_request = {
        **changed,
        "preview_id": changed_preview["preview_id"],
        "confirmation_id": identifiers["confirmation_id"],
        "confirm": True,
    }
    with pytest.raises(ValueError, match="verändertem Inhalt"):
        confirm_policy(
            cast(Connection, _RaceConnection(conn)), conflicting_request
        )
    assert conn.execute("SELECT COUNT(*) FROM portfolio_policies").fetchone()[0] == 1
    assert conn.execute("SELECT COUNT(*) FROM audit_log").fetchone()[0] == 1


def test_new_confirmation_allows_same_content_and_a_b_a_creates_three_versions():
    conn = db()
    client = client_for(conn)
    policy_a = payload(objective="A")
    policy_b = payload(objective="B")

    first = confirm(client, policy_a).json()
    second = confirm(client, policy_b).json()
    third = confirm(client, policy_a).json()

    assert [first["version"], second["version"], third["version"]] == [1, 2, 3]
    assert len({first["policy_id"], second["policy_id"], third["policy_id"]}) == 3
    rows = conn.execute(
        """SELECT policy_id, version, previous_policy_id, is_active, objective
           FROM portfolio_policies ORDER BY version"""
    ).fetchall()
    assert rows[0]["previous_policy_id"] is None
    assert rows[1]["previous_policy_id"] == rows[0]["policy_id"]
    assert rows[2]["previous_policy_id"] == rows[1]["policy_id"]
    assert [row["is_active"] for row in rows] == [0, 0, 1]
    assert [row["objective"] for row in rows] == ["A", "B", "A"]


def test_audit_has_exact_allowlist_and_history_content_is_immutable():
    conn = db()
    client = client_for(conn)
    confirmed = confirm(client, payload()).json()
    audit = conn.execute(
        """SELECT entity_id, action, confirmed, new_values_json
           FROM audit_log WHERE audit_id=?""",
        (confirmed["audit_id"],),
    ).fetchone()
    details = json.loads(audit["new_values_json"])
    assert audit["entity_id"] == confirmed["policy_id"]
    assert audit["action"] == "portfolio_policy_confirmed"
    assert audit["confirmed"] == 1
    assert set(details) == {"version", "previous_policy_id"}
    assert details == {"version": 1, "previous_policy_id": None}
    serialized = audit["new_values_json"]
    for forbidden in (
        "CHF", "target_pct", "benchmarks", "objective", "preview",
        "confirmation", "payload", "hash", "monthly_contribution",
    ):
        assert forbidden not in serialized

    with pytest.raises(Exception, match="immutable"):
        conn.execute(
            "UPDATE portfolio_policies SET objective='changed' WHERE policy_id=?",
            (confirmed["policy_id"],),
        )
    with pytest.raises(Exception, match="cannot be deleted"):
        conn.execute(
            "DELETE FROM portfolio_policy_allocations WHERE policy_id=?",
            (confirmed["policy_id"],),
        )


def test_decimal_null_percentage_and_benchmark_validation():
    conn = db()
    client = client_for(conn)

    valid = payload(
        liquidity_reserve=None,
        monthly_contribution="0.10",
        min_transaction_amount="0",
        max_single_position_pct="100",
        rebalance_tolerance_pct="0.01",
        benchmarks=[{"reference": "SPI"}],
    )
    confirmed = confirm(client, valid)
    assert confirmed.status_code == 200
    row = conn.execute(
        """SELECT liquidity_reserve, monthly_contribution, min_transaction_amount
           FROM portfolio_policies WHERE policy_id=?""",
        (confirmed.json()["policy_id"],),
    ).fetchone()
    assert row["liquidity_reserve"] is None
    assert row["monthly_contribution"] == "0.10"
    assert row["min_transaction_amount"] == "0"

    cases = [
        (payload(liquidity_reserve="-0.01"), "liquidity_reserve"),
        (payload(monthly_contribution="-1"), "monthly_contribution"),
        (payload(max_crypto_pct="100.01"), "100 Prozent"),
        (payload(max_single_position_pct="101"), "max_single_position_pct"),
        (payload(benchmarks=[{"reference": "SPI", "weight": "100"}]), "Gewichte"),
        (payload(benchmarks=[{"weight": None}]), "Referenz"),
    ]
    for invalid, expected in cases:
        response = client.post("/api/portfolio/policy/preview", json=invalid)
        assert response.status_code == 200
        assert response.json()["valid"] is False
        assert expected in "; ".join(response.json()["errors"])

    direct_float = preview_policy(conn, payload(monthly_contribution=0.1))
    assert direct_float["valid"] is False
    assert "exakter Decimal-Text" in "; ".join(direct_float["errors"])


def test_allocation_percentages_bounds_total_and_crypto_cap():
    conn = db()
    client = client_for(conn)
    cases = [
        (
            [{"asset_class": "cash", "target_pct": "101", "lower_pct": "0", "upper_pct": "101"}],
            "100 Prozent",
        ),
        (
            [{"asset_class": "cash", "target_pct": "100", "lower_pct": "101", "upper_pct": "100"}],
            "Bandgrenzen",
        ),
        (
            [{"asset_class": "cash", "target_pct": "99.9", "lower_pct": "0", "upper_pct": "100"}],
            "exakt 100",
        ),
        (
            [
                {"asset_class": "cash", "target_pct": "70", "lower_pct": "0", "upper_pct": "100"},
                {"asset_class": "crypto", "target_pct": "30", "lower_pct": "0", "upper_pct": "100"},
            ],
            "Crypto-Zielallokation",
        ),
    ]
    for allocations, expected in cases:
        response = client.post(
            "/api/portfolio/policy/preview",
            json=payload(allocations=allocations),
        )
        assert response.status_code == 200
        assert response.json()["valid"] is False
        assert expected in "; ".join(response.json()["errors"])


def test_migration_adds_request_idempotency_columns_to_sprint4_schema():
    conn = connect_memory()
    conn.executescript(
        """
        CREATE TABLE portfolio_policies (
          policy_id TEXT PRIMARY KEY, version INTEGER NOT NULL UNIQUE,
          is_active INTEGER NOT NULL, effective_from TEXT NOT NULL,
          previous_policy_id TEXT, base_currency TEXT NOT NULL, horizon TEXT,
          objective TEXT, liquidity_reserve TEXT, monthly_contribution TEXT,
          max_single_position_pct TEXT, max_crypto_pct TEXT,
          rebalance_tolerance_pct TEXT, min_transaction_amount TEXT,
          benchmarks_json TEXT NOT NULL DEFAULT '[]',
          restrictions_json TEXT NOT NULL DEFAULT '[]',
          request_fingerprint TEXT NOT NULL UNIQUE, audit_id TEXT NOT NULL,
          created_at TEXT NOT NULL
        );
        INSERT INTO portfolio_policies(
          policy_id, version, is_active, effective_from, base_currency,
          request_fingerprint, audit_id, created_at
        ) VALUES ('old-policy', 1, 1, '2026-07-23', 'CHF', 'old-hash', 'old-audit', 'now');
        """
    )
    apply_migrations(conn)
    columns = {
        row["name"] for row in conn.execute("PRAGMA table_info(portfolio_policies)")
    }
    assert {"confirmation_id", "payload_hash"} <= columns
    old = conn.execute(
        """SELECT policy_id, request_fingerprint, confirmation_id, payload_hash
           FROM portfolio_policies WHERE policy_id='old-policy'"""
    ).fetchone()
    assert dict(old) == {
        "policy_id": "old-policy",
        "request_fingerprint": "old-hash",
        "confirmation_id": None,
        "payload_hash": None,
    }
    indexes = {
        row["name"] for row in conn.execute("PRAGMA index_list(portfolio_policies)")
    }
    assert "idx_portfolio_policies_confirmation_id" in indexes


def test_policy_detail_returns_requested_historical_and_active_versions_read_only():
    conn = db()
    client = client_for(conn)
    first = confirm(client, payload(objective="Historisch")).json()
    second_data = payload(
        objective="Aktiv",
        allocations=[
            {
                "asset_class": "cash",
                "target_pct": "50",
                "lower_pct": "40",
                "upper_pct": "60",
            },
            {
                "asset_class": "equity",
                "target_pct": "50",
                "lower_pct": "40",
                "upper_pct": "60",
            },
        ],
    )
    second = confirm(client, second_data).json()
    before_changes = conn.total_changes
    before_policies = [
        tuple(row)
        for row in conn.execute(
            "SELECT * FROM portfolio_policies ORDER BY version"
        ).fetchall()
    ]
    before_allocations = [
        tuple(row)
        for row in conn.execute(
            "SELECT * FROM portfolio_policy_allocations ORDER BY policy_id, asset_class"
        ).fetchall()
    ]
    before_audits = [
        tuple(row)
        for row in conn.execute("SELECT * FROM audit_log ORDER BY audit_id").fetchall()
    ]

    historical = client.get(f"/api/portfolio/policy/{first['policy_id']}")
    active = client.get(f"/api/portfolio/policy/{second['policy_id']}")
    repeated = client.get(f"/api/portfolio/policy/{first['policy_id']}")

    assert historical.status_code == active.status_code == repeated.status_code == 200
    assert historical.json()["version"] == 1
    assert historical.json()["is_active"] is False
    assert historical.json()["objective"] == "Historisch"
    assert {
        (row["asset_class"], row["target_pct"])
        for row in historical.json()["allocations"]
    } == {("cash", "60"), ("equity", "40")}
    assert active.json()["version"] == 2
    assert active.json()["is_active"] is True
    assert active.json()["previous_policy_id"] == first["policy_id"]
    assert {
        (row["asset_class"], row["target_pct"])
        for row in active.json()["allocations"]
    } == {("cash", "50"), ("equity", "50")}
    assert conn.total_changes == before_changes
    assert [
        tuple(row)
        for row in conn.execute(
            "SELECT * FROM portfolio_policies ORDER BY version"
        ).fetchall()
    ] == before_policies
    assert [
        tuple(row)
        for row in conn.execute(
            "SELECT * FROM portfolio_policy_allocations ORDER BY policy_id, asset_class"
        ).fetchall()
    ] == before_allocations
    assert [
        tuple(row)
        for row in conn.execute("SELECT * FROM audit_log ORDER BY audit_id").fetchall()
    ] == before_audits


def test_policy_detail_404_has_no_identifier_and_exposes_no_write_route():
    conn = db()
    client = client_for(conn)
    missing_id = "policy-secret-do-not-echo"
    response = client.get(f"/api/portfolio/policy/{missing_id}")
    assert response.status_code == 404
    assert response.json() == {"detail": "Policy-Version nicht gefunden"}
    assert missing_id not in response.text

    path = "/api/portfolio/policy/{policy_id}"
    operations = create_app(write_mode="test").openapi()["paths"][path]
    assert set(operations) == {"get"}


def test_policy_openapi_and_write_security_cover_preview_confirm_and_detail():
    conn = db()
    app = create_app(write_mode="disabled")
    app.dependency_overrides[get_db] = lambda: conn
    remote = TestClient(app, client=("100.64.0.10", 50000))
    assert remote.post("/api/portfolio/policy/confirm", json={}).status_code == 403

    schema = app.openapi()
    preview_schema = schema["components"]["schemas"]["PolicyPreviewResponse"]
    confirm_schema = schema["components"]["schemas"]["PolicyConfirmRequest"]
    assert {"preview_id", "confirmation_id"} <= set(preview_schema["required"])
    assert {
        "preview_id", "confirmation_id", "confirm", "allocations",
        "base_currency", "effective_from",
    } <= set(confirm_schema["required"])
    detail_get = schema["paths"]["/api/portfolio/policy/{policy_id}"]["get"]
    assert detail_get["responses"]["200"]["content"]["application/json"]["schema"] == {
        "$ref": "#/components/schemas/PortfolioPolicyResponse"
    }
    assert "404" in detail_get["responses"]


def test_confirm_requires_complete_request_and_explicit_confirmation():
    conn = db()
    client = client_for(conn)
    data = payload()
    identifiers = preview(client, data)
    for omitted in ("preview_id", "confirmation_id", "allocations"):
        request = {**data, **identifiers, "confirm": True}
        request.pop(omitted)
        assert client.post("/api/portfolio/policy/confirm", json=request).status_code == 422
    rejected = client.post(
        "/api/portfolio/policy/confirm",
        json={**data, **identifiers, "confirm": False},
    )
    assert rejected.status_code == 400
    assert rejected.json() == {"detail": "Explizite Bestätigung ist erforderlich"}