DataOps-platform/tests/knowledge/test_chunking.py

from __future__ import annotations


def test_chunks_are_deterministic_and_trace_back_to_points():
    from app.core.knowledge.chunking import build_chunks
    from app.core.knowledge.point_builder import build_knowledge_snapshot

    snapshot = build_knowledge_snapshot(
        "DataFlow",
        {
            "uid": "01900000-0000-7000-8000-000000000901",
            "version": 1,
            "name": "sync",
            "purpose": "sync customer data",
            "owner": "governance",
        },
    )
    first = build_chunks(snapshot)
    second = build_chunks(snapshot)

    assert first == second
    assert len(first) == len(snapshot.points)
    assert all(chunk.point_keys == (chunk.primary_point_key,) for chunk in first)
    assert {chunk.primary_point_key for chunk in first} == {
        point.point_key for point in snapshot.points
    }


def test_embedding_input_hash_changes_with_profile_but_not_source_revision():
    from app.core.knowledge.chunking import build_chunks, embedding_input_hash
    from app.core.knowledge.point_builder import build_knowledge_snapshot

    first_snapshot = build_knowledge_snapshot(
        "DataFlow",
        {
            "uid": "01900000-0000-7000-8000-000000000902",
            "version": 1,
            "purpose": "same purpose",
        },
    )
    second_snapshot = build_knowledge_snapshot(
        "DataFlow",
        {
            "uid": "01900000-0000-7000-8000-000000000902",
            "version": 2,
            "purpose": "same purpose",
        },
    )
    first_chunk = build_chunks(first_snapshot)[0]
    second_chunk = build_chunks(second_snapshot)[0]

    assert first_chunk.content_hash == second_chunk.content_hash
    assert embedding_input_hash(first_chunk, "qwen:1024:v1") == embedding_input_hash(
        second_chunk, "qwen:1024:v1"
    )
    assert embedding_input_hash(first_chunk, "qwen:1024:v1") != embedding_input_hash(
        first_chunk, "qwen:1024:v2"
    )