Codex
82e2c91f42
Publish documentation / publish (push) Successful in 1m27s
Add PostgreSQL-backed memory, editable evidence with source review and activation, and human-approved archive repairs across the harness, API, and UI. Include migrations, deployment support, regression coverage, and validation documentation. Refresh permissions from validated session roles so existing administrator logins can access newly deployed archive management features.
81 lines
3.6 KiB
Python
81 lines
3.6 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.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_save_one_passes_the_selected_source_to_authoritative_service(monkeypatch, capsys):
|
|
snapshot = object()
|
|
calls = []
|
|
service = SimpleNamespace(close=lambda: None, promote=lambda source, seqs:
|
|
calls.append((source, seqs)) or [{"indexed": False, "saved": True}])
|
|
monkeypatch.setattr(memory_cmd, "_load_config_or_exit", lambda path: object())
|
|
monkeypatch.setattr(memory_cmd, "load_snapshot_or_exit", lambda cfg, session: snapshot)
|
|
monkeypatch.setattr(memory_cmd, "memory_service", lambda cfg: service)
|
|
memory_cmd.save_one_cmd(session="s1", decision=7, json_out=True)
|
|
assert calls == [(snapshot, [7])]
|
|
assert '"indexed": false' in capsys.readouterr().out
|
|
|
|
|
|
def test_solved_recovery_uses_authority_without_recreating_historical_content(monkeypatch):
|
|
snapshot = object()
|
|
calls = []
|
|
service = SimpleNamespace(close=lambda: None, retry_solved=lambda source:
|
|
calls.append(source) or {"indexed": True, "action": "upsert"})
|
|
monkeypatch.setattr(memory_cmd, "load_snapshot_or_exit", lambda cfg, session: snapshot)
|
|
monkeypatch.setattr(memory_cmd, "memory_service", lambda cfg: service)
|
|
assert memory_cmd.index_solved_session(object(), "s1") == 1
|
|
assert calls == [snapshot]
|