115 lines
5.3 KiB
Python
115 lines
5.3 KiB
Python
from datetime import UTC, datetime
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import typer
|
|
|
|
from tht.cli import db_cmd, memory_cmd
|
|
from tht.cli.lsh_cmd import _extract_lsh_values
|
|
from tht.memory import MemoryRecord
|
|
from tht.mschema.models import Annotations, ColumnPhysical, PhysicalSchema, TablePhysical
|
|
from tht.ports.dwh import DistinctValues, DwhHealth
|
|
|
|
|
|
def _ping(monkeypatch, health, capsys):
|
|
monkeypatch.setattr(db_cmd, "load_config", lambda path: SimpleNamespace(database=SimpleNamespace(user="u")))
|
|
monkeypatch.setattr(db_cmd, "build_dwh", lambda cfg: SimpleNamespace(health=lambda: health))
|
|
try:
|
|
db_cmd.ping_cmd()
|
|
except typer.Exit as exc:
|
|
code = exc.exit_code
|
|
else:
|
|
code = 0
|
|
return code, capsys.readouterr()
|
|
|
|
|
|
def test_db_ping_public_health_success(monkeypatch, capsys):
|
|
code, output = _ping(monkeypatch, DwhHealth(ok=True, database="d", schema="s", read_only=True), capsys)
|
|
assert code == 0
|
|
assert "OK: connesso a d (schema s)" in output.out
|
|
|
|
|
|
def test_db_ping_rest_inaccessible_historical_wording(monkeypatch, capsys):
|
|
code, output = _ping(monkeypatch, DwhHealth(ok=False, detail="{'db_connected': False}", error_kind="inaccessible"), capsys)
|
|
assert code == 1
|
|
assert "ERRORE: DWH non accessibile via REST (risposta: {'db_connected': False})." in output.err
|
|
|
|
|
|
def test_db_ping_direct_connection_historical_wording(monkeypatch, capsys):
|
|
code, output = _ping(monkeypatch, DwhHealth(ok=False, detail="connection refused", error_kind="connection"), capsys)
|
|
assert code == 1
|
|
assert "ERRORE di connessione: connection refused" in output.err
|
|
|
|
|
|
@pytest.mark.parametrize(("limit", "truncated"), [(7, False), (1201, True)])
|
|
def test_lsh_extraction_honors_configured_limit(limit, truncated):
|
|
physical = PhysicalSchema(database="d", schema="s", introspected_at=datetime(2026, 1, 1, tzinfo=UTC), tables={
|
|
"t": TablePhysical(columns={"c": ColumnPhysical(type="text", eligible=True)})
|
|
})
|
|
calls = []
|
|
class Dwh:
|
|
def distinct_values(self, table, column, *, limit):
|
|
calls.append(limit)
|
|
return DistinctValues(values=list(range(limit)), truncated=truncated)
|
|
values, _, reports = _extract_lsh_values(Dwh(), physical, Annotations(), limit)
|
|
assert calls == [limit]
|
|
assert len(values["t"]["c"]) == limit
|
|
assert [report.indexed for report in reports] == ([limit] if truncated else [])
|
|
|
|
|
|
def test_memory_command_writes_through_factory_vector_store(monkeypatch):
|
|
store = SimpleNamespace(existing_hashes=lambda *args: {}, upsert=lambda table, rows: 1)
|
|
captured = []
|
|
original_upsert = store.upsert
|
|
store.upsert = lambda table, rows: captured.extend(rows) or original_upsert(table, rows)
|
|
# Legacy server deployments wrote directly through the factory and intentionally
|
|
# did not configure the workstation-only REST writer key.
|
|
cfg = SimpleNamespace(profile="server", embeddings=object(), vector_write_rest=None)
|
|
manifest = SimpleNamespace(id="s1")
|
|
snapshot = SimpleNamespace(manifest=manifest, decisions=[], artifacts={})
|
|
record = MemoryRecord(
|
|
id="m1", ts=datetime(2026, 1, 1, tzinfo=UTC), session_id="s1",
|
|
decision_seq=7, type="concept_clarified", subject="paziente attivo",
|
|
detail="flag_attivo = TRUE", question_context="q",
|
|
)
|
|
monkeypatch.setattr(memory_cmd, "_load_config_or_exit", lambda path: cfg)
|
|
monkeypatch.setattr(memory_cmd, "load_snapshot_or_exit", lambda cfg, session: snapshot)
|
|
monkeypatch.setattr(memory_cmd, "registry_path", lambda cfg: None)
|
|
monkeypatch.setattr("tht.adapters.factory.build_vector_store", lambda cfg, require_write: store)
|
|
monkeypatch.setattr("tht.cli.vector_cmd.make_embedder",
|
|
lambda cfg: SimpleNamespace(embed_documents=lambda texts: [[0.1]]))
|
|
monkeypatch.setattr("tht.memory.promote_snapshot", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr("tht.memory.load_registry", lambda path: [record])
|
|
memory_cmd.save_one_cmd(session="s1", decision=7, json_out=True)
|
|
from tht.ports.vector import VectorWriteRecord
|
|
assert len(captured) == 1 and isinstance(captured[0], VectorWriteRecord)
|
|
|
|
|
|
def test_solved_index_writes_through_writer_only_factory_store(monkeypatch):
|
|
writer_only_store = SimpleNamespace(
|
|
capabilities=SimpleNamespace(search=False, upsert=True),
|
|
existing_hashes=lambda *args: {},
|
|
upsert=lambda table, rows: 1,
|
|
)
|
|
cfg = SimpleNamespace(embeddings=object(), vector_write_rest=object())
|
|
manifest = SimpleNamespace(id="s1")
|
|
snapshot = SimpleNamespace(manifest=manifest, decisions=[], artifacts={})
|
|
calls = []
|
|
|
|
monkeypatch.setattr(memory_cmd, "load_snapshot_or_exit", lambda cfg, session: snapshot)
|
|
monkeypatch.setattr(
|
|
"tht.adapters.factory.build_vector_store",
|
|
lambda cfg, require_write: calls.append(require_write) or writer_only_store,
|
|
)
|
|
monkeypatch.setattr("tht.cli.sql_cmd.promoted_tables_for", lambda *args: [])
|
|
monkeypatch.setattr(
|
|
"tht.memory.index_solved_question",
|
|
lambda loaded, tables, *, store, embedder: int(
|
|
loaded is snapshot and tables == [] and store is writer_only_store
|
|
),
|
|
)
|
|
monkeypatch.setattr("tht.cli.vector_cmd.make_embedder", lambda cfg: object())
|
|
|
|
assert memory_cmd.index_solved_session(cfg, "s1") == 1
|
|
assert calls == [True]
|