DataOps-platform/tests/acceptance/test_data_research_v60_v65.py

from __future__ import annotations

from pathlib import Path


ROOT = Path(__file__).resolve().parents[2]


def valid_graph():
    return {
        "classes": [{"uid": "customer", "name": "Customer"}],
        "properties": [
            {"uid": "customer-id", "name": "customerId", "class_uid": "customer", "required": True, "cardinality": "1", "data_element_uid": "element-1"}
        ],
        "relations": [],
        "constraints": [],
        "domain_links": [
            {"domain_uid": "sales", "role": "owner"},
            {"domain_uid": "service", "role": "contributor"},
        ],
        "element_mappings": [{"property_uid": "customer-id", "data_element_uid": "element-1"}],
    }


def test_v60_v62_acceptance_multi_source_policy_evidence_and_redaction():
    from app.core.data_research.artifacts import redact_excerpt
    from app.core.data_research.extractors.registry import default_registry
    from app.core.data_research.file_policy import FilePolicy

    registry = default_registry()
    assert registry.resolve("text/sql", "schema.sql").parser_version.startswith("sql")
    assert registry.resolve("text/csv", "elements.csv").parser_version.startswith("csv")
    assert registry.resolve("application/pdf", "standard.pdf").parser_version.startswith("pdf")
    assert registry.resolve(
        "application/vnd.openxmlformats-officedocument.wordprocessingml.document", "standard.docx"
    ).parser_version.startswith("docx")
    assert set(FilePolicy.MEDIA_TYPES) == {".sql", ".csv", ".xls", ".xlsx", ".docx", ".pdf", ".png", ".jpg", ".jpeg"}
    redacted = redact_excerpt("phone=13800138000 token=top-secret")
    assert "138****8000" in redacted and "top-secret" not in redacted


def test_v61_v64_acceptance_element_ontology_dynamic_publish_rollback_and_knowledge():
    from app.core.data_research.data_elements import DataElementService
    from app.core.data_research.ontology.change_sets import ChangeSetService
    from app.core.data_research.ontology.knowledge_sync import OntologyKnowledgeSync
    from app.core.data_research.ontology.publication import OntologyPublicationService
    from app.core.data_research.ontology.repository import MemoryOntologyRepository
    from app.core.data_research.ontology.suggestions import OntologySuggestionEngine

    class Elements:
        def __init__(self): self.records = {}
        def get(self, uid): return self.records.get(uid)
        def get_by_code(self, code): return next((x for x in self.records.values() if x.code == code), None)
        def save(self, record): self.records[record.uid] = record; return record

    element_events = []
    elements = DataElementService(Elements(), uid_factory=lambda: "element-1", outbox_enqueue=lambda **event: element_events.append(event))
    element = elements.create_draft(
        {"code": "CUSTOMER_ID", "name_zh": "客户编号", "name_en": "customer_id", "definition": "客户唯一编号", "logical_type": "string", "evidence_uids": ["evidence-1"], "business_domain_uids": ["sales", "service"]},
        "editor-1",
    )
    elements.transition(element.uid, "in_review", expected_version=1, actor_uid="editor-1")
    published_element = elements.transition(element.uid, "published", expected_version=2, actor_uid="admin-1")
    assert published_element.current_version == 3 and element_events[0]["event_type"] == "data_element.version_published"

    ids = iter(("ontology-1", "version-1", "version-2"))
    repository = MemoryOntologyRepository(uid_factory=ids.__next__)
    ontology = repository.create(code="CUSTOMER", name="客户本体", owner_uid="admin-1", domain_links=valid_graph()["domain_links"])
    assert {link.domain_uid for link in ontology.domain_links} == {"sales", "service"}
    version = repository.save_draft(ontology.uid, valid_graph(), expected_revision=0, actor_uid="editor-1")
    ontology_events = []
    publication = OntologyPublicationService(repository, outbox_enqueue=lambda **event: ontology_events.append(event))
    published = publication.publish(version.uid, idempotency_key="publish-1", actor_uid="admin-1")
    rollback = publication.create_rollback(ontology.uid, target_version_uid=published.uid, expected_revision=1, actor_uid="admin-1")
    assert published.status == "published" and rollback.version == 2

    suggestions = OntologySuggestionEngine().generate(
        domains=[{"uid": "sales", "name": "销售"}],
        data_elements=[{"uid": "element-1", "name": "客户编号", "status": "published"}],
        foreign_keys=[],
    )
    changes = ChangeSetService(uid_factory=lambda: "change-1")
    change_set = changes.create(ontology.uid, suggestions, actor_uid="editor-1")
    reviewed = changes.decide(change_set, [{"suggestion_uid": suggestions[0].uid, "decision": "accept", "reason": "已核验证据"}], actor_uid="reviewer-1")
    applied = changes.apply(valid_graph(), reviewed)
    assert len(applied["classes"]) == 2

    class KnowledgeRepo:
        def __init__(self): self.docs = {}; self.chunks = {}
        def upsert_document(self, doc):
            key = (doc["object_uid"], doc["object_version"]); changed = key not in self.docs; self.docs[key] = doc; return changed
        def replace_chunks(self, key, chunks): self.chunks[key] = chunks

    synced = OntologyKnowledgeSync(KnowledgeRepo()).sync({"uid": ontology.uid, "name": ontology.name, "version": published.version, "status": "published", "graph_document": valid_graph()})
    assert synced["document"]["object_version"] == 1


def test_v65_acceptance_rbac_reconciliation_and_release_parity():
    from app.core.system.permissions import permission_for_request

    assert permission_for_request("/api/development/v1/ontologies/o/publish", "POST") == ("ontologies:publish",)
    assert permission_for_request("/api/development/v1/sources/files", "POST") == ("ingestion:run",)
    assert permission_for_request("/api/development/v1/evidence/e", "GET") == ("governance:read",)

    for relative in (
        "core/data_research/ontology/publication.py",
        "core/data_research/operations.py",
        "api/data_development/routes.py",
    ):
        assert (ROOT / "deployment/app" / relative).read_bytes() == (ROOT / "app" / relative).read_bytes()

    overview = (ROOT / "docs/architecture/ARCHITECTURE_OVERVIEW.md").read_text(encoding="utf-8")
    data_model = (ROOT / "docs/architecture/DATA_MODEL.md").read_text(encoding="utf-8")
    roadmap = (ROOT / "docs/architecture/NEXT_ITERATION_ROADMAP.md").read_text(encoding="utf-8")
    assert "V60-V65 数据研发" in overview
    assert "ontology_versions" in data_model
    assert "V65" in roadmap and "已完成" in roadmap