auto-sync: 2026-05-18 00:51:40
This commit is contained in:
@@ -424,3 +424,83 @@ class TestV27Migration:
|
||||
assert "stages_json" in cols
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
# ===================================================================
|
||||
# Mail API
|
||||
# ===================================================================
|
||||
|
||||
class TestMailRoutes:
|
||||
"""Mail Tab API 测试(用 Blackboard 直接操作 _mail DB)"""
|
||||
|
||||
def test_mail_create_and_list(self, tmp_path):
|
||||
from src.blackboard.operations import Blackboard
|
||||
from src.blackboard.db import init_db
|
||||
mail_db = tmp_path / "_mail" / "blackboard.db"
|
||||
mail_db.parent.mkdir(parents=True)
|
||||
init_db(mail_db)
|
||||
bb = Blackboard(mail_db)
|
||||
|
||||
meta = {"from": "pangtong", "type": "review_request", "is_read": False}
|
||||
bb.create_task(Task(
|
||||
id="mail-001", title="请评审v2.7代码",
|
||||
description="代码已提交,请评审",
|
||||
assignee="simayi-challenger",
|
||||
assigned_by="pangtong-fujunshi",
|
||||
must_haves=json.dumps(meta),
|
||||
card_id="default",
|
||||
task_type="mail",
|
||||
))
|
||||
|
||||
tasks = bb.list_tasks()
|
||||
assert len(tasks) == 1
|
||||
assert tasks[0].title == "请评审v2.7代码"
|
||||
assert tasks[0].assignee == "simayi-challenger"
|
||||
|
||||
# 检查 must_haves 元数据
|
||||
t = bb.get_task("mail-001")
|
||||
m = json.loads(t.must_haves)
|
||||
assert m["from"] == "pangtong"
|
||||
assert m["is_read"] is False
|
||||
|
||||
def test_mail_mark_read(self, tmp_path):
|
||||
from src.blackboard.operations import Blackboard
|
||||
from src.blackboard.db import init_db
|
||||
mail_db = tmp_path / "_mail" / "blackboard.db"
|
||||
mail_db.parent.mkdir(parents=True)
|
||||
init_db(mail_db)
|
||||
bb = Blackboard(mail_db)
|
||||
|
||||
meta = {"from": "zhangfei", "is_read": False}
|
||||
bb.create_task(Task(
|
||||
id="mail-002", title="测试",
|
||||
must_haves=json.dumps(meta), card_id="default",
|
||||
))
|
||||
|
||||
# 更新 is_read
|
||||
conn = bb._conn()
|
||||
try:
|
||||
meta["is_read"] = True
|
||||
conn.execute("UPDATE tasks SET must_haves=? WHERE id=?",
|
||||
(json.dumps(meta), "mail-002"))
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
t = bb.get_task("mail-002")
|
||||
assert json.loads(t.must_haves)["is_read"] is True
|
||||
|
||||
def test_mail_filter_by_to(self, tmp_path):
|
||||
from src.blackboard.operations import Blackboard
|
||||
from src.blackboard.db import init_db
|
||||
mail_db = tmp_path / "_mail" / "blackboard.db"
|
||||
mail_db.parent.mkdir(parents=True)
|
||||
init_db(mail_db)
|
||||
bb = Blackboard(mail_db)
|
||||
|
||||
bb.create_task(Task(id="m1", title="A", assignee="simayi", card_id="default"))
|
||||
bb.create_task(Task(id="m2", title="B", assignee="zhangfei", card_id="default"))
|
||||
bb.create_task(Task(id="m3", title="C", assignee="simayi", card_id="default"))
|
||||
|
||||
tasks = bb.list_tasks(assignee="simayi")
|
||||
assert len(tasks) == 2
|
||||
|
||||
Reference in New Issue
Block a user