305 lines
12 KiB
Python
305 lines
12 KiB
Python
"""push.py 单测：订阅增删查、偏好过滤、分发错误不静默丢。
|
|
|
|
只用标准库，不联网、不向真实推送服务发任何请求：所有发送都用
|
|
假 sender（可调用对象或按通道分的 dict）注入；webpush/apns/fcm
|
|
只断言缺凭证报错，不调用任何 sender。
|
|
本文件不含任何真实密钥、token、证书或私钥。
|
|
"""
|
|
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
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_subscriptions,
|
|
)
|
|
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_subscriptions,
|
|
)
|
|
|
|
|
|
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_webhook_without_sender_returns_no_sender_error_not_silent_ok(self):
|
|
self._sub(channel="webhook")
|
|
result = dispatch_event(
|
|
self.conn, "completed", "proj-a", {"task_id": "t-1"}, sender=None)
|
|
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|