DataOps-platform/tests/integration/test_device_semantics_postgres.py

from __future__ import annotations

import os
import uuid

import pytest

pytestmark = pytest.mark.integration


def test_device_semantic_codes_keep_versions_reviews_and_search(
    monkeypatch,
):
    platform_url = os.environ.get("TEST_DATABASE_URL")
    if not platform_url:
        pytest.skip("TEST_DATABASE_URL is required")

    monkeypatch.setenv("DATABASE_URL", platform_url)
    from app import create_app, db
    from app.core.data_research.device_semantic_repository import (
        SqlAlchemyDeviceSemanticCodeRepository,
    )
    from app.core.data_research.device_semantics import (
        DeviceSemanticCodeService,
    )
    from app.models.data_research import (
        DeviceSemanticCode,
        DeviceSemanticCodeReview,
        DeviceSemanticCodeVersion,
        OntologyModel,
    )

    app = create_app()
    app.config.update(TESTING=True)
    ontology_uid = None
    created_ontology = False
    canonical_code = f"F-PG-{uuid.uuid4().hex[:8].upper()}"
    code_uid = None
    try:
        with app.app_context():
            existing = OntologyModel.query.filter_by(
                code="DEVICE_SEMANTIC"
            ).first()
            if existing is None:
                ontology_uid = str(uuid.uuid4())
                db.session.add(
                    OntologyModel(
                        uid=ontology_uid,
                        code="DEVICE_SEMANTIC",
                        name="设备语义集成验收本体",
                        owner_uid="integration-test",
                        status="draft",
                        draft_revision=0,
                        created_by="integration-test",
                    )
                )
                created_ontology = True
            else:
                ontology_uid = str(existing.uid)
            db.session.commit()

            repository = SqlAlchemyDeviceSemanticCodeRepository(
                db.session
            )
            approvals = []
            service = DeviceSemanticCodeService(
                repository,
                review_authorizer=lambda ontology, actor: approvals.append(
                    (ontology, actor)
                ),
                commit=db.session.commit,
                rollback=db.session.rollback,
            )
            created = service.create(
                {
                    "ontology_uid": ontology_uid,
                    "code_type": "fault",
                    "canonical_code": canonical_code,
                    "canonical_name": "轴承过热",
                    "definition": "集成验收首版定义",
                    "suggestion_source": "rule",
                    "evidence_uids": ["integration-evidence-1"],
                },
                actor_uid="integration-editor",
            )
            code_uid = created.uid
            revised = service.revise(
                created.uid,
                {
                    "canonical_name": "轴承温度过高",
                    "definition": "集成验收修订定义",
                },
                expected_version=1,
                actor_uid="integration-editor",
            )
            service.submit(
                created.uid,
                expected_version=2,
                actor_uid="integration-editor",
            )
            published, review = service.review(
                created.uid,
                {
                    "decision": "approve",
                    "reason": "企业故障字典验收通过",
                    "expected_version": 2,
                },
                actor_uid="integration-manager",
            )

            records, total = service.search(
                {
                    "ontology_uid": ontology_uid,
                    "code_type": "fault",
                    "status": "published",
                    "keyword": canonical_code,
                },
                page=1,
                page_size=20,
            )
            versions = service.versions(created.uid)
            reviews = service.reviews(created.uid)

            assert revised.current_version == 2
            assert published.status == "published"
            assert review.version == 2
            assert total == 1
            assert records[0].uid == created.uid
            assert [item.version for item in versions] == [2, 1]
            assert versions[0].snapshot["canonical_name"] == "轴承温度过高"
            assert versions[1].snapshot["canonical_name"] == "轴承过热"
            assert reviews == [review]
            assert approvals == [
                (ontology_uid, "integration-manager")
            ]
    finally:
        with app.app_context():
            if code_uid:
                db.session.query(DeviceSemanticCodeReview).filter_by(
                    code_uid=code_uid
                ).delete(synchronize_session=False)
                db.session.query(DeviceSemanticCodeVersion).filter_by(
                    code_uid=code_uid
                ).delete(synchronize_session=False)
                db.session.query(DeviceSemanticCode).filter_by(
                    uid=code_uid
                ).delete(synchronize_session=False)
            if created_ontology:
                db.session.query(OntologyModel).filter_by(
                    uid=ontology_uid
                ).delete(synchronize_session=False)
            db.session.commit()