Chiusura fase di ristrutturazione e modularizzazione del workflow per favorire sviluppo modulare

This commit is contained in:
2026-08-24 13:32:20 +02:00
parent fa2298653b
commit 6062cb010e
96 changed files with 304 additions and 259 deletions
+4 -4
View File
@@ -6,6 +6,7 @@ is not assumed reliable' gains real teeth for the data layer.
"""
import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError
from tht.db.connection import can_create_in_schema, ping, writable_tables
@@ -43,10 +44,9 @@ def test_read_only_role_cannot_insert(ro_url):
by our engine)."""
engine = create_engine(ro_url)
try:
with pytest.raises(Exception):
with engine.begin() as conn:
conn.execute(text('INSERT INTO dw.dim_pazienti VALUES (999, %s, %s)'),
("test", "test"))
with pytest.raises(SQLAlchemyError), engine.begin() as conn:
conn.execute(text('INSERT INTO dw.dim_pazienti VALUES (999, %s, %s)'),
("test", "test"))
finally:
engine.dispose()
+1 -1
View File
@@ -40,7 +40,7 @@ def test_unique_values_for_lsh_returns_most_frequent(admin_engine):
schema = introspect(admin_engine, "testdb", "dw")
# Before classify_all, all text columns are eligible=True by default. Sampling
# only touches text types regardless.
values, skipped, truncated = unique_values_for_lsh(
values, _skipped, _truncated = unique_values_for_lsh(
admin_engine, schema, LshConfig(max_values_per_column=100)
)
# dim_pazienti.citta: Milano, Bergamo, Brescia (3 distinct, all eligible text)
@@ -38,7 +38,7 @@ def test_ablazione_returns_multiple_columns(l2_env):
schema_name = ws.database.db_schema
try:
lsh, minhashes, meta = load_index(index_dir, schema_name)
except Exception as e:
except Exception as e: # noqa: BLE001 - any unusable external index skips this L2 probe
pytest.skip(f"LSH index not built yet (run tht preprocess dwh --steps lsh -c {WORKSPACE}): {e}")
hits = query_index(lsh, minhashes, "ablazione", meta, top_n=20)
+1 -1
View File
@@ -1,7 +1,7 @@
import json
import uuid
from datetime import UTC, datetime
from types import SimpleNamespace
import uuid
from typer.testing import CliRunner
@@ -1,7 +1,7 @@
import uuid
from datetime import UTC, datetime
from pathlib import Path
from types import SimpleNamespace
import uuid
from tht.cli import session_cmd
from tht.decisions import DecisionInput
+4 -2
View File
@@ -1,6 +1,7 @@
from tht.decisions import DecisionType
import typing
from tht.decisions import DecisionType
def test_column_decision_types_exist():
allowed = set(typing.get_args(DecisionType))
@@ -9,8 +10,9 @@ def test_column_decision_types_exist():
def test_f4_emits_column_types():
import yaml
from pathlib import Path
import yaml
wf = yaml.safe_load(Path("workflow.yaml").read_text())
f4 = next(p for p in wf["phases"] if p["id"] == "F4")
assert "column_promoted" in f4["emits"]
+2 -2
View File
@@ -1,7 +1,5 @@
import pytest
from tht.evidence.adapters import FilesystemEvidenceSource, HttpManifestEvidenceSource
from tht.evidence import build_sources
from tht.config import (
ConfigError,
PgvectorDirectConfig,
@@ -12,6 +10,8 @@ from tht.config import (
load_config,
workspace_id_for_config,
)
from tht.evidence import build_sources
from tht.evidence.adapters import FilesystemEvidenceSource, HttpManifestEvidenceSource
def test_direct_vector_passwords_load_from_file_references(monkeypatch, tmp_path):
+4 -1
View File
@@ -209,7 +209,10 @@ def test_model_copy_revalidates_records_and_manifests():
def test_manifest_datetimes_are_aware_and_normalized_to_utc():
with pytest.raises(ValidationError, match="timezone-aware"):
CorpusManifest(created_at=datetime(2026, 7, 12), pipeline_version="evidence-v1")
CorpusManifest(
created_at=datetime(2026, 7, 12), # noqa: DTZ001 - verifies rejection
pipeline_version="evidence-v1",
)
plus_two = datetime(2026, 7, 12, 12, tzinfo=timezone(timedelta(hours=2)))
manifest = CorpusManifest(created_at=plus_two, pipeline_version="evidence-v1")
+1 -1
View File
@@ -3,8 +3,8 @@ from datetime import UTC, datetime
import pytest
from tht.evidence.corpus.normalize import MAX_DOCUMENT_BYTES, PermanentNormalizationError, normalize
from tht.evidence.contracts import AcquiredDocument, SourceObject
from tht.evidence.corpus.normalize import MAX_DOCUMENT_BYTES, PermanentNormalizationError, normalize
def acquired(content: bytes, *, media_type: str = "text/markdown") -> AcquiredDocument:
+20 -17
View File
@@ -2,11 +2,11 @@ from datetime import UTC, datetime, timedelta
import pytest
from tht.evidence.contracts import AcquiredDocument, SourceObject
from tht.evidence.corpus.chunk import ChunkPolicy
from tht.evidence.corpus.models import CanonicalChunk, CanonicalDocument, CorpusManifest
from tht.evidence.corpus.pipeline import CorpusPipeline, PipelineError, PipelineResult
from tht.evidence.corpus.store import CorpusStore
from tht.evidence.corpus.models import CanonicalChunk, CanonicalDocument, CorpusManifest
from tht.evidence.contracts import AcquiredDocument, SourceObject
from tht.ports.vector import VectorCapabilities, VectorHealth
@@ -273,6 +273,7 @@ def test_gc_preserves_vector_dependencies_of_retained_manifests(tmp_path):
def test_active_searcher_without_active_fails_closed_for_evidence(tmp_path):
from types import SimpleNamespace
from tht.evidence.search import active_searcher
class Delegate:
@@ -287,6 +288,7 @@ def test_active_searcher_without_active_fails_closed_for_evidence(tmp_path):
def test_active_searcher_splits_default_and_mixed_kinds_before_global_limit(tmp_path):
from types import SimpleNamespace
from tht.evidence.search import ActiveEvidenceSearcher
store = CorpusStore(tmp_path / "corpus")
@@ -319,6 +321,7 @@ def test_active_searcher_splits_default_and_mixed_kinds_before_global_limit(tmp_
def test_active_evidence_query_holds_lock_against_publish(tmp_path):
import threading
from types import SimpleNamespace
from tht.evidence.search import ActiveEvidenceSearcher
first_pipeline = pipeline(tmp_path, Source([(item("one", "a"), "old")]), vectors=Vectors())
@@ -519,9 +522,9 @@ def test_unchanged_job_reuses_active_generation_without_new_directory(tmp_path):
"hello",
)])
candidate = pipeline(tmp_path, source)
args = dict(workspace_id="demo", workspace_root=tmp_path,
config_fingerprint="sha256:" + "1" * 64,
input_fingerprint="sha256:" + "2" * 64)
args = {"workspace_id": "demo", "workspace_root": tmp_path,
"config_fingerprint": "sha256:" + "1" * 64,
"input_fingerprint": "sha256:" + "2" * 64}
first = candidate.run_as_job(**args)
snapshot = first.manifest.metadata["source_snapshot"]["fs:one"]
assert snapshot == {
@@ -546,9 +549,9 @@ def test_job_source_snapshot_change_forces_publish_with_same_fingerprint(tmp_pat
metadata={"media_type": "text/markdown", "size": 5, "label": "original"},
)
vectors = Vectors()
args = dict(workspace_id="demo", workspace_root=tmp_path,
config_fingerprint="sha256:" + "1" * 64,
input_fingerprint="sha256:" + "2" * 64)
args = {"workspace_id": "demo", "workspace_root": tmp_path,
"config_fingerprint": "sha256:" + "1" * 64,
"input_fingerprint": "sha256:" + "2" * 64}
first = pipeline(tmp_path, Source([(original, "hello")]), vectors=vectors).run_as_job(**args)
updates = {
"uri": "file:///safe/renamed.md",
@@ -567,9 +570,9 @@ def test_job_source_snapshot_change_forces_publish_with_same_fingerprint(tmp_pat
def test_job_binding_change_forces_publish(tmp_path, fingerprint_name):
vectors = Vectors()
source_object = item("one", "a")
args = dict(workspace_id="demo", workspace_root=tmp_path,
config_fingerprint="sha256:" + "1" * 64,
input_fingerprint="sha256:" + "2" * 64)
args = {"workspace_id": "demo", "workspace_root": tmp_path,
"config_fingerprint": "sha256:" + "1" * 64,
"input_fingerprint": "sha256:" + "2" * 64}
first = pipeline(tmp_path, Source([(source_object, "hello")]), vectors=vectors).run_as_job(**args)
args[fingerprint_name] = "sha256:" + "3" * 64
source = Source([(source_object, "hello")])
@@ -587,9 +590,9 @@ def test_job_incomplete_active_contract_never_noops(tmp_path, damage):
vectors = Vectors()
source_object = item("one", "a")
args = dict(workspace_id="demo", workspace_root=tmp_path,
config_fingerprint="sha256:" + "1" * 64,
input_fingerprint="sha256:" + "2" * 64)
args = {"workspace_id": "demo", "workspace_root": tmp_path,
"config_fingerprint": "sha256:" + "1" * 64,
"input_fingerprint": "sha256:" + "2" * 64}
candidate = pipeline(tmp_path, Source([(source_object, "hello")]), vectors=vectors)
first = candidate.run_as_job(**args)
if damage == "legacy_metadata":
@@ -628,9 +631,9 @@ def test_job_corrupt_canonical_document_or_chunk_never_noops(tmp_path, damage):
modified_at=datetime(2026, 1, 1, tzinfo=UTC),
metadata={"media_type": "text/markdown", "size": 11, "owner": "docs"},
)
args = dict(workspace_id="demo", workspace_root=tmp_path,
config_fingerprint="sha256:" + "1" * 64,
input_fingerprint="sha256:" + "2" * 64)
args = {"workspace_id": "demo", "workspace_root": tmp_path,
"config_fingerprint": "sha256:" + "1" * 64,
"input_fingerprint": "sha256:" + "2" * 64}
candidate = pipeline(
tmp_path, Source([(source_object, "hello world")]), vectors=vectors,
policy=ChunkPolicy(version="chunk-v1", max_chars=6),
+6 -3
View File
@@ -1,6 +1,7 @@
import pytest
import os
import pytest
from tht.evidence.corpus.models import CorpusManifest
from tht.evidence.corpus.store import CorpusStore, UnsafeCorpusPath
@@ -111,9 +112,10 @@ def test_published_inventory_excludes_staged_and_invalid_newer_directories(tmp_p
def test_owned_copy_uses_validated_descriptor_bytes_when_source_is_replaced(tmp_path, monkeypatch):
from tht.evidence.corpus.models import CanonicalDocument
import hashlib
from tht.evidence.corpus.models import CanonicalDocument
content = "active bytes"
document = CanonicalDocument(
document_id="doc:" + "c" * 64, source_id="fs:copy", source_uri="file:///copy",
@@ -140,9 +142,10 @@ def test_owned_copy_uses_validated_descriptor_bytes_when_source_is_replaced(tmp_
def test_materialized_snapshot_uses_identified_manifest_when_active_changes(tmp_path):
from tht.evidence.corpus.models import CanonicalDocument
import hashlib
from tht.evidence.corpus.models import CanonicalDocument
def doc(content, fingerprint):
return CanonicalDocument(
document_id="doc:" + hashlib.sha256(content.encode()).hexdigest(),
+7 -7
View File
@@ -6,7 +6,7 @@ cte_plan_doc.json entry, and the last CteTestRecord for the CTE. --json output m
pristine (only valid JSON on stdout).
"""
import json
from datetime import datetime
from datetime import UTC, datetime
from typer.testing import CliRunner
@@ -18,7 +18,7 @@ from tht.session.store import create_session
def _db():
return DatabaseConfig(database="testdb", user="u", password="p", **{"schema": "public"}) # noqa: S106
return DatabaseConfig(database="testdb", user="u", password="p", schema="public")
def _patch_cfg(monkeypatch, tmp_path):
@@ -46,7 +46,7 @@ def test_info_happy_path_json(tmp_path, monkeypatch):
sid, sdir = _make_session(tmp_path, ["a", "b"])
(sdir / "ctes" / "b.sql").write_text("WITH b AS (SELECT 1)")
append_cte_test(sdir, CteTestRecord(
name="b", ts=datetime(2025, 1, 1, 12, 0), sql_hash="h1", status="ok",
name="b", ts=datetime(2025, 1, 1, 12, 0, tzinfo=UTC), sql_hash="h1", status="ok",
columns=["x"], row_sample=1, execution_ms=5, preview_rows=[[1]],
))
_patch_cfg(monkeypatch, tmp_path)
@@ -105,10 +105,10 @@ def test_info_last_test_is_most_recent_record(tmp_path, monkeypatch):
sid, sdir = _make_session(tmp_path, ["a"])
(sdir / "ctes" / "a.sql").write_text("WITH a AS (SELECT 1)")
append_cte_test(sdir, CteTestRecord(
name="a", ts=datetime(2025, 1, 1), sql_hash="old", status="ok",
name="a", ts=datetime(2025, 1, 1, tzinfo=UTC), sql_hash="old", status="ok",
))
append_cte_test(sdir, CteTestRecord(
name="a", ts=datetime(2025, 1, 2), sql_hash="new", status="ok",
name="a", ts=datetime(2025, 1, 2, tzinfo=UTC), sql_hash="new", status="ok",
))
_patch_cfg(monkeypatch, tmp_path)
@@ -134,7 +134,7 @@ def test_info_missing_plan_exit_1(tmp_path, monkeypatch):
def test_info_name_not_in_plan_exit_1(tmp_path, monkeypatch):
sid, sdir = _make_session(tmp_path, ["a"])
sid, _sdir = _make_session(tmp_path, ["a"])
_patch_cfg(monkeypatch, tmp_path)
res = CliRunner().invoke(cte_app, ["info", "zzz", "--session", sid, "--json"])
assert res.exit_code == 1
@@ -142,7 +142,7 @@ def test_info_name_not_in_plan_exit_1(tmp_path, monkeypatch):
def test_info_missing_sql_file_exit_1(tmp_path, monkeypatch):
sid, sdir = _make_session(tmp_path, ["a"])
sid, _sdir = _make_session(tmp_path, ["a"])
_patch_cfg(monkeypatch, tmp_path)
res = CliRunner().invoke(cte_app, ["info", "a", "--session", sid, "--json"])
assert res.exit_code == 1
+1 -1
View File
@@ -10,7 +10,7 @@ from tht.session.store import create_session
def _db():
return DatabaseConfig(database="testdb", user="u", password="p", **{"schema": "public"}) # noqa: S106
return DatabaseConfig(database="testdb", user="u", password="p", schema="public")
def _patch_cfg(monkeypatch, tmp_path):
+1 -1
View File
@@ -17,7 +17,7 @@ from tht.session.store import create_session
def _db():
return DatabaseConfig(database="testdb", user="u", password="p", **{"schema": "public"}) # noqa: S106
return DatabaseConfig(database="testdb", user="u", password="p", schema="public")
def _patch_cfg(monkeypatch, tmp_path):
+6 -6
View File
@@ -7,7 +7,7 @@ and the ledger I/O (load/append, tolerant of JSON-array and JSONL formats).
Pure logic, no DB.
"""
import json
from datetime import date, datetime
from datetime import UTC, date, datetime
from decimal import Decimal
import pytest
@@ -82,10 +82,10 @@ def test_build_test_sql_single_cte():
# --- ledger I/O: load_cte_tests / append_cte_test ---------------------------
def _record(**kw) -> CteTestRecord:
base = dict(
name="ablazione_q", ts=datetime(2025, 1, 1, 12, 0), sql_hash="abc123",
status="ok", columns=["x"], row_sample=5, execution_ms=42,
)
base = {
"name": "ablazione_q", "ts": datetime(2025, 1, 1, 12, 0, tzinfo=UTC), "sql_hash": "abc123",
"status": "ok", "columns": ["x"], "row_sample": 5, "execution_ms": 42,
}
base.update(kw)
return CteTestRecord(**base)
@@ -147,7 +147,7 @@ def test_jsonable_passes_through_native_types():
def test_jsonable_coerces_decimal_and_date_to_str():
assert _jsonable(Decimal("12.34")) == "12.34"
assert _jsonable(date(2025, 1, 1)) == "2025-01-01"
assert _jsonable(datetime(2025, 1, 1, 12, 0, 0)) == "2025-01-01 12:00:00"
assert _jsonable(datetime(2025, 1, 1, 12, 0, 0)) == "2025-01-01 12:00:00" # noqa: DTZ001
def test_jsonable_truncates_long_strings():
+1 -1
View File
@@ -17,7 +17,7 @@ def _concurrent_append_worker(session, subject, start, ready, done):
try:
append_decision(session, type="concept_clarified", subject=subject)
done.put((subject, None))
except Exception as error: # pragma: no cover - surfaced through the parent assertion
except Exception as error: # noqa: BLE001 # pragma: no cover - sent to parent
done.put((subject, repr(error)))
+2 -1
View File
@@ -57,8 +57,9 @@ def test_empty_session_returns_empty_list(tmp_path):
def test_decision_type_literal_includes_retracted():
"""decision_retracted e' un tipo valido (pydantic lo accetta)."""
from datetime import UTC, datetime
from tht.decisions import DecisionRecord
from datetime import datetime, UTC
d = DecisionRecord(
seq=1, ts=datetime.now(UTC), type="decision_retracted",
subject="phase:4", retracts=1,
-1
View File
@@ -5,7 +5,6 @@ from typer.testing import CliRunner
from tht.cli import app
runner = CliRunner()
+1 -1
View File
@@ -1,12 +1,12 @@
import pytest
from sqlalchemy.exc import OperationalError
from tht.adapters.dwh import PostgresDwhAdapter
from tht.config import DatabaseConfig, RestConfig
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():
+2 -2
View File
@@ -5,10 +5,10 @@ import pytest
from tht.execute import ExecResult, PlanSummary
from tht.mschema.models import PhysicalSchema
from tht.ports.dwh import (
DistinctValues,
DwhAdapter,
DwhCapabilities,
DwhHealth,
DistinctValues,
UnsupportedCapability,
)
@@ -55,10 +55,10 @@ def test_contract_types_are_public_and_capabilities_are_immutable():
def test_all_contract_types_are_exported_from_public_package():
from tht.ports import DistinctValues as PublicDistinctValues
from tht.ports import DwhAdapter as PublicDwhAdapter
from tht.ports import DwhCapabilities as PublicDwhCapabilities
from tht.ports import DwhHealth as PublicDwhHealth
from tht.ports import DistinctValues as PublicDistinctValues
from tht.ports import UnsupportedCapability as PublicUnsupportedCapability
result = PublicDistinctValues(values=["a"], truncated=True)
+23 -15
View File
@@ -6,13 +6,16 @@ from typer.testing import CliRunner
from tht.cli import app
from tht.config import load_config
from tht.jobs.dwh_pipeline import DwhPreprocessPipeline
from tht.jobs.dwh_pipeline import active_generation_dir, config_dwh_binding, fingerprint
from tht.jobs.dwh_pipeline import resolve_dwh_snapshot
from tht.jobs.dwh_pipeline import lease_dwh_snapshot
from tht.jobs.dwh_pipeline import (
DwhPreprocessPipeline,
active_generation_dir,
config_dwh_binding,
fingerprint,
lease_dwh_snapshot,
resolve_dwh_snapshot,
)
from tht.jobs.locking import _lock_name
FP = "sha256:" + hashlib.sha256(b"test").hexdigest()
@@ -104,9 +107,8 @@ def test_unowned_reads_fail_closed_without_creating_any_files(tmp_path):
cfg = snapshot_config(tmp_path)
with pytest.raises(Exception, match="not initialized"):
resolve_dwh_snapshot(cfg)
with pytest.raises(Exception, match="not initialized"):
with lease_dwh_snapshot(cfg):
pass
with pytest.raises(Exception, match="not initialized"), lease_dwh_snapshot(cfg):
pass
assert not (tmp_path / ".tht-dwh").exists()
@@ -125,7 +127,7 @@ def test_writer_claim_allows_only_lock_and_empty_generations(tmp_path):
pipeline = DwhPreprocessPipeline(
workspace_id="demo", workspace_root=root.parent,
config_fingerprint=FP, input_fingerprint=FP,
introspect=lambda output: calls.append("called"),
introspect=lambda output, calls=calls: calls.append("called"),
build_lsh=lambda physical, output: None,
)
with pytest.raises(Exception, match="unbound"):
@@ -188,6 +190,7 @@ def test_owner_publication_remains_on_locked_root_when_path_is_swapped(
monkeypatch, tmp_path,
):
import pytest
import tht.jobs.dwh_pipeline as module
real_replace = module.os.replace
@@ -211,7 +214,7 @@ def test_owner_publication_remains_on_locked_root_when_path_is_swapped(
introspect=lambda output: (_ for _ in ()).throw(AssertionError("callback called")),
build_lsh=lambda physical, output: None,
)
with pytest.raises(Exception):
with pytest.raises(Exception, match="root"):
pipeline.run()
assert swapped
assert (moved / "OWNER.json").is_file()
@@ -394,7 +397,7 @@ def test_missing_active_with_generations_and_symlink_owner_marker_fail_closed(tm
def _capture_error(operation):
try:
return operation()
except Exception as error:
except Exception as error: # noqa: BLE001 - helper returns the exact injected failure
return error
@@ -513,6 +516,7 @@ def test_unsafe_lsh_filename_is_rejected(tmp_path):
def test_active_fsync_failure_restores_previous_pointer(monkeypatch, tmp_path):
import os
import tht.jobs.dwh_pipeline as module
def build(physical, output):
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json"):
@@ -585,7 +589,7 @@ def test_snapshot_root_swap_after_lease_never_reads_replacement(monkeypatch, tmp
try:
with lease_dwh_snapshot(snapshot_config(tmp_path)) as snapshot:
assert snapshot.physical.read_text() == "trusted"
except Exception as error:
except Exception as error: # noqa: BLE001 - either safe refusal path is acceptable
assert "ACTIVE" in str(error) or "root" in str(error)
assert swapped
assert (replacement / "sentinel").read_text() == "replacement-secret"
@@ -623,7 +627,9 @@ def test_snapshot_copies_each_validated_artifact_once_without_reopen(monkeypatch
def test_reconcile_mismatch_closes_active_generation_fd(monkeypatch, tmp_path):
import os
from types import SimpleNamespace
import pytest
import tht.jobs.dwh_pipeline as module
pipeline = DwhPreprocessPipeline(
@@ -637,7 +643,7 @@ def test_reconcile_mismatch_closes_active_generation_fd(monkeypatch, tmp_path):
real_active = module._active_generation_fd
def mismatched_active(root_fd, binding):
generation, generation_fd = real_active(root_fd, binding)
_generation, generation_fd = real_active(root_fd, binding)
return "f" * 32, generation_fd
monkeypatch.setattr(module, "_active_generation_fd", mismatched_active)
@@ -676,9 +682,10 @@ def test_pipeline_releases_materialized_snapshot_after_every_run(tmp_path):
def test_corrupt_resume_checkpoint_releases_materialized_snapshot(tmp_path):
import tht.jobs.dwh_pipeline as module
import pytest
import tht.jobs.dwh_pipeline as module
pipeline = DwhPreprocessPipeline(
workspace_id="demo", workspace_root=tmp_path,
config_fingerprint=FP, input_fingerprint=FP,
@@ -699,9 +706,10 @@ def test_corrupt_resume_checkpoint_releases_materialized_snapshot(tmp_path):
def test_job_spec_construction_failure_releases_materialized_snapshot(monkeypatch, tmp_path):
import tht.jobs.dwh_pipeline as module
import pytest
import tht.jobs.dwh_pipeline as module
pipeline = DwhPreprocessPipeline(
workspace_id="demo", workspace_root=tmp_path,
config_fingerprint=FP, input_fingerprint=FP,
@@ -1,13 +1,11 @@
from datetime import UTC, datetime
import hashlib
import inspect
from datetime import UTC, datetime
from pathlib import Path
from types import SimpleNamespace
import pytest
from tht.evidence.corpus.models import CanonicalDocument, CorpusManifest
from tht.evidence.corpus.store import CorpusStore
from tht.decisions import DecisionRecord
from tht.evidence import (
acquire,
@@ -25,6 +23,8 @@ from tht.evidence.contracts import (
EvidenceSourceErrorCategory,
SourceObject,
)
from tht.evidence.corpus.models import CanonicalDocument, CorpusManifest
from tht.evidence.corpus.store import CorpusStore
from tht.session.models import Candidate, SchemaLinking
-1
View File
@@ -1,6 +1,5 @@
from pathlib import Path
HARNESS_ROOT = Path(__file__).resolve().parents[1]
LEGACY_PATHS = (
"tht/ports/evidence.py",
+6 -2
View File
@@ -154,7 +154,7 @@ def test_datetimes_must_be_aware_and_are_normalized_to_utc():
source_id="source:a",
uri="https://host/a",
fingerprint="etag:abc",
modified_at=datetime(2026, 7, 12),
modified_at=datetime(2026, 7, 12), # noqa: DTZ001 - verifies rejection
)
source = SourceObject(
@@ -236,4 +236,8 @@ def test_model_copy_revalidates_source_and_acquired_records():
with pytest.raises(ValidationError, match="namespaced"):
source.model_copy(update={"source_id": "invalid"})
with pytest.raises(ValidationError, match="timezone-aware"):
acquired.model_copy(update={"acquired_at": datetime(2026, 7, 12)})
acquired.model_copy(
update={
"acquired_at": datetime(2026, 7, 12) # noqa: DTZ001 - verifies rejection
}
)
+1 -1
View File
@@ -46,7 +46,7 @@ def test_union_query_accepts_limit():
def test_with_cte_query_accepts_limit():
sql = "WITH cte AS (SELECT 1) SELECT * FROM cte"
out, injected = _inject_limit(sql, 10)
_out, injected = _inject_limit(sql, 10)
assert injected is True
+1
View File
@@ -70,6 +70,7 @@ def test_retrieve_empty_on_missing_dir(tmp_path):
def test_concept_formula_decision_types_exist():
import typing
from tht.decisions import DecisionType
args = typing.get_args(DecisionType)
assert "concept_formula_approved" in args
+6 -5
View File
@@ -1,6 +1,7 @@
import threading
import socket
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import ClassVar
import pytest
@@ -12,8 +13,8 @@ class Handler(BaseHTTPRequestHandler):
etag_requests = 0
etag_body_responses = 0
redirect_target = "/redirected-v1"
redirect_request_validators = []
final_request_validators = []
redirect_request_validators: ClassVar[list[str | None]] = []
final_request_validators: ClassVar[list[tuple[str, str | None]]] = []
def do_GET(self):
if self.path.startswith("/etag"):
@@ -198,7 +199,6 @@ class FakeSocket:
class FakeResponse:
status_code = 200
headers = {}
is_redirect = False
def __init__(self, *, peer="127.0.0.1", stream_error=None, location=None):
@@ -214,6 +214,7 @@ class FakeResponse:
self.is_redirect = False
self.status_code = 200
self.headers = {}
self.headers = {}
def iter_content(self, chunk_size):
if self.stream_error:
@@ -265,7 +266,7 @@ def test_http_closes_response_when_streaming_fails():
source = HttpManifestEvidenceSource(
["https://example.test/doc"], allow_private_hosts=True
)
response = FakeResponse(stream_error=socket.timeout("read timed out"))
response = FakeResponse(stream_error=TimeoutError("read timed out"))
source._session = FakeSession(response)
with pytest.raises(EvidenceSourceError):
list(source.discover())
+5 -3
View File
@@ -37,9 +37,11 @@ def test_same_workspace_and_job_are_exclusive_across_processes(tmp_path):
def test_evidence_and_dwh_jobs_have_distinct_locks(tmp_path):
with WorkspaceJobLock(tmp_path, "demo", "evidence"):
with WorkspaceJobLock(tmp_path, "demo", "dwh"):
pass
with (
WorkspaceJobLock(tmp_path, "demo", "evidence"),
WorkspaceJobLock(tmp_path, "demo", "dwh"),
):
pass
def test_lock_keys_cannot_escape_lock_directory(tmp_path):
+2 -1
View File
@@ -1,11 +1,12 @@
import json
import os
import pytest
from pydantic import ValidationError
import tht.jobs.runner as runner_module
from tht.jobs.models import JobSpec
from tht.jobs.runner import CorruptCheckpointError, StageArtifacts, run_job
import tht.jobs.runner as runner_module
def _spec(tmp_path, **updates):
+2 -6
View File
@@ -3,7 +3,7 @@ import hashlib
import pytest
from tht.jobs.dwh_pipeline import DwhPreprocessPipeline
from tht.jobs.runner import CorruptCheckpointError
FP = "sha256:" + hashlib.sha256(b"test").hexdigest()
@@ -67,12 +67,8 @@ def test_resume_rejects_a_different_stage_selection(tmp_path):
introspect=lambda output: output.write_text("catalog"),
build_lsh=lambda physical, output: None,
)
try:
with pytest.raises(CorruptCheckpointError, match="incompatible"):
pipeline.run(("lsh",), resume_run_id=failed.run_id)
except Exception as error:
assert "incompatible" in str(error)
else:
raise AssertionError("resume with different stages must fail")
def test_resume_rejects_tampered_succeeded_stage_artifact(tmp_path):
+1 -1
View File
@@ -1,5 +1,5 @@
from tht.session.store import create_session
from tht.session.models import SessionManifest
from tht.session.store import create_session
def test_manifest_persists_pi_fields(tmp_path):
+7 -7
View File
@@ -5,18 +5,18 @@ fonte delle memory. L'hit di search_similar deve bastare per ricostruire la deci
completa. Per questo memory_vector_records mette subject/detail/rationale nel metadata
del VectorRecord (pack_metadata li serializza nel jsonb via **record.metadata).
"""
from datetime import datetime
from datetime import UTC, datetime
from tht.memory import MemoryRecord, memory_vector_records
def _record(**kw) -> MemoryRecord:
base = dict(
id="mem-x", ts=datetime(2025, 1, 1), session_id="s", decision_seq=1,
type="concept_clarified", subject="paziente attivo", detail="flag_attivo = TRUE",
rationale="perche' serve", question_context="dammi pazienti",
tables=[], concepts=["paziente attivo"],
)
base = {
"id": "mem-x", "ts": datetime(2025, 1, 1, tzinfo=UTC), "session_id": "s", "decision_seq": 1,
"type": "concept_clarified", "subject": "paziente attivo", "detail": "flag_attivo = TRUE",
"rationale": "perche' serve", "question_context": "dammi pazienti",
"tables": [], "concepts": ["paziente attivo"],
}
base.update(kw)
return MemoryRecord(**base)
+8 -8
View File
@@ -6,7 +6,7 @@ la rende un passo del workflow. Questi test fissano il contratto harness-side:
- i candidati rifiutati al gate (memory_promotion_declined, detail "seq:<n>") non
vengono riproposti da reusable_promotions/preview_promotions.
"""
from datetime import datetime
from datetime import UTC, datetime
from tht.decisions import DecisionRecord, append_decision
from tht.memory import (
@@ -23,7 +23,7 @@ from tht.workflow import load_workflow
def test_promotion_decision_types_are_valid():
for t in ("memory_promoted", "memory_promotion_declined"):
d = DecisionRecord(
seq=1, ts=datetime(2026, 1, 1), type=t, subject="fact_x", detail="seq:5"
seq=1, ts=datetime(2026, 1, 1, tzinfo=UTC), type=t, subject="fact_x", detail="seq:5"
)
assert d.type == t
@@ -36,14 +36,14 @@ def test_promotion_decision_types_min_phase_is_f8():
def _manifest() -> SessionManifest:
return SessionManifest(
id="s1", created_at=datetime(2026, 1, 1), question="domanda originale",
id="s1", created_at=datetime(2026, 1, 1, tzinfo=UTC), question="domanda originale",
database="db", schema="public",
)
def test_declined_promotion_seqs_parses_seq_detail():
d = DecisionRecord(
seq=9, ts=datetime(2026, 1, 1), type="memory_promotion_declined",
seq=9, ts=datetime(2026, 1, 1, tzinfo=UTC), type="memory_promotion_declined",
subject="fact_x", detail="seq:5",
)
assert declined_promotion_seqs([d]) == {5}
@@ -51,9 +51,9 @@ def test_declined_promotion_seqs_parses_seq_detail():
def test_declined_promotion_seqs_ignores_malformed_and_other_types():
ds = [
DecisionRecord(seq=1, ts=datetime(2026, 1, 1),
DecisionRecord(seq=1, ts=datetime(2026, 1, 1, tzinfo=UTC),
type="memory_promotion_declined", subject="x", detail=""),
DecisionRecord(seq=2, ts=datetime(2026, 1, 1),
DecisionRecord(seq=2, ts=datetime(2026, 1, 1, tzinfo=UTC),
type="table_promoted", subject="x", detail="seq:3"),
]
assert declined_promotion_seqs(ds) == set()
@@ -100,11 +100,11 @@ def test_only_concept_clarified_is_proposed_or_promoted(tmp_path):
def test_legacy_table_records_are_not_published_as_memory_vectors():
records = [
MemoryRecord(
id="mem-0001", ts=datetime(2026, 1, 1), session_id="s1",
id="mem-0001", ts=datetime(2026, 1, 1, tzinfo=UTC), session_id="s1",
decision_seq=1, type="table_promoted", subject="fact_a",
),
MemoryRecord(
id="mem-0002", ts=datetime(2026, 1, 1), session_id="s1",
id="mem-0002", ts=datetime(2026, 1, 1, tzinfo=UTC), session_id="s1",
decision_seq=2, type="concept_clarified", subject="paziente attivo",
detail="flag_attivo = TRUE",
),
-1
View File
@@ -7,7 +7,6 @@ from pathlib import Path
import pytest
THT_ROOT = Path(__file__).resolve().parents[1] / "tht"
DOMAIN_PACKAGES = ("evidence", "memory")
+5 -5
View File
@@ -4,7 +4,7 @@ Wide text (lettere di dimissione, note, anamnesi) is excluded everywhere; data
comes only from numerics, enums, temporals, booleans, and short text. Annotation
override wins over the physical classification.
"""
from datetime import datetime
from datetime import UTC, datetime
from tht.config import EligibilityConfig
from tht.mschema.eligibility import classify_all, classify_column, effective_eligibility
@@ -70,16 +70,16 @@ def test_array_type_is_wide_text():
def _schema_with(**columns) -> PhysicalSchema:
return PhysicalSchema(
database="db", schema="dw", introspected_at=datetime(2025, 1, 1),
database="db", schema="dw", introspected_at=datetime(2025, 1, 1, tzinfo=UTC),
tables={"t": TablePhysical(columns={k: ColumnPhysical(**v) for k, v in columns.items()})},
)
def test_classify_all_marks_wide_text_and_clears_examples():
schema = _schema_with(
note=dict(type="text", examples=["a" * 500, "b" * 400]),
cod=dict(type="varchar(10)", examples=["X", "Y"]),
etl_last_update=dict(type="timestamp", examples=[]),
note={"type": "text", "examples": ["a" * 500, "b" * 400]},
cod={"type": "varchar(10)", "examples": ["X", "Y"]},
etl_last_update={"type": "timestamp", "examples": []},
)
classify_all(schema, CFG)
cols = schema.tables["t"].columns
+2 -2
View File
@@ -3,7 +3,7 @@
Catches port breaks in the render layer (markdown reviewer report, mschema-text
ThothAI style, schema-dict for AV-SQL). Pure logic, no I/O.
"""
from datetime import datetime
from datetime import UTC, datetime
from tht.mschema.models import (
Annotations,
@@ -20,7 +20,7 @@ def _fake_schema() -> PhysicalSchema:
return PhysicalSchema(
database="testdb",
schema="dw",
introspected_at=datetime(2025, 1, 1, 0, 0, 0),
introspected_at=datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC),
tables={
"dim_pazienti": TablePhysical(
comment="Anagrafica pazienti",
+1 -1
View File
@@ -4,9 +4,9 @@ from types import SimpleNamespace
from typer.testing import CliRunner
from tht.config import EmbeddingsConfig
from tht.cli import ollama_cmd
from tht.cli.ollama_cmd import ensure_ollama, ollama_app
from tht.config import EmbeddingsConfig
def _cfg(**kw):
+1 -1
View File
@@ -1,4 +1,4 @@
from tht.decisions import append_decision, DecisionRecord
from tht.decisions import DecisionRecord, append_decision
from tht.phase import current_phase, effective_decisions
+1 -1
View File
@@ -81,7 +81,7 @@ def test_successful_reopen_records_ledger_and_runs_teardown(tmp_path, monkeypatc
calls = []
class _Report:
deleted_files = []
deleted_files = ()
def _spy(_repository, snapshot, phase):
# By the time teardown runs, the ledger already holds the reopen decision.
@@ -9,7 +9,6 @@ from tht.pi_skill_projection import (
render_projection,
)
BASELINE_SHA256 = "626a794071c095a4f20fffabb3bab901f05c101590adbdc58e45adfae56f3219"
+2 -1
View File
@@ -1,4 +1,5 @@
import json
from tht.cli.sql_cmd import promoted_columns_for
@@ -21,6 +22,6 @@ def test_promoted_columns_for(tmp_path):
}))
class Cfg:
class paths: # noqa: N801
class paths:
sessions = tmp_path
assert promoted_columns_for(Cfg, sid) == {"dim_patient.cod_paz"}
@@ -6,10 +6,10 @@ import yaml
from pydantic import SecretStr
from typer.testing import CliRunner
from tht.evidence.adapters import HttpManifestEvidenceSource
from tht.evidence import build_sources
from tht.cli import app
from tht.config import ConfigError, load_config
from tht.evidence import build_sources
from tht.evidence.adapters import HttpManifestEvidenceSource
SIGNED_CANARY = "SIGNED-CANARY-QUERY"
ACCESS_CANARY = "ACCESS-CANARY"
+3 -1
View File
@@ -99,7 +99,9 @@ def test_s3_rejects_leading_slash_prefix_empty_and_control_keys():
S3EvidenceSource(bucket="evidence", prefix="/clinical", client=Client())
for key in ("", "clinical/a\x00.md", "clinical/a\x7f.md"):
client = Client()
client.list_objects_v2 = lambda **kwargs: {"Contents": [{"Key": key, "ETag": '"x"'}]}
client.list_objects_v2 = lambda key=key, **kwargs: {
"Contents": [{"Key": key, "ETag": '"x"'}]
}
with pytest.raises(EvidenceSourceError):
list(S3EvidenceSource(bucket="evidence", prefix="clinical/", client=client).discover())
+3 -3
View File
@@ -1,5 +1,5 @@
import json
from datetime import datetime
from datetime import UTC, datetime
from typer.testing import CliRunner
@@ -9,7 +9,7 @@ from tht.mschema.models import ColumnPhysical, PhysicalSchema, TablePhysical
def _write_catalog(tmp_path):
phys = PhysicalSchema(
database="d", schema="s", introspected_at=datetime(2026, 1, 1),
database="d", schema="s", introspected_at=datetime(2026, 1, 1, tzinfo=UTC),
tables={
"dim_patient": TablePhysical(
comment="Anagrafica",
@@ -48,7 +48,7 @@ def _write_family_catalog(tmp_path):
)
phys = PhysicalSchema(
database="d", schema="s", introspected_at=datetime(2026, 1, 1),
database="d", schema="s", introspected_at=datetime(2026, 1, 1, tzinfo=UTC),
tables={
"fact_sost_impianto_pmk": _t("Sost PMK"),
"fact_sost_impianto_crt_d": _t("Sost CRT-D"),
@@ -1,16 +1,16 @@
from datetime import datetime
from datetime import UTC, datetime
from typer.testing import CliRunner
from tht.cli import app
from tht.mschema.models import ColumnPhysical, PhysicalSchema, TablePhysical
from tht.config import ExamplesConfig
from tht.cli.schema_cmd import _add_examples
from tht.config import ExamplesConfig
from tht.mschema.models import ColumnPhysical, PhysicalSchema, TablePhysical
def _write_catalog(tmp_path):
phys = PhysicalSchema(
database="d", schema="s", introspected_at=datetime(2026, 1, 1),
database="d", schema="s", introspected_at=datetime(2026, 1, 1, tzinfo=UTC),
tables={
"dim_patient": TablePhysical(
comment="Anagrafica",
@@ -48,7 +48,7 @@ def test_introspect_fresh_root_initializes_through_writer_job(tmp_path, monkeypa
cfg = _write_config(tmp_path)
physical = PhysicalSchema(
database="d", schema="s", introspected_at=datetime(2026, 1, 1),
database="d", schema="s", introspected_at=datetime(2026, 1, 1, tzinfo=UTC),
tables={"dim_patient": TablePhysical(columns={"id": ColumnPhysical(type="bigint")})},
)
@@ -90,7 +90,7 @@ def test_render_without_catalog_guides_fallback(tmp_path):
def test_examples_skip_one_unreadable_column_and_continue(caplog):
physical = PhysicalSchema(
database="d", schema="s", introspected_at=datetime(2026, 1, 1),
database="d", schema="s", introspected_at=datetime(2026, 1, 1, tzinfo=UTC),
tables={"t": TablePhysical(columns={
"bad": ColumnPhysical(type="text"), "good": ColumnPhysical(type="text")
})},
+11 -6
View File
@@ -1,20 +1,25 @@
"""Tests for `tht session documents --json` and build_documents."""
import json
from datetime import datetime
from datetime import UTC, datetime
from typer.testing import CliRunner
from tht.cli.session_cmd import session_app
from tht.config import DatabaseConfig
from tht.decisions import DecisionRecord
from tht.session.models import SessionSnapshot
from tht.session.store import build_documents, build_snapshot_documents, create_session, new_session_manifest
from tht.cli.session_cmd import session_app
from tht.session.store import (
build_documents,
build_snapshot_documents,
create_session,
new_session_manifest,
)
def _db():
return DatabaseConfig(
database="testdb", user="u", password="p", # noqa: S106
**{"schema": "public"},
database="testdb", user="u", password="p",
schema="public",
)
@@ -53,7 +58,7 @@ def test_build_documents_includes_existing_artifacts_only(tmp_path):
def _decision(seq: int, type_: str, subject: str, detail: str = "", rationale: str = ""):
return DecisionRecord(
seq=seq,
ts=datetime(2026, 1, 1),
ts=datetime(2026, 1, 1, tzinfo=UTC),
type=type_,
subject=subject,
detail=detail,
+2 -2
View File
@@ -14,8 +14,8 @@ def _make_db():
return DatabaseConfig(
database="testdb",
user="testuser",
password="testpass", # noqa: S106
**{"schema": "public"},
password="testpass",
schema="public",
)
+2 -2
View File
@@ -17,8 +17,8 @@ from tht.session.store import (
def _db():
return DatabaseConfig(
database="testdb", user="u", password="p", # noqa: S106
**{"schema": "public"},
database="testdb", user="u", password="p",
schema="public",
)
+4 -4
View File
@@ -8,15 +8,15 @@ import json
from typer.testing import CliRunner
from tht.config import DatabaseConfig
from tht.cli.session_cmd import session_app
from tht.config import DatabaseConfig
from tht.session.store import _extract_name, _summarize, load_session
def _db():
return DatabaseConfig(
database="testdb", user="u", password="p", # noqa: S106
**{"schema": "public"},
database="testdb", user="u", password="p",
schema="public",
)
@@ -41,7 +41,7 @@ def test_extract_name_is_a_3_to_5_word_italian_summary():
def test_extract_name_falls_back_to_summarize_on_failure(monkeypatch):
import tht.session.store as store
from tht.session import store
def boom(*a, **k):
raise RuntimeError("yake down")
+2 -3
View File
@@ -7,12 +7,11 @@ from tht.decisions import DecisionInput
from tht.session.filesystem_repository import FilesystemSessionRepository
from tht.session.models import PrincipalContext, SessionManifest, local_principal
from tht.session.repository import build_session_repository, resolve_principal
from tht.session.store import SessionError
from tht.session.store import create_session
from tht.session.store import SessionError, create_session
def _db() -> DatabaseConfig:
return DatabaseConfig(database="testdb", schema="public", user="u", password="p") # noqa: S106
return DatabaseConfig(database="testdb", schema="public", user="u", password="p")
def _config(tmp_path):
@@ -1,7 +1,7 @@
import uuid
from tht.decisions import DecisionInput
from tht.phase import current_phase, cte_plan, next_cte
from tht.phase import cte_plan, current_phase, next_cte
from tht.session.filesystem_repository import FilesystemSessionRepository
from tht.session.models import PrincipalContext, SessionManifest
from tht.session.store import persist_verified_finalization
+1 -1
View File
@@ -10,7 +10,7 @@ from tht.session.store import create_session, set_schema_linking
def _db():
return DatabaseConfig(database="testdb", user="u", password="p", **{"schema": "public"}) # noqa: S106
return DatabaseConfig(database="testdb", user="u", password="p", schema="public")
def test_writes_and_revalidates(tmp_path):
+1 -1
View File
@@ -9,7 +9,7 @@ from tht.session.store import create_session
def _db():
return DatabaseConfig(database="testdb", user="u", password="p", **{"schema": "public"}) # noqa: S106
return DatabaseConfig(database="testdb", user="u", password="p", schema="public")
def _patch_cfg(monkeypatch, tmp_path):
-1
View File
@@ -6,7 +6,6 @@ Pure-logic tests (no DB needed):
"""
import json
from tht.execute.limit import inject_limit_offset
+2 -2
View File
@@ -17,12 +17,12 @@ from tht.sqlcheck import validate_sql
def _schema() -> PhysicalSchema:
"""A tiny known schema for the object-existence checks."""
from datetime import datetime
from datetime import UTC, datetime
return PhysicalSchema(
database="db",
schema="dw",
introspected_at=datetime(2025, 1, 1),
introspected_at=datetime(2025, 1, 1, tzinfo=UTC),
tables={
"dim_pazienti": TablePhysical(
columns={
+2 -2
View File
@@ -8,8 +8,8 @@ from tht.session.store import create_session, set_schema_linking, sync_schema_li
def _db():
return DatabaseConfig(
database="testdb", user="u", password="p", # noqa: S106
**{"schema": "public"},
database="testdb", user="u", password="p",
schema="public",
)
+2 -1
View File
@@ -64,7 +64,8 @@ def test_empty_hits_returns_empty():
def test_value_grounded_decision_type_exists():
# D14a adds the value_grounded decision type so the gate can record the
# reviewer's choice of which column(s) anchor a cited value.
from tht.decisions import DecisionType
import typing
from tht.decisions import DecisionType
args = typing.get_args(DecisionType)
assert "value_grounded" in args
@@ -1,15 +1,15 @@
"""Executable baseline for persisted workflow behavior touched by the refactor."""
import hashlib
from dataclasses import asdict
from datetime import UTC, datetime
import hashlib
import pytest
from tht.evidence.corpus.models import CanonicalDocument, CorpusManifest
from tht.evidence.corpus.store import CorpusStore
from tht.decisions import DecisionRecord, append_decision
from tht.evidence import project_session
from tht.evidence.corpus.models import CanonicalDocument, CorpusManifest
from tht.evidence.corpus.store import CorpusStore
from tht.phase import current_phase, effective_decisions
from tht.session.models import Candidate, SchemaLinking
from tht.workflow import load_workflow