DataOps-platform/tests/knowledge/test_impact.py

from __future__ import annotations


def _dependency(from_key: str, to_key: str, relation: str = "uses"):
    from app.core.knowledge.contracts import KnowledgeDependencyDraft

    return KnowledgeDependencyDraft(
        from_point_key=from_key,
        to_point_key=to_key,
        relation_type=relation,
        source="governance",
    )


def test_impact_propagates_reverse_dependencies_and_records_path():
    from app.core.knowledge.impact import propagate_impact

    root = "DataMeta/field-1/definition"
    standard = "DataStandard/std-1/rules/rule-1"
    flow = "DataFlow/flow-1/purpose"
    result = propagate_impact(
        {root},
        [_dependency(standard, root, "governs"), _dependency(flow, standard)],
        allowed_relations={"governs", "uses"},
        max_hops=3,
        max_points=20,
        max_objects=20,
    )

    assert result.impacted_point_keys == frozenset({standard, flow})
    assert result.paths[flow] == (root, standard, flow)
    assert result.truncated is False


def test_impact_stops_cycles_and_ignores_disallowed_relations():
    from app.core.knowledge.impact import propagate_impact

    a = "DataMeta/a/definition"
    b = "DataFlow/b/purpose"
    c = "DataStandard/c/definition"
    result = propagate_impact(
        {a},
        [
            _dependency(b, a),
            _dependency(a, b),
            _dependency(c, b, "lightrag_extracted"),
        ],
        allowed_relations={"uses"},
        max_hops=5,
        max_points=20,
        max_objects=20,
    )
    assert result.impacted_point_keys == frozenset({b})
    assert result.truncated is False


def test_impact_marks_result_truncated_instead_of_publishing_partial_scope():
    from app.core.knowledge.impact import propagate_impact

    root = "DataMeta/root/definition"
    dependencies = [
        _dependency(f"DataFlow/flow-{index}/purpose", root) for index in range(3)
    ]
    result = propagate_impact(
        {root},
        dependencies,
        allowed_relations={"uses"},
        max_hops=2,
        max_points=1,
        max_objects=10,
    )
    assert result.truncated is True
    assert len(result.impacted_point_keys) <= 1