DataOps-platform/tests/knowledge/test_cache_invalidation.py

from __future__ import annotations


def test_cache_invalidation_targets_direct_and_impacted_points():
    from app.core.knowledge.cache_invalidation import (
        CacheDependency,
        plan_cache_invalidation,
    )

    dependencies = [
        CacheDependency("cache-a", "DataMeta/a/definition", "domain-a", 1),
        CacheDependency("cache-b", "DataFlow/b/purpose", "domain-a", 1),
        CacheDependency("cache-c", "DataMeta/c/definition", "domain-b", 1),
    ]
    plan = plan_cache_invalidation(
        dependencies,
        changed_point_keys={"DataMeta/a/definition"},
        impacted_point_keys={"DataFlow/b/purpose"},
    )
    assert plan.cache_key_hashes == frozenset({"cache-a", "cache-b"})
    assert plan.business_domains_to_flush == frozenset()


def test_missing_dependency_metadata_falls_back_to_business_domain_flush():
    from app.core.knowledge.cache_invalidation import plan_cache_invalidation

    plan = plan_cache_invalidation(
        [],
        changed_point_keys={"DataMeta/a/definition"},
        impacted_point_keys=set(),
        dependency_metadata_complete=False,
        affected_business_domains={"domain-a"},
    )
    assert plan.cache_key_hashes == frozenset()
    assert plan.business_domains_to_flush == frozenset({"domain-a"})