"""Cloud-subscription External API — /api/v1/open/cloud/subscriptions.

The cmdb data surface gated by the standalone API-client credential
(``helix_api_``). Pins:

* read scope gates list/get; a client without it → 403
* fields: id / external_id / provider / name / environment / status /
  owner{name,email,department} / budget
* owner = name + work email + primary department (not masked)
* filters provider / environment / status / owner_email / q + pagination
* {key} resolves Helix UUID and cloud-native external id
* only ``helix_api_`` tokens accepted (PAT / svc bearer → 401)
* write scope gates PATCH; read-only client → 403
* PATCH updates owner(by email)/environment/display_name/notes/budget,
  idempotent, audits ``cloud.scope.open_update`` as actor_type=service
  with client IP and no actor_user_id
* write guards: closed → 409, leaver owner → 409, unknown email → 422,
  empty body → 422
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime

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

from app.api_clients.models import ApiClient
from app.api_clients.scopes import (
    SCOPE_CLOUD_SUBSCRIPTIONS_READ,
    SCOPE_CLOUD_SUBSCRIPTIONS_WRITE,
)
from app.api_clients.tokens import ApiClientTokenService
from app.audit.models import AuditLog
from app.domains.cmdb.models import CmdbCloudScope
from app.identity.person_models import IdentityOrgMembership, IdentityPerson
from app.integrations.models import Integration

pytestmark = pytest.mark.asyncio

BASE = "/api/v1/open/cloud/subscriptions"


def _headers(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


async def _mk_person(
    db: AsyncSession, *, email: str, name: str = "Owner", status: str = "active",
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=name,
        employment_status=status,
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:10]}",
    )
    db.add(p)
    await db.flush()
    return p


async def _set_primary_dept(
    db: AsyncSession, person: IdentityPerson, dept_name: str,
    *, org_path: str | None = None,
) -> None:
    db.add(IdentityOrgMembership(
        id=uuid.uuid4(),
        person_id=person.id,
        org_node_source_id=f"dept-{uuid.uuid4().hex[:8]}",
        org_node_type="department",
        org_node_name=dept_name,
        org_path=org_path,
        is_primary=True,
        membership_status="active",
    ))
    await db.flush()


async def _mk_client(db: AsyncSession, *, scopes: list[str]) -> str:
    owner = await _mk_person(db, email=f"co-{uuid.uuid4().hex[:8]}@pingcap.com")
    client = ApiClient(
        id=uuid.uuid4(),
        name="BI Platform",
        slug=f"bi-{uuid.uuid4().hex[:6]}",
        owner_person_id=owner.id,
        scopes=scopes,
    )
    db.add(client)
    await db.flush()
    await db.refresh(client)
    _row, plaintext = await ApiClientTokenService(db).mint(
        client=client, name="t", expires_in_days=None, created_by=None,
    )
    await db.commit()
    return plaintext


async def _mk_scope(
    db: AsyncSession,
    *,
    provider: str = "aws",
    scope_type: str = "aws_account",
    external_id: str | None = None,
    display_name: str = "RD Compute",
    environment: str = "dev",
    status: str = "active",
    owner: IdentityPerson | None = None,
    budget: str | None = None,
    cloud_email: str | None = None,
    account_email_override: str | None = None,
) -> CmdbCloudScope:
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider=provider,
        name=f"{provider}-{uuid.uuid4().hex[:6]}",
        slug=f"{provider}-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db.add(integ)
    await db.flush()
    from decimal import Decimal
    scope = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integ.id),
        scope_type=scope_type,
        external_id=external_id or f"acct-{uuid.uuid4().hex[:10]}",
        display_name=display_name,
        owner_person_id=str(owner.id) if owner else None,
        status=status,
        environment=environment,
        budget_usd_monthly=Decimal(budget) if budget is not None else None,
        cloud_account_email=cloud_email,
        account_email_override=account_email_override,
        first_seen_at=datetime.now(UTC),
        last_synced_at=datetime.now(UTC),
    )
    db.add(scope)
    await db.commit()
    return scope


# ── Read ──────────────────────────────────────────────────────────────


async def test_list_requires_read_scope(client, db_session):
    await _mk_scope(db_session)
    # A client with an unrelated (write-only-impossible) grant: use a
    # client that holds no cloud scope at all is impossible (need ≥1),
    # so grant only write… but write implies read at create. Simulate a
    # no-cloud client by directly seeding scopes=[] is rejected; instead
    # assert the inverse below. Here: a client with read CAN list.
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ])
    r = await client.get(BASE, headers=_headers(token))
    assert r.status_code == 200, r.text


async def test_write_only_grant_is_rejected_at_creation_so_read_always_present():
    # Belt-and-suspenders on the scope rule rather than the HTTP path.
    from app.api_clients.scopes import validate_scopes
    with pytest.raises(ValueError, match="add-on"):
        validate_scopes([SCOPE_CLOUD_SUBSCRIPTIONS_WRITE])


async def test_list_fields_and_owner(client, db_session):
    owner = await _mk_person(db_session, email="zhangsan@pingcap.com", name="Zhang San")
    await _set_primary_dept(
        db_session, owner, "RD Infra",
        org_path="PingCAP / Engineering Group / TiDB / RD Infra",
    )
    await _mk_scope(
        db_session, provider="aws", external_id="123456789012",
        display_name="Prod Account", environment="prod", owner=owner, budget="5000",
    )
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ])
    r = await client.get(BASE, headers=_headers(token))
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["meta"]["total"] == 1
    item = body["data"][0]
    assert item["external_id"] == "123456789012"
    assert item["provider"] == "aws"
    assert item["name"] == "Prod Account"
    assert item["environment"] == "prod"
    assert item["status"] == "active"
    assert item["budget_usd_monthly"] == 5000.0
    # Full department chain, tenant root "PingCAP" stripped; leaf name
    # unchanged under `department`.
    assert item["owner"] == {
        "name": "Zhang San", "email": "zhangsan@pingcap.com", "department": "RD Infra",
        "department_full_path": "Engineering Group / TiDB / RD Infra",
        "department_path_segments": ["Engineering Group", "TiDB", "RD Infra"],
    }


@pytest.mark.asyncio
async def test_owner_department_path_null_when_no_membership(client, db_session):
    """Owner present but no primary org membership → path fields degrade to
    null / [] (not absent), and `department` stays null too."""
    owner = await _mk_person(db_session, email="nopath@pingcap.com", name="No Path")
    await _mk_scope(db_session, owner=owner, external_id="999999999999")
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ])
    r = await client.get(BASE, headers=_headers(token))
    assert r.status_code == 200, r.text
    item = r.json()["data"][0]
    assert item["owner"]["department"] is None
    assert item["owner"]["department_full_path"] is None
    assert item["owner"]["department_path_segments"] == []


async def test_list_filters_and_pagination(client, db_session):
    await _mk_scope(db_session, provider="aws", environment="dev")
    await _mk_scope(db_session, provider="gcp", scope_type="gcp_project", environment="prod")
    await _mk_scope(db_session, provider="gcp", scope_type="gcp_project", environment="dev")
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ])

    r = await client.get(BASE, headers=_headers(token), params={"provider": "gcp"})
    assert r.json()["meta"]["total"] == 2
    r = await client.get(BASE, headers=_headers(token),
                         params={"provider": "gcp", "environment": "prod"})
    assert r.json()["meta"]["total"] == 1
    r = await client.get(BASE, headers=_headers(token), params={"page_size": 1})
    body = r.json()
    assert len(body["data"]) == 1
    assert body["meta"]["page_size"] == 1 and body["meta"]["total"] == 3
    r = await client.get(BASE, headers=_headers(token), params={"environment": "qa"})
    assert r.status_code == 422


async def test_list_filter_owner_email_and_keyword(client, db_session):
    owner = await _mk_person(db_session, email="alice@pingcap.com", name="Alice")
    await _mk_scope(db_session, owner=owner, display_name="Analytics Prod",
                    external_id="acct-analytics-1")
    await _mk_scope(db_session, display_name="Marketing Sandbox", external_id="acct-mkt-9")
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ])

    r = await client.get(BASE, headers=_headers(token),
                         params={"owner_email": "Alice@pingcap.com"})
    assert r.json()["meta"]["total"] == 1
    r = await client.get(BASE, headers=_headers(token),
                         params={"owner_email": "nobody@pingcap.com"})
    assert r.status_code == 200 and r.json()["meta"]["total"] == 0
    r = await client.get(BASE, headers=_headers(token), params={"q": "analytics"})
    assert r.json()["meta"]["total"] == 1
    r = await client.get(BASE, headers=_headers(token), params={"q": "acct-mkt"})
    assert r.json()["meta"]["total"] == 1


async def test_get_by_uuid_and_external_id(client, db_session):
    scope = await _mk_scope(db_session, external_id="987654321098")
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ])
    r = await client.get(f"{BASE}/{scope.id}", headers=_headers(token))
    assert r.status_code == 200 and r.json()["data"]["external_id"] == "987654321098"
    r = await client.get(f"{BASE}/987654321098", headers=_headers(token))
    assert r.status_code == 200 and r.json()["data"]["id"] == str(scope.id)
    r = await client.get(f"{BASE}/does-not-exist", headers=_headers(token))
    assert r.status_code == 404


async def test_non_api_client_token_rejected(client, db_session):
    await _mk_scope(db_session)
    # PAT-prefixed, svc-prefixed, garbage, and no header all → 401.
    for bad in ("helix_pat_deadbeefdeadbeef", "helix_svc_deadbeefdeadbeef", "garbage"):
        r = await client.get(BASE, headers=_headers(bad))
        assert r.status_code == 401, (bad, r.text)
    r = await client.get(BASE)
    assert r.status_code == 401


# ── Write ─────────────────────────────────────────────────────────────


async def test_write_requires_write_scope(client, db_session):
    scope = await _mk_scope(db_session)
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ])
    r = await client.patch(f"{BASE}/{scope.id}", headers=_headers(token),
                           json={"environment": "prod"})
    assert r.status_code == 403, r.text


async def test_write_updates_fields_and_audits(client, db_session):
    scope = await _mk_scope(db_session, environment="dev", display_name="old", budget=None)
    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ, SCOPE_CLOUD_SUBSCRIPTIONS_WRITE],
    )
    r = await client.patch(
        f"{BASE}/{scope.id}", headers=_headers(token),
        json={"environment": "staging", "display_name": "new name",
              "notes": "managed by partner X", "budget_usd_monthly": 1200},
    )
    assert r.status_code == 200, r.text
    data = r.json()["data"]
    assert data["environment"] == "staging"
    assert data["name"] == "new name"
    assert data["budget_usd_monthly"] == 1200.0

    rows = (await db_session.execute(
        select(AuditLog).where(AuditLog.action == "cloud.scope.open_update")
    )).scalars().all()
    assert len(rows) == 1
    audit = rows[0]
    assert audit.actor_type == "service"
    assert audit.actor_user_id is None
    assert audit.resource_id == str(scope.id)
    assert "client_slug" in audit.detail
    assert "environment" in audit.detail["changes"]
    assert audit.ip_address is not None


async def test_write_is_idempotent(client, db_session):
    scope = await _mk_scope(db_session, environment="dev")
    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ, SCOPE_CLOUD_SUBSCRIPTIONS_WRITE],
    )
    payload = {"environment": "prod"}
    assert (await client.patch(f"{BASE}/{scope.id}", headers=_headers(token), json=payload)).status_code == 200
    assert (await client.patch(f"{BASE}/{scope.id}", headers=_headers(token), json=payload)).status_code == 200
    rows = (await db_session.execute(
        select(AuditLog).where(AuditLog.action == "cloud.scope.open_update")
    )).scalars().all()
    assert len(rows) == 1


async def test_write_owner_by_email(client, db_session):
    scope = await _mk_scope(db_session)
    await _mk_person(db_session, email="newowner@pingcap.com", name="New Owner")
    await db_session.commit()
    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ, SCOPE_CLOUD_SUBSCRIPTIONS_WRITE],
    )
    r = await client.patch(f"{BASE}/{scope.id}", headers=_headers(token),
                           json={"owner_email": "NewOwner@pingcap.com"})
    assert r.status_code == 200, r.text
    assert r.json()["data"]["owner"]["email"] == "newowner@pingcap.com"
    r = await client.patch(f"{BASE}/{scope.id}", headers=_headers(token),
                           json={"owner_email": "ghost@pingcap.com"})
    assert r.status_code == 422


async def test_write_owner_leaver_rejected(client, db_session):
    scope = await _mk_scope(db_session)
    await _mk_person(db_session, email="leaver@pingcap.com", name="Leaver", status="left")
    await db_session.commit()
    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ, SCOPE_CLOUD_SUBSCRIPTIONS_WRITE],
    )
    r = await client.patch(f"{BASE}/{scope.id}", headers=_headers(token),
                           json={"owner_email": "leaver@pingcap.com"})
    assert r.status_code == 409, r.text


async def test_write_closed_scope_rejected(client, db_session):
    scope = await _mk_scope(db_session, status="closed")
    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ, SCOPE_CLOUD_SUBSCRIPTIONS_WRITE],
    )
    r = await client.patch(f"{BASE}/{scope.id}", headers=_headers(token),
                           json={"environment": "prod"})
    assert r.status_code == 409, r.text


async def test_write_empty_body_rejected(client, db_session):
    scope = await _mk_scope(db_session)
    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ, SCOPE_CLOUD_SUBSCRIPTIONS_WRITE],
    )
    r = await client.patch(f"{BASE}/{scope.id}", headers=_headers(token), json={})
    assert r.status_code == 422, r.text


# ── Account email (migration 450) ────────────────────────────────────


async def test_read_exposes_account_email_fields(client, db_session):
    """The cloud-synced email is surfaced; with no override, the
    effective ``account_email`` equals ``cloud_account_email``."""
    await _mk_scope(
        db_session, external_id="123456789012",
        cloud_email="aws-rd@pingcap.com",
    )
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ])
    r = await client.get(f"{BASE}/123456789012", headers=_headers(token))
    assert r.status_code == 200, r.text
    data = r.json()["data"]
    assert data["cloud_account_email"] == "aws-rd@pingcap.com"
    assert data["account_email_override"] is None
    assert data["account_email"] == "aws-rd@pingcap.com"


async def test_read_effective_email_prefers_override(client, db_session):
    await _mk_scope(
        db_session, external_id="223456789012",
        cloud_email="cloud@pingcap.com",
        account_email_override="billing@pingcap.com",
    )
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ])
    r = await client.get(f"{BASE}/223456789012", headers=_headers(token))
    data = r.json()["data"]
    assert data["cloud_account_email"] == "cloud@pingcap.com"
    assert data["account_email_override"] == "billing@pingcap.com"
    assert data["account_email"] == "billing@pingcap.com"


async def test_write_sets_and_clears_email_override(client, db_session):
    scope = await _mk_scope(db_session, cloud_email="cloud@pingcap.com")
    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ, SCOPE_CLOUD_SUBSCRIPTIONS_WRITE],
    )
    # Set override — effective flips, cloud value preserved.
    r = await client.patch(
        f"{BASE}/{scope.id}", headers=_headers(token),
        json={"account_email_override": "override@pingcap.com"},
    )
    assert r.status_code == 200, r.text
    data = r.json()["data"]
    assert data["account_email"] == "override@pingcap.com"
    assert data["cloud_account_email"] == "cloud@pingcap.com"
    assert data["account_email_override"] == "override@pingcap.com"

    # Clear via empty string — reverts to the cloud value.
    r = await client.patch(
        f"{BASE}/{scope.id}", headers=_headers(token),
        json={"account_email_override": ""},
    )
    assert r.status_code == 200, r.text
    data = r.json()["data"]
    assert data["account_email_override"] is None
    assert data["account_email"] == "cloud@pingcap.com"


async def test_write_invalid_email_override_rejected(client, db_session):
    scope = await _mk_scope(db_session)
    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ, SCOPE_CLOUD_SUBSCRIPTIONS_WRITE],
    )
    r = await client.patch(
        f"{BASE}/{scope.id}", headers=_headers(token),
        json={"account_email_override": "not-an-email"},
    )
    assert r.status_code == 422, r.text