diff --git a/harness/tests/test_adapter_command_regressions.py b/harness/tests/test_adapter_command_regressions.py new file mode 100644 index 00000000..ad52c825 --- /dev/null +++ b/harness/tests/test_adapter_command_regressions.py @@ -0,0 +1,84 @@ +from types import SimpleNamespace + +import pytest +import typer + +from tht.cli import db_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 +from tht.cli import memory_cmd +from tht.memory import MemoryRecord +from datetime import datetime + + +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), 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) + cfg = SimpleNamespace(embeddings=object(), vector_write_rest=object()) + manifest = SimpleNamespace(id="s1") + record = MemoryRecord(id="m1", ts=datetime(2026, 1, 1), session_id="s1", + decision_seq=7, type="table_promoted", subject="t", + question_context="q") + monkeypatch.setattr(memory_cmd, "_load_config_or_exit", lambda path: cfg) + monkeypatch.setattr(memory_cmd, "load_session_or_exit", lambda cfg, session: manifest) + monkeypatch.setattr(memory_cmd, "require_vector_write_allowed", lambda *args: None) + monkeypatch.setattr(memory_cmd, "has_vector_write_rest", lambda cfg: True) + monkeypatch.setattr(memory_cmd, "session_dir", lambda *args: None) + 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", 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) diff --git a/harness/tests/test_dwh_adapters.py b/harness/tests/test_dwh_adapters.py index 985b9d4b..8ffa8604 100644 --- a/harness/tests/test_dwh_adapters.py +++ b/harness/tests/test_dwh_adapters.py @@ -6,6 +6,7 @@ from tht.db.sampling import distinct_values_rest, sample_column_rest from tht.execute import ExecutionError from tht.ports import DistinctValues, DwhAdapter from tht.rest.client import RestError +from tht.adapters.dwh import PostgresDwhAdapter def postgres_factory(): @@ -145,3 +146,18 @@ def test_postgres_health_only_normalizes_database_errors(monkeypatch): ) with pytest.raises(ValueError, match="programming bug"): adapter.health() + + +def test_non_default_timeout_reaches_query_and_explain(monkeypatch): + adapter = PostgresDwhAdapter( + DatabaseConfig(database="analytics", schema="dw", user="reader", password="secret"), + statement_timeout_ms=12_345, + ) + calls = [] + monkeypatch.setattr("tht.adapters.dwh.postgres.execute.run_query", + lambda engine, sql, *, limit, timeout_ms: calls.append(("run", timeout_ms))) + monkeypatch.setattr("tht.adapters.dwh.postgres.execute.explain", + lambda engine, sql, *, timeout_ms: calls.append(("explain", timeout_ms))) + adapter.run_query("select 1", limit=2) + adapter.explain("select 1") + assert calls == [("run", 12_345), ("explain", 12_345)] diff --git a/harness/tht/adapters/dwh/postgres.py b/harness/tht/adapters/dwh/postgres.py index 58356b6b..a926005e 100644 --- a/harness/tht/adapters/dwh/postgres.py +++ b/harness/tht/adapters/dwh/postgres.py @@ -1,7 +1,7 @@ """Direct PostgreSQL implementation of the DWH port.""" from tht.config import DatabaseConfig -from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.exc import OperationalError, SQLAlchemyError from tht.db import execute, sampling from tht.db.connection import can_create_in_schema, make_engine, ping, writable_tables @@ -22,8 +22,10 @@ class PostgresDwhAdapter: def health(self) -> DwhHealth: try: ping(self._engine) + except OperationalError as exc: + return DwhHealth(ok=False, detail=str(exc.orig), error_kind="connection") except SQLAlchemyError as exc: - return DwhHealth(ok=False, detail=str(exc)) + return DwhHealth(ok=False, detail=str(exc), error_kind="connection") writable = tuple(writable_tables(self._engine, self._config.db_schema)) can_create = can_create_in_schema(self._engine, self._config.db_schema) return DwhHealth(ok=True, database=self._config.database, schema=self._config.db_schema, diff --git a/harness/tht/adapters/dwh/thoth_rest.py b/harness/tht/adapters/dwh/thoth_rest.py index e1f1c321..367ac156 100644 --- a/harness/tht/adapters/dwh/thoth_rest.py +++ b/harness/tht/adapters/dwh/thoth_rest.py @@ -21,11 +21,12 @@ class ThothRestDwhAdapter: try: result = self._client.ping() except RestError as exc: - return DwhHealth(ok=False, detail=str(exc)) + return DwhHealth(ok=False, detail=str(exc), error_kind="connection") ok = bool(result.get("db_connected") and result.get("schema_accessible")) return DwhHealth(ok=ok, detail=None if ok else str(result), database=self._database.database, schema=self._database.db_schema, - endpoint=self._client.cfg.base_url, read_only=True) + endpoint=self._client.cfg.base_url, read_only=True, + error_kind=None if ok else "inaccessible") def introspect(self) -> PhysicalSchema: return introspect_rest( diff --git a/harness/tht/cli/db_cmd.py b/harness/tht/cli/db_cmd.py index a1d06b65..73d1b8df 100644 --- a/harness/tht/cli/db_cmd.py +++ b/harness/tht/cli/db_cmd.py @@ -19,7 +19,12 @@ def ping_cmd(config: Path = CONFIG_OPT) -> None: raise typer.Exit(code=1) health = build_dwh(cfg).health() if not health.ok: - typer.secho(f"ERRORE di connessione: {health.detail}", fg=typer.colors.RED, err=True) + message = ( + f"ERRORE: DWH non accessibile via REST (risposta: {health.detail})." + if health.error_kind == "inaccessible" + else f"ERRORE di connessione: {health.detail}" + ) + typer.secho(message, fg=typer.colors.RED, err=True) raise typer.Exit(code=1) if health.endpoint: typer.secho(f"OK: connesso via REST a {health.endpoint} (schema {health.schema})", diff --git a/harness/tht/cli/lsh_cmd.py b/harness/tht/cli/lsh_cmd.py index f1916fcc..ffedbdf1 100644 --- a/harness/tht/cli/lsh_cmd.py +++ b/harness/tht/cli/lsh_cmd.py @@ -12,6 +12,30 @@ def _lsh_dir(cfg) -> Path: return cfg.paths.indexes / "lsh" +def _extract_lsh_values(dwh, physical, annotations, limit): + from tht.db.sampling import SkippedColumn, TruncatedColumn, is_text_type + from tht.mschema.eligibility import effective_eligibility + + values, skipped, truncated = {}, [], [] + for table_name, table in physical.tables.items(): + table_ann = annotations.tables.get(table_name) + for column_name, column in table.columns.items(): + ann_col = table_ann.columns.get(column_name) if table_ann else None + if not is_text_type(column.type) or not effective_eligibility(column, ann_col)[0]: + continue + try: + distinct = dwh.distinct_values(table_name, column_name, limit=limit) + except Exception as exc: + skipped.append(SkippedColumn(table_name, column_name, f"errore: {exc}")) + continue + vals = [str(value) for value in distinct.values if value not in (None, "")] + if vals: + values.setdefault(table_name, {})[column_name] = vals + if distinct.truncated: + truncated.append(TruncatedColumn(table_name, column_name, len(vals))) + return values, skipped, truncated + + @lsh_app.command("build") def build_cmd(config: Path = CONFIG_OPT) -> None: """Costruisce l'indice LSH dai valori del database e lo salva su pickle.""" @@ -33,31 +57,10 @@ def build_cmd(config: Path = CONFIG_OPT) -> None: typer.echo("Estrazione valori (i più frequenti) dalle colonne testuali eligible...") from tht.adapters.factory import build_dwh - from tht.db.sampling import SkippedColumn, TruncatedColumn, is_text_type - from tht.mschema.eligibility import effective_eligibility - dwh = build_dwh(cfg) - values: dict[str, dict[str, list[str]]] = {} - skipped: list[SkippedColumn] = [] - truncated: list[TruncatedColumn] = [] - for table_name, table in physical.tables.items(): - table_ann = annotations.tables.get(table_name) - for column_name, column in table.columns.items(): - ann_col = table_ann.columns.get(column_name) if table_ann else None - if not is_text_type(column.type) or not effective_eligibility(column, ann_col)[0]: - continue - try: - distinct = dwh.distinct_values( - table_name, column_name, limit=cfg.lsh.max_values_per_column - ) - except Exception as e: - skipped.append(SkippedColumn(table_name, column_name, f"errore: {e}")) - continue - vals = [str(value) for value in distinct.values if value not in (None, "")] - if vals: - values.setdefault(table_name, {})[column_name] = vals - if distinct.truncated or len(vals) >= cfg.lsh.max_values_per_column: - truncated.append(TruncatedColumn(table_name, column_name, len(vals))) + values, skipped, truncated = _extract_lsh_values( + dwh, physical, annotations, cfg.lsh.max_values_per_column + ) n_values = sum(len(v) for t in values.values() for v in t.values()) typer.echo(f" {n_values} valori da {sum(len(t) for t in values.values())} colonne") for s in skipped: diff --git a/harness/tht/ports/dwh.py b/harness/tht/ports/dwh.py index d30dc71a..10bfebb2 100644 --- a/harness/tht/ports/dwh.py +++ b/harness/tht/ports/dwh.py @@ -25,6 +25,7 @@ class DwhHealth: read_only: bool | None = None writable_tables: tuple[str, ...] = () can_create: bool = False + error_kind: str | None = None @dataclass(frozen=True)