Spaces:
Sleeping
Sleeping
File size: 4,134 Bytes
1605cbb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | """
Pytest fixtures for FALSIFY.
The core belief-revision logic (refutation propagation, orphan forget, hypothesis
promotion) is pure graph reasoning: given nodes, typed edges, and truth-states, it
decides what to flip and what to delete. We test that logic against an in-memory
:class:`FakeGraph` that stands in for Cognee's graph/vector engines — so the whole
suite runs with **no API key, no database, no network**.
Only the handful of *engine-touching* helpers in :mod:`falsify.graph_ops` are
monkeypatched (``load_graph``, ``get_truth``, ``set_state``, ``set_weight``,
``delete_from_both_stores``, ``get_vector_engine``). The pure adjacency helpers
(``dependents_of``, ``incoming``, ``outgoing``, ``node_label``) are exercised as-is,
so the tests cover the real code paths.
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional, Tuple
import pytest
from falsify import graph_ops
from falsify.models import TruthState
Edge = Tuple[str, str, str, Dict[str, Any]]
Node = Tuple[str, Dict[str, Any]]
class FakeGraph:
"""A minimal in-memory graph mimicking the subset of the engine FALSIFY uses.
Nodes are ``{id: props}``; edges are ``(src, dst, rel, props)`` tuples; truth is
``{id: {"truth_alignment": [...], "truth_epoch": n}}``. Deletions remove nodes and
record deleted ids/collections so tests can assert vector cleanup happened.
"""
def __init__(self) -> None:
self.nodes: Dict[str, Dict[str, Any]] = {}
self.edges: List[Edge] = []
self.truth: Dict[str, Dict[str, Any]] = {}
self.weights: Dict[str, float] = {}
self.deleted: List[str] = []
self.deleted_from_collections: List[str] = []
# -- builders -------------------------------------------------------
def add_node(self, nid: str, **props: Any) -> str:
self.nodes[nid] = props
self.truth.setdefault(nid, {"truth_alignment": [TruthState.ALIVE.value], "truth_epoch": 0})
return nid
def add_edge(self, src: str, dst: str, rel: str, **props: Any) -> None:
self.edges.append((src, dst, rel, props))
# -- engine-shaped async API (monkeypatched onto graph_ops) ---------
async def load_graph(self) -> Tuple[List[Node], List[Edge]]:
nodes = [(nid, dict(props)) for nid, props in self.nodes.items()]
return nodes, list(self.edges)
async def get_truth(self, node_ids: List[str]) -> Dict[str, List[str]]:
out: Dict[str, List[str]] = {}
for nid in node_ids:
entry = self.truth.get(str(nid))
out[str(nid)] = list(entry["truth_alignment"]) if entry else [TruthState.ALIVE.value]
return out
async def set_state(self, node_id: str, state: TruthState, epoch: int) -> None:
value = state.value if isinstance(state, TruthState) else str(state)
self.truth[str(node_id)] = {"truth_alignment": [value], "truth_epoch": int(epoch)}
async def set_weight(self, node_id: str, weight: float) -> None:
self.weights[str(node_id)] = float(weight)
async def delete_from_both_stores(self, node_ids: List[str], collections: List[str]) -> int:
for nid in node_ids:
self.nodes.pop(str(nid), None)
self.truth.pop(str(nid), None)
self.deleted.append(str(nid))
self.edges = [e for e in self.edges if e[0] not in node_ids and e[1] not in node_ids]
self.deleted_from_collections.extend(collections)
return len(node_ids)
@pytest.fixture
def fake_graph(monkeypatch) -> FakeGraph:
"""Install a :class:`FakeGraph` in place of the real engine helpers.
Returns the graph so a test can build topology (``add_node``/``add_edge``) and
later assert on final truth-states and deletions.
"""
g = FakeGraph()
monkeypatch.setattr(graph_ops, "load_graph", g.load_graph)
monkeypatch.setattr(graph_ops, "get_truth", g.get_truth)
monkeypatch.setattr(graph_ops, "set_state", g.set_state)
monkeypatch.setattr(graph_ops, "set_weight", g.set_weight)
monkeypatch.setattr(graph_ops, "delete_from_both_stores", g.delete_from_both_stores)
return g
|