DataOps-platform/tests/data_research/test_ontology_knowledge_sync.py

from __future__ import annotations

import pytest

from tests.data_research.test_ontology_validation import base_graph


class Repository:
    def __init__(self):
        self.documents = {}
        self.chunks = {}

    def upsert_document(self, document):
        key = (document["object_uid"], document["object_version"])
        existing = self.documents.get(key)
        self.documents[key] = document
        return existing is None or existing["content_hash"] != document["content_hash"]

    def replace_chunks(self, key, chunks):
        self.chunks[key] = list(chunks)


def test_only_published_versions_sync_as_versioned_idempotent_documents():
    from app.core.data_research.ontology.knowledge_sync import OntologyKnowledgeSync

    repository = Repository()
    service = OntologyKnowledgeSync(repository, max_chunk_chars=80)
    payload = {
        "uid": "ontology-1",
        "name": "客户本体",
        "version": 3,
        "status": "published",
        "graph_document": base_graph(),
        "token": "must-not-sync",
    }
    first = service.sync(payload)
    second = service.sync(payload)

    assert first["changed"] is True
    assert second["changed"] is False
    assert first["document"]["object_version"] == 3
    assert "must-not-sync" not in first["document"]["content"]
    assert len(repository.chunks[("ontology-1", 3)]) > 1

    with pytest.raises(ValueError, match="published"):
        service.sync({**payload, "status": "draft"})