79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
"""L1: coppia domanda->SQL risolta (kind solved_question) — memoria attiva parte B.
|
|
|
|
Una sessione finalizzata produce UN record nel vectordb (tabella `memory`, kind
|
|
dedicato): embedding = domanda riscritta, metadata = {question, sql, tables,
|
|
session_id}. Upsert one-row stile D11 (mai sync: il suo delete-stale cancellerebbe
|
|
i record delle altre sessioni). L'hash di dedup copre domanda+SQL, cosi' un
|
|
re-finalize che cambia solo l'SQL aggiorna comunque la riga.
|
|
"""
|
|
from unittest.mock import MagicMock
|
|
|
|
from tht.solved import (
|
|
SOLVED_KIND,
|
|
_solved_hash,
|
|
save_solved_question,
|
|
solved_question_record,
|
|
)
|
|
from tht.vectorstore.reader import tables_for_kinds
|
|
|
|
|
|
def _rec(**kw):
|
|
base = dict(
|
|
session_id="s1", question="quante ablazioni nel 2023",
|
|
sql="SELECT count(*) FROM fact_seeablazione", tables=["fact_seeablazione"],
|
|
)
|
|
base.update(kw)
|
|
return solved_question_record(**base)
|
|
|
|
|
|
def test_record_shape():
|
|
r = _rec()
|
|
assert r.id == "solved:s1"
|
|
assert r.kind == SOLVED_KIND
|
|
assert r.content == "quante ablazioni nel 2023" # embedding = solo la domanda
|
|
assert r.metadata["sql"].startswith("SELECT")
|
|
assert r.metadata["tables"] == ["fact_seeablazione"]
|
|
assert r.metadata["session_id"] == "s1"
|
|
|
|
|
|
def test_solved_kind_maps_to_memory_table():
|
|
assert tables_for_kinds([SOLVED_KIND]) == ["memory"]
|
|
|
|
|
|
def test_save_upserts_single_row_into_memory_table():
|
|
writer = MagicMock()
|
|
writer.existing_hashes.return_value = {}
|
|
writer.upsert_records.return_value = 1
|
|
embedder = MagicMock()
|
|
embedder.embed_documents.return_value = [[0.1] * 8]
|
|
|
|
assert save_solved_question(_rec(), writer=writer, embedder=embedder) == 1
|
|
writer.sync.assert_not_called()
|
|
table, rows = writer.upsert_records.call_args[0]
|
|
assert table == "memory"
|
|
assert len(rows) == 1
|
|
assert rows[0]["record_key"] == "solved:s1"
|
|
assert rows[0]["metadata"]["kind"] == SOLVED_KIND
|
|
assert rows[0]["metadata"]["sql"].startswith("SELECT")
|
|
|
|
|
|
def test_save_skips_when_question_and_sql_unchanged():
|
|
r = _rec()
|
|
writer = MagicMock()
|
|
writer.existing_hashes.return_value = {r.id: _solved_hash(r)}
|
|
embedder = MagicMock()
|
|
assert save_solved_question(r, writer=writer, embedder=embedder) == 0
|
|
embedder.embed_documents.assert_not_called()
|
|
writer.upsert_records.assert_not_called()
|
|
|
|
|
|
def test_sql_change_alone_triggers_reupsert():
|
|
old = _rec()
|
|
new = _rec(sql="SELECT 1") # stessa domanda, SQL diverso
|
|
writer = MagicMock()
|
|
writer.existing_hashes.return_value = {old.id: _solved_hash(old)}
|
|
writer.upsert_records.return_value = 1
|
|
embedder = MagicMock()
|
|
embedder.embed_documents.return_value = [[0.0] * 4]
|
|
assert save_solved_question(new, writer=writer, embedder=embedder) == 1
|