"""push.py 单测：订阅增删查、偏好过滤、分发错误不静默丢。

不连外网、不向真实推送服务发任何请求：注入 sender 的测试
用假 sender；真实发送端的测试用本地回环 HTTP 服务器
（127.0.0.1 临时端口）收 POST，断言方法、头与 body。
本文件不含任何真实密钥、token、证书或私钥。
"""

import http.server
import json
import os
import sqlite3
import sys
import threading
import unittest
from pathlib import Path

try:
    from cloud.push import (
        ALLOWED_PUSH_EVENTS,
        CHANNELS,
        FAILURE_EVENTS,
        PREFERENCES,
        create_subscription,
        delete_subscription,
        dispatch_event,
        list_outbox,
        list_pending_digest,
        list_subscriptions,
        process_outbox_due,
        send_digest_due,
        send_webhook,
    )
except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from cloud.push import (
        ALLOWED_PUSH_EVENTS,
        CHANNELS,
        FAILURE_EVENTS,
        PREFERENCES,
        create_subscription,
        delete_subscription,
        dispatch_event,
        list_outbox,
        list_pending_digest,
        list_subscriptions,
        process_outbox_due,
        send_digest_due,
        send_webhook,
    )


class PushTestCase(unittest.TestCase):
    def setUp(self):
        # 内存库 + 单一连接：init 与查询共用同一连接，避免 :memory: 换连接丢表。
        self.conn = sqlite3.connect(":memory:")

    def tearDown(self):
        self.conn.close()

    def _sub(self, project="proj-a", user="user-1", channel="webhook",
             target="https://example.invalid/hook", preference="instant"):
        return create_subscription(
            self.conn, project, user, channel, target, preference)

    def _calls(self):
        calls = []

        def sender(message):
            calls.append(message)

        return calls, sender


class TestSubscriptionCrud(PushTestCase):
    def test_create_and_list(self):
        sub = self._sub()
        self.assertEqual(sub["project_id"], "proj-a")
        self.assertEqual(sub["user_id"], "user-1")
        self.assertEqual(sub["channel"], "webhook")
        self.assertEqual(sub["preference"], "instant")
        listed = list_subscriptions(self.conn, "proj-a", "user-1")
        self.assertEqual(len(listed), 1)
        self.assertEqual(listed[0]["id"], sub["id"])

    def test_create_duplicate_returns_existing_not_duplicate_row(self):
        first = self._sub()
        second = self._sub()
        self.assertEqual(first["id"], second["id"])
        self.assertEqual(
            len(list_subscriptions(self.conn, "proj-a", "user-1")), 1)

    def test_create_invalid_channel_or_preference_rejected(self):
        with self.assertRaises(ValueError):
            self._sub(channel="sms")
        with self.assertRaises(ValueError):
            self._sub(preference="weekly")
        self.assertEqual(
            len(list_subscriptions(self.conn, "proj-a", "user-1")), 0)

    def test_delete_by_id_then_no_longer_listed(self):
        sub = self._sub()
        self.assertTrue(delete_subscription(self.conn, sub["id"]))
        self.assertEqual(
            list_subscriptions(self.conn, "proj-a", "user-1"), [])
        # 再次删除无匹配，返回 False，不报错。
        self.assertFalse(delete_subscription(self.conn, sub["id"]))

    def test_constants_cover_task_contract(self):
        self.assertEqual(
            set(CHANNELS), {"webhook", "slack", "webpush", "apns", "fcm"})
        self.assertEqual(set(PREFERENCES), {"instant", "digest", "silent"})
        self.assertEqual(
            ALLOWED_PUSH_EVENTS,
            {"completed", "failed", "goal-unmet", "merge-failed",
             "review-requested"},
        )
        self.assertEqual(FAILURE_EVENTS, {"failed", "goal-unmet", "merge-failed"})


class TestDispatchPreferences(PushTestCase):
    def test_silent_preference_never_sends(self):
        self._sub(preference="silent")
        calls, sender = self._calls()
        result = dispatch_event(
            self.conn, "failed", "proj-a", {"task_id": "t-1"}, sender=sender)
        self.assertEqual(calls, [])
        self.assertEqual(result["sent"], [])
        self.assertEqual(len(result["skipped_silent"]), 1)
        self.assertEqual(result["skipped_silent"][0]["preference"], "silent")

    def test_failure_event_sends_immediately_even_with_digest(self):
        # 失败类：digest 也即时发，不进 skipped_for_digest。
        for event_type in ("failed", "goal-unmet", "merge-failed"):
            with self.subTest(event_type=event_type):
                conn = sqlite3.connect(":memory:")
                try:
                    create_subscription(
                        conn, "proj-a", "user-1", "webhook",
                        "https://example.invalid/hook", "digest")
                    calls, sender = self._calls()
                    result = dispatch_event(
                        conn, event_type, "proj-a", {"task_id": "t-1"},
                        sender=sender)
                    self.assertEqual(len(calls), 1)
                    self.assertEqual(calls[0]["event_type"], event_type)
                    self.assertEqual(len(result["sent"]), 1)
                    self.assertEqual(result["skipped_for_digest"], [])
                finally:
                    conn.close()

    def test_completed_event_filtered_by_digest_returns_pending_list(self):
        self._sub(preference="digest")
        calls, sender = self._calls()
        result = dispatch_event(
            self.conn, "completed", "proj-a", {"task_id": "t-1"},
            sender=sender)
        # 不立即发，但不是静默丢弃：明确列出待摘要。
        self.assertEqual(calls, [])
        self.assertEqual(result["sent"], [])
        self.assertEqual(len(result["skipped_for_digest"]), 1)
        self.assertEqual(
            result["skipped_for_digest"][0]["preference"], "digest")
        self.assertEqual(result["errors"], [])

    def test_review_requested_digest_also_deferred(self):
        self._sub(preference="digest")
        calls, sender = self._calls()
        result = dispatch_event(
            self.conn, "review-requested", "proj-a", {"task_id": "t-1"},
            sender=sender)
        self.assertEqual(calls, [])
        self.assertEqual(len(result["skipped_for_digest"]), 1)

    def test_instant_completed_sends(self):
        self._sub(preference="instant")
        calls, sender = self._calls()
        result = dispatch_event(
            self.conn, "completed", "proj-a", {"task_id": "t-1"},
            sender=sender)
        self.assertEqual(len(calls), 1)
        self.assertEqual(calls[0]["channel"], "webhook")
        self.assertEqual(calls[0]["payload"], {"task_id": "t-1"})
        self.assertEqual(len(result["sent"]), 1)
        self.assertEqual(result["errors"], [])

    def test_invalid_event_rejected_and_sender_not_called(self):
        self._sub(preference="instant")
        calls, sender = self._calls()
        result = dispatch_event(
            self.conn, "running", "proj-a", {"task_id": "t-1"}, sender=sender)
        self.assertEqual(calls, [])
        self.assertEqual(result["sent"], [])
        self.assertEqual(result["reason"], "event not allowed")
        self.assertTrue(
            any("event not allowed" in e["reason"] for e in result["errors"]))


class TestDispatchSendersAndCredentials(PushTestCase):
    def test_explicit_sender_missing_channel_returns_no_sender_error_not_silent_ok(self):
        # sender=None 时内置真实发送端会真 POST（见 TestRealSenders），
        # “无 sender”只发生在显式 sender 里没有该通道时，仍不静默成功。
        self._sub(channel="webhook")
        result = dispatch_event(
            self.conn, "completed", "proj-a", {"task_id": "t-1"},
            sender={})
        self.assertEqual(result["sent"], [])
        self.assertEqual(len(result["errors"]), 1)
        self.assertIn("no sender", result["errors"][0]["reason"])

    def test_slack_via_channel_dict_sender(self):
        create_subscription(
            self.conn, "proj-a", "user-1", "slack", "C123", "instant")
        slack_calls = []

        def slack_sender(message):
            slack_calls.append(message)

        result = dispatch_event(
            self.conn, "completed", "proj-a", {"task_id": "t-1"},
            sender={"slack": slack_sender})
        self.assertEqual(len(slack_calls), 1)
        self.assertEqual(slack_calls[0]["channel"], "slack")
        self.assertEqual(len(result["sent"]), 1)
        self.assertEqual(result["errors"], [])

    def test_webpush_apns_fcm_missing_credential_error_not_silent_drop(self):
        cases = {
            "webpush": "VAPID",
            "apns": "APNs",
            "fcm": "FCM",
        }
        for channel, keyword in cases.items():
            with self.subTest(channel=channel):
                conn = sqlite3.connect(":memory:")
                try:
                    create_subscription(
                        conn, "proj-a", "user-1", channel,
                        "device-or-endpoint", "instant")
                    calls, sender = self._calls()
                    result = dispatch_event(
                        conn, "failed", "proj-a", {"task_id": "t-1"},
                        sender=sender)
                    # 绝不走假 sender、绝不假装成功：明确报错。
                    self.assertEqual(calls, [])
                    self.assertEqual(result["sent"], [])
                    self.assertEqual(len(result["errors"]), 1)
                    self.assertIn("missing credential", result["errors"][0]["reason"])
                    self.assertIn(keyword, result["errors"][0]["reason"])
                finally:
                    conn.close()

    def test_sender_exception_recorded_not_raised_and_others_continue(self):
        self._sub(user="user-1", target="https://example.invalid/hook-1")

        def flaky(message):
            if message["target"] == "https://example.invalid/hook-1":
                raise RuntimeError("fake send failure")
            calls.append(message)

        calls = []
        # 第二个订阅用另一个 target，同一 sender 只对第一个抛错。
        create_subscription(
            self.conn, "proj-a", "user-2", "webhook",
            "https://example.invalid/hook-2", "instant")
        result = dispatch_event(
            self.conn, "completed", "proj-a", {"task_id": "t-1"},
            sender=flaky)
        self.assertEqual(len(result["sent"]), 1)
        self.assertEqual(len(result["errors"]), 1)
        self.assertIn("sender failed", result["errors"][0]["reason"])
        self.assertEqual(len(calls), 1)


class TestDispatchIsolation(PushTestCase):
    def test_deleted_subscription_no_longer_receives(self):
        sub = self._sub()
        self.assertTrue(delete_subscription(self.conn, sub["id"]))
        calls, sender = self._calls()
        result = dispatch_event(
            self.conn, "failed", "proj-a", {"task_id": "t-1"}, sender=sender)
        self.assertEqual(calls, [])
        self.assertEqual(result["sent"], [])
        self.assertEqual(result["errors"], [])

    def test_cross_project_subscriptions_do_not_mix(self):
        self._sub(project="proj-a", user="user-1")
        self._sub(project="proj-b", user="user-1",
                 target="https://example.invalid/hook-b")
        calls, sender = self._calls()
        # B 项目事件只发给 B 的订阅，A 的订阅收不到。
        result_b = dispatch_event(
            self.conn, "failed", "proj-b", {"task_id": "t-1"}, sender=sender)
        self.assertEqual(len(calls), 1)
        self.assertEqual(calls[0]["target"], "https://example.invalid/hook-b")
        self.assertEqual(len(result_b["sent"]), 1)
        # A 项目事件只发给 A 的订阅。
        calls.clear()
        result_a = dispatch_event(
            self.conn, "failed", "proj-a", {"task_id": "t-1"}, sender=sender)
        self.assertEqual(len(calls), 1)
        self.assertEqual(
            calls[0]["target"], "https://example.invalid/hook")
        self.assertEqual(len(result_a["sent"]), 1)

    def test_list_subscriptions_scoped_by_project_and_user(self):
        self._sub(project="proj-a", user="user-1")
        self._sub(project="proj-a", user="user-2",
                 target="https://example.invalid/hook-2")
        self._sub(project="proj-b", user="user-1",
                 target="https://example.invalid/hook-b")
        self.assertEqual(len(list_subscriptions(self.conn, "proj-a")), 2)
        self.assertEqual(
            len(list_subscriptions(self.conn, "proj-a", "user-1")), 1)
        self.assertEqual(
            len(list_subscriptions(self.conn, "proj-b", "user-1")), 1)


class _RecordingHandler(http.server.BaseHTTPRequestHandler):
    """本地回环接收器：记录每次 POST 的头与 body，按类属性回状态。"""
    requests = []
    status = 200

    def do_POST(self):
        length = int(self.headers.get("Content-Length") or 0)
        body = self.rfile.read(length) if length else b""
        _RecordingHandler.requests.append({
            "path": self.path,
            "content_type": self.headers.get("Content-Type"),
            "body": body,
        })
        self.send_response(_RecordingHandler.status)
        self.send_header("Content-Length", "2")
        self.end_headers()
        self.wfile.write(b"{}")

    def log_message(self, *args):
        pass


class TestRealSenders(PushTestCase):
    """内置真实发送端：对本地回环服务器真 POST，断言方法/头/body；
    不连外网。"""

    def setUp(self):
        super().setUp()
        # P1 SSRF 门默认挡 loopback；本地录制服务器测试显式放行，
        # 生产默认值仍关闭（不要把本开关当成生产默认）。
        self._old_push_allow = os.environ.get(
            "CODEX_PUSH_ALLOW_PRIVATE_TARGETS")
        os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1"
        _RecordingHandler.requests = []
        _RecordingHandler.status = 200
        self.server = http.server.ThreadingHTTPServer(
            ("127.0.0.1", 0), _RecordingHandler)
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(
            target=self.server.serve_forever, daemon=True)
        self.thread.start()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)
        if getattr(self, "_old_push_allow", None) is None:
            os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None)
        else:
            os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = (
                self._old_push_allow)
        super().tearDown()

    def _url(self, path="/hook"):
        return "http://127.0.0.1:%d%s" % (self.port, path)

    def test_webhook_posts_json_via_default_sender(self):
        create_subscription(
            self.conn, "proj-a", "user-1", "webhook", self._url(),
            "instant")
        result = dispatch_event(
            self.conn, "completed", "proj-a", {"task_id": "t-1"},
            sender=None)
        self.assertEqual(len(result["sent"]), 1)
        self.assertEqual(result["errors"], [])
        self.assertEqual(len(_RecordingHandler.requests), 1)
        req = _RecordingHandler.requests[0]
        self.assertEqual(req["content_type"], "application/json")
        body = json.loads(req["body"].decode("utf-8"))
        self.assertEqual(body["event_type"], "completed")
        self.assertEqual(body["project_id"], "proj-a")
        self.assertEqual(body["task_id"], "t-1")
        self.assertEqual(body["payload"], {"task_id": "t-1"})

    def test_webhook_non_2xx_is_error_not_sent(self):
        _RecordingHandler.status = 500
        create_subscription(
            self.conn, "proj-a", "user-1", "webhook", self._url(),
            "instant")
        result = dispatch_event(
            self.conn, "failed", "proj-a", {"task_id": "t-1"},
            sender=None)
        self.assertEqual(result["sent"], [])
        self.assertEqual(len(result["errors"]), 1)
        self.assertIn("delivery failed", result["errors"][0]["reason"])
        self.assertEqual(len(_RecordingHandler.requests), 1)

    def test_webhook_non_http_target_is_error(self):
        create_subscription(
            self.conn, "proj-a", "user-1", "webhook",
            "ftp://example.invalid/hook", "instant")
        with self.assertRaises(ValueError):
            send_webhook({
                "event_type": "completed", "project_id": "proj-a",
                "payload": {"task_id": "t-1"},
                "channel": "webhook",
                "target": "ftp://example.invalid/hook",
                "user_id": "user-1", "subscription_id": "s-1"})
        result = dispatch_event(
            self.conn, "completed", "proj-a", {"task_id": "t-1"},
            sender=None)
        self.assertEqual(result["sent"], [])
        self.assertIn("http(s) URL", result["errors"][0]["reason"])

    def test_slack_posts_text_with_task_and_error(self):
        create_subscription(
            self.conn, "proj-a", "user-1", "slack", self._url(),
            "instant")
        result = dispatch_event(
            self.conn, "failed", "proj-a",
            {"task_id": "t-9", "error": "boom happened"},
            sender=None)
        self.assertEqual(len(result["sent"]), 1)
        body = json.loads(_RecordingHandler.requests[0]["body"].decode())
        self.assertIn("t-9", body["text"])
        self.assertIn("failed", body["text"])
        self.assertIn("boom happened", body["text"])

    def test_webpush_still_missing_credential_with_default_sender(self):
        create_subscription(
            self.conn, "proj-a", "user-1", "webpush",
            "https://example.invalid/endpoint", "instant")
        result = dispatch_event(
            self.conn, "failed", "proj-a", {"task_id": "t-1"},
            sender=None)
        self.assertEqual(result["sent"], [])
        self.assertIn("missing credential", result["errors"][0]["reason"])
        self.assertEqual(_RecordingHandler.requests, [])

    def test_explicit_sender_wins_over_default(self):
        create_subscription(
            self.conn, "proj-a", "user-1", "webhook", self._url(),
            "instant")
        calls = []

        def fake(message):
            calls.append(message)

        result = dispatch_event(
            self.conn, "completed", "proj-a", {"task_id": "t-1"},
            sender={"webhook": fake})
        self.assertEqual(len(calls), 1)
        self.assertEqual(len(result["sent"]), 1)
        self.assertEqual(_RecordingHandler.requests, [])


class TestDigestPersistence(PushTestCase):
    """G11：digest 非失败事件必须持久化、按组汇总发一次。"""

    def test_digest_events_persisted_and_sent_once_per_group(self):
        self._sub(preference="digest")
        calls, sender = self._calls()
        result = dispatch_event(
            self.conn, "completed", "proj-a", {"task_id": "t-1"},
            sender=sender)
        self.assertEqual(calls, [])
        self.assertEqual(len(result["skipped_for_digest"]), 1)
        dispatch_event(
            self.conn, "completed", "proj-a", {"task_id": "t-2"},
            sender=sender)
        pending = list_pending_digest(self.conn, "proj-a")
        self.assertEqual(len(pending), 2)
        digest = send_digest_due(self.conn, sender=sender)
        self.assertEqual(len(digest["sent"]), 1)
        self.assertEqual(digest["sent"][0]["count"], 2)
        self.assertEqual(len(calls), 1)
        message = calls[0]
        self.assertEqual(message["event_type"], "digest")
        self.assertTrue(message["payload"]["digest"])
        self.assertEqual(message["payload"]["count"], 2)
        self.assertEqual(list_pending_digest(self.conn, "proj-a"), [])
        # 第二次无待发事件，不再重复发。
        again = send_digest_due(self.conn, sender=sender)
        self.assertEqual(again["sent"], [])
        self.assertEqual(len(calls), 1)

    def test_failure_events_skip_digest_and_send_immediately(self):
        self._sub(preference="digest")
        calls, sender = self._calls()
        result = dispatch_event(
            self.conn, "failed", "proj-a", {"task_id": "t-1"},
            sender=sender)
        self.assertEqual(len(calls), 1)
        self.assertEqual(result["skipped_for_digest"], [])
        self.assertEqual(list_pending_digest(self.conn, "proj-a"), [])


class TestOutboxRetry(PushTestCase):
    """G12：推送失败进 outbox、到期重试、3 次后丢弃。"""

    def _failing(self):
        def sender(message):
            raise RuntimeError("downstream down")
        return sender

    def test_failed_send_enqueued_then_retried_successfully(self):
        self._sub(preference="instant")
        result = dispatch_event(
            self.conn, "failed", "proj-a", {"task_id": "t-1"},
            sender=self._failing())
        self.assertEqual(len(result["errors"]), 1)
        self.assertEqual(len(result["outbox"]), 1)
        rows = list_outbox(self.conn, "proj-a")
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]["attempts"], 0)
        # 未到期不重试。
        early = process_outbox_due(self.conn, sender=self._failing(),
                                   now=rows[0]["next_attempt_at"] - 1)
        self.assertEqual(early["sent"], [])
        self.assertEqual(len(list_outbox(self.conn, "proj-a")), 1)
        # 到期后成功发送，行删掉。
        calls, sender = self._calls()
        done = process_outbox_due(
            self.conn, sender=sender,
            now=rows[0]["next_attempt_at"] + 1)
        self.assertEqual(len(done["sent"]), 1)
        self.assertEqual(len(calls), 1)
        self.assertEqual(calls[0]["event_type"], "failed")
        self.assertEqual(list_outbox(self.conn, "proj-a"), [])

    def test_three_failures_drop_the_row(self):
        self._sub(preference="instant")
        dispatch_event(
            self.conn, "failed", "proj-a", {"task_id": "t-1"},
            sender=self._failing())
        self.assertEqual(len(list_outbox(self.conn, "proj-a")), 1)
        # now 取远未来 + backoff=0，让三次重试都在同一 now 发生
        # （单测确定性；真实退避由上一例的 next_attempt_at 覆盖）。
        for i in range(2):
            out = process_outbox_due(
                self.conn, sender=self._failing(), now=10 ** 10,
                backoff_seconds=0)
            self.assertEqual(out["failed"], [])
            self.assertEqual(len(list_outbox(self.conn, "proj-a")), 1)
        out = process_outbox_due(
            self.conn, sender=self._failing(), now=10 ** 10,
            backoff_seconds=0)
        self.assertEqual(len(out["failed"]), 1)
        self.assertEqual(out["failed"][0]["attempts"], 3)
        self.assertEqual(list_outbox(self.conn, "proj-a"), [])

    def test_missing_credential_does_not_enter_outbox(self):
        create_subscription(
            self.conn, "proj-a", "user-1", "webpush",
            "https://example.invalid/endpoint", "instant")
        result = dispatch_event(
            self.conn, "failed", "proj-a", {"task_id": "t-1"},
            sender=self._calls()[1])
        self.assertEqual(len(result["errors"]), 1)
        self.assertEqual(result["outbox"], [])
        self.assertEqual(list_outbox(self.conn, "proj-a"), [])


if __name__ == "__main__":
    unittest.main()