"""Alerting tests — v2 schema aligned.

Webhook payload + POST tests stay unchanged (unrelated to schema). Threshold
check tests (R1/R2/R3) now build crawl_site + crawl_target rows and reference
them via int FKs; assertions use `site_code` / `target_code` surfaced via joins.
"""
from datetime import datetime, timedelta
from unittest.mock import MagicMock

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from govcrawler.alerting import build_payload, run_checks
from govcrawler.alerting.webhook import send_alert
from govcrawler.alerting import webhook as webhook_mod
from govcrawler.models import Article, Base, CrawlLog

from tests._v2fixtures import make_site_and_target


# ---------- webhook payload + POST ----------

def test_feishu_payload_shape():
    p = build_payload("hello", "feishu")
    assert p == {"msg_type": "text", "content": {"text": "hello"}}


def test_wechat_payload_shape():
    p = build_payload("hello", "wechat")
    assert p == {"msgtype": "text", "text": {"content": "hello"}}


def test_unknown_provider_raises():
    with pytest.raises(ValueError):
        build_payload("x", "slack")


def test_send_alert_disabled_when_url_empty(monkeypatch):
    monkeypatch.setenv("DB_URL", "postgresql+psycopg://x/x")
    monkeypatch.setenv("ALERT_WEBHOOK_URL", "")
    assert send_alert("x") is False


def test_send_alert_posts_payload(monkeypatch):
    monkeypatch.setenv("DB_URL", "postgresql+psycopg://x/x")
    monkeypatch.setenv("ALERT_WEBHOOK_URL", "https://open.feishu.cn/bot/v2/xxx")
    monkeypatch.setenv("ALERT_PROVIDER", "feishu")

    captured: dict = {}

    def fake_post(url, json=None, timeout=10):
        captured["url"] = url
        captured["json"] = json
        r = MagicMock()
        r.status_code = 200
        return r

    monkeypatch.setattr(webhook_mod.httpx, "post", fake_post)
    assert send_alert("成功率偏低") is True
    assert captured["url"] == "https://open.feishu.cn/bot/v2/xxx"
    assert captured["json"] == {"msg_type": "text", "content": {"text": "成功率偏低"}}


def test_send_alert_returns_false_on_non_2xx(monkeypatch):
    monkeypatch.setenv("DB_URL", "postgresql+psycopg://x/x")
    r = MagicMock()
    r.status_code = 500
    r.text = "boom"
    monkeypatch.setattr(webhook_mod.httpx, "post", lambda *a, **kw: r)
    assert send_alert("x", url="https://x", provider="feishu") is False


def test_send_alert_returns_false_on_network_error(monkeypatch):
    monkeypatch.setenv("DB_URL", "postgresql+psycopg://x/x")

    def boom(*a, **kw):
        raise ConnectionError("down")

    monkeypatch.setattr(webhook_mod.httpx, "post", boom)
    assert send_alert("x", url="https://x", provider="feishu") is False


# ---------- threshold checks ----------

@pytest.fixture
def session_and_ids(monkeypatch, tmp_path):
    """Yields (session, site_pk, target_id) — the alerting module JOINs
    crawl_site + crawl_target so we need real rows to see site_code/target_code.
    """
    monkeypatch.setenv("DB_URL", "sqlite:///" + str(tmp_path / "alert.db"))
    monkeypatch.setenv("USER_AGENT", "TestBot/1.0")
    engine = create_engine("sqlite:///" + str(tmp_path / "alert.db"), future=True)
    Base.metadata.create_all(engine)
    SM = sessionmaker(bind=engine, expire_on_commit=False)
    with SM() as s:
        site, target = make_site_and_target(s, site_code="a", column_id="c")
        s.commit()
        yield s, site.id, target.id


_log_id_counter = {"n": 0}


def _add_log(s, *, site_pk, target_id, ok, status_code, when):
    _log_id_counter["n"] += 1
    s.add(CrawlLog(
        id=_log_id_counter["n"],
        site_id=site_pk, target_id=target_id,
        article_url=f"https://site{site_pk}/x",
        strategy="httpx", http_status=status_code,
        duration_ms=10, success=ok, error_msg=None,
        occurred_at=when,
    ))


def test_r1_success_rate_triggers(session_and_ids):
    session, site_pk, target_id = session_and_ids
    now = datetime(2026, 4, 22, 10, 0, 0)
    # 10 logs in last 24h, 3 succeed → 30%
    for i in range(10):
        _add_log(
            session, site_pk=site_pk, target_id=target_id,
            ok=(i < 3), status_code=200 if i < 3 else 500,
            when=now - timedelta(hours=1),
        )
    session.commit()
    rules = run_checks(now=now, session=session)
    codes = [r.code for r in rules]
    assert "R1_SUCCESS_RATE" in codes
    # Codes surfaced, not FK ids
    r1 = next(r for r in rules if r.code == "R1_SUCCESS_RATE")
    assert r1.site_code == "a"
    assert r1.target_code == "a__c"


def test_r1_ignores_low_sample(session_and_ids):
    session, site_pk, target_id = session_and_ids
    now = datetime(2026, 4, 22, 10, 0, 0)
    for i in range(3):   # below R1_MIN_SAMPLES=5
        _add_log(
            session, site_pk=site_pk, target_id=target_id,
            ok=False, status_code=500, when=now - timedelta(hours=1),
        )
    session.commit()
    rules = run_checks(now=now, session=session)
    assert all(r.code != "R1_SUCCESS_RATE" for r in rules)


def test_r2_block_rate_triggers(session_and_ids):
    session, site_pk, target_id = session_and_ids
    now = datetime(2026, 4, 22, 10, 0, 0)
    # 15 logs in last 1h; 6 are 412 → 40%
    for i in range(15):
        _add_log(
            session, site_pk=site_pk, target_id=target_id,
            ok=(i >= 6), status_code=412 if i < 6 else 200,
            when=now - timedelta(minutes=30),
        )
    session.commit()
    rules = run_checks(now=now, session=session)
    codes = [r.code for r in rules]
    assert "R2_BLOCK_RATE" in codes
    r2 = next(r for r in rules if r.code == "R2_BLOCK_RATE")
    assert r2.site_code == "a"
    assert r2.target_code is None  # site-wide rule


def test_r3_stale_target_triggers(session_and_ids):
    session, site_pk, target_id = session_and_ids
    now = datetime(2026, 4, 22, 10, 0, 0)
    # Prior 24-48h: active
    session.add(Article(
        id=1, site_id=site_pk, target_id=target_id,
        url="https://a/1", url_hash="a" * 64,
        status="ready", has_attachment=False,
        fetched_at=now - timedelta(hours=30),
    ))
    # Current 24h: nothing (simulates selector break)
    session.commit()
    rules = run_checks(now=now, session=session)
    codes = [(r.code, r.site_code, r.target_code) for r in rules]
    assert ("R3_STALE_TARGET", "a", "a__c") in codes


def test_r3_no_alert_when_current_active(session_and_ids):
    session, site_pk, target_id = session_and_ids
    now = datetime(2026, 4, 22, 10, 0, 0)
    session.add(Article(
        id=1, site_id=site_pk, target_id=target_id,
        url="https://a/1", url_hash="a" * 64,
        status="ready", has_attachment=False,
        fetched_at=now - timedelta(hours=30),
    ))
    session.add(Article(
        id=2, site_id=site_pk, target_id=target_id,
        url="https://a/2", url_hash="b" * 64,
        status="ready", has_attachment=False,
        fetched_at=now - timedelta(hours=3),
    ))
    session.commit()
    rules = run_checks(now=now, session=session)
    assert all(r.code != "R3_STALE_TARGET" for r in rules)