"""Integration tests for the cloud-efficiency surface — RBAC, list/summary
endpoints, rule-threshold editing, and the end-to-end compute path
(metrics_daily + cost_facts → efficiency rows)."""
from __future__ import annotations

import uuid
from datetime import UTC, date, datetime, timedelta

import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.efficiency import compute_efficiency_for_integration
from app.domains.cmdb.models import (
    CmdbCloudScope,
    CmdbCostFact,
    CmdbServer,
    CmdbServerEfficiency,
    CmdbServerMetricsDaily,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration, PlatformSetting


async def _integration(db: AsyncSession, provider: str = "aws") -> Integration:
    integ = Integration(
        id=uuid.uuid4(), kind="cloud", provider=provider,
        name=f"{provider}-eff", slug=f"{provider}-{uuid.uuid4().hex[:6]}",
        enabled=True, is_primary=False, capabilities=["servers", "metrics"],
    )
    db.add(integ)
    await db.flush()
    return integ


async def _person(db: AsyncSession, email: str) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(), work_email=email, display_name=email.split("@")[0],
        employment_status="active", source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db.add(p)
    await db.flush()
    return p


async def _server(db: AsyncSession, integ: Integration, **kw) -> CmdbServer:
    now = datetime.now(UTC)
    d = {
        "id": uuid.uuid4(),
        "instance_id": kw.pop("instance_id", f"i-{uuid.uuid4().hex[:10]}"),
        "name": kw.pop("name", "srv"),
        "status": kw.pop("status", "running"),
        "region": kw.pop("region", "us-east-1"),
        "integration_id": str(integ.id),
        "first_seen_at": now - timedelta(days=100),
        "last_synced_at": now,
    }
    d.update(kw)
    s = CmdbServer(**d)
    db.add(s)
    await db.flush()
    return s


async def _efficiency(db: AsyncSession, integ: Integration, server: CmdbServer, **kw):
    d = {
        "id": uuid.uuid4(),
        "server_id": str(server.id),
        "integration_id": str(integ.id),
        "verdict": kw.pop("verdict", "healthy"),
        "recommendation": kw.pop("recommendation", "keep"),
        "computed_at": datetime.now(UTC),
    }
    d.update(kw)
    e = CmdbServerEfficiency(**d)
    db.add(e)
    await db.flush()
    return e


async def _metric_day(db, integ, server, day, **kw):
    d = {
        "id": uuid.uuid4(),
        "server_id": str(server.id),
        "integration_id": str(integ.id),
        "metric_date": day,
        "sample_count": kw.pop("sample_count", 24),
    }
    d.update(kw)
    m = CmdbServerMetricsDaily(**d)
    db.add(m)
    await db.flush()
    return m


async def _cost_fact(
    db, integ, resource_id, amount, day, *, currency="USD", amount_native=None,
):
    f = CmdbCostFact(
        fact_key=uuid.uuid4().hex,
        date=day,
        integration_id=str(integ.id),
        resource_id=resource_id,
        resource_type="Instance",
        service="AmazonEC2",
        amount_usd=amount,
        amount_native=amount if amount_native is None else amount_native,
        currency=currency,
    )
    db.add(f)
    await db.flush()
    return f


# ── RBAC + summary ───────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_summary_admin_sees_rollup(
    client: AsyncClient, db_session: AsyncSession, cloud_admin_headers,
):
    integ = await _integration(db_session)
    s1 = await _server(db_session, integ, name="idle-box")
    s2 = await _server(db_session, integ, name="busy-box")
    await _efficiency(
        db_session, integ, s1, verdict="idle", recommendation="stop",
        estimated_monthly_savings_usd=120, monthly_cost_usd=120,
    )
    await _efficiency(db_session, integ, s2, verdict="busy", recommendation="keep")
    await db_session.commit()

    r = await client.get("/api/v1/cmdb/efficiency/summary", headers=cloud_admin_headers)
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["totals"]["servers"] == 2
    assert data["totals"]["with_recommendation"] == 1
    assert float(data["totals"]["estimated_monthly_savings_usd"]) == 120.0
    verdicts = {row["verdict"]: row["count"] for row in data["by_verdict"]}
    assert verdicts.get("idle") == 1 and verdicts.get("busy") == 1


@pytest.mark.asyncio
async def test_instances_filter_by_verdict(
    client: AsyncClient, db_session: AsyncSession, cloud_admin_headers,
):
    integ = await _integration(db_session)
    s1 = await _server(db_session, integ)
    s2 = await _server(db_session, integ)
    await _efficiency(db_session, integ, s1, verdict="idle", recommendation="stop",
                      estimated_monthly_savings_usd=50)
    await _efficiency(db_session, integ, s2, verdict="healthy")
    await db_session.commit()

    r = await client.get(
        "/api/v1/cmdb/efficiency/instances?verdict=idle",
        headers=cloud_admin_headers,
    )
    assert r.status_code == 200
    items = r.json()["data"]
    assert len(items) == 1
    assert items[0]["verdict"] == "idle"


@pytest.mark.asyncio
async def test_employee_scoped_to_owned_servers(
    client: AsyncClient, db_session: AsyncSession, employee_user, employee_headers,
):
    person = employee_user[0] if isinstance(employee_user, tuple) else employee_user
    integ = await _integration(db_session)
    mine = await _server(db_session, integ, owner_person_id=str(person.id))
    other = await _server(db_session, integ, owner_person_id=None)
    await _efficiency(db_session, integ, mine, verdict="idle", recommendation="stop",
                      estimated_monthly_savings_usd=30)
    await _efficiency(db_session, integ, other, verdict="idle", recommendation="stop",
                      estimated_monthly_savings_usd=999)
    await db_session.commit()

    r = await client.get("/api/v1/cmdb/efficiency/instances", headers=employee_headers)
    assert r.status_code == 200
    items = r.json()["data"]
    server_ids = {i["server_id"] for i in items}
    assert str(mine.id) in server_ids
    assert str(other.id) not in server_ids

    # Detail of a server they don't own → 404 (same as non-existent, so the
    # endpoint isn't an existence oracle; see efficiency_router SEC fix).
    r2 = await client.get(
        f"/api/v1/cmdb/efficiency/instances/{other.id}", headers=employee_headers,
    )
    assert r2.status_code == 404


@pytest.mark.asyncio
async def test_instance_detail_includes_series(
    client: AsyncClient, db_session: AsyncSession, cloud_admin_headers,
):
    integ = await _integration(db_session)
    s = await _server(db_session, integ)
    await _efficiency(db_session, integ, s, verdict="oversized", recommendation="downsize")
    today = date.today()
    for i in range(3):
        await _metric_day(
            db_session, integ, s, today - timedelta(days=i),
            cpu_pct_avg=10.0, cpu_pct_p95=15.0, cpu_pct_max=20.0,
        )
    await db_session.commit()

    r = await client.get(
        f"/api/v1/cmdb/efficiency/instances/{s.id}", headers=cloud_admin_headers,
    )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["verdict"] == "oversized"
    assert len(data["metrics_series"]) == 3


# ── neutralized boxes / unattributed count / owner surface ───────────
@pytest.mark.asyncio
async def test_summary_excludes_neutralized_stopped_box(
    client: AsyncClient, db_session: AsyncSession, cloud_admin_headers,
):
    """An idle/oversized box neutralized to recommendation='none' (already
    stopped) keeps its verdict for visibility but must NOT inflate the
    可优化机器/待处理 counts."""
    integ = await _integration(db_session)
    live = await _server(db_session, integ, name="live-idle")
    dead = await _server(db_session, integ, name="stopped-idle", status="stopped")
    await _efficiency(
        db_session, integ, live, verdict="idle", recommendation="stop",
        estimated_monthly_savings_usd=40, monthly_cost_usd=40,
    )
    await _efficiency(
        db_session, integ, dead, verdict="idle", recommendation="none",
        estimated_monthly_savings_usd=0, monthly_cost_usd=40,
    )
    await db_session.commit()

    r = await client.get("/api/v1/cmdb/efficiency/summary", headers=cloud_admin_headers)
    data = r.json()["data"]
    # Both idle in the donut, but only the live one is actionable/pending.
    assert data["totals"]["servers"] == 2
    assert data["totals"]["with_recommendation"] == 1
    assert data["totals"]["pending"] == 1

    # The neutralized row is still listed, but flagged non-actionable.
    r = await client.get(
        "/api/v1/cmdb/efficiency/instances?verdict=idle",
        headers=cloud_admin_headers,
    )
    by_name = {i["name"]: i for i in r.json()["data"]}
    assert by_name["live-idle"]["actionable"] is True
    assert by_name["stopped-idle"]["actionable"] is False


@pytest.mark.asyncio
async def test_summary_counts_unattributed_cost(
    client: AsyncClient, db_session: AsyncSession, cloud_admin_headers,
):
    """Actionable machines with no attributed per-instance cost (savings
    NULL) are counted in cost_unattributed so the UI can say '未归属成本'
    instead of implying $0."""
    integ = await _integration(db_session)
    priced = await _server(db_session, integ)
    unpriced = await _server(db_session, integ)
    await _efficiency(
        db_session, integ, priced, verdict="idle", recommendation="stop",
        estimated_monthly_savings_usd=30, monthly_cost_usd=30,
    )
    await _efficiency(
        db_session, integ, unpriced, verdict="oversized", recommendation="downsize",
        estimated_monthly_savings_usd=None, monthly_cost_usd=None,
    )
    await db_session.commit()

    r = await client.get("/api/v1/cmdb/efficiency/summary", headers=cloud_admin_headers)
    data = r.json()["data"]
    assert data["totals"]["with_recommendation"] == 2
    assert data["totals"]["cost_unattributed"] == 1


@pytest.mark.asyncio
async def test_instances_filter_by_owner_stored_and_inherited(
    client: AsyncClient, db_session: AsyncSession, cloud_admin_headers,
):
    """The owner filter matches both a stored server owner AND the owner
    inherited from the server's cloud subscription."""
    integ = await _integration(db_session)
    alice = await _person(db_session, f"alice-{uuid.uuid4().hex[:6]}@x.com")
    bob = await _person(db_session, f"bob-{uuid.uuid4().hex[:6]}@x.com")

    # Alice owns s1 directly.
    s1 = await _server(
        db_session, integ, name="alice-box", owner_person_id=str(alice.id),
    )
    await _efficiency(db_session, integ, s1, verdict="idle", recommendation="stop",
                      estimated_monthly_savings_usd=10, monthly_cost_usd=10)
    # s2 has no owner but is attributed (cost fact) to a scope Bob owns.
    s2 = await _server(db_session, integ, name="bob-inherited", instance_id="i-bobs00001")
    scope = CmdbCloudScope(
        id=uuid.uuid4(), integration_id=str(integ.id), scope_type="aws_account",
        external_id=f"acct-{uuid.uuid4().hex[:8]}", display_name="bob-sub",
        owner_person_id=str(bob.id), environment="dev",
    )
    db_session.add(scope)
    await db_session.flush()
    await _cost_fact(db_session, integ, "i-bobs00001", 20, date.today())
    # Point that cost fact at Bob's scope (the inherited-owner link).
    row = (await db_session.execute(
        select(CmdbCostFact).where(CmdbCostFact.resource_id == "i-bobs00001")
    )).scalar_one()
    row.scope_id = str(scope.id)
    await _efficiency(db_session, integ, s2, verdict="idle", recommendation="stop",
                      estimated_monthly_savings_usd=20, monthly_cost_usd=20)
    await db_session.commit()

    r = await client.get(
        f"/api/v1/cmdb/efficiency/instances?owner_person_id={alice.id}",
        headers=cloud_admin_headers,
    )
    names = {i["name"] for i in r.json()["data"]}
    assert names == {"alice-box"}

    r = await client.get(
        f"/api/v1/cmdb/efficiency/instances?owner_person_id={bob.id}",
        headers=cloud_admin_headers,
    )
    got = r.json()["data"]
    assert {i["name"] for i in got} == {"bob-inherited"}
    # The row surfaces the inherited owner PERSON, not just the subscription.
    assert got[0]["effective_owner_label"] == bob.display_name
    assert got[0]["owner_inherited"] is True


@pytest.mark.asyncio
async def test_instance_row_resolves_stored_owner_person(
    client: AsyncClient, db_session: AsyncSession, cloud_admin_headers,
):
    """A stored-owner row exposes the owner person's label (who to contact),
    not just an owner_person_id — previously the 归属 column showed '—'."""
    integ = await _integration(db_session)
    alice = await _person(db_session, f"alice-{uuid.uuid4().hex[:6]}@x.com")
    s = await _server(db_session, integ, owner_person_id=str(alice.id))
    await _efficiency(db_session, integ, s, verdict="idle", recommendation="stop",
                      estimated_monthly_savings_usd=15, monthly_cost_usd=15)
    await db_session.commit()

    r = await client.get(
        f"/api/v1/cmdb/efficiency/instances/{s.id}", headers=cloud_admin_headers,
    )
    data = r.json()["data"]
    assert data["effective_owner_label"] == alice.display_name
    assert data["effective_owner_person_id"] == str(alice.id)
    assert data["owner_inherited"] is False


# ── rule thresholds ──────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_rules_put_requires_manage(
    client: AsyncClient, db_session: AsyncSession, employee_headers,
):
    r = await client.put(
        "/api/v1/cmdb/efficiency/rules",
        json={"idle_cpu_p95_max": 3.0}, headers=employee_headers,
    )
    assert r.status_code == 403


@pytest.mark.asyncio
async def test_rules_put_persists(
    client: AsyncClient, db_session: AsyncSession, cloud_admin_headers,
):
    r = await client.put(
        "/api/v1/cmdb/efficiency/rules",
        json={"idle_cpu_p95_max": 3.0, "downsize_savings_ratio": 0.4},
        headers=cloud_admin_headers,
    )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["idle_cpu_p95_max"] == 3.0
    assert data["downsize_savings_ratio"] == 0.4

    row = await db_session.get(PlatformSetting, "cloud_efficiency.rules")
    assert row is not None
    assert row.value["idle_cpu_p95_max"] == 3.0


# ── end-to-end compute ───────────────────────────────────────────────
@pytest.mark.asyncio
async def test_compute_idle_with_cost(
    db_session: AsyncSession,
):
    integ = await _integration(db_session)
    s = await _server(db_session, integ, instance_id="i-idle0001aa")
    today = date.today()
    # 10 days of dead-idle CPU + silent network.
    for i in range(10):
        await _metric_day(
            db_session, integ, s, today - timedelta(days=i),
            cpu_pct_avg=1.0, cpu_pct_p95=2.0, cpu_pct_max=3.0,
            net_in_bytes_avg=100, net_out_bytes_avg=100, sample_count=24,
        )
    # $90 over the trailing 30d window → normalized to a ~$90 month.
    await _cost_fact(db_session, integ, "i-idle0001aa", 90, today - timedelta(days=1))
    await db_session.commit()

    result = await compute_efficiency_for_integration(db_session, str(integ.id), today=today)
    assert result["computed"] == 1

    row = (
        await db_session.execute(
            select(CmdbServerEfficiency).where(
                CmdbServerEfficiency.server_id == str(s.id)
            )
        )
    ).scalar_one()
    assert row.verdict == "idle"
    assert row.recommendation == "stop"
    assert float(row.monthly_cost_usd) == 90.0
    assert float(row.estimated_monthly_savings_usd) == 90.0
    assert row.cost_currency == "USD"


@pytest.mark.asyncio
async def test_compute_stopped_instance_not_recommended(db_session: AsyncSession):
    # A stopped box with idle metrics earlier in the window must NOT be told
    # to "stop" (no-op) nor count phantom savings.
    integ = await _integration(db_session)
    s = await _server(db_session, integ, instance_id="i-stopped01", status="stopped")
    today = date.today()
    for i in range(10):
        await _metric_day(
            db_session, integ, s, today - timedelta(days=i),
            cpu_pct_avg=1.0, cpu_pct_p95=2.0, cpu_pct_max=3.0,
            net_in_bytes_avg=100, net_out_bytes_avg=100, sample_count=24,
        )
    await _cost_fact(db_session, integ, "i-stopped01", 90, today - timedelta(days=1))
    await db_session.commit()

    await compute_efficiency_for_integration(db_session, str(integ.id), today=today, narrate=False)
    row = (
        await db_session.execute(
            select(CmdbServerEfficiency).where(
                CmdbServerEfficiency.server_id == str(s.id)
            )
        )
    ).scalar_one()
    assert row.recommendation == "none"
    assert float(row.estimated_monthly_savings_usd) == 0.0
    assert "instance_not_running" in (row.verdict_reasons or [])


@pytest.mark.asyncio
async def test_compute_insufficient_data(db_session: AsyncSession):
    integ = await _integration(db_session)
    s = await _server(db_session, integ)
    today = date.today()
    # Only 1 day of samples → insufficient_data.
    await _metric_day(
        db_session, integ, s, today, cpu_pct_avg=1.0, cpu_pct_p95=1.0,
        cpu_pct_max=1.0, sample_count=24,
    )
    await db_session.commit()

    await compute_efficiency_for_integration(db_session, str(integ.id), today=today)
    row = (
        await db_session.execute(
            select(CmdbServerEfficiency).where(
                CmdbServerEfficiency.server_id == str(s.id)
            )
        )
    ).scalar_one()
    assert row.verdict == "insufficient_data"


async def _idle_metrics(db, integ, s, today, days=10):
    for i in range(days):
        await _metric_day(
            db, integ, s, today - timedelta(days=i),
            cpu_pct_avg=1.0, cpu_pct_p95=2.0, cpu_pct_max=3.0,
            net_in_bytes_avg=100, net_out_bytes_avg=100, sample_count=24,
        )


@pytest.mark.asyncio
async def test_compute_unconverted_native_cost_is_unattributed(
    db_session: AsyncSession,
):
    """未换算 guard: a resource whose only spend is unconverted native
    (amount_usd=0, amount_native>0, currency!=USD) must be UNATTRIBUTED —
    monthly_cost NULL, not an attributed $0 mislabeled as USD."""
    integ = await _integration(db_session)
    s = await _server(db_session, integ, instance_id="i-cny00001aa")
    today = date.today()
    await _idle_metrics(db_session, integ, s, today)
    # Unconverted CNY line: no USD value, real native spend.
    await _cost_fact(
        db_session, integ, "i-cny00001aa", 0, today - timedelta(days=1),
        currency="CNY", amount_native=650,
    )
    await db_session.commit()

    await compute_efficiency_for_integration(
        db_session, str(integ.id), today=today, narrate=False,
    )
    row = (
        await db_session.execute(
            select(CmdbServerEfficiency).where(
                CmdbServerEfficiency.server_id == str(s.id)
            )
        )
    ).scalar_one()
    assert row.verdict == "idle"                 # utilization still renders
    assert row.monthly_cost_usd is None          # not an attributed $0
    assert row.estimated_monthly_savings_usd is None
    assert row.cost_currency is None


@pytest.mark.asyncio
async def test_compute_converted_foreign_currency_stamps_real_currency(
    db_session: AsyncSession,
):
    """A CNY-billed box that WAS converted keeps its USD figure but is
    labeled with its real billing currency, not an unconditional 'USD'."""
    integ = await _integration(db_session)
    s = await _server(db_session, integ, instance_id="i-cny00002bb")
    today = date.today()
    await _idle_metrics(db_session, integ, s, today)
    await _cost_fact(
        db_session, integ, "i-cny00002bb", 90, today - timedelta(days=1),
        currency="CNY", amount_native=650,
    )
    await db_session.commit()

    await compute_efficiency_for_integration(
        db_session, str(integ.id), today=today, narrate=False,
    )
    row = (
        await db_session.execute(
            select(CmdbServerEfficiency).where(
                CmdbServerEfficiency.server_id == str(s.id)
            )
        )
    ).scalar_one()
    assert float(row.monthly_cost_usd) == 90.0
    assert row.cost_currency == "CNY"


@pytest.mark.asyncio
async def test_compute_name_collision_does_not_steal_cost(
    db_session: AsyncSession,
):
    """A server's display NAME must not exact-match another resource's real
    cost resource_id and steal its spend (name is suffix-phase only)."""
    integ = await _integration(db_session)
    today = date.today()
    # Resource B: its real cost resource_id is literally "webserver01".
    b = await _server(db_session, integ, instance_id="webserver01", name="b-box")
    # Resource A: DISPLAY NAME collides with B's cost id; own id has no cost.
    a = await _server(
        db_session, integ, instance_id="i-aaaa1111bb", name="webserver01",
    )
    for srv in (a, b):
        await _idle_metrics(db_session, integ, srv, today)
    await _cost_fact(db_session, integ, "webserver01", 120, today - timedelta(days=1))
    await db_session.commit()

    await compute_efficiency_for_integration(
        db_session, str(integ.id), today=today, narrate=False,
    )
    rows = {
        r.server_id: r
        for r in (
            await db_session.execute(
                select(CmdbServerEfficiency).where(
                    CmdbServerEfficiency.server_id.in_([str(a.id), str(b.id)])
                )
            )
        ).scalars().all()
    }
    # B keeps its own cost; A did NOT steal it via the name collision.
    assert float(rows[str(b.id)].monthly_cost_usd) == 120.0
    assert rows[str(a.id)].monthly_cost_usd is None