100 lines
3.9 KiB
Python
100 lines
3.9 KiB
Python
"""L1: `tht memory solved-search` — degrado gentile e mapping dei risultati.
|
|
|
|
SKILL.md prescrive solved-search in F4/F6/F7 di OGNI sessione: a vectordb
|
|
irraggiungibile (VPN giu', Ollama spento) il comando non deve morire con un
|
|
traceback grezzo ma degradare a un avviso di una riga su stderr, con stdout
|
|
puro (`[]` in modalita' --json) ed exit 0, cosi' il modello prosegue senza
|
|
exemplar. Il finalize-hook gestisce gia' lo stesso scenario in modo analogo.
|
|
"""
|
|
import json
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli import app
|
|
from tht.ports.vector import VectorReadUnavailable
|
|
from tht.vectorstore.rest_client import VectorRestError
|
|
from tht.vectorstore.store import VectorHit
|
|
|
|
|
|
def _cfg(tmp_path):
|
|
cfg = tmp_path / "workspace.yaml"
|
|
cfg.write_text(
|
|
"database: {database: d, schema: s, user: u, password: p, transport: direct}\n"
|
|
f"paths: {{artifacts: {tmp_path/'a'}, indexes: {tmp_path/'i'}, sessions: {tmp_path/'se'}}}\n"
|
|
"vector_db: {database: v, schema: vectors, user: u, password: p, transport: direct}\n"
|
|
"embeddings: {base_url: 'http://localhost:11434', model: nomic-embed-text, dim: 8}\n"
|
|
)
|
|
return cfg
|
|
|
|
|
|
def test_solved_search_degrades_when_vectordb_unreachable(tmp_path, monkeypatch):
|
|
def boom(cfg):
|
|
raise VectorRestError("Vector REST non raggiungibile su https://v/ (rpc search_similar)")
|
|
|
|
monkeypatch.setattr("tht.cli.vector_cmd.open_searcher", boom)
|
|
res = CliRunner().invoke(
|
|
app, ["memory", "solved-search", "quante ablazioni", "--json", "-c", str(_cfg(tmp_path))]
|
|
)
|
|
assert res.exit_code == 0, res.output
|
|
assert json.loads(res.stdout) == [] # stdout puro: JSON valido
|
|
assert "exemplar non disponibili" in res.stderr
|
|
|
|
|
|
def test_solved_search_degrades_direct_vector_read_error(tmp_path, monkeypatch):
|
|
def boom(cfg):
|
|
raise VectorReadUnavailable("Vector read operation unavailable")
|
|
|
|
monkeypatch.setattr("tht.cli.vector_cmd.open_searcher", boom)
|
|
res = CliRunner().invoke(
|
|
app, ["memory", "solved-search", "quante ablazioni", "--json", "-c", str(_cfg(tmp_path))]
|
|
)
|
|
assert res.exit_code == 0, res.output
|
|
assert json.loads(res.stdout) == []
|
|
assert "exemplar non disponibili" in res.stderr
|
|
|
|
|
|
def test_solved_search_degrades_human_mode(tmp_path, monkeypatch):
|
|
def boom(cfg):
|
|
raise VectorRestError("Vector REST non raggiungibile")
|
|
|
|
monkeypatch.setattr("tht.cli.vector_cmd.open_searcher", boom)
|
|
res = CliRunner().invoke(
|
|
app, ["memory", "solved-search", "quante ablazioni", "-c", str(_cfg(tmp_path))]
|
|
)
|
|
assert res.exit_code == 0, res.output
|
|
assert "exemplar non disponibili" in res.stderr
|
|
assert "Traceback" not in res.stderr
|
|
|
|
|
|
def test_solved_search_json_maps_hit_metadata(tmp_path, monkeypatch):
|
|
hit = VectorHit(
|
|
id="solved:s1", kind="solved_question", ref="s1", title="quante ablazioni nel 2023",
|
|
content="quante ablazioni nel 2023",
|
|
metadata={
|
|
"session_id": "s1", "question": "quante ablazioni nel 2023",
|
|
"sql": "SELECT 1", "tables": ["fact_seeablazione"],
|
|
},
|
|
similarity=0.91,
|
|
)
|
|
|
|
class FakeSearcher:
|
|
def search(self, vec, top_n=10, kinds=None):
|
|
assert kinds == ["solved_question"]
|
|
return [hit]
|
|
|
|
class FakeEmbedder:
|
|
def embed_query(self, text):
|
|
return [0.1] * 8
|
|
|
|
monkeypatch.setattr("tht.cli.vector_cmd.open_searcher", lambda cfg: FakeSearcher())
|
|
monkeypatch.setattr("tht.cli.vector_cmd.make_embedder", lambda e: FakeEmbedder())
|
|
res = CliRunner().invoke(
|
|
app, ["memory", "solved-search", "quante ablazioni", "--json", "-c", str(_cfg(tmp_path))]
|
|
)
|
|
assert res.exit_code == 0, res.output
|
|
data = json.loads(res.stdout)
|
|
assert data == [{
|
|
"session_id": "s1", "question": "quante ablazioni nel 2023",
|
|
"sql": "SELECT 1", "tables": ["fact_seeablazione"], "score": 0.91,
|
|
}]
|