"""Tests for the certificate expiry-alert path after the unified-engine
rewrite (cert-expiry-governance PR2):

* the kill switch still short-circuits ``run_cert_expiry_alert``;
* ``build_cert_expiry_items`` collapses the many probed/cloud rows of one
  logical cert to a SINGLE item keyed on covered-name set, warns about the
  SOONEST expiry, resolves ownership live (DNS record → domain → network
  fallback), and survives a naive ``first_seen_at``;
* the builder output drives the engine end-to-end (owner gets escalated in
  the last-mile window).

The graduated cadence / escalation / first-enable suppression / ack state
machine itself is covered by ``test_expiry_alert_engine.py``; owner
resolution by ``test_cert_owner_resolver.py``.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

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

from app.domains.cmdb.expiry_alert_engine import (
    build_cert_expiry_items,
    build_domain_expiry_items,
    run_expiry_alerts,
)
from app.domains.cmdb.models import (
    CERT_SOURCE_CLOUD,
    CERT_SOURCE_PROBED,
    EXPIRY_RESOURCE_CERTIFICATE,
    CmdbCertificate,
    CmdbDnsRecord,
    CmdbDomain,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration, PlatformSetting
from app.scheduled_jobs.handlers import run_cert_expiry_alert


class _Recorder:
    def __init__(self):
        self.calls = []

    async def __call__(self, item, recipients, bucket):
        self.calls.append((item.label, bucket, sorted(r.role for r in recipients)))
        return True


@pytest_asyncio.fixture
async def integ(db_session: AsyncSession) -> Integration:
    i = Integration(
        id=uuid.uuid4(), kind="cloud", provider="aliyun",
        name="pytest-cert-exp", slug=f"cert-exp-{uuid.uuid4().hex[:8]}",
        enabled=True, is_primary=False, capabilities=["domains", "dns_records"],
    )
    db_session.add(i)
    await db_session.commit()
    return i


async def _person(db, name, *, manager_id=None) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(), source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", display_name=name,
        work_email=f"{name.lower()}-{uuid.uuid4().hex[:6]}@pingcap.com",
        employment_status="active", manager_person_id=manager_id,
    )
    db.add(p)
    await db.flush()
    return p


@pytest.mark.asyncio
async def test_kill_switch_off_skips(db_session: AsyncSession):
    db_session.add(PlatformSetting(
        key="network.user_notifications_enabled", value=False,
    ))
    await db_session.commit()
    res = await run_cert_expiry_alert(db_session, {})
    assert res["status"] == "skipped"
    assert "user_notifications_enabled" in res["reason"]


@pytest.mark.asyncio
async def test_builder_collapses_dedup_soonest_and_resolves_owner(
    db_session: AsyncSession, integ: Integration,
):
    rec_owner = await _person(db_session, "RecOwner")
    db_session.add(CmdbDomain(
        id=uuid.uuid4(), integration_id=integ.id, domain_name="example.com",
        owner_person_id=str(rec_owner.id),
    ))
    db_session.add(CmdbDnsRecord(
        integration_id=integ.id, zone_name="example.com", record_id="r1",
        name="svc", record_type="A", value="1.2.3.4", status="enable",
        owner_person_id=str(rec_owner.id),
    ))
    soon = datetime.now(UTC) + timedelta(days=3)
    later = datetime.now(UTC) + timedelta(days=40)
    # Two rows of ONE logical cert (same covered names): a probed row
    # (soonest) + a cloud row (later expiry). Must collapse to one item
    # warning about the SOONEST.
    db_session.add(CmdbCertificate(
        id=uuid.uuid4(), dedup_key="probed:svc.example.com:443",
        source=CERT_SOURCE_PROBED, probe_host="svc.example.com",
        common_name="svc.example.com", sans=["svc.example.com"],
        not_after=soon, status="expiring", first_seen_at=datetime.now(UTC),
    ))
    db_session.add(CmdbCertificate(
        id=uuid.uuid4(), dedup_key=f"{CERT_SOURCE_CLOUD}:{integ.id}:c1",
        source=CERT_SOURCE_CLOUD, integration_id=str(integ.id),
        common_name="svc.example.com", sans=["svc.example.com"],
        not_after=later, status="valid", first_seen_at=datetime.now(UTC),
    ))
    await db_session.commit()

    items = await build_cert_expiry_items(db_session)
    mine = [it for it in items if it.label == "svc.example.com"]
    assert len(mine) == 1, "probed + cloud rows must collapse to one item"
    assert mine[0].expiry_date == soon.date(), "soonest expiry wins"
    assert mine[0].primary_owner_id == str(rec_owner.id)


@pytest.mark.asyncio
async def test_builder_naive_first_seen_does_not_crash(
    db_session: AsyncSession, integ: Integration,
):
    # ``first_seen_at`` written naive (asyncmy strips tzinfo on fresh load).
    db_session.add(CmdbCertificate(
        id=uuid.uuid4(), dedup_key="probed:naive.example.com:443",
        source=CERT_SOURCE_PROBED, probe_host="naive.example.com",
        common_name="naive.example.com",
        not_after=datetime.now(UTC) + timedelta(days=2), status="expiring",
        first_seen_at=datetime.utcnow(),  # noqa: DTZ003 — intentionally naive
    ))
    await db_session.commit()
    items = await build_cert_expiry_items(db_session)  # must not raise
    assert any(it.label == "naive.example.com" for it in items)


@pytest.mark.asyncio
async def test_builder_drives_engine_escalation(
    db_session: AsyncSession, integ: Integration,
):
    lead = await _person(db_session, "Lead")
    owner = await _person(db_session, "Owner", manager_id=str(lead.id))
    db_session.add(CmdbDomain(
        id=uuid.uuid4(), integration_id=integ.id, domain_name="esc.example.com",
        owner_person_id=str(owner.id),
    ))
    db_session.add(CmdbDnsRecord(
        integration_id=integ.id, zone_name="esc.example.com", record_id="r1",
        name="app", record_type="A", value="1.2.3.4", status="enable",
        owner_person_id=str(owner.id),
    ))
    # Fresh + already inside the ≤7d window → fresh-urgent emit on first
    # sight, escalating to owner + lead (no separate domain owner here).
    db_session.add(CmdbCertificate(
        id=uuid.uuid4(), dedup_key="probed:app.esc.example.com:443",
        source=CERT_SOURCE_PROBED, probe_host="app.esc.example.com",
        common_name="app.esc.example.com",
        not_after=datetime.now(UTC) + timedelta(days=4), status="expiring",
        first_seen_at=datetime.now(UTC),
    ))
    await db_session.commit()

    items = await build_cert_expiry_items(db_session)
    rec = _Recorder()
    summary = await run_expiry_alerts(
        db_session, resource_type=EXPIRY_RESOURCE_CERTIFICATE,
        items=items, config={}, notify=rec,
    )
    mine = [c for c in rec.calls if c[0] == "app.esc.example.com"]
    assert len(mine) == 1
    assert "owner" in mine[0][2] and "lead" in mine[0][2]
    assert summary["escalated"] >= 1


@pytest.mark.asyncio
async def test_prod_certificate_excluded_from_reminders(
    db_session: AsyncSession, integ: Integration,
):
    """Certs tagged ``environment='prod'`` are out of scope for the reminder
    lifecycle — prod TLS renewal is owned outside Helix self-service. A
    single prod-tagged member suppresses the whole logical cert group, while
    an unrelated non-prod cert still surfaces."""
    owner = await _person(db_session, "ProdOwner")
    soon = datetime.now(UTC) + timedelta(days=3)
    # Logical prod cert with two rows: one tagged prod, one untagged. The
    # prod tag on either row must suppress the whole group (no leak via the
    # untagged sibling).
    db_session.add(CmdbCertificate(
        id=uuid.uuid4(), dedup_key="probed:prod.tidbcloud.com:443",
        source=CERT_SOURCE_PROBED, probe_host="prod.tidbcloud.com",
        common_name="prod.tidbcloud.com", sans=["prod.tidbcloud.com"],
        environment="prod", not_after=soon, status="expiring",
        first_seen_at=datetime.now(UTC),
    ))
    db_session.add(CmdbCertificate(
        id=uuid.uuid4(), dedup_key=f"{CERT_SOURCE_CLOUD}:{integ.id}:prodc",
        source=CERT_SOURCE_CLOUD, integration_id=str(integ.id),
        common_name="prod.tidbcloud.com", sans=["prod.tidbcloud.com"],
        environment=None, not_after=soon, status="valid",
        first_seen_at=datetime.now(UTC), owner_person_id=str(owner.id),
    ))
    # A separate non-prod cert must still be reminded.
    db_session.add(CmdbCertificate(
        id=uuid.uuid4(), dedup_key="probed:dev.tidbcloud.com:443",
        source=CERT_SOURCE_PROBED, probe_host="dev.tidbcloud.com",
        common_name="dev.tidbcloud.com", sans=["dev.tidbcloud.com"],
        environment="dev", not_after=soon, status="expiring",
        first_seen_at=datetime.now(UTC),
    ))
    await db_session.commit()

    labels = {it.label for it in await build_cert_expiry_items(db_session)}
    assert labels == {"dev.tidbcloud.com"}


@pytest.mark.asyncio
async def test_prod_domain_excluded_from_reminders(
    db_session: AsyncSession, integ: Integration,
):
    """Domains tagged ``environment='prod'`` are out of scope for expiry
    reminders; non-prod / untagged domains are unaffected."""
    owner = await _person(db_session, "DomOwner")
    exp = (datetime.now(UTC) + timedelta(days=5)).date()
    db_session.add(CmdbDomain(
        id=uuid.uuid4(), integration_id=integ.id, domain_name="tidbcloud.com",
        owner_person_id=str(owner.id), environment="prod", expiry_date=exp,
    ))
    db_session.add(CmdbDomain(
        id=uuid.uuid4(), integration_id=integ.id,
        domain_name="staging.tidbcloud.com",
        owner_person_id=str(owner.id), environment="staging", expiry_date=exp,
    ))
    await db_session.commit()

    labels = {it.label for it in await build_domain_expiry_items(db_session)}
    assert labels == {"staging.tidbcloud.com"}