silk-graph 0.2.6

"""Compaction policies — automatic oplog management.

Tests for IntervalPolicy, ThresholdPolicy, and the CompactionPolicy protocol.
"""

import time
from silk import GraphStore, CompactionPolicy, IntervalPolicy, ThresholdPolicy

ONTOLOGY = {
    "node_types": {"entity": {"properties": {}}},
    "edge_types": {}
}


def _store():
    return GraphStore("test", ONTOLOGY)


# -- ThresholdPolicy --


def test_threshold_no_compact_below():
    """Below threshold: no compaction."""
    store = _store()
    store.add_node("n1", "entity", "A")
    policy = ThresholdPolicy(max_entries=100)
    result = policy.check(store)
    assert result is None
    assert store.len() > 1  # not compacted


def test_threshold_compacts_above():
    """Above threshold: compaction triggered."""
    store = _store()
    for i in range(20):
        store.add_node(f"n{i}", "entity", f"Node {i}")
    assert store.len() > 20  # genesis + 20 nodes

    policy = ThresholdPolicy(max_entries=10)
    result = policy.check(store)
    assert result is not None  # compaction happened
    assert len(result) == 64  # hex hash
    assert store.len() == 1  # compacted to checkpoint


def test_threshold_preserves_data():
    """Compaction via policy preserves all nodes."""
    store = _store()
    for i in range(15):
        store.add_node(f"n{i}", "entity", f"Node {i}")

    policy = ThresholdPolicy(max_entries=5)
    policy.check(store)

    for i in range(15):
        assert store.get_node(f"n{i}") is not None


def test_threshold_should_compact():
    """should_compact returns bool correctly."""
    store = _store()
    for i in range(10):
        store.add_node(f"n{i}", "entity", f"Node {i}")

    policy = ThresholdPolicy(max_entries=100)
    assert policy.should_compact(store) is False

    policy2 = ThresholdPolicy(max_entries=5)
    assert policy2.should_compact(store) is True


# -- IntervalPolicy --


def test_interval_first_check_compacts():
    """First check always compacts (last_compact is 0)."""
    store = _store()
    store.add_node("n1", "entity", "A")

    policy = IntervalPolicy(seconds=0.01)
    time.sleep(0.02)
    result = policy.check(store)
    assert result is not None
    assert store.len() == 1


def test_interval_second_check_skips():
    """Second check within interval: no compaction."""
    store = _store()
    store.add_node("n1", "entity", "A")

    policy = IntervalPolicy(seconds=10.0)  # 10 second interval
    policy.check(store)  # first: compacts

    store.add_node("n2", "entity", "B")
    result = policy.check(store)  # second: too soon
    assert result is None
    assert store.len() == 2  # checkpoint + new node, not re-compacted


def test_interval_compacts_after_wait():
    """After interval passes: compaction happens again."""
    store = _store()
    store.add_node("n1", "entity", "A")

    policy = IntervalPolicy(seconds=0.01)
    policy.check(store)

    store.add_node("n2", "entity", "B")
    time.sleep(0.02)
    result = policy.check(store)
    assert result is not None
    assert store.len() == 1  # re-compacted


# -- CompactionPolicy protocol --


def test_custom_policy():
    """Custom policy via the protocol."""
    class AlwaysCompact:
        def should_compact(self, store):
            return True

        def check(self, store):
            if self.should_compact(store):
                return store.compact()
            return None

    store = _store()
    store.add_node("n1", "entity", "A")

    policy = AlwaysCompact()
    result = policy.check(store)
    assert result is not None
    assert store.len() == 1


def test_custom_policy_never():
    """Custom policy that never compacts."""
    class NeverCompact:
        def should_compact(self, store):
            return False

        def check(self, store):
            return None

    store = _store()
    store.add_node("n1", "entity", "A")

    policy = NeverCompact()
    result = policy.check(store)
    assert result is None
    assert store.len() > 1


def test_protocol_check():
    """CompactionPolicy is runtime-checkable."""
    class Good:
        def should_compact(self, store): return False
        def check(self, store): return None

    class Bad:
        pass

    assert isinstance(Good(), CompactionPolicy)
    assert not isinstance(Bad(), CompactionPolicy)


def test_builtin_policies_implement_protocol():
    """Built-in policies satisfy the protocol."""
    assert isinstance(IntervalPolicy(seconds=60), CompactionPolicy)
    assert isinstance(ThresholdPolicy(max_entries=100), CompactionPolicy)


# -- reclaim_disk pass-through --


def test_policies_pass_reclaim_disk(tmp_path):
    """Policies with reclaim_disk=True shrink the file when they compact."""
    import os

    path = str(tmp_path / "policy.redb")
    ontology = {
        "node_types": {"entity": {"properties": {"status": {"value_type": "string"}}}},
        "edge_types": {}
    }
    store = GraphStore("test", ontology, path=path)
    filler = "x" * 200
    for i in range(300):
        store.add_node(f"n{i}", "entity", f"Node {i}", {"status": filler})
    for i in range(300):
        store.remove_node(f"n{i}")

    size_before = os.path.getsize(path)
    policy = ThresholdPolicy(max_entries=10, reclaim_disk=True)
    result = policy.check(store)
    assert result is not None
    assert store.len() == 1
    assert os.path.getsize(path) < size_before / 2


def test_interval_policy_accepts_reclaim_disk():
    """IntervalPolicy carries the flag; in-memory store is a no-op."""
    store = _store()
    store.add_node("n1", "entity", "A")
    policy = IntervalPolicy(seconds=0, reclaim_disk=True)
    result = policy.check(store)
    assert result is not None
    assert store.len() == 1