govoplan-campaign/tests/test_campaign_transfers.py

from __future__ import annotations

import copy
from types import SimpleNamespace
from unittest.mock import patch

import pytest
from fastapi import HTTPException
from sqlalchemy import Column, String, Table, create_engine
from sqlalchemy.orm import Session, sessionmaker
from govoplan_campaign.backend.campaign.transfers import (
    DEFAULT_PORTABLE_CAMPAIGN_SCOPES,
    build_campaign_portable_package,
    canonical_sha256,
    inspect_campaign_portable_package,
)
from govoplan_campaign.backend.db.models import (
    Campaign,
    CampaignIssue,
    CampaignJob,
    CampaignShare,
    CampaignVersion,
)
from govoplan_campaign.backend.persistence.versions import minimal_campaign_json
from govoplan_campaign.backend.routes.transfers import (
    export_campaign_package,
    import_campaign_package,
    preview_campaign_import,
)
from govoplan_campaign.backend.schemas import (
    CampaignExportRequest,
    CampaignImportApplyRequest,
    CampaignImportPreviewRequest,
    CampaignPortablePackageResponse,
)
from govoplan_core.core.change_sequence import ChangeSequenceEntry
from govoplan_core.db.base import Base


def _source() -> tuple[Campaign, CampaignVersion]:
    raw_json = minimal_campaign_json(
        external_id="monthly-notice",
        name="Monthly notice",
        description="Portable source",
    )
    raw_json["fields"] = [
        {"name": "case_id", "type": "string"},
        {"name": "private_code", "type": "password"},
    ]
    raw_json["global_values"] = {
        "office": "Permits",
        "private_code": "must-not-leave-the-source",
    }
    raw_json["server"] = {
        "mail_profile_id": "mail-profile-source",
        "smtp_server_id": "smtp-source",
        "smtp_credential_id": "credential-source",
    }
    raw_json["template"] = {
        "subject": "Case {{case_id}}",
        "text": "Hello",
        "html": None,
    }
    raw_json["attachments"]["global"] = [
        {"base_dir": ".", "file_filter": "notice.pdf", "required": True}
    ]
    raw_json["entries"]["inline"] = [
        {
            "id": "recipient-1",
            "to": [{"email": "person@example.test"}],
            "fields": {
                "case_id": "A-1",
                "private_code": "recipient-secret",
            },
            "attachments": [
                {"base_dir": ".", "file_filter": "A-1.pdf", "required": True}
            ],
        }
    ]
    campaign = Campaign(
        id="campaign-source",
        tenant_id="tenant-source",
        external_id="monthly-notice",
        name="Monthly notice",
        description="Portable source",
        status="completed",
        settings={
            "retention_days": 90,
            "provider_token": "must-not-export",
        },
        mail_profile_policy={
            "profile_id": "mail-profile-source",
            "credential_id": "credential-source",
        },
    )
    version = CampaignVersion(
        id="version-source",
        campaign_id=campaign.id,
        version_number=4,
        raw_json=raw_json,
        schema_version="1.0",
        workflow_state="completed",
        validation_summary={"ok": True, "error_count": 0},
        build_summary={"built_count": 1},
        editor_state={
            "review_send": {
                "inspection_complete": True,
                "reviewed_message_keys": ["message-1"],
                "issue_decisions": [
                    {
                        "decision": "accept",
                        "issue_codes": ["attachment_warning"],
                        "issue_fingerprint": "fingerprint-1",
                        "message_sha256": "a" * 64,
                        "reason": "Verified manually",
                    }
                ],
            }
        },
    )
    return campaign, version


def _job(campaign: Campaign, version: CampaignVersion) -> CampaignJob:
    return CampaignJob(
        id="job-1",
        tenant_id=campaign.tenant_id,
        campaign_id=campaign.id,
        campaign_version_id=version.id,
        entry_index=0,
        entry_id="recipient-1",
        recipient_email="person@example.test",
        message_id_header="<message@example.test>",
        eml_sha256="b" * 64,
        build_status="built",
        validation_status="ready",
        queue_status="cancelled",
        send_status="smtp_accepted",
        postbox_status="not_requested",
        print_status="not_requested",
        imap_status="appended",
        attempt_count=1,
        delivery_provenance={"route": "mail", "storage_key": "hidden"},
    )


def test_privacy_default_export_is_configuration_only_and_redacts_secrets() -> None:
    campaign, version = _source()

    package = build_campaign_portable_package(
        campaign=campaign,
        version=version,
        scopes=DEFAULT_PORTABLE_CAMPAIGN_SCOPES,
        module_version="0.1.24",
    )

    assert package["scopes"] == ["metadata", "template_config"]
    assert set(package["payload"]) == {"metadata", "template_config"}
    assert package["manifest"]["secrets_included"] is False
    assert package["manifest"]["redactions"] == {
        "deployment_credential_reference": 1,
        "password_field_value": 2,
        "sensitive_setting": 2,
    }
    template = package["payload"]["template_config"]
    assert "private_code" not in template["configuration"]["global_values"]
    assert "smtp_credential_id" not in template["configuration"]["server"]
    assert "provider_token" not in template["campaign_settings"]
    assert "credential_id" not in template["mail_profile_policy"]
    serialized = CampaignPortablePackageResponse.model_validate(package).model_dump(
        mode="json"
    )
    assert inspect_campaign_portable_package(
        serialized,
        selected_scopes=None,
        external_id="serialized-import",
        name="Serialized import",
    ).preview["compatible"] is True


def test_full_export_import_applies_configuration_but_never_replays_evidence() -> None:
    campaign, version = _source()
    package = build_campaign_portable_package(
        campaign=campaign,
        version=version,
        scopes=(
            "metadata",
            "template_config",
            "recipients",
            "attachments",
            "review_state",
            "delivery_history",
        ),
        jobs=(_job(campaign, version),),
        issues=(
            CampaignIssue(
                id="issue-1",
                tenant_id=campaign.tenant_id,
                campaign_id=campaign.id,
                campaign_version_id=version.id,
                severity="warning",
                code="attachment_warning",
                message="Review attachment",
            ),
        ),
        module_version="0.1.24",
    )

    inspection = inspect_campaign_portable_package(
        package,
        selected_scopes=None,
        external_id="monthly-notice-import",
        name="Imported monthly notice",
    )

    assert inspection.preview["compatible"] is True
    assert inspection.configuration is not None
    assert inspection.configuration["campaign"] == {
        "id": "monthly-notice-import",
        "name": "Imported monthly notice",
        "description": "Portable source",
        "mode": "draft",
    }
    assert inspection.configuration["server"] == {}
    assert inspection.configuration["entries"]["inline"][0]["to"] == [
        {"email": "person@example.test"}
    ]
    assert inspection.configuration["entries"]["inline"][0]["attachments"][0][
        "file_filter"
    ] == "A-1.pdf"
    assert "private_code" not in inspection.configuration["entries"]["inline"][0][
        "fields"
    ]
    skipped_codes = {item["code"] for item in inspection.preview["will_skip"]}
    assert skipped_codes == {
        "deployment_bound_mail_profile",
        "operational_evidence_not_replayed",
    }
    assert package["payload"]["review_state"]["decision_count"] == 1
    assert package["payload"]["delivery_history"]["jobs"][0][
        "recipient_email"
    ] == "person@example.test"
    assert "storage_key" not in package["payload"]["delivery_history"]["jobs"][0][
        "delivery_provenance"
    ]


def test_import_preview_reports_unselected_recipient_attachment_rules() -> None:
    campaign, version = _source()
    package = build_campaign_portable_package(
        campaign=campaign,
        version=version,
        scopes=("metadata", "attachments", "recipients"),
        module_version="0.1.24",
    )

    inspection = inspect_campaign_portable_package(
        package,
        selected_scopes=("metadata", "attachments"),
        external_id="attachment-import",
        name="Attachment import",
    )

    assert inspection.preview["compatible"] is True
    skipped = {item["code"]: item for item in inspection.preview["will_skip"]}
    assert skipped["recipient_scope_required"]["item_count"] == 1
    assert skipped["scope_not_selected"]["scope"] == "recipients"
    assert inspection.configuration is not None
    assert inspection.configuration["entries"]["inline"] == []


def test_import_preview_fails_closed_when_package_is_tampered() -> None:
    campaign, version = _source()
    package = build_campaign_portable_package(
        campaign=campaign,
        version=version,
        scopes=DEFAULT_PORTABLE_CAMPAIGN_SCOPES,
        module_version="0.1.24",
    )
    tampered = copy.deepcopy(package)
    tampered["payload"]["metadata"]["name"] = "Tampered"

    inspection = inspect_campaign_portable_package(
        tampered,
        selected_scopes=None,
        external_id="tampered-import",
        name="Tampered",
    )

    assert inspection.preview["compatible"] is False
    assert inspection.configuration is None
    assert any("integrity checksum" in error for error in inspection.preview["errors"])


def test_import_preview_rejects_unsupported_campaign_schema_even_with_valid_checksum() -> None:
    campaign, version = _source()
    package = build_campaign_portable_package(
        campaign=campaign,
        version=version,
        scopes=DEFAULT_PORTABLE_CAMPAIGN_SCOPES,
        module_version="0.1.24",
    )
    package["source"]["campaign_schema_version"] = "2.0"
    package["integrity"]["package_sha256"] = canonical_package_hash(package)

    inspection = inspect_campaign_portable_package(
        package,
        selected_scopes=None,
        external_id="future-import",
        name="Future import",
    )

    assert inspection.preview["compatible"] is False
    assert any("schema version" in error for error in inspection.preview["errors"])


def canonical_package_hash(package: dict[str, object]) -> str:
    content = copy.deepcopy(package)
    content.pop("integrity", None)
    return canonical_sha256(content)


class _Principal:
    tenant_id = "tenant-1"
    api_key = None

    def __init__(self, *scopes: str) -> None:
        self.user = SimpleNamespace(id="user-1", display_name="Importer")
        self.scopes = frozenset(scopes)

    def has(self, scope: str) -> bool:
        return scope in self.scopes or "tenant:*" in self.scopes


@pytest.fixture()
def route_session() -> Session:
    engine = create_engine("sqlite+pysqlite:///:memory:")
    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),
        )
    access_groups = Base.metadata.tables.get("access_groups")
    if access_groups is None:
        access_groups = Table(
            "access_groups",
            Base.metadata,
            Column("id", String(36), primary_key=True),
        )
    Base.metadata.create_all(
        engine,
        tables=[
            access_users,
            access_groups,
            Campaign.__table__,
            CampaignVersion.__table__,
            CampaignShare.__table__,
            CampaignJob.__table__,
            CampaignIssue.__table__,
            ChangeSequenceEntry.__table__,
        ],
    )
    session_factory = sessionmaker(bind=engine, class_=Session, expire_on_commit=False)
    database = session_factory()
    user_values = {"id": "user-1"}
    if "tenant_id" in access_users.c:
        user_values.update(
            tenant_id="tenant-1",
            account_id="account-1",
            email="user-1@example.test",
        )
    database.execute(access_users.insert().values(**user_values))
    raw_json = minimal_campaign_json(external_id="source", name="Source")
    raw_json["entries"]["inline"] = [
        {"id": "one", "to": [{"email": "one@example.test"}]}
    ]
    source = Campaign(
        id="source-campaign",
        tenant_id="tenant-1",
        created_by_user_id="user-1",
        owner_user_id="user-1",
        external_id="source",
        name="Source",
        status="draft",
        current_version_id="source-version",
    )
    source_version = CampaignVersion(
        id="source-version",
        campaign_id=source.id,
        version_number=1,
        raw_json=raw_json,
    )
    database.add_all((source, source_version))
    database.commit()
    try:
        yield database
    finally:
        database.close()
        engine.dispose()


def test_export_route_enforces_recipient_export_scope(route_session: Session) -> None:
    principal = _Principal(
        "campaigns:campaign:read",
        "campaigns:campaign:export",
        "campaigns:recipient:read",
    )

    with pytest.raises(HTTPException) as denied:
        export_campaign_package(
            "source-campaign",
            "source-version",
            CampaignExportRequest(scopes=["metadata", "recipients"]),
            session=route_session,
            principal=principal,
        )

    assert denied.value.status_code == 403
    assert denied.value.detail == "Missing scope: campaigns:recipient:export"


def test_export_preview_and_apply_routes_keep_matching_provenance(
    route_session: Session,
) -> None:
    exporter = _Principal(
        "campaigns:campaign:read",
        "campaigns:campaign:export",
        "campaigns:recipient:read",
        "campaigns:recipient:export",
    )

    def commit_audit(active_session: Session, *_args, **_kwargs) -> None:
        active_session.commit()

    with patch(
        "govoplan_campaign.backend.routes.transfers.audit_from_principal",
        side_effect=commit_audit,
    ):
        package = export_campaign_package(
            "source-campaign",
            "source-version",
            CampaignExportRequest(scopes=["metadata", "template_config", "recipients"]),
            session=route_session,
            principal=exporter,
        )

    limited_importer = _Principal(
        "campaigns:campaign:create",
        "campaigns:campaign:import",
    )
    limited_preview = preview_campaign_import(
        CampaignImportPreviewRequest(package=package),
        session=route_session,
        principal=limited_importer,
    )
    assert limited_preview["compatible"] is True
    assert "recipients" in limited_preview["selected_scopes"]

    importer = _Principal(
        "campaigns:campaign:create",
        "campaigns:campaign:import",
        "campaigns:recipient:write",
        "campaigns:recipient:import",
    )
    preview = preview_campaign_import(
        CampaignImportPreviewRequest(package=package),
        session=route_session,
        principal=importer,
    )
    assert preview["compatible"] is True
    assert preview["destination"]["external_id"] == "source-import"

    def create_import(active_session: Session, **kwargs):
        raw_json = kwargs["raw_json"]
        destination = Campaign(
            id="imported-campaign",
            tenant_id="tenant-1",
            created_by_user_id="user-1",
            owner_user_id="user-1",
            external_id=raw_json["campaign"]["id"],
            name=raw_json["campaign"]["name"],
            status="draft",
            current_version_id="imported-version",
        )
        version = CampaignVersion(
            id="imported-version",
            campaign_id=destination.id,
            version_number=1,
            raw_json=raw_json,
        )
        active_session.add_all((destination, version))
        active_session.flush()
        return destination, version

    with (
        patch(
            "govoplan_campaign.backend.routes.transfers.create_campaign_version_from_json",
            side_effect=create_import,
        ),
        patch(
            "govoplan_campaign.backend.routes.transfers.audit_from_principal",
            side_effect=commit_audit,
        ),
        patch(
            "govoplan_campaign.backend.routes.transfers._write_current_version_snapshot_if_available"
        ),
    ):
        response = import_campaign_package(
            CampaignImportApplyRequest(
                package=package,
                selected_scopes=preview["selected_scopes"],
                external_id=preview["destination"]["external_id"],
                name=preview["destination"]["name"],
                expected_package_sha256=preview["package_sha256"],
            ),
            session=route_session,
            principal=importer,
        )

    assert response.campaign.external_id == "source-import"
    assert response.receipt["package_id"] == package["package_id"]
    assert response.receipt["package_sha256"] == package["integrity"]["package_sha256"]
    imported = route_session.get(Campaign, "imported-campaign")
    assert imported is not None
    assert imported.settings["portable_import"]["package_id"] == package["package_id"]
    assert route_session.query(CampaignJob).filter_by(campaign_id=imported.id).count() == 0