Lars/mindnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 21 Wiki Activity

tests/test_edges_all.py aktualisiert
All checks were successful
Deploy mindnet to llm-node / deploy (push) Successful in 3s
Details

Browse Source
This commit is contained in:
Lars 2025-11-17 15:31:36 +01:00
parent 0cbc4a2d72
commit e4115e705e
1 changed files with 20 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
tests/test_edges_all.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import sys, json
from collections import Counter
from app.core.qdrant import QdrantConfig, get_client
@ -21,6 +22,12 @@ def fetch_all(client, col):
break
return points
def is_callout_rule(rule_id: str) -> bool:
if not rule_id:
return False
r = rule_id.lower()
return r.startswith("callout:edge:v1") or ("callout" in r)
def main():
cfg = QdrantConfig.from_env()
cl = get_client(cfg)
@ -33,20 +40,25 @@ def main():
edges = fetch_all(cl, ce)
chunks_by_note = Counter([c.payload.get("note_id") for c in chunks])
belongs_by_note = Counter(); next_by_note = Counter(); prev_by_note = Counter()
belongs_by_note = Counter()
next_by_note = Counter()
prev_by_note = Counter()
for e in edges:
pl = e.payload
nid = pl.get("note_id")
k = pl.get("kind") or pl.get("relation")
if k == "belongs_to": belongs_by_note[nid] += 1
elif k == "next": next_by_note[nid] += 1
elif k == "prev": prev_by_note[nid] += 1
if k == "belongs_to":
belongs_by_note[nid] += 1
elif k == "next":
next_by_note[nid] += 1
elif k == "prev":
prev_by_note[nid] += 1
for nid, ccount in chunks_by_note.items():
if belongs_by_note[nid] != ccount:
fail("belongs_to != chunks", {"note_id": nid, "chunks": ccount, "belongs_to": belongs_by_note[nid]})
if not (next_by_note[nid] == prev_by_note[nid] == max(ccount-1, 0)):
if not (next_by_note[nid] == prev_by_note[nid] == max(ccount - 1, 0)):
fail("next/prev mismatch", {"note_id": nid, "chunks": ccount, "next": next_by_note[nid], "prev": prev_by_note[nid]})
# Dubletten
@ -62,11 +74,10 @@ def main():
seen.add(key)
# Wenn Callouts vorhanden: mindestens eine Mehrfach-Ziel-Zeile muss erkannt worden sein
callouts = [e for e in edges if (e.payload.get("rule_id") or "").startswith("callout:edge:v1")]
callouts = [e for e in edges if is_callout_rule(e.payload.get("rule_id") or "")]
if callouts:
from collections import Counter
cnt = Counter((e.payload.get("chunk_id"), e.payload.get("kind") or e.payload.get("relation")) for e in callouts)
if max(cnt.values() or [0]) < 2:
ck = Counter((e.payload.get("chunk_id"), (e.payload.get("kind") or e.payload.get("relation"))) for e in callouts)
if max(ck.values() or [0]) < 2:
fail("callout edges present but no multi-target callout detected")
print(json.dumps({"ok": True, "notes_checked": len(chunks_by_note)}, ensure_ascii=False))