group_fqcd_jr/tests/unit/service/test_offsite_mail_adapter.py

from __future__ import annotations

import os
from datetime import datetime
from email.message import EmailMessage
from pathlib import Path

from app.core.config import Settings
from app.service.offsite_mail_adapter import (
    OffsiteImapReceiver,
    OffsiteMailStorage,
    RawMailAttachment,
    RawMailMessage,
)


class FakeImapConnection:
    def __init__(self, messages: dict[str, bytes] | None = None) -> None:
        self.messages = messages or {}
        self.noop_count = 0
        self.logout_count = 0
        self.selected: list[tuple[str, bool]] = []
        self.uid_calls: list[tuple[str, tuple[object, ...]]] = []

    def login(self, user: str, password: str) -> tuple[str, list[bytes]]:
        return "OK", [b"LOGIN completed"]

    def select(self, mailbox: str, readonly: bool = False) -> tuple[str, list[bytes]]:
        self.selected.append((mailbox, readonly))
        return "OK", [b"2"]

    def uid(self, command: str, *args: object) -> tuple[str, list[object]]:
        self.uid_calls.append((command, args))
        if command == "search":
            return "OK", [b" ".join(uid.encode("ascii") for uid in self.messages)]
        if command == "fetch":
            uid = str(args[0])
            return "OK", [(b"RFC822", self.messages[uid]), b")"]
        return "NO", []

    def noop(self) -> tuple[str, list[bytes]]:
        self.noop_count += 1
        return "OK", [b"NOOP completed"]

    def logout(self) -> tuple[str, list[bytes]]:
        self.logout_count += 1
        return "BYE", [b"LOGOUT completed"]


def test_health_check_returns_disabled_when_imap_switch_is_off() -> None:
    connection = FakeImapConnection()
    receiver = OffsiteImapReceiver(_settings(offsite_imap_enabled=False), connection)

    assert receiver.health_check() == {"status": "disabled", "message": "场外 IMAP 未启用"}
    assert connection.noop_count == 0


def test_health_check_reports_missing_required_imap_config() -> None:
    connection = FakeImapConnection()
    receiver = OffsiteImapReceiver(
        _settings(
            offsite_imap_enabled=True,
            offsite_imap_host="",
            offsite_imap_username="",
            offsite_imap_password="",
        ),
        connection,
    )

    result = receiver.health_check()

    assert result["status"] == "misconfigured"
    assert result["missing"] == [
        "OFFSITE_IMAP_HOST",
        "OFFSITE_IMAP_USERNAME",
        "OFFSITE_IMAP_PASSWORD",
    ]
    assert connection.noop_count == 0


def test_fetch_since_selects_inbox_and_filters_sender_whitelist() -> None:
    accepted = _raw_email("15008108550@163.com", "<accepted@example.local>")
    rejected = _raw_email("blocked@example.com", "<blocked@example.local>")
    connection = FakeImapConnection({"5": accepted, "6": rejected})
    receiver = OffsiteImapReceiver(_settings(offsite_imap_enabled=True), connection)

    messages = receiver.fetch_since("4", limit=20)

    assert connection.selected == [("INBOX", True)]
    assert ("search", (None, "UID 5:*")) in connection.uid_calls
    assert ("fetch", ("5", "(RFC822)")) in connection.uid_calls
    assert ("fetch", ("6", "(RFC822)")) in connection.uid_calls
    assert len(messages) == 1
    assert messages[0].imap_uid == "5"
    assert messages[0].message_id == "<accepted@example.local>"
    assert messages[0].sender == "15008108550@163.com"
    assert messages[0].return_path == "15008108550@163.com"
    assert messages[0].attachments[0].filename == "申购申请单.pdf"
    assert messages[0].auth_result["dkim_signature_present"] is True


def test_mail_storage_saves_eml_and_attachments_without_overwriting(tmp_path: Path) -> None:
    storage = OffsiteMailStorage(tmp_path)
    mail = _raw_mail_message(
        raw_message=b"From: first@example.com\r\n\r\nfirst",
        attachment_payload=b"first attachment",
    )

    first = storage.save(mail)
    second = storage.save(
        _raw_mail_message(
            raw_message=b"From: second@example.com\r\n\r\nsecond",
            attachment_payload=b"second attachment",
        )
    )

    eml_path = Path(first.eml_path)
    first_attachment_path = Path(first.attachments[0].original_file_path)
    second_attachment_path = Path(second.attachments[0].original_file_path)
    assert eml_path.read_bytes() == b"From: first@example.com\r\n\r\nfirst"
    assert first_attachment_path.read_bytes() == b"first attachment"
    assert second_attachment_path.read_bytes() == b"second attachment"
    assert first_attachment_path != second_attachment_path
    assert first.attachments[0].file_hash != second.attachments[0].file_hash

    for path in (eml_path, first_attachment_path, second_attachment_path):
        os.chmod(path, 0o666)


def _settings(**updates: object) -> Settings:
    values: dict[str, object] = {
        "jwt_issuer": "jr-local",
        "jwt_audience": "jr-agent-platform",
        "mysql_dsn": "sqlite+aiosqlite:///test.db",
        "redis_url": "redis://127.0.0.1:6379/0",
        "milvus_uri": "http://127.0.0.1:19530",
        "neo4j_uri": "bolt://127.0.0.1:7687",
        "offsite_imap_host": "imap.example.local",
        "offsite_imap_username": "15273589815@163.com",
        "offsite_imap_password": "test-auth-code",
        "offsite_allowed_senders": ("15008108550@163.com",),
    }
    values.update(updates)
    return Settings(**values)


def _raw_email(sender: str, message_id: str) -> bytes:
    message = EmailMessage()
    message["From"] = sender
    message["To"] = "15273589815@163.com"
    message["Message-ID"] = message_id
    message["Return-Path"] = sender
    message["Authentication-Results"] = "mx.example.local; spf=pass"
    message["Received-SPF"] = "pass"
    message["DKIM-Signature"] = "v=1; a=rsa-sha256; b=test"
    message.set_content("场外基金业务邮件")
    message.add_attachment(
        b"pdf-bytes",
        maintype="application",
        subtype="pdf",
        filename="申购申请单.pdf",
    )
    return message.as_bytes()


def _raw_mail_message(raw_message: bytes, attachment_payload: bytes) -> RawMailMessage:
    return RawMailMessage(
        imap_uid="8",
        message_id="<same@example.local>",
        sender="15008108550@163.com",
        return_path="15008108550@163.com",
        auth_result={"spf": "pass"},
        raw_message=raw_message,
        received_at=datetime(2026, 9, 10, 9, 30, 0),
        attachments=(
            RawMailAttachment(
                filename="申购申请单.pdf",
                media_type="application/pdf",
                payload=attachment_payload,
            ),
        ),
    )