151 lines
4.6 KiB
Python
151 lines
4.6 KiB
Python
import pytest
|
|
|
|
from app.services.human_handoff_service import (
|
|
MESSAGE_SCOPE_EXTERNAL,
|
|
MESSAGE_SCOPE_HANDOFF,
|
|
_body_has_quote_trail,
|
|
_mirror_from_identity,
|
|
build_handoff_subject,
|
|
maybe_trigger_handoff,
|
|
process_human_reply,
|
|
redact_human_address,
|
|
resolve_handoff_from_inbound,
|
|
select_matching_handoff_rule,
|
|
)
|
|
|
|
|
|
def test_scope_constants_are_distinct():
|
|
assert MESSAGE_SCOPE_EXTERNAL != MESSAGE_SCOPE_HANDOFF
|
|
|
|
|
|
def test_human_handoff_runtime_is_enabled():
|
|
assert callable(maybe_trigger_handoff)
|
|
assert callable(process_human_reply)
|
|
assert callable(resolve_handoff_from_inbound)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mirror_body_appends_external_history_when_missing(monkeypatch):
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
from app.services.human_handoff_service import _mirror_body_with_trail
|
|
|
|
parent = {
|
|
"_id": "parent",
|
|
"sender": {"name": "Parth Dixit", "address": "parthdixit091@gmail.com"},
|
|
"body_text": "sender: C\n\nOn earlier Suyash Dixit wrote:\n> responder: B",
|
|
"body_html": None,
|
|
"sent_at": None,
|
|
}
|
|
monkeypatch.setattr(
|
|
"app.services.human_handoff_service._latest_thread_message",
|
|
AsyncMock(return_value=parent),
|
|
)
|
|
monkeypatch.setattr(
|
|
"app.services.human_handoff_service.load_mirror_payload",
|
|
AsyncMock(return_value=SimpleNamespace(body_text=parent["body_text"], body_html=None)),
|
|
)
|
|
text, html = await _mirror_body_with_trail(
|
|
SimpleNamespace(),
|
|
account_id="account-1",
|
|
handoff={"external_thread_id": "thread"},
|
|
source={"_id": "child", "new_reply_content": "responder: D", "new_reply_html": None},
|
|
payload=SimpleNamespace(body_text="responder: D", body_html=None),
|
|
)
|
|
assert text.startswith("responder: D")
|
|
assert "parthdixit091@gmail.com" in text
|
|
assert "> sender: C" in text
|
|
assert "responder: B" in text
|
|
assert html and "gmail_quote" in html
|
|
|
|
|
|
def test_quote_trail_is_detected_in_forwarded_history():
|
|
assert _body_has_quote_trail(
|
|
"responder: B\n\nOn Tue, Oct 6, 2026 at 7:44 AM Suyash Dixit wrote:\n> sender: A",
|
|
None,
|
|
)
|
|
assert _body_has_quote_trail(
|
|
"sender: C",
|
|
'<div>sender: C</div><div class="gmail_quote">sender: A</div>',
|
|
)
|
|
assert _body_has_quote_trail("sender: E", None) is False
|
|
|
|
|
|
def test_mirror_from_stays_on_gentic_address():
|
|
name, address = _mirror_from_identity(
|
|
{
|
|
"direction": "inbound",
|
|
"sender": {
|
|
"name": "Parth Dixit",
|
|
"address": "parthdixit091@gmail.com",
|
|
},
|
|
},
|
|
"Suyash Dixit",
|
|
"suyash@gentic.email",
|
|
)
|
|
assert name == "Parth Dixit"
|
|
assert address == "suyash@gentic.email"
|
|
assert "gmail.com" not in address
|
|
|
|
agent_name, agent_address = _mirror_from_identity(
|
|
{"direction": "outbound", "sender": {"address": "suyash@gentic.email"}},
|
|
"Suyash Dixit",
|
|
"suyash@gentic.email",
|
|
)
|
|
assert agent_name == "Suyash Dixit"
|
|
assert agent_address == "suyash@gentic.email"
|
|
|
|
|
|
def test_private_subject_contains_customer_name_and_external_subject():
|
|
subject = build_handoff_subject(
|
|
sender_name="Parth Dixit",
|
|
external_sender="parthdixit091@gmail.com",
|
|
external_subject="Invoice 42",
|
|
)
|
|
assert subject == (
|
|
"Parth Dixit · handoff · parthdixit091@gmail.com · Invoice 42"
|
|
)
|
|
|
|
|
|
def test_private_subject_removes_header_line_breaks():
|
|
subject = build_handoff_subject(
|
|
sender_name="Urgent\r\nBcc: attacker@example.com",
|
|
external_sender="customer@example.com",
|
|
external_subject="Help",
|
|
)
|
|
assert "\r" not in subject
|
|
assert "\n" not in subject
|
|
|
|
|
|
def test_human_address_is_replaced_before_external_relay():
|
|
value = "Regards,\nOwner <private.owner@example.com>"
|
|
assert redact_human_address(
|
|
value,
|
|
human_email="private.owner@example.com",
|
|
agent_email="agent@gentic.email",
|
|
) == "Regards,\nOwner <agent@gentic.email>"
|
|
|
|
|
|
def test_lowest_priority_matching_rule_wins():
|
|
rules = [
|
|
{
|
|
"name": "Backup",
|
|
"priority": 20,
|
|
"labels": ["All mail"],
|
|
"email_types": ["all"],
|
|
"instructions": "",
|
|
},
|
|
{
|
|
"name": "Primary",
|
|
"priority": 1,
|
|
"labels": ["All mail"],
|
|
"email_types": ["all"],
|
|
"instructions": "",
|
|
},
|
|
]
|
|
selected = select_matching_handoff_rule(
|
|
rules,
|
|
email={"subject": "Hello", "body_text": "Need help"},
|
|
)
|
|
assert selected["name"] == "Primary"
|