agentic.email/backend/tests/test_external_reply_service.py
2026-10-06 02:44:54 +05:30

57 lines
2 KiB
Python

from bson import ObjectId
from app.services.external_reply_service import build_external_reply_plan
from app.services.thread_service import threading_fields_from_parent
def test_build_external_reply_references_gmail_parent():
parent = {
"_id": ObjectId(),
"internet_message_id": "<gmail-a@mail.gmail.com>",
"message_id": "<gmail-a@mail.gmail.com>",
"references": [],
"subject": "lets try one more time",
}
plan = build_external_reply_plan(
parent_external_message=parent,
body_text="I hear you.",
recipients=["customer@gmail.com"],
)
assert plan.threading.in_reply_to == "<gmail-a@mail.gmail.com>"
assert plan.threading.references == ["<gmail-a@mail.gmail.com>"]
assert plan.subject == "Re: lets try one more time"
assert plan.threading.replying_to_internet_message_id == "<gmail-a@mail.gmail.com>"
def test_build_external_reply_chains_references_for_multi_round():
parent = {
"_id": ObjectId(),
"internet_message_id": "<a2@gmail.com>",
"message_id": "<a2@gmail.com>",
"references": ["<a1@gmail.com>", "<g1@gentic.email>"],
"subject": "Re: lets try one more time",
}
threading = threading_fields_from_parent(parent)
assert threading["in_reply_to"] == "<a2@gmail.com>"
assert threading["references"] == [
"<a1@gmail.com>",
"<g1@gentic.email>",
"<a2@gmail.com>",
]
def test_external_reply_never_uses_internal_object_id_as_in_reply_to():
parent = {
"_id": ObjectId(),
"internet_message_id": "<gmail-a@mail.gmail.com>",
"message_id": "<gmail-a@mail.gmail.com>",
"references": [],
"subject": "hello",
}
plan = build_external_reply_plan(
parent_external_message=parent,
body_text="reply",
recipients=["customer@gmail.com"],
)
assert str(parent["_id"]) not in (plan.threading.in_reply_to or "")
assert all(str(parent["_id"]) not in ref for ref in plan.threading.references)