DataOps-platform/tests/test_governance_identifiers.py

from __future__ import annotations

import uuid

import pytest


def test_generated_governance_uid_is_uuid7_compatible():
    from app.core.common.identifiers import new_governance_uid

    value = new_governance_uid()
    parsed = uuid.UUID(value)

    assert parsed.version == 7
    assert parsed.variant == uuid.RFC_4122


def test_ensure_governance_uid_preserves_retry_identity():
    from app.core.common.identifiers import ensure_governance_uid

    properties = {"name_zh": "客户域"}
    first = ensure_governance_uid(properties)
    second = ensure_governance_uid(properties)

    assert first == second
    assert properties["uid"] == first


def test_ensure_governance_uid_rejects_invalid_or_non_v7_values():
    from app.core.common.identifiers import ensure_governance_uid

    with pytest.raises(ValueError, match="UUIDv7"):
        ensure_governance_uid({"uid": "not-a-uuid"})
    with pytest.raises(ValueError, match="UUIDv7"):
        ensure_governance_uid({"uid": str(uuid.uuid4())})


class RecordingSession:
    def __init__(self):
        self.calls = []

    def run(self, query, parameters=None, **kwargs):
        self.calls.append((query, parameters or kwargs))
        return []


def test_reconciliation_defines_uid_constraints_and_reports_without_writing():
    from app.commands.reconcile_governance_uids import (
        ensure_uid_constraints,
        report_missing_uids,
    )

    session = RecordingSession()
    ensure_uid_constraints(session)
    report_missing_uids(session, limit=25)

    queries = "\n".join(query for query, _ in session.calls)
    assert "business_domain_uid" in queries
    assert "data_meta_uid" in queries
    assert "data_flow_uid" in queries
    assert "datasource_uid" in queries
    assert "IS UNIQUE" in queries
    assert "SET n.uid" not in queries
    assert "n.uid IS NULL" in queries


def test_primary_governance_creation_paths_attach_uid():
    from pathlib import Path

    root = Path(__file__).resolve().parents[1]
    business_domain = (root / "app/core/business_domain/business_domain.py").read_text(
        encoding="utf-8"
    )
    dataflow = (root / "app/core/data_flow/dataflows.py").read_text(encoding="utf-8")
    metadata_routes = (root / "app/api/meta_data/routes.py").read_text(
        encoding="utf-8"
    )

    assert business_domain.count("ensure_governance_uid(") >= 4
    assert dataflow.count("ensure_governance_uid(") >= 1
    assert metadata_routes.count("ensure_governance_uid(") >= 3