DataOps-platform/tests/data_research/test_data_element_lifecycle.py

from __future__ import annotations

from dataclasses import replace

import pytest


class MemoryElementRepository:
    def __init__(self):
        self.records = {}

    def get(self, uid):
        return self.records.get(uid)

    def get_by_code(self, code):
        return next((item for item in self.records.values() if item.code == code), None)

    def list(self, *, status=None, business_domain_uid=None):
        records = list(self.records.values())
        if status:
            records = [item for item in records if item.status == status]
        if business_domain_uid:
            records = [
                item
                for item in records
                if business_domain_uid
                in set(item.snapshot.get("business_domain_uids") or [])
            ]
        return records

    def save(self, record):
        self.records[record.uid] = record
        return record


def valid_payload(**overrides):
    payload = {
        "code": "ORDER_ID",
        "name_zh": "订单编号",
        "name_en": "order_id",
        "definition": "订单的唯一业务标识",
        "logical_type": "string",
        "evidence_uids": ["evidence-1"],
        "business_domain_uids": ["domain-1"],
    }
    payload.update(overrides)
    return payload


def test_data_element_requires_definition_and_starts_as_draft():
    from app.core.data_research.data_elements import DataElementService
    from app.core.data_research.errors import DataElementInvalid

    repository = MemoryElementRepository()
    service = DataElementService(repository, uid_factory=lambda: "element-1")

    with pytest.raises(DataElementInvalid, match="definition"):
        service.create_draft(valid_payload(definition=""), actor_uid="editor-1")

    element = service.create_draft(valid_payload(), actor_uid="editor-1")
    assert element.uid == "element-1"
    assert element.status == "draft"
    assert element.current_version == 1
    assert element.snapshot["evidence_uids"] == ["evidence-1"]


def test_lifecycle_uses_optimistic_version_and_emits_publish_event():
    from app.core.data_research.data_elements import DataElementService
    from app.core.data_research.errors import DataElementVersionConflict

    repository = MemoryElementRepository()
    events = []
    service = DataElementService(
        repository,
        uid_factory=lambda: "element-1",
        outbox_enqueue=lambda **event: events.append(event),
    )
    element = service.create_draft(valid_payload(), actor_uid="editor-1")
    review = service.transition(
        element.uid,
        "in_review",
        expected_version=1,
        actor_uid="editor-1",
    )
    published = service.transition(
        element.uid,
        "published",
        expected_version=2,
        actor_uid="admin-1",
    )

    assert review.current_version == 2
    assert published.current_version == 3
    assert events == [
        {
            "aggregate_type": "data_element",
            "aggregate_id": "element-1",
            "event_type": "data_element.version_published",
            "payload": {"uid": "element-1", "version": 3},
        }
    ]
    with pytest.raises(DataElementVersionConflict):
        service.transition(
            element.uid,
            "deprecated",
            expected_version=2,
            actor_uid="admin-1",
        )


def test_invalid_lifecycle_transition_is_rejected():
    from app.core.data_research.data_elements import DataElementService
    from app.core.data_research.errors import DataElementTransitionInvalid

    repository = MemoryElementRepository()
    service = DataElementService(repository, uid_factory=lambda: "element-1")
    element = service.create_draft(valid_payload(), actor_uid="editor-1")

    with pytest.raises(DataElementTransitionInvalid, match="draft -> published"):
        service.transition(
            element.uid,
            "published",
            expected_version=1,
            actor_uid="admin-1",
        )


def test_elements_can_be_filtered_for_ontology_mapping():
    from app.core.data_research.data_elements import DataElementService

    repository = MemoryElementRepository()
    service = DataElementService(
        repository, uid_factory=iter(("element-1", "element-2")).__next__
    )
    first = service.create_draft(valid_payload(), actor_uid="editor-1")
    service.transition(
        first.uid, "in_review", expected_version=1, actor_uid="editor-1"
    )
    service.transition(
        first.uid, "published", expected_version=2, actor_uid="admin-1"
    )
    service.create_draft(
        valid_payload(code="OTHER_ID", business_domain_uids=["domain-2"]),
        actor_uid="editor-1",
    )

    assert [
        item.uid
        for item in service.list(
            status="published", business_domain_uid="domain-1"
        )
    ] == ["element-1"]