from __future__ import annotations import base64 from datetime import UTC, datetime import poplib import ssl from types import SimpleNamespace import unittest from unittest.mock import patch from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from govoplan_access.backend.db.models import Account, User from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef from govoplan_core.db.base import Base from govoplan_core.security.secrets import decrypt_secret from govoplan_mail.backend.config import Pop3Config, TransportSecurity from govoplan_mail.backend.db.models import ( MailPop3Import, MailServerEndpoint, MailServerProfile, ) from govoplan_mail.backend.pop3_imports import ( Pop3ImportResult, create_pop3_imports, list_pop3_imports, ) from govoplan_mail.backend.router import import_profile_pop3_messages from govoplan_mail.backend.schemas import MailPop3ImportRequest from govoplan_mail.backend.sending.pop3 import ( Pop3ConfigurationError, Pop3DownloadedMessage, Pop3MessageSummary, Pop3ProviderError, _open_pop3, delete_pop3_messages, download_pop3_messages, preview_pop3_messages, ) _RAW = ( b"Subject: Legacy notice\r\n" b"From: Office \r\n" b"To: Subject \r\n" b"Message-ID: \r\n" b"\r\n" b"A bounded legacy message.\r\n" ) class _Pop3Client: def __init__(self, *, quit_error: Exception | None = None) -> None: self.deletions: list[int] = [] self.quit_calls = 0 self.rset_calls = 0 self.close_calls = 0 self.quit_error = quit_error def stat(self): return 1, len(_RAW) def uidl(self): return b"+OK", [b"1 uid-1"], 1 def list(self): return b"+OK", [f"1 {len(_RAW)}".encode("ascii")], 1 def top(self, _number, _lines): return b"+OK", _RAW.rstrip(b"\r\n").split(b"\r\n"), len(_RAW) def retr(self, _number): return b"+OK", _RAW.rstrip(b"\r\n").split(b"\r\n"), len(_RAW) def dele(self, number): self.deletions.append(number) def quit(self): self.quit_calls += 1 if self.quit_error is not None: raise self.quit_error return b"+OK" def rset(self): self.rset_calls += 1 def close(self): self.close_calls += 1 def _config(**changes) -> Pop3Config: values = { "host": "pop3.example.test", "security": "tls", "username": "legacy-user", "password": "legacy-password", "legacy_import_enabled": True, } values.update(changes) return Pop3Config.model_validate(values) def _download(uidl: str = "uid-1") -> Pop3DownloadedMessage: summary = Pop3MessageSummary( message_number=1, uidl=uidl, subject="Legacy notice", from_header="Office ", to_header="Subject ", date="Sat, 22 Aug 2026 10:00:00 +0200", message_id="", size_bytes=len(_RAW), body_preview="A bounded legacy message.", ) import hashlib return Pop3DownloadedMessage( message_number=1, uidl=uidl, raw=_RAW, raw_sha256=hashlib.sha256(_RAW).hexdigest(), summary=summary, ) class Pop3TransportTests(unittest.TestCase): def test_legacy_import_is_disabled_until_explicitly_enabled(self) -> None: with self.assertRaisesRegex(Pop3ConfigurationError, "disabled"): preview_pop3_messages( pop3_config=_config(legacy_import_enabled=False), limit=10, ) with self.assertRaisesRegex(ValueError, "batch size limit"): _config(max_message_bytes=2 * 1024 * 1024, max_batch_bytes=1024 * 1024) def test_preview_and_download_are_non_destructive(self) -> None: preview_client = _Pop3Client() with patch( "govoplan_mail.backend.sending.pop3._open_pop3", return_value=preview_client, ): preview = preview_pop3_messages(pop3_config=_config(), limit=10) self.assertEqual(["uid-1"], [item.uidl for item in preview.messages]) self.assertEqual([], preview_client.deletions) self.assertEqual(1, preview_client.quit_calls) download_client = _Pop3Client() with patch( "govoplan_mail.backend.sending.pop3._open_pop3", return_value=download_client, ): downloaded = download_pop3_messages( pop3_config=_config(), uidls=("uid-1",) ) self.assertEqual(_RAW, downloaded[0].raw) self.assertEqual([], download_client.deletions) self.assertEqual(1, download_client.quit_calls) def test_source_deletion_needs_policy_and_commits_with_quit(self) -> None: with self.assertRaisesRegex(Pop3ConfigurationError, "disabled"): delete_pop3_messages(pop3_config=_config(), uidls=("uid-1",)) client = _Pop3Client() with patch( "govoplan_mail.backend.sending.pop3._open_pop3", return_value=client ): result = delete_pop3_messages( pop3_config=_config(allow_delete_after_import=True), uidls=("uid-1",), ) self.assertEqual(("uid-1",), result.deleted_uidls) self.assertEqual([1], client.deletions) self.assertEqual(1, client.quit_calls) def test_quit_failure_marks_deletion_outcome_unknown(self) -> None: client = _Pop3Client(quit_error=poplib.error_proto("connection lost")) with ( patch( "govoplan_mail.backend.sending.pop3._open_pop3", return_value=client, ), self.assertRaises(Pop3ProviderError) as captured, ): delete_pop3_messages( pop3_config=_config(allow_delete_after_import=True), uidls=("uid-1",), ) self.assertTrue(captured.exception.outcome_unknown) self.assertEqual([1], client.deletions) def test_tls_and_authentication_failures_are_sanitized(self) -> None: with ( patch( "govoplan_mail.backend.sending.pop3.validate_outbound_host" ), patch( "govoplan_mail.backend.sending.pop3._OutboundPolicyPOP3SSL", side_effect=ssl.SSLError("private TLS detail"), ), self.assertRaisesRegex(Pop3ProviderError, "TLS negotiation failed"), ): _open_pop3(_config()) auth_client = _Pop3Client() auth_client.user = lambda _value: None # type: ignore[attr-defined] auth_client.pass_ = lambda _value: (_ for _ in ()).throw( # type: ignore[attr-defined] poplib.error_proto("private auth detail") ) with ( patch( "govoplan_mail.backend.sending.pop3.validate_outbound_host" ), patch( "govoplan_mail.backend.sending.pop3._OutboundPolicyPOP3", return_value=auth_client, ), self.assertRaisesRegex(Pop3ProviderError, "authentication failed"), ): _open_pop3(_config(security=TransportSecurity.PLAIN)) class Pop3PersistenceTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:", future=True) Base.metadata.create_all( self.engine, tables=( Account.__table__, User.__table__, MailServerProfile.__table__, MailServerEndpoint.__table__, MailPop3Import.__table__, ), ) self.session = sessionmaker(bind=self.engine, expire_on_commit=False)() self.profile = MailServerProfile( id="profile-1", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", name="Legacy source", slug="legacy-source", smtp_config={}, ) self.server = MailServerEndpoint( id="server-1", profile_id=self.profile.id, tenant_id="tenant-1", protocol="pop3", name="Legacy POP3", config={"legacy_import_enabled": True}, scope_type="tenant", scope_id="tenant-1", transport_revision="revision-1", ) self.session.add_all((self.profile, self.server)) self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def test_import_is_encrypted_and_duplicate_uidl_is_reused(self) -> None: first = create_pop3_imports( self.session, tenant_id="tenant-1", profile_id=self.profile.id, pop3_server_id=self.server.id, pop3_credential_id=None, transport_revision="revision-1", messages=(_download(),), user_id=None, deletion_requested=False, ) self.session.commit() self.assertEqual(1, len(first.imported)) encrypted = first.imported[0].raw_message_encrypted self.assertNotIn("Legacy notice", encrypted) self.assertEqual( _RAW, base64.b64decode(decrypt_secret(encrypted) or ""), ) self.assertEqual("not_requested", first.imported[0].deletion_status) repeated = create_pop3_imports( self.session, tenant_id="tenant-1", profile_id=self.profile.id, pop3_server_id=self.server.id, pop3_credential_id=None, transport_revision="revision-1", messages=(_download(),), user_id=None, deletion_requested=False, ) self.assertEqual((), repeated.imported) self.assertEqual((first.imported[0].id,), tuple(row.id for row in repeated.duplicates)) self.assertEqual( (first.imported[0].id,), tuple( row.id for row in list_pop3_imports( self.session, tenant_id="tenant-1", profile_ids=(self.profile.id,), ) ), ) self.assertEqual( (), list_pop3_imports( self.session, tenant_id="tenant-1", profile_ids=("unrelated-profile",), ), ) class _RouteSession: def __init__(self, events: list[str]) -> None: self.events = events def commit(self) -> None: self.events.append("commit") def rollback(self) -> None: self.events.append("rollback") class Pop3ImportRouteTests(unittest.TestCase): def test_local_import_and_audit_commit_before_source_deletion(self) -> None: events: list[str] = [] now = datetime.now(UTC) row = SimpleNamespace( id="import-1", profile_id="profile-1", pop3_server_id="server-1", transport_revision="revision-1", provider_uidl="uid-1", message_id="", subject="Legacy notice", from_header="office@example.test", to_header="subject@example.test", date="2026-08-22", body_preview="A bounded legacy message.", size_bytes=len(_RAW), raw_sha256=_download().raw_sha256, status="pending_review", imported_at=now, deletion_requested=True, deletion_status="pending", deletion_attempted_at=None, deletion_error=None, ) resolved = SimpleNamespace( config=_config(allow_delete_after_import=True), server=SimpleNamespace(id="server-1"), credential=None, transport_revision="revision-1", ) principal = ApiPrincipal( principal=PrincipalRef( account_id="account-1", membership_id="user-1", tenant_id="tenant-1", scopes=frozenset( { "mail:profile:use", "mail:pop3:import", "mail:pop3:delete", } ), ), account=SimpleNamespace(id="account-1"), user=SimpleNamespace(id="user-1"), ) payload = MailPop3ImportRequest( server_id="server-1", expected_transport_revision="revision-1", uidls=["uid-1"], delete_after_import=True, ) def audit(*_args, **_kwargs) -> None: events.append("audit") def delete(**_kwargs): self.assertEqual(["audit", "commit"], events) events.append("delete") def mark(*_args, **_kwargs): events.append("mark") row.deletion_status = "succeeded" row.deletion_attempted_at = now return (row,) with ( patch( "govoplan_mail.backend.router._resolve_profile_pop3_transport", return_value=(SimpleNamespace(id="profile-1"), resolved), ), patch( "govoplan_mail.backend.router.download_pop3_messages", return_value=(_download(),), ), patch( "govoplan_mail.backend.router.create_pop3_imports", return_value=Pop3ImportResult(imported=(row,), duplicates=()), ), patch("govoplan_mail.backend.router.audit_event", side_effect=audit), patch("govoplan_mail.backend.router.delete_pop3_messages", side_effect=delete), patch( "govoplan_mail.backend.router.mark_pop3_deletion_result", side_effect=mark, ), ): result = import_profile_pop3_messages( "profile-1", payload, principal=principal, session=_RouteSession(events), # type: ignore[arg-type] ) self.assertEqual("succeeded", result.deletion_status) self.assertEqual( ["audit", "commit", "delete", "mark", "commit", "audit", "commit"], events, ) if __name__ == "__main__": unittest.main()