"""Local smoke test: signature verification, payload routing, abuse policy.

Run after `pip install -r requirements.txt`:
    WEBHOOK_SECRET=testsecret BOT_ACCESS_TOKEN=x python test_local.py

It uses FastAPI's TestClient and monkeypatches every outbound Mastodon call,
so it makes no network requests.
"""

import hashlib
import hmac
import json
import os
from datetime import datetime, timedelta, timezone

os.environ.setdefault("WEBHOOK_SECRET", "testsecret")
os.environ.setdefault("BOT_ACCESS_TOKEN", "dummy")
os.environ.setdefault("ABUSE_BOT_TOKEN", "dummy-admin")
os.environ.setdefault("ABUSE_DRY_RUN", "false")
os.environ.setdefault("MOD_ALERT_ACCT", "pete")
os.environ.setdefault("DB_PATH", "/tmp/welcomebot-test.db")

if os.path.exists(os.environ["DB_PATH"]):
    os.remove(os.environ["DB_PATH"])

from fastapi.testclient import TestClient  # noqa: E402

import app.main as main  # noqa: E402

# Capture welcome calls instead of hitting the network.
sent = []
main.send_welcome = lambda account_id, acct: sent.append((account_id, acct))

# Capture report handling at the route boundary for the routing test.
reports = []
_real_handle_report = main.handle_report
main.handle_report = lambda *a: reports.append(a)

client = TestClient(main.app)


def sign(body: bytes) -> str:
    return "sha256=" + hmac.new(b"testsecret", body, hashlib.sha256).hexdigest()


def post(payload, secret_ok=True):
    body = json.dumps(payload).encode()
    sig = sign(body) if secret_ok else "sha256=deadbeef"
    return client.post(
        "/webhook", content=body,
        headers={"X-Hub-Signature": sig, "Content-Type": "application/json"},
    )


def days_ago(n):
    return (datetime.now(timezone.utc) - timedelta(days=n)).isoformat()


def routing_tests():
    # 1. health
    assert client.get("/healthz").json() == {"ok": True}

    # 2. bad signature rejected
    r = post({"event": "account.created", "object": {}}, secret_ok=False)
    assert r.status_code == 401, r.status_code

    # 3. missing signature rejected
    body = json.dumps({"event": "account.created"}).encode()
    assert client.post("/webhook", content=body).status_code == 401

    # 4. valid account.created -> queued
    payload = {
        "event": "account.created",
        "object": {
            "id": "1001", "username": "alice", "domain": None,
            "account": {"id": "1001", "username": "alice", "acct": "alice",
                        "url": "https://yttrx.com/@alice"},
        },
    }
    r = post(payload)
    assert r.status_code == 200, r.text
    assert r.json().get("queued") == "alice", r.json()
    assert sent == [("1001", "alice")], sent

    # 5. report.created -> routed to handle_report with extracted fields
    r = post({
        "event": "report.created",
        "object": {
            "id": "55", "category": "spam",
            "target_account": {
                "id": "777", "username": "spammer", "domain": None,
                "suspended": False, "silenced": False,
                "account": {"id": "777", "acct": "spammer"},
            },
        },
    })
    assert r.status_code == 200, r.text
    assert r.json().get("report") == "55", r.json()
    # (report_id, target_id, acct, is_local, suspended, silenced, privileged, email)
    assert reports == [("55", "777", "spammer", True, False, False, False, "")], reports

    # 5b. report against a staff account -> privileged=True extracted from role
    r = post({
        "event": "report.created",
        "object": {
            "id": "56",
            "target_account": {
                "id": "778", "username": "waffles", "domain": None,
                "suspended": False, "silenced": False,
                "role": {"id": 3, "name": "Owner", "permissions": "1"},
                "account": {"id": "778", "acct": "waffles"},
            },
        },
    })
    assert reports[-1] == ("56", "778", "waffles", True, False, False, True, ""), reports[-1]

    # 6. account.approved also welcomes — but dedup means alice (id 1001,
    #    already welcomed via account.created above) is not welcomed twice.
    r = post({
        "event": "account.approved",
        "object": {"id": "1001", "username": "alice", "domain": None,
                   "account": {"id": "1001", "acct": "alice"}},
    })
    assert r.status_code == 200, r.text
    assert r.json().get("queued") == "alice", r.json()
    # send_welcome is stubbed to append; real dedup is exercised in policy run.
    # A brand-new account via account.approved should also route to welcome.
    r = post({
        "event": "account.approved",
        "object": {"id": "2002", "username": "carol", "domain": None,
                   "account": {"id": "2002", "acct": "carol"}},
    })
    assert r.json().get("queued") == "carol", r.json()
    assert ("2002", "carol") in sent, sent

    # 7. truly unknown event ignored
    r = post({"event": "status.updated", "object": {"id": "5"}})
    assert r.json().get("ignored") == "status.updated", r.json()

    # 8. remote account skipped when LOCAL_ONLY (welcome)
    r = post({
        "event": "account.created",
        "object": {"id": "9", "username": "bob", "domain": "other.social",
                   "account": {"id": "9", "acct": "bob@other.social"}},
    })
    assert r.json().get("skipped"), r.json()


def policy_tests():
    """Drive handle_report directly with stubbed Mastodon calls."""
    main.handle_report = _real_handle_report

    actions, dms, user_dms = [], [], []
    main.apply_action = lambda target_id, action, text: actions.append((target_id, action))
    main.dm_moderator = lambda message: dms.append(message)
    main.dm_silenced_user = lambda acct: user_dms.append(acct)

    def run(target_id, acct, *, profile, sources, **kw):
        main.classify_account = lambda tid: profile
        main.count_distinct_reporters = lambda tid: sources
        defaults = dict(is_local=True, suspended=False, silenced=False, privileged=False)
        defaults.update(kw)
        main.handle_report("R", target_id, acct, defaults["is_local"],
                           defaults["suspended"], defaults["silenced"],
                           defaults["privileged"])

    young = {"has_posts": True, "young": True, "dormant": False,
             "newest_at": None, "oldest_seen": None, "truncated": False}
    active = {"has_posts": True, "young": False, "dormant": False,
              "newest_at": None, "oldest_seen": None, "truncated": False}
    dormant = {"has_posts": True, "young": False, "dormant": True,
               "newest_at": None, "oldest_seen": None, "truncated": False}
    noposts = {"has_posts": False, "young": False, "dormant": False,
               "newest_at": None, "oldest_seen": None, "truncated": False}

    # young, 1 source -> below threshold (needs 2), no action
    run("1", "y1", profile=young, sources=1)
    assert actions == [], actions

    # young, 2 sources -> silenced; both the mod and the user are DMed
    run("2", "y2", profile=young, sources=2)
    assert ("2", "silence") in actions, actions
    assert any("y2" in d for d in dms), dms
    assert "y2" in user_dms, user_dms

    # dormant, 2 sources -> silenced
    run("3", "d1", profile=dormant, sources=2)
    assert ("3", "silence") in actions, actions

    # active, 2 sources -> below threshold (needs 3), no action
    run("4", "a1", profile=active, sources=2)
    assert ("4", "silence") not in actions, actions

    # active, 3 sources -> silenced
    run("5", "a2", profile=active, sources=3)
    assert ("5", "silence") in actions, actions

    # no-posts, 2 sources -> silenced (grouped with young/dormant)
    run("6", "n1", profile=noposts, sources=2)
    assert ("6", "silence") in actions, actions

    # already silenced -> skipped even with enough sources
    run("7", "s1", profile=young, sources=5, silenced=True)
    assert ("7", "silence") not in actions, actions

    # remote -> skipped (ABUSE_LOCAL_ONLY)
    run("8", "r1@elsewhere", profile=young, sources=5, is_local=False)
    assert ("8", "silence") not in actions, actions

    # privileged (staff role) -> skipped even with plenty of sources
    run("9", "waffles", profile=young, sources=9, privileged=True)
    assert ("9", "silence") not in actions, actions

    # dedup: re-running an already-actioned target takes no new action
    before = len(actions)
    run("2", "y2", profile=young, sources=9)
    assert len(actions) == before, "expected dedup to suppress repeat action"


def ip_scrutiny_tests():
    """Drive process_signup + the abuse threshold override, network stubbed out."""
    sent = []
    dms = []
    ipblocks = []
    main.send_welcome = lambda account_id, acct: sent.append((account_id, acct))
    main.dm_moderator = lambda message: dms.append(message)
    main.register_ip_block = lambda ip, acct, org: ipblocks.append((ip, acct, org))
    # A flagged-but-not-held signup starts the suspicious watch inline, which
    # would otherwise hit the network for a baseline snapshot — stub it.
    main.fetch_account_counts = lambda account_id: (0, 0)

    def classify(ip):
        return classifications[ip]

    classifications = {
        "203.0.113.10": ("residential", "Example Residential ISP"),
        "198.51.100.20": ("datacenter", "Example Cloud Hosting Inc"),
        "198.51.100.21": ("datacenter", "Example Cloud Hosting Inc"),
    }
    main.classify_signup_ip = classify

    main.IP_SCRUTINY_ENABLED = True

    # A. residential IP -> welcomed immediately, no ip-block, no flag recorded.
    main.IP_SCRUTINY_DRY_RUN = False
    main.IP_SCRUTINY_HOLD_WELCOME = True
    main.IP_SCRUTINY_AUTO_IPBLOCK = True
    main.process_signup("101", "res1", "203.0.113.10")
    assert ("101", "res1") in sent, sent
    assert ipblocks == [], ipblocks
    assert main.get_signup_flag("101") is False, "residential signup should not be flagged"

    # B. datacenter IP, HOLD_WELCOME + AUTO_IPBLOCK, live (not dry-run) ->
    #    welcome held, ip-block registered, moderator DM'd; account.approved
    #    later releases the welcome.
    sent.clear(); dms.clear(); ipblocks.clear()
    main.process_signup("102", "dc1", "198.51.100.20")
    assert ("102", "dc1") not in sent, "welcome should be held for a flagged signup"
    assert ipblocks == [("198.51.100.20", "dc1", "Example Cloud Hosting Inc")], ipblocks
    assert any("dc1" in d and "held" in d for d in dms), dms
    assert main.get_signup_flag("102") is True

    main.send_welcome("102", "dc1")  # simulate account.approved's unconditional welcome
    assert ("102", "dc1") in sent, sent

    # C. same datacenter IP, but IP_SCRUTINY_DRY_RUN -> welcomed immediately
    #    (no hold), no ip-block write, DM carries the dry-run prefix.
    sent.clear(); dms.clear(); ipblocks.clear()
    main.IP_SCRUTINY_DRY_RUN = True
    main.process_signup("103", "dc2", "198.51.100.21")
    assert ("103", "dc2") in sent, "dry-run must not hold the welcome"
    assert ipblocks == [], "dry-run must not write an ip_block"
    assert any("[DRY-RUN]" in d for d in dms), dms

    # D. abuse threshold override: a flagged account needs only
    #    IP_SCRUTINY_ABUSE_THRESHOLD (1) distinct reporter to silence, vs. the
    #    normal young-tier threshold of 2 for an unflagged account.
    main.IP_SCRUTINY_DRY_RUN = False
    main.dm_moderator = lambda message: dms.append(message)
    main.handle_report = _real_handle_report
    actions = []
    main.apply_action = lambda target_id, action, text: actions.append((target_id, action))
    main.dm_silenced_user = lambda acct: None
    young = {"has_posts": True, "young": True, "dormant": False,
             "newest_at": None, "oldest_seen": None, "truncated": False}
    main.classify_account = lambda tid: young
    main.count_distinct_reporters = lambda tid: 1

    # "102" (dc1) was flagged above -> 1 reporter is enough.
    main.handle_report("R2", "102", "dc1", True, False, False, False)
    assert ("102", "silence") in actions, actions

    # "101" (res1) was not flagged -> 1 reporter stays below the tier's usual
    # threshold of 2.
    main.handle_report("R3", "101", "res1", True, False, False, False)
    assert ("101", "silence") not in actions, actions


def email_scrutiny_tests():
    """Drive process_signup's email-domain path + the report-triggered
    suspend+domain-block override, network stubbed out."""
    sent = []
    dms = []
    domain_blocks = []
    main.send_welcome = lambda account_id, acct: sent.append((account_id, acct))
    main.dm_moderator = lambda message: dms.append(message)
    main.register_email_domain_block = lambda domain, acct: domain_blocks.append((domain, acct))
    # A flagged-but-not-held signup starts the suspicious watch inline, which
    # would otherwise hit the network for a baseline snapshot — stub it.
    main.fetch_account_counts = lambda account_id: (0, 0)

    classifications = {
        "gmail.com": (False, 5),
        "temp-mail.org": (True, 99),
        "sketchy-disposable.example": (True, 95),
    }
    main.classify_email_domain = lambda domain: classifications[domain]

    main.CHECK_MAIL_ENABLED = True
    main.CHECK_MAIL_API_KEY = "test-key"  # required for the signup-path gate
    main.IP_SCRUTINY_ENABLED = False  # isolate the email-only signal

    # A. clean domain -> welcomed immediately, no domain-block, not flagged.
    main.CHECK_MAIL_DRY_RUN = False
    main.CHECK_MAIL_HOLD_WELCOME = True
    main.CHECK_MAIL_AUTO_DOMAIN_BLOCK = True
    main.process_signup("201", "clean1", "", "clean1@gmail.com")
    assert ("201", "clean1") in sent, sent
    assert domain_blocks == [], domain_blocks
    assert main.get_signup_email("201") == ("gmail.com", False)

    # B. disposable domain, HOLD_WELCOME + AUTO_DOMAIN_BLOCK, live -> welcome
    #    held, domain blocked, moderator DM'd; account.approved later releases
    #    the welcome.
    sent.clear(); dms.clear(); domain_blocks.clear()
    main.process_signup("202", "spam1", "", "spam1@temp-mail.org")
    assert ("202", "spam1") not in sent, "welcome should be held for a flagged signup"
    assert domain_blocks == [("temp-mail.org", "spam1")], domain_blocks
    assert any("spam1" in d and "held" in d for d in dms), dms
    assert main.get_signup_email("202") == ("temp-mail.org", True)

    main.send_welcome("202", "spam1")  # simulate account.approved's unconditional welcome
    assert ("202", "spam1") in sent, sent

    # C. same idea, but CHECK_MAIL_DRY_RUN -> welcomed immediately (no hold),
    #    no domain-block write, DM carries the dry-run marker.
    sent.clear(); dms.clear(); domain_blocks.clear()
    main.CHECK_MAIL_DRY_RUN = True
    main.process_signup("203", "spam2", "", "spam2@sketchy-disposable.example")
    assert ("203", "spam2") in sent, "dry-run must not hold the welcome"
    assert domain_blocks == [], "dry-run must not write a domain block"
    assert any("[DRY-RUN]" in d for d in dms), dms

    # D0. CHECK_MAIL_DRY_RUN alone holds the report-triggered suspend even
    #     though ABUSE_DRY_RUN=False in this test env (set in the module
    #     preamble) — the two dry-run switches are independent; either one
    #     holding is enough to keep this brand-new action path inert.
    actions = []
    main.apply_action = lambda target_id, action, text: actions.append((target_id, action))
    main.dm_silenced_user = lambda acct: None
    domain_blocks.clear()

    main.handle_report("R9", "997", "heldbydryrun", True, False, False, False,
                        "heldbydryrun@temp-mail.org")
    assert ("997", "suspend") not in actions, actions
    assert domain_blocks == [], "CHECK_MAIL_DRY_RUN alone must prevent the domain block too"

    # D. report-triggered override: ANY report against an account with a
    #    high-risk email domain suspends immediately (no distinct-reporter
    #    threshold) and blocks the domain — classified LIVE from the report
    #    payload's target_account.email, independent of whatever (if
    #    anything) process_signup recorded at signup time. This account was
    #    never seen by process_signup at all.
    main.CHECK_MAIL_DRY_RUN = False
    actions.clear()
    domain_blocks.clear()

    main.handle_report("R10", "999", "neversignedup", True, False, False, False,
                        "neversignedup@temp-mail.org")
    assert ("999", "suspend") in actions, actions
    assert domain_blocks == [("temp-mail.org", "neversignedup")], domain_blocks

    # E. report against an account with a clean email domain -> untouched by
    #    this path, falls through to the normal reporter-count/tier policy.
    actions.clear(); domain_blocks.clear()
    noposts = {"has_posts": False, "young": False, "dormant": False,
               "newest_at": None, "oldest_seen": None, "truncated": False}
    main.classify_account = lambda tid: noposts
    main.count_distinct_reporters = lambda tid: 1
    main.handle_report("R11", "998", "clean2", True, False, False, False,
                        "clean2@gmail.com")
    assert ("998", "suspend") not in actions, actions
    assert domain_blocks == [], domain_blocks


def suspicious_watch_tests():
    """Drive maybe_start_suspicious_watch: baseline capture on flagged
    signups only, never on clean ones, and never twice for the same account."""
    main.SUSPICIOUS_SWEEP_ENABLED = True
    counts = {"301": (2, 5), "302": (0, 0)}
    main.fetch_account_counts = lambda account_id: counts[account_id]

    # A. flagged by IP only -> watch started with that baseline.
    main.record_signup_ip("301", "watched1", "198.51.100.30", "datacenter", "Cloud Inc", True)
    main.maybe_start_suspicious_watch("301", "watched1")
    with main._db() as conn:
        row = conn.execute(
            "SELECT reasons, baseline_statuses_count, baseline_following_count, status "
            "FROM suspicious_watch WHERE account_id = ?", ("301",),
        ).fetchone()
    assert row == ("ip", 2, 5, "pending"), row

    # B. calling again (e.g. account.created then account.approved both
    #    firing) does not reset/duplicate the baseline.
    counts["301"] = (99, 99)
    main.maybe_start_suspicious_watch("301", "watched1")
    with main._db() as conn:
        row = conn.execute(
            "SELECT baseline_statuses_count FROM suspicious_watch WHERE account_id = ?", ("301",),
        ).fetchone()
    assert row == (2,), "re-calling must not overwrite the original baseline"

    # C. never flagged -> no watch row at all.
    main.maybe_start_suspicious_watch("302", "clean3")
    with main._db() as conn:
        row = conn.execute(
            "SELECT 1 FROM suspicious_watch WHERE account_id = ?", ("302",),
        ).fetchone()
    assert row is None, "an unflagged signup must never be watched"


def suspicious_sweep_tests():
    """Drive app.suspicious_sweep.sweep() directly against rows seeded into
    the shared sqlite store, with the Mastodon admin-account fetch and
    apply_action/dm_moderator stubbed out (no network)."""
    import app.suspicious_sweep as sweep_mod

    sweep_mod.SUSPICIOUS_SWEEP_ENABLED = True
    sweep_mod.SUSPICIOUS_GRACE_HOURS = 24
    sweep_mod.SUSPICIOUS_ACTION = "suspend"
    sweep_mod.SUSPICIOUS_DRY_RUN = False
    sweep_mod.ABUSE_SKIP_PRIVILEGED = True
    sweep_mod.ABUSE_ALLOWLIST = {"trustedstaff"}

    def seed(account_id, acct, reasons, baseline_statuses, baseline_following, hours_ago):
        started_at = (datetime.now(timezone.utc) - timedelta(hours=hours_ago)).isoformat()
        with main._db() as conn:
            conn.execute(
                "INSERT INTO suspicious_watch "
                "(account_id, acct, reasons, started_at, baseline_statuses_count, "
                " baseline_following_count) VALUES (?, ?, ?, ?, ?, ?)",
                (account_id, acct, reasons, started_at, baseline_statuses, baseline_following),
            )

    def status_of(account_id):
        with main._db() as conn:
            row = conn.execute(
                "SELECT status FROM suspicious_watch WHERE account_id = ?", (account_id,),
            ).fetchone()
        return row[0] if row else None

    admin_accounts = {}
    sweep_mod._fetch_admin_account = lambda account_id: admin_accounts[account_id]

    actions = []
    dms = []
    sweep_mod.apply_action = lambda target_id, action, text: actions.append((target_id, action))
    sweep_mod.dm_moderator = lambda message: dms.append(message)

    def public(statuses_count=0, following_count=0, **extra):
        acct = {"statuses_count": statuses_count, "following_count": following_count}
        return {"account": acct, **extra}

    # A. not yet due (started 1h ago, 24h grace) -> untouched, still pending.
    seed("401", "toosoon", "ip", 0, 0, hours_ago=1)
    admin_accounts["401"] = public(0, 0)

    # B. due, zero new posts/follows since baseline -> suspended.
    seed("402", "dormantbot", "ip", 0, 0, hours_ago=25)
    admin_accounts["402"] = public(0, 0)

    # C. due, but posted since baseline -> cleared, not suspended.
    seed("403", "postedsince", "email", 3, 1, hours_ago=25)
    admin_accounts["403"] = public(4, 1)

    # D. due, but followed someone since baseline -> cleared, not suspended.
    seed("404", "followedsince", "ip,email", 0, 0, hours_ago=25)
    admin_accounts["404"] = public(0, 1)

    # E. due, but already suspended/silenced by another mechanism -> skipped.
    seed("405", "alreadylimited", "ip", 0, 0, hours_ago=25)
    admin_accounts["405"] = public(0, 0, suspended=True)

    # F. due, but holds a staff role -> skipped, never suspended.
    seed("406", "staffer", "ip", 0, 0, hours_ago=25)
    admin_accounts["406"] = public(0, 0, role={"id": 3, "name": "Owner"})

    # G. due, allowlisted by handle -> skipped.
    seed("407", "trustedstaff", "ip", 0, 0, hours_ago=25)
    admin_accounts["407"] = public(0, 0)

    sweep_mod.sweep()

    assert status_of("401") == "pending", "not-yet-due watch must be left alone"
    assert ("402", "suspend") in actions, actions
    assert status_of("402") == "suspend"
    assert any("dormantbot" in d for d in dms), dms

    assert ("403", "suspend") not in actions, actions
    assert status_of("403") == "cleared"

    assert ("404", "suspend") not in actions, actions
    assert status_of("404") == "cleared"

    assert ("405", "suspend") not in actions, actions
    assert status_of("405") == "skipped-already-limited"

    assert ("406", "suspend") not in actions, actions
    assert status_of("406") == "skipped-privileged"

    assert ("407", "suspend") not in actions, actions
    assert status_of("407") == "skipped-privileged"

    # H. re-running the sweep must not re-suspend or re-DM an already-resolved row.
    before_actions, before_dms = len(actions), len(dms)
    sweep_mod.sweep()
    assert len(actions) == before_actions, "resolved rows must not be re-actioned"
    assert len(dms) == before_dms, "resolved rows must not be re-DMed"

    # I. dry-run mode logs/DMs but takes no real action.
    seed("408", "dryrunbot", "ip", 0, 0, hours_ago=25)
    admin_accounts["408"] = public(0, 0)
    sweep_mod.SUSPICIOUS_DRY_RUN = True
    sweep_mod.sweep()
    assert ("408", "suspend") not in actions, "dry-run must not call apply_action"
    assert status_of("408") == "dry-run:suspend"
    assert any("[DRY-RUN]" in d and "dryrunbot" in d for d in dms), dms


if __name__ == "__main__":
    routing_tests()
    policy_tests()
    ip_scrutiny_tests()
    email_scrutiny_tests()
    suspicious_watch_tests()
    suspicious_sweep_tests()
    print("ALL TESTS PASSED")