govoplan-campaign/tests/test_calendar_invitations.py

from __future__ import annotations

import tempfile
import unittest
from email import policy
from email.message import EmailMessage
from email.parser import BytesParser
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

from govoplan_campaign.backend.campaign.models import (
    BuildStatus,
    CampaignConfig,
    SendStatus,
)
from govoplan_campaign.backend.campaign.validation import validate_campaign_config
from govoplan_campaign.backend.messages.builder import BuiltMessage
from govoplan_campaign.backend.messages.models import (
    ImapStatus,
    MessageAddress,
    MessageDraft,
    MessageValidationStatus,
)
from govoplan_campaign.backend.persistence.campaigns import (
    _prepare_built_calendar_invitations,
)
from govoplan_campaign.backend.sending.jobs import _mark_accepted_job_artifacts
from govoplan_core.core.calendar import CalendarInvitationRef


def _config() -> CampaignConfig:
    return CampaignConfig.model_validate(
        {
            "version": "1.0",
            "campaign": {
                "id": "campaign-1",
                "name": "Invitation campaign",
                "mode": "send",
            },
            "template": {"subject": "Planning", "text": "Please reply."},
            "entries": {
                "inline": [
                    {
                        "id": "recipient-1",
                        "name": "Ada",
                        "to": [{"email": "ada@example.test", "name": "Ada"}],
                        "fields": {"appointment_start": "2026-08-05T09:00:00"},
                    }
                ]
            },
            "delivery": {
                "channel_policy": "mail",
                "calendar_invitation": {
                    "enabled": True,
                    "calendar_id": "calendar-1",
                    "summary_template": "Planning with {{name}}",
                    "start_at_template": "{{appointment_start}}",
                    "end_at_template": "2026-08-05T10:00:00",
                    "timezone": "Europe/Berlin",
                },
            },
        }
    )


class _CalendarIntegration:
    available = True

    def request_from_payload(self, payload):
        return payload

    def render_invitation(self, request):
        assert request["correlation_id"].startswith("campaign:version-1:")
        return "\r\n".join(
            (
                "BEGIN:VCALENDAR",
                "VERSION:2.0",
                "METHOD:REQUEST",
                "BEGIN:VEVENT",
                "UID:invitation-1@govoplan.local",
                "SUMMARY:Planning with Ada",
                "END:VEVENT",
                "END:VCALENDAR",
                "",
            )
        )


class _MirroringIntegration:
    available = True

    def __init__(self) -> None:
        self.upsert_count = 0

    def request_from_payload(self, payload):
        return SimpleNamespace(
            metadata=payload.get("metadata") or {},
            payload=payload,
        )

    def upsert_invitation(self, _session, **_kwargs):
        self.upsert_count += 1
        return CalendarInvitationRef(
            event_id="event-1",
            calendar_id="calendar-1",
            uid="invitation-1@govoplan.local",
            correlation_id="campaign:version-1:entry-1",
            source_module="campaigns",
            source_resource_type="campaign_version",
            source_resource_id="version-1",
        )


class _Session:
    def __init__(self) -> None:
        self.added: list[object] = []

    def add(self, value) -> None:
        self.added.append(value)


class CampaignCalendarInvitationTests(unittest.TestCase):
    def test_validation_requires_calendar_capability(self) -> None:
        report = validate_campaign_config(_config(), calendar_available=False)

        self.assertIn(
            "calendar_invitation_unavailable",
            {issue.code for issue in report.issues},
        )

    def test_build_freezes_individual_request_and_ics_attachment(self) -> None:
        config = _config()
        entry = config.entries.inline[0]  # type: ignore[index]
        message = EmailMessage()
        message["From"] = "Organizer <organizer@example.test>"
        message["To"] = "Ada <ada@example.test>"
        message["Subject"] = "Planning"
        message.set_content("Please reply.")
        with tempfile.TemporaryDirectory() as temp_dir:
            eml_path = Path(temp_dir) / "message.eml"
            eml_path.write_bytes(bytes(message))
            draft = MessageDraft(
                entry_index=1,
                entry_id="recipient-1",
                active=True,
                build_status=BuildStatus.BUILT,
                validation_status=MessageValidationStatus.READY,
                send_status=SendStatus.DRAFT,
                imap_status=ImapStatus.NOT_REQUESTED,
                subject="Planning",
                **{
                    "from": MessageAddress(
                        email="organizer@example.test",
                        name="Organizer",
                    )
                },
                to=[MessageAddress(email="ada@example.test", name="Ada")],
                eml_path=str(eml_path),
            )
            provenance: dict[int, dict[str, object]] = {1: {}}

            with patch(
                "govoplan_campaign.backend.persistence.campaigns.calendar_integration",
                return_value=_CalendarIntegration(),
            ):
                _prepare_built_calendar_invitations(
                    version=SimpleNamespace(
                        id="version-1",
                        campaign_id="campaign-1",
                    ),
                    config=config,
                    built_messages=[BuiltMessage(draft=draft, mime=message)],
                    entries_by_index={1: entry},
                    delivery_provenance_by_index=provenance,
                    user_id="user-1",
                )

            parsed = BytesParser(policy=policy.default).parsebytes(
                eml_path.read_bytes()
            )
            attachments = list(parsed.iter_attachments())
            self.assertEqual("text/calendar", attachments[0].get_content_type())
            self.assertEqual("REQUEST", attachments[0].get_param("method"))
            invitation = provenance[1]["calendar_invitation"]
            self.assertEqual("prepared", invitation["state"])
            self.assertEqual(
                "2026-08-05T09:00:00+02:00",
                invitation["request"]["start_at"],
            )

    def test_delivery_acceptance_mirrors_once_without_reopening_delivery(self) -> None:
        integration = _MirroringIntegration()
        job = SimpleNamespace(
            tenant_id="tenant-1",
            delivery_provenance={
                "calendar_invitation": {
                    "state": "prepared",
                    "request": {
                        "metadata": {"prepared_by_user_id": "user-1"},
                    },
                }
            },
        )
        session = _Session()

        with (
            patch(
                "govoplan_campaign.backend.sending.jobs.calendar_integration",
                return_value=integration,
            ),
            patch(
                "govoplan_campaign.backend.sending.jobs.files_integration"
            ),
        ):
            _mark_accepted_job_artifacts(session, job)
            _mark_accepted_job_artifacts(session, job)

        invitation = job.delivery_provenance["calendar_invitation"]
        self.assertEqual("mirrored", invitation["state"])
        self.assertEqual("event-1", invitation["event_id"])
        self.assertEqual(1, integration.upsert_count)
        self.assertEqual([job], session.added)


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