agentic.email/backend/tests/test_handoff_mirror.py

import pytest

from app.services.handoff_email_templates import (
    external_reply_subject,
    format_sender_line,
    handoff_email_subject,
)
from app.services.handoff_mirror import load_mirror_payload
from app.services.html_sanitizer import sanitize_email_html
from app.services.mime_body_extractor import (
    extract_mirror_content,
    extract_new_reply_mime_content,
)
from app.services.thread_service import threading_fields_from_parent


@pytest.mark.asyncio
async def test_mirror_body_is_unchanged_without_gentic_wrapper():
    payload = await load_mirror_payload(
        {
            "body_text": "one last attempt. i am tired",
            "body_html": "<p><strong>one last attempt.</strong> i am tired</p>",
        }
    )
    assert payload.body_text == "one last attempt. i am tired"
    assert payload.body_html == (
        "<p><strong>one last attempt.</strong> i am tired</p>"
    )
    assert "Human Handoff" not in payload.body_text
    assert "gentic-handoff" not in (payload.body_html or "")


@pytest.mark.asyncio
async def test_mirror_plain_text_fallback_is_unchanged():
    payload = await load_mirror_payload(
        {
            "body_text": "one last attempt. i am tired",
            "body_html": None,
        }
    )
    assert payload.body_text == "one last attempt. i am tired"
    assert payload.body_html is None


def test_format_sender_with_name():
    assert (
        format_sender_line({"name": "Rahul Sharma", "address": "rahul@example.com"})
        == "Rahul Sharma <rahul@example.com>"
    )


def test_quoted_history_uses_original_sender_not_human_owner():
    from datetime import datetime, timezone

    from app.services.handoff_email_templates import (
        attach_quoted_history,
        build_gmail_quote,
    )

    quoted_text, quoted_html = build_gmail_quote(
        sender={"name": "Parth Dixit", "address": "parthdixit091@gmail.com"},
        event_at=datetime(2026, 10, 6, 7, 29, tzinfo=timezone.utc),
        body_text="sender: C",
        body_html="<div>sender: C</div>",
    )
    body_text, body_html = attach_quoted_history(
        new_text="responder: D",
        new_html="<div>responder: D</div>",
        quoted_text=quoted_text,
        quoted_html=quoted_html,
    )
    assert body_text.startswith("responder: D")
    assert "Parth Dixit <parthdixit091@gmail.com>" in body_text
    assert "> sender: C" in body_text
    assert "parthdixit091@gmail.com" in body_html
    assert "Suyash AIML" not in body_text


def test_external_threading_headers_from_customer_message():
    parent = {
        "_id": "mongo-id",
        "message_id": "<a@gmail.com>",
        "references": [],
    }
    threading = threading_fields_from_parent(parent)
    assert threading["in_reply_to"] == "<a@gmail.com>"
    assert threading["references"] == ["<a@gmail.com>"]


def test_external_reply_subject_uses_customer_subject():
    assert external_reply_subject("lets try one more time") == "Re: lets try one more time"
    assert handoff_email_subject("lets try one more time") == "Human Handoff · lets try one more time"


def test_sanitize_removes_script():
    dirty = '<p>Hello</p><script>alert(1)</script><img src="x" onerror="alert(1)">'
    clean = sanitize_email_html(dirty)
    assert "<script" not in clean.lower()
    assert "onerror" not in clean.lower()
    assert "Hello" in clean


def test_extract_html_from_multipart():
    raw = (
        b"From: sender@example.com\r\n"
        b"To: agent@gentic.email\r\n"
        b"Subject: Invoice approval\r\n"
        b'MIME-Version: 1.0\r\nContent-Type: multipart/alternative; boundary="b"\r\n\r\n'
        b"--b\r\nContent-Type: text/plain; charset=utf-8\r\n\r\nPlain fallback\r\n"
        b"--b\r\nContent-Type: text/html; charset=utf-8\r\n\r\n"
        b"<h2>Invoice</h2><table><tr><td>Amount</td></tr></table>\r\n"
        b"--b--\r\n"
    )
    extracted = extract_mirror_content(raw)
    assert "Invoice" in (extracted.body_html or "")
    assert "<table>" in (extracted.body_html or "")


def test_rich_human_reply_keeps_new_html_and_removes_private_quote():
    raw = (
        b"From: owner@example.com\r\n"
        b"To: agent@gentic.email\r\n"
        b"Subject: Re: Finance handoff\r\n"
        b'MIME-Version: 1.0\r\nContent-Type: multipart/alternative; boundary="b"\r\n\r\n'
        b"--b\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n"
        b"Approved with urgency.\r\n\r\n"
        b"On Tue, Oct 6, 2026 at 6:00 AM agent@gentic.email wrote:\r\n"
        b"> private mirror\r\n"
        b"--b\r\nContent-Type: text/html; charset=utf-8\r\n\r\n"
        b'<div><strong>Approved</strong> with <em>urgency</em>.</div>'
        b'<div class="gmail_quote">private mirror</div>\r\n'
        b"--b--\r\n"
    )
    extracted = extract_new_reply_mime_content(raw)
    assert extracted.body_text == "Approved with urgency."
    assert "<strong>Approved</strong>" in (extracted.body_html or "")
    assert "private mirror" not in (extracted.body_html or "")


def test_rich_human_reply_keeps_new_attachment():
    raw = (
        b"From: owner@example.com\r\n"
        b"To: agent@gentic.email\r\n"
        b"MIME-Version: 1.0\r\n"
        b'Content-Type: multipart/mixed; boundary="m"\r\n\r\n'
        b"--m\r\nContent-Type: text/plain; charset=utf-8\r\n\r\nApproved\r\n"
        b"--m\r\nContent-Type: application/pdf\r\n"
        b'Content-Disposition: attachment; filename="approval.pdf"\r\n\r\nPDF\r\n'
        b"--m--\r\n"
    )
    extracted = extract_new_reply_mime_content(raw)
    assert extracted.body_text == "Approved"
    assert [part.filename for part in extracted.file_attachments] == [
        "approval.pdf"
    ]