govoplan-mail/tests/test_recovery.py

from __future__ import annotations

import json
from pathlib import Path
import tempfile
import unittest

from sqlalchemy import Column, String, Table, create_engine, select
from sqlalchemy.orm import Session, sessionmaker

from govoplan_core.core.recovery import (
    RecoveryCheckpoint,
    RecoveryGuaranteeError,
    RecoveryOperation,
    RecoveryStatus,
)
from govoplan_core.core.runtime_coordination import (
    DistributedLease,
    RuntimeIdentity,
    bind_process_runtime_identity,
)
from govoplan_core.db.base import Base
from govoplan_core.db.session import configure_database, reset_database
from govoplan_mail.backend.db.models import (
    MailMailboxFolderIndex,
    MailMailboxMessageIndex,
    MailServerProfile,
)
from govoplan_mail.backend.mailbox_index import cache_mailbox_folders
from govoplan_mail.backend.recovery import (
    MailRecoveryError,
    MailboxRefreshBusy,
    begin_mailbox_refresh_recovery,
    begin_provider_effect_recovery,
    reconcile_outbox_provider_effect,
)
from govoplan_mail.backend.sending.imap import (
    ImapFolderListResult,
    ImapMailboxInfo,
)


class MailRecoveryTests(unittest.TestCase):
    def setUp(self) -> None:
        self.tempdir = tempfile.TemporaryDirectory()
        self.addCleanup(self.tempdir.cleanup)
        database_path = Path(self.tempdir.name) / "mail-recovery.sqlite3"
        self.engine = create_engine(f"sqlite:///{database_path}")
        access_users = Base.metadata.tables.get("access_users")
        if access_users is None:
            access_users = Table(
                "access_users",
                Base.metadata,
                Column("id", String(36), primary_key=True),
            )
        Base.metadata.create_all(
            self.engine,
            tables=[
                access_users,
                DistributedLease.__table__,
                RecoveryOperation.__table__,
                RecoveryCheckpoint.__table__,
                MailServerProfile.__table__,
                MailMailboxFolderIndex.__table__,
                MailMailboxMessageIndex.__table__,
            ],
        )
        configure_database(
            f"sqlite:///{database_path}",
            engine=self.engine,
            dispose_previous=True,
        )
        bind_process_runtime_identity(
            RuntimeIdentity(
                installation_id="mail-recovery-tests",
                node_id="mail-test-node",
                incarnation="mail-test-incarnation",
                role="worker",
                software_version="test",
                composition_hash="a" * 64,
            )
        )
        self.SessionLocal = sessionmaker(
            bind=self.engine,
            class_=Session,
            expire_on_commit=False,
        )
        with self.SessionLocal() as session:
            session.add(
                MailServerProfile(
                    id="profile-1",
                    tenant_id="tenant-1",
                    scope_type="tenant",
                    scope_id="tenant-1",
                    name="Recovery profile",
                    slug="recovery-profile",
                    smtp_config={},
                )
            )
            session.commit()
        self.addCleanup(self._cleanup_runtime)

    def _cleanup_runtime(self) -> None:
        bind_process_runtime_identity(None)
        reset_database()
        self.engine.dispose()

    def test_smtp_effect_is_durable_before_provider_and_redacted_on_success(self) -> None:
        recovery = begin_provider_effect_recovery(
            kind="smtp-delivery",
            effect_id="outbox:command-1:smtp-attempt:1",
            tenant_id="tenant-1",
            profile_id="profile-1",
            message_bytes=b"Subject: Recovery\r\n\r\nBody",
            expected_transport_revision="revision-1",
            recipient_count=1,
            resource_type="mail_delivery_command",
            resource_id="command-1",
        )
        assert recovery is not None and recovery.operation is not None
        with self.SessionLocal() as session:
            operation = session.get(RecoveryOperation, recovery.operation_id)
            assert operation is not None
            self.assertEqual(RecoveryStatus.RUNNING.value, operation.status)

        recovery.succeed_smtp(
            accepted_count=1,
            refused_recipients={},
        )

        with self.SessionLocal() as session:
            operation = session.get(RecoveryOperation, recovery.operation_id)
            assert operation is not None
            self.assertEqual(RecoveryStatus.SUCCEEDED.value, operation.status)
            evidence = json.dumps(
                [
                    item.evidence
                    for item in session.scalars(
                        select(RecoveryCheckpoint).where(
                            RecoveryCheckpoint.operation_id == operation.id
                        )
                    )
                ]
            )
            self.assertNotIn("recipient@example.test", evidence)
            self.assertNotIn("Subject: Recovery", evidence)

    def test_unknown_smtp_outcome_blocks_replay_until_reconciled(self) -> None:
        kwargs = {
            "kind": "smtp-delivery",
            "effect_id": "outbox:command-2:smtp-attempt:1",
            "tenant_id": "tenant-1",
            "profile_id": "profile-1",
            "message_bytes": b"Subject: Unknown\r\n\r\nBody",
            "expected_transport_revision": "revision-1",
            "recipient_count": 1,
            "resource_type": "mail_delivery_command",
            "resource_id": "command-2",
        }
        recovery = begin_provider_effect_recovery(**kwargs)
        assert recovery is not None
        recovery.unknown(
            code="socket_closed_after_data",
            summary="SMTP outcome is unknown",
        )

        with self.assertRaises(MailRecoveryError):
            begin_provider_effect_recovery(**kwargs)

        self.assertTrue(
            reconcile_outbox_provider_effect(
                command_id="command-2",
                effect_occurred=False,
                evidence_reference="provider-case-42",
                user_id="operator-1",
            )
        )
        with self.SessionLocal() as session:
            operation = session.get(RecoveryOperation, recovery.operation_id)
            assert operation is not None
            self.assertEqual(RecoveryStatus.RECOVERED.value, operation.status)

    def test_mailbox_refresh_verifies_the_committed_index(self) -> None:
        result = ImapFolderListResult(
            host="imap.example.test",
            port=993,
            security="tls",
            folders=[
                ImapMailboxInfo(
                    name="INBOX",
                    flags=["\\HasNoChildren"],
                    message_count=4,
                    unseen_count=1,
                )
            ],
        )
        recovery = begin_mailbox_refresh_recovery(
            tenant_id="tenant-1",
            profile_id="profile-1",
            folder="*",
            purpose="folders",
        )
        with self.SessionLocal() as session:
            session.add(
                MailMailboxFolderIndex(
                    tenant_id="tenant-1",
                    profile_id="profile-1",
                    folder="Removed",
                    flags=[],
                )
            )
            session.commit()
            cache_mailbox_folders(
                session,
                tenant_id="tenant-1",
                profile_id="profile-1",
                result=result,
            )
            session.commit()
        recovery.complete_folders(result)

        with self.SessionLocal() as session:
            self.assertEqual(
                ["INBOX"],
                list(
                    session.scalars(
                        select(MailMailboxFolderIndex.folder).order_by(
                            MailMailboxFolderIndex.folder
                        )
                    )
                ),
            )
            operation = session.get(
                RecoveryOperation,
                recovery.operation.operation_id,
            )
            assert operation is not None
            self.assertEqual(RecoveryStatus.SUCCEEDED.value, operation.status)

    def test_missing_runtime_identity_fails_before_a_provider_effect_can_start(self) -> None:
        bind_process_runtime_identity(None)
        with self.assertRaises(MailRecoveryError):
            begin_provider_effect_recovery(
                kind="imap-append",
                effect_id="campaign-job:job-1:imap-attempt:1",
                tenant_id="tenant-1",
                profile_id="profile-1",
                message_bytes=b"message",
                expected_transport_revision="revision-1",
                folder="Sent",
            )

    def test_tampered_provider_evidence_cannot_be_marked_successful(self) -> None:
        recovery = begin_provider_effect_recovery(
            kind="smtp-delivery",
            effect_id="outbox:command-3:smtp-attempt:1",
            tenant_id="tenant-1",
            profile_id="profile-1",
            message_bytes=b"message",
            expected_transport_revision="revision-1",
            recipient_count=1,
            resource_type="mail_delivery_command",
            resource_id="command-3",
        )
        assert recovery is not None
        with self.SessionLocal() as session:
            checkpoint = session.scalar(
                select(RecoveryCheckpoint)
                .where(RecoveryCheckpoint.operation_id == recovery.operation_id)
                .order_by(RecoveryCheckpoint.sequence)
                .limit(1)
            )
            assert checkpoint is not None
            checkpoint.summary = "tampered"
            session.commit()

        with self.assertRaises(RecoveryGuaranteeError):
            recovery.succeed_smtp(accepted_count=1, refused_recipients={})
        with self.SessionLocal() as session:
            operation = session.get(RecoveryOperation, recovery.operation_id)
            assert operation is not None
            self.assertNotEqual(RecoveryStatus.SUCCEEDED.value, operation.status)

    def test_mailbox_refresh_has_a_cross_runtime_fence(self) -> None:
        recovery = begin_mailbox_refresh_recovery(
            tenant_id="tenant-1",
            profile_id="profile-1",
            folder="INBOX",
            purpose="messages",
        )
        bind_process_runtime_identity(
            RuntimeIdentity(
                installation_id="mail-recovery-tests",
                node_id="mail-test-node-2",
                incarnation="mail-test-incarnation-2",
                role="worker",
                software_version="test",
                composition_hash="a" * 64,
            )
        )
        with self.assertRaises(MailboxRefreshBusy):
            begin_mailbox_refresh_recovery(
                tenant_id="tenant-1",
                profile_id="profile-1",
                folder="INBOX",
                purpose="messages",
            )
        recovery.reject(summary="Test refresh stopped", code="test")


if __name__ == "__main__":
    unittest.main()