ixoblakp
8dec8f6038
11 plugins from github.com/librefang/librefang-registry plugins/. index.json / index.json.sig are signed with Arka's Ed25519 key (not upstream stats.librefang.ai). Private key is not in this repo.
115 lines
3.5 KiB
Python
115 lines
3.5 KiB
Python
"""Tests for the ingest hook (stdin/stdout interface)."""
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
HOOK = Path(__file__).parent.parent / "hooks" / "ingest.py"
|
|
|
|
|
|
def run_hook(payload: dict) -> dict:
|
|
result = subprocess.run(
|
|
[sys.executable, str(HOOK)],
|
|
input=json.dumps(payload),
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return json.loads(result.stdout.strip())
|
|
|
|
|
|
def test_bad_json_returns_empty():
|
|
result = subprocess.run(
|
|
[sys.executable, str(HOOK)],
|
|
input="not json",
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
out = json.loads(result.stdout.strip())
|
|
assert out == {"memories": []}
|
|
|
|
|
|
def test_empty_message_returns_empty():
|
|
out = run_hook({"type": "ingest", "agent_id": "a1", "message": ""})
|
|
assert out == {"memories": []}
|
|
|
|
|
|
def test_short_message_returns_empty():
|
|
out = run_hook({"type": "ingest", "agent_id": "a1", "message": "hi"})
|
|
assert out == {"memories": []}
|
|
|
|
|
|
def test_no_mempalace_returns_error_not_crash():
|
|
"""Without mempalace installed, ingest returns error field but doesn't crash."""
|
|
out = run_hook({"type": "ingest", "agent_id": "a1", "message": "What are my upcoming meetings?"})
|
|
assert "memories" in out
|
|
assert isinstance(out["memories"], list)
|
|
assert "error" in out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Similarity filtering (unit-level, no mempalace needed)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _load_ingest_module(min_similarity="0.3"):
|
|
saved = os.environ.get("MEMPALACE_MIN_SIMILARITY")
|
|
os.environ["MEMPALACE_MIN_SIMILARITY"] = min_similarity
|
|
spec = importlib.util.spec_from_file_location("ingest", HOOK)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
if saved is None:
|
|
os.environ.pop("MEMPALACE_MIN_SIMILARITY", None)
|
|
else:
|
|
os.environ["MEMPALACE_MIN_SIMILARITY"] = saved
|
|
return mod
|
|
|
|
|
|
def test_truncate_at_word_boundary():
|
|
mod = _load_ingest_module()
|
|
text = "one two three four five six seven"
|
|
result = mod.truncate_at_word(text, 15)
|
|
assert result.endswith("...")
|
|
assert len(result) <= 18
|
|
|
|
|
|
def test_truncate_short_unchanged():
|
|
mod = _load_ingest_module()
|
|
assert mod.truncate_at_word("hello", 100) == "hello"
|
|
|
|
|
|
def test_similarity_threshold_filters_results():
|
|
"""Simulate the similarity filtering logic directly."""
|
|
mod = _load_ingest_module(min_similarity="0.5")
|
|
|
|
results = [
|
|
{"text": "good match", "source_file": "s1", "wing": "w1", "similarity": 0.8},
|
|
{"text": "bad match", "source_file": "s2", "wing": "w1", "similarity": 0.2},
|
|
{"text": "no score", "source_file": "s3", "wing": "w1"},
|
|
]
|
|
|
|
MIN_SIMILARITY = mod.MIN_SIMILARITY
|
|
memories = []
|
|
for r in results:
|
|
text = r.get("text", "")
|
|
similarity = r.get("similarity")
|
|
if not text:
|
|
continue
|
|
if similarity is not None and MIN_SIMILARITY > 0 and similarity < MIN_SIMILARITY:
|
|
continue
|
|
memories.append(text)
|
|
|
|
assert "good match" in memories
|
|
assert "bad match" not in memories
|
|
assert "no score" in memories # passthrough when similarity is absent
|
|
|
|
|
|
def test_similarity_zero_disables_filtering():
|
|
mod = _load_ingest_module(min_similarity="0")
|
|
assert mod.MIN_SIMILARITY == 0.0
|
|
|
|
|
|
def test_min_similarity_default():
|
|
mod = _load_ingest_module()
|
|
assert mod.MIN_SIMILARITY == 0.3
|