DataOps-platform/tests/integration/test_domain_template_postgres.py

from __future__ import annotations

import json
import os
import uuid
from pathlib import Path

import pytest
from sqlalchemy import text

pytestmark = pytest.mark.integration
ROOT = Path(__file__).resolve().parents[2]


def test_domain_template_import_diff_and_rollback_are_atomic(monkeypatch):
    database_url = os.environ.get("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")

    monkeypatch.setenv("DATABASE_URL", database_url)
    from app import create_app, db
    from app.core.governance.domain_template_repository import (
        SqlAlchemyDomainTemplateRepository,
    )
    from app.core.governance.domain_templates import DomainTemplateService

    app = create_app()
    app.config.update(TESTING=True)
    actor_uid = str(uuid.uuid4())
    definition = json.loads(
        (
            ROOT / "docs/phase2/P2_WP01_SPARE_PARTS_DOMAIN_TEMPLATE.json"
        ).read_text(encoding="utf-8")
    )
    definition["template_code"] = f"spare_parts_{uuid.uuid4().hex[:8]}"
    definition["name"] = f"集成测试模板 {definition['template_code']}"
    template_uid = None
    try:
        with app.app_context():
            db.session.execute(
                text(
                    """
                    INSERT INTO public.users (
                        id, username, display_name, password_hash, status
                    ) VALUES (
                        CAST(:id AS uuid), :username, :username,
                        'p2-wp01-integration-hash', 'active'
                    )
                    """
                ),
                {"id": actor_uid, "username": f"p2-wp01-{actor_uid[:8]}"},
            )
            db.session.commit()

            service = DomainTemplateService(
                SqlAlchemyDomainTemplateRepository(db.session)
            )
            first = service.import_template(definition, actor_uid=actor_uid)
            db.session.commit()
            template_uid = first["uid"]

            changed = dict(definition)
            changed["description"] = "集成测试第二版"
            preview = service.dry_run(changed)
            assert preview["diff"]["added_object_types"] == []

            class FailingRepository(SqlAlchemyDomainTemplateRepository):
                def apply_import(self, record):
                    super().apply_import(record)
                    raise RuntimeError("simulated failure after persistence")

            failing_service = DomainTemplateService(FailingRepository(db.session))
            with pytest.raises(RuntimeError, match="simulated failure"):
                failing_service.import_template(changed, actor_uid=actor_uid)
            db.session.rollback()
            assert service.get_template(
                definition["template_code"]
            )["current_version"] == 1

            second = service.import_template(changed, actor_uid=actor_uid)
            db.session.commit()
            rolled_back = service.rollback(
                definition["template_code"],
                target_version=1,
                actor_uid=actor_uid,
            )
            db.session.commit()

            assert first["current_version"] == 1
            assert second["current_version"] == 2
            assert rolled_back["current_version"] == 3
            assert rolled_back["description"] == definition["description"]
            assert len(rolled_back["object_types"]) == 6
            assert [item["operation"] for item in service.list_imports(
                definition["template_code"]
            )] == ["rollback", "import", "import"]

            counts = db.session.execute(
                text(
                    """
                    SELECT
                      (SELECT count(*) FROM public.governance_object_types
                       WHERE template_uid = CAST(:uid AS uuid)) AS object_types,
                      (SELECT count(*) FROM public.governance_domain_template_versions
                       WHERE template_uid = CAST(:uid AS uuid)) AS versions,
                      (SELECT count(*) FROM public.governance_domain_template_imports
                       WHERE template_uid = CAST(:uid AS uuid)) AS imports
                    """
                ),
                {"uid": template_uid},
            ).mappings().one()
            assert dict(counts) == {
                "object_types": 6,
                "versions": 3,
                "imports": 3,
            }
    finally:
        with app.app_context():
            db.session.rollback()
            if template_uid:
                for statement in (
                    "DELETE FROM public.governance_domain_template_imports "
                    "WHERE template_uid = CAST(:uid AS uuid)",
                    "DELETE FROM public.governance_object_types "
                    "WHERE template_uid = CAST(:uid AS uuid)",
                    "DELETE FROM public.governance_domain_template_versions "
                    "WHERE template_uid = CAST(:uid AS uuid)",
                    "DELETE FROM public.governance_domain_templates "
                    "WHERE uid = CAST(:uid AS uuid)",
                ):
                    db.session.execute(
                        text(statement),
                        {"uid": template_uid},
                    )
            db.session.execute(
                text("DELETE FROM public.users WHERE id = CAST(:id AS uuid)"),
                {"id": actor_uid},
            )
            db.session.commit()