Lars
79dcfdcf9a
All checks were successful
Deploy mindnet to llm-node / deploy (push) Successful in 3s
59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
tests/assert_payload_schema.py
|
|
|
|
Prüft, ob die erwarteten Payload-Indizes (payload_schema) auf den drei Collections vorhanden sind.
|
|
- mindnet_notes : note_id, type, title, updated, tags
|
|
- mindnet_chunks : note_id, chunk_id, index, type, tags
|
|
- mindnet_edges : note_id, kind, scope, source_id, target_id, chunk_id
|
|
|
|
Aufruf:
|
|
python3 -m tests.assert_payload_schema
|
|
"""
|
|
from __future__ import annotations
|
|
import json, sys
|
|
from typing import Dict, List
|
|
from app.core.qdrant import QdrantConfig, get_client, collection_names
|
|
|
|
REQUIRED = {
|
|
"notes": ["note_id", "type", "title", "updated", "tags"],
|
|
"chunks": ["note_id", "chunk_id", "index", "type", "tags"],
|
|
"edges": ["note_id", "kind", "scope", "source_id", "target_id", "chunk_id"],
|
|
}
|
|
|
|
def fetch_schema(client, name: str) -> Dict[str, dict]:
|
|
info = client.get_collection(collection_name=name)
|
|
d = info.dict() if hasattr(info, "dict") else info
|
|
return (d.get("result", {}) or {}).get("payload_schema") or {}
|
|
|
|
def ensure_fields(schema: Dict[str, dict], required: List[str]) -> Dict[str, bool]:
|
|
return {k: (k in schema) for k in required}
|
|
|
|
def main():
|
|
cfg = QdrantConfig.from_env()
|
|
client = get_client(cfg)
|
|
notes, chunks, edges = collection_names(cfg.prefix)
|
|
names = {"notes": notes, "chunks": chunks, "edges": edges}
|
|
|
|
report = {}
|
|
ok_all = True
|
|
for kind, col in names.items():
|
|
sch = fetch_schema(client, col)
|
|
checks = ensure_fields(sch, REQUIRED[kind])
|
|
ok = all(checks.values())
|
|
ok_all = ok_all and ok
|
|
report[kind] = {
|
|
"collection": col,
|
|
"ok": ok,
|
|
"missing": [k for k,v in checks.items() if not v],
|
|
"present": [k for k,v in checks.items() if v],
|
|
}
|
|
|
|
print(json.dumps({"prefix": cfg.prefix, "ok": ok_all, "report": report}, ensure_ascii=False, indent=2))
|
|
if not ok_all:
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|