996 lines
37 KiB
Python
996 lines
37 KiB
Python
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
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.locking import _lock_name
|
|
|
|
|
|
FP = "sha256:" + hashlib.sha256(b"test").hexdigest()
|
|
|
|
|
|
def _runtime_config(tmp_path, filename, revision, database="warehouse"):
|
|
path = tmp_path / filename
|
|
path.write_text(f"""
|
|
runtime_identity:
|
|
workspace_id: demo
|
|
workspace_revision: {revision}
|
|
source_identity: workspace://demo
|
|
dwh:
|
|
type: postgres_direct
|
|
connection: {{database: {database}, schema: analytics, user: reader, password: secret}}
|
|
roots:
|
|
sessions: {tmp_path / 'sessions'}
|
|
artifacts: {tmp_path / 'artifacts'}
|
|
indexes: {tmp_path / 'indexes'}
|
|
""")
|
|
return path
|
|
|
|
|
|
def test_runtime_lease_path_and_revision_metadata_do_not_change_dwh_binding(tmp_path, monkeypatch):
|
|
monkeypatch.delenv("THT_HOME", raising=False)
|
|
monkeypatch.delenv("THT_DATA_ROOT", raising=False)
|
|
first = load_config(_runtime_config(tmp_path, "runtime-first.yaml", "a" * 40))
|
|
second = load_config(_runtime_config(tmp_path, "runtime-second.yaml", "a" * 40))
|
|
metadata_only_revision = load_config(
|
|
_runtime_config(tmp_path, "runtime-third.yaml", "b" * 40)
|
|
)
|
|
|
|
expected = config_dwh_binding(first)
|
|
|
|
assert expected == config_dwh_binding(second)
|
|
assert expected == config_dwh_binding(metadata_only_revision)
|
|
assert expected["input_fingerprint"] == fingerprint("workspace://demo")
|
|
|
|
|
|
def test_effective_dwh_change_invalidates_runtime_binding(tmp_path, monkeypatch):
|
|
monkeypatch.delenv("THT_HOME", raising=False)
|
|
monkeypatch.delenv("THT_DATA_ROOT", raising=False)
|
|
original = load_config(_runtime_config(tmp_path, "runtime-first.yaml", "a" * 40))
|
|
changed = load_config(
|
|
_runtime_config(tmp_path, "runtime-second.yaml", "b" * 40, database="warehouse_v2")
|
|
)
|
|
|
|
assert (
|
|
config_dwh_binding(original)["config_fingerprint"]
|
|
!= config_dwh_binding(changed)["config_fingerprint"]
|
|
)
|
|
|
|
|
|
def snapshot_config(tmp_path, workspace_id="demo"):
|
|
from types import SimpleNamespace
|
|
|
|
cfg = SimpleNamespace(
|
|
paths=SimpleNamespace(artifacts=tmp_path / "artifacts", indexes=tmp_path / "indexes"),
|
|
_workspace_id=workspace_id,
|
|
_config_source="test",
|
|
)
|
|
cfg.model_dump_json = lambda: "test"
|
|
return cfg
|
|
|
|
|
|
def test_dwh_and_evidence_jobs_have_distinct_lock_names():
|
|
assert _lock_name("demo", "dwh") != _lock_name("demo", "evidence")
|
|
|
|
|
|
def test_session_storage_does_not_change_the_dwh_artifact_binding(tmp_path):
|
|
from types import SimpleNamespace
|
|
|
|
def config(session_storage=None):
|
|
payload = {"dwh": {"connection": {"database": "warehouse"}}}
|
|
if session_storage is not None:
|
|
payload["session_storage"] = session_storage
|
|
cfg = SimpleNamespace(_workspace_id="demo", _config_source="test")
|
|
cfg.model_dump = lambda **_kwargs: payload
|
|
cfg.model_dump_json = lambda: json.dumps(payload, separators=(",", ":"))
|
|
return cfg
|
|
|
|
without_session_storage = config()
|
|
with_session_storage = config({"type": "postgres_direct", "connection": {"database": "sessions"}})
|
|
|
|
assert config_dwh_binding(without_session_storage) == config_dwh_binding(with_session_storage)
|
|
|
|
|
|
def test_unowned_reads_fail_closed_without_creating_any_files(tmp_path):
|
|
import pytest
|
|
|
|
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
|
|
assert not (tmp_path / ".tht-dwh").exists()
|
|
|
|
|
|
def test_writer_claim_allows_only_lock_and_empty_generations(tmp_path):
|
|
import pytest
|
|
|
|
for name, make_entry in (
|
|
("unexpected", lambda root: (root / "unexpected").write_text("x")),
|
|
("stale-temp", lambda root: (root / ".OWNER.json.stale.tmp").write_text("x")),
|
|
("unexpected-dir", lambda root: (root / "other").mkdir()),
|
|
):
|
|
root = tmp_path / name / ".tht-dwh"
|
|
root.mkdir(parents=True, mode=0o700)
|
|
make_entry(root)
|
|
calls = []
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=root.parent,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: calls.append("called"),
|
|
build_lsh=lambda physical, output: None,
|
|
)
|
|
with pytest.raises(Exception, match="unbound"):
|
|
pipeline.run()
|
|
assert calls == []
|
|
assert not (root / "OWNER.json").exists()
|
|
|
|
allowed = tmp_path / "allowed"
|
|
(allowed / ".tht-dwh" / "generations").mkdir(parents=True, mode=0o700)
|
|
report = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=allowed,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("catalog"),
|
|
build_lsh=lambda physical, output: _write_lsh([], physical, output),
|
|
).run()
|
|
assert report.status == "succeeded"
|
|
|
|
|
|
def test_writer_rejects_unbound_legacy_artifacts_before_building(tmp_path):
|
|
import pytest
|
|
|
|
legacy = tmp_path / "artifacts" / "mschema" / "physical.yaml"
|
|
legacy.parent.mkdir(parents=True)
|
|
legacy.write_text("legacy")
|
|
calls = []
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: calls.append("called"),
|
|
build_lsh=lambda physical, output: None,
|
|
current_physical=legacy,
|
|
)
|
|
with pytest.raises(Exception, match="legacy artifacts are unbound"):
|
|
pipeline.run()
|
|
assert calls == []
|
|
assert not (tmp_path / ".tht-dwh" / "OWNER.json").exists()
|
|
|
|
|
|
def test_writer_rejects_dangling_legacy_symlinks_before_claim_or_callback(tmp_path):
|
|
import pytest
|
|
|
|
legacy = tmp_path / "artifacts" / "mschema" / "physical.yaml"
|
|
legacy.parent.mkdir(parents=True)
|
|
legacy.symlink_to(tmp_path / "missing-catalog")
|
|
calls = []
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: calls.append("called"),
|
|
build_lsh=lambda physical, output: None,
|
|
current_physical=legacy,
|
|
)
|
|
with pytest.raises(Exception, match="legacy artifacts are unbound"):
|
|
pipeline.run()
|
|
assert calls == []
|
|
assert not (tmp_path / ".tht-dwh" / "OWNER.json").exists()
|
|
|
|
|
|
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
|
|
moved = tmp_path / "locked-root"
|
|
replacement = tmp_path / ".tht-dwh"
|
|
swapped = False
|
|
|
|
def swapping_replace(source, destination, *args, **kwargs):
|
|
nonlocal swapped
|
|
if destination == "OWNER.json" and kwargs.get("dst_dir_fd") is not None:
|
|
swapped = True
|
|
replacement.rename(moved)
|
|
replacement.mkdir(mode=0o700)
|
|
(moved / "generation.lock").rename(replacement / "generation.lock")
|
|
return real_replace(source, destination, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(module.os, "replace", swapping_replace)
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: (_ for _ in ()).throw(AssertionError("callback called")),
|
|
build_lsh=lambda physical, output: None,
|
|
)
|
|
with pytest.raises(Exception):
|
|
pipeline.run()
|
|
assert swapped
|
|
assert (moved / "OWNER.json").is_file()
|
|
assert not (replacement / "OWNER.json").exists()
|
|
assert (replacement / "generation.lock").is_file()
|
|
|
|
|
|
def test_owner_requires_exact_read_only_owner_mode_and_active_requires_binding(tmp_path):
|
|
import pytest
|
|
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("catalog"),
|
|
build_lsh=lambda physical, output: _write_lsh([], physical, output),
|
|
)
|
|
pipeline.run()
|
|
cfg = snapshot_config(tmp_path)
|
|
binding = config_dwh_binding(cfg)
|
|
assert active_generation_dir(tmp_path, binding) is not None
|
|
with pytest.raises(Exception, match="different workspace configuration"):
|
|
active_generation_dir(tmp_path, {**binding, "workspace_id": "other"})
|
|
|
|
marker = tmp_path / ".tht-dwh" / "OWNER.json"
|
|
marker.chmod(0o440)
|
|
with pytest.raises(Exception, match="ownership marker"):
|
|
resolve_dwh_snapshot(cfg)
|
|
|
|
|
|
def test_selected_dwh_stages_run_in_declared_order(tmp_path):
|
|
calls = []
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo",
|
|
workspace_root=tmp_path,
|
|
config_fingerprint=FP,
|
|
input_fingerprint=FP,
|
|
introspect=lambda output: (calls.append("introspect"), output.write_text("catalog")),
|
|
build_lsh=lambda physical, output: _write_lsh(calls, physical, output),
|
|
)
|
|
|
|
report = pipeline.run(("introspect", "lsh"))
|
|
|
|
assert report.status == "succeeded"
|
|
assert calls == ["introspect", "lsh"]
|
|
assert [stage.name for stage in report.stages] == ["introspect", "lsh"]
|
|
active = (tmp_path / ".tht-dwh" / "ACTIVE").read_text().strip()
|
|
published = tmp_path / ".tht-dwh" / "generations" / active
|
|
assert (published / "physical.yaml").read_text() == "catalog"
|
|
assert sorted(path.name for path in published.iterdir()) == [
|
|
"demo_lsh.pkl", "demo_meta.json", "demo_minhashes.pkl",
|
|
"generation-manifest.json", "physical.yaml",
|
|
]
|
|
|
|
|
|
def test_shared_root_rejects_other_workspace_before_builder_or_read(tmp_path):
|
|
calls = []
|
|
owner = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("catalog"),
|
|
build_lsh=lambda physical, output: _write_lsh([], physical, output),
|
|
)
|
|
published = owner.run()
|
|
contender = DwhPreprocessPipeline(
|
|
workspace_id="other", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: calls.append("introspect"),
|
|
build_lsh=lambda physical, output: calls.append("lsh"),
|
|
)
|
|
|
|
import pytest
|
|
with pytest.raises(Exception, match="different workspace configuration"):
|
|
contender.run()
|
|
with pytest.raises(Exception, match="different workspace configuration"):
|
|
resolve_dwh_snapshot(snapshot_config(tmp_path, "other"))
|
|
|
|
assert calls == []
|
|
assert resolve_dwh_snapshot(snapshot_config(tmp_path)).generation == published.run_id
|
|
|
|
|
|
def test_shared_root_mismatch_fails_without_deadlock_while_owner_reader_is_active(tmp_path):
|
|
import threading
|
|
|
|
owner = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("catalog"),
|
|
build_lsh=lambda physical, output: _write_lsh([], physical, output),
|
|
)
|
|
owner.run()
|
|
contender = DwhPreprocessPipeline(
|
|
workspace_id="other", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: (_ for _ in ()).throw(AssertionError("builder called")),
|
|
build_lsh=lambda physical, output: None,
|
|
)
|
|
finished = threading.Event()
|
|
errors = []
|
|
with lease_dwh_snapshot(snapshot_config(tmp_path)):
|
|
thread = threading.Thread(
|
|
target=lambda: (errors.append(_capture_error(contender.run)), finished.set())
|
|
)
|
|
thread.start()
|
|
assert not finished.wait(0.1)
|
|
thread.join(2)
|
|
assert finished.is_set()
|
|
assert "different workspace configuration" in str(errors[0])
|
|
|
|
|
|
def test_concurrent_brand_new_shared_root_has_one_atomic_owner_and_loser_never_builds(tmp_path):
|
|
import threading
|
|
|
|
calls = {"alpha": 0, "beta": 0}
|
|
results = []
|
|
barrier = threading.Barrier(2)
|
|
|
|
def run(workspace):
|
|
def introspect(output):
|
|
calls[workspace] += 1
|
|
output.write_text("catalog")
|
|
|
|
def build(physical, output):
|
|
calls[workspace] += 1
|
|
_write_lsh([], physical, output)
|
|
|
|
candidate = DwhPreprocessPipeline(
|
|
workspace_id=workspace, workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=introspect, build_lsh=build,
|
|
)
|
|
barrier.wait()
|
|
results.append((workspace, _capture_error(candidate.run)))
|
|
|
|
threads = [threading.Thread(target=run, args=(name,)) for name in ("alpha", "beta")]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join(5)
|
|
assert all(not thread.is_alive() for thread in threads)
|
|
winner = next(name for name, result in results if not isinstance(result, Exception))
|
|
loser = next(name for name, result in results if isinstance(result, Exception))
|
|
assert calls[winner] == 2
|
|
assert calls[loser] == 0
|
|
|
|
|
|
def test_missing_active_with_generations_and_symlink_owner_marker_fail_closed(tmp_path):
|
|
import pytest
|
|
|
|
calls = []
|
|
owner = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("catalog"),
|
|
build_lsh=lambda physical, output: _write_lsh([], physical, output),
|
|
)
|
|
owner.run()
|
|
(tmp_path / ".tht-dwh" / "ACTIVE").unlink()
|
|
owner.introspect = lambda output: calls.append("called")
|
|
with pytest.raises(Exception, match="without a consistent ACTIVE"):
|
|
owner.run()
|
|
assert calls == []
|
|
|
|
other_root = tmp_path / "other"
|
|
marker_root = other_root / ".tht-dwh"
|
|
marker_root.mkdir(parents=True)
|
|
external = tmp_path / "external-owner"
|
|
external.write_text("foreign")
|
|
(marker_root / "OWNER.json").symlink_to(external)
|
|
contender = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=other_root,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: calls.append("symlink-called"),
|
|
build_lsh=lambda physical, output: None,
|
|
)
|
|
with pytest.raises(Exception, match="ownership marker"):
|
|
contender.run()
|
|
assert calls == []
|
|
|
|
|
|
def _capture_error(operation):
|
|
try:
|
|
return operation()
|
|
except Exception as error:
|
|
return error
|
|
|
|
|
|
def _write_lsh(calls, physical: Path, output: Path):
|
|
calls.append("lsh")
|
|
assert physical.read_text() == "catalog"
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json"):
|
|
(output / name).write_text(name)
|
|
|
|
|
|
def test_preprocess_dwh_json_is_pristine(monkeypatch, tmp_path):
|
|
import tht.cli.preprocess_cmd as command
|
|
|
|
class Report:
|
|
status = "succeeded"
|
|
|
|
def model_dump(self, mode=None):
|
|
return {"status": "succeeded", "run_id": "a" * 32, "stages": []}
|
|
|
|
seen = {}
|
|
|
|
def run(config, *, steps, resume):
|
|
seen.update(config=config, steps=steps, resume=resume)
|
|
return Report()
|
|
|
|
monkeypatch.setattr(command, "run_dwh_from_config", run)
|
|
response = CliRunner().invoke(
|
|
app,
|
|
[
|
|
"preprocess", "dwh", "--steps", "introspect,lsh", "--json",
|
|
"-c", str(tmp_path / "workspace.yaml"),
|
|
],
|
|
)
|
|
|
|
assert response.exit_code == 0, response.output
|
|
assert json.loads(response.output)["run_id"] == "a" * 32
|
|
assert seen["steps"] == ("introspect", "lsh")
|
|
|
|
|
|
def test_preprocess_dwh_rejects_unknown_or_duplicate_steps(monkeypatch, tmp_path):
|
|
import tht.cli.preprocess_cmd as command
|
|
|
|
called = False
|
|
|
|
def forbidden(*args, **kwargs):
|
|
nonlocal called
|
|
called = True
|
|
|
|
monkeypatch.setattr(command, "run_dwh_from_config", forbidden)
|
|
runner = CliRunner()
|
|
for value in ("introspect,unknown", "lsh,lsh", ""):
|
|
response = runner.invoke(
|
|
app,
|
|
["preprocess", "dwh", "--steps", value, "--json", "-c", str(tmp_path / "w.yaml")],
|
|
)
|
|
assert response.exit_code == 2
|
|
assert json.loads(response.output)["status"] == "failed"
|
|
assert called is False
|
|
|
|
|
|
def test_failed_multi_file_build_never_replaces_active_generation(tmp_path):
|
|
def catalog(output):
|
|
output.write_text("old-catalog")
|
|
|
|
first = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=catalog,
|
|
build_lsh=lambda physical, output: [
|
|
(output / name).write_text(name)
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
|
|
],
|
|
).run(("introspect", "lsh"))
|
|
assert first.status == "succeeded"
|
|
old_active = (tmp_path / ".tht-dwh" / "ACTIVE").read_text()
|
|
|
|
def partial_lsh(physical, output):
|
|
(output / "demo_lsh.pkl").write_text("new-but-partial")
|
|
raise RuntimeError("crash between LSH files")
|
|
|
|
failed = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("new-catalog"),
|
|
build_lsh=partial_lsh,
|
|
).run(("introspect", "lsh"))
|
|
|
|
assert failed.status == "failed"
|
|
assert (tmp_path / ".tht-dwh" / "ACTIVE").read_text() == old_active
|
|
|
|
resumed = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: (_ for _ in ()).throw(
|
|
AssertionError("completed introspection must not repeat")
|
|
),
|
|
build_lsh=lambda physical, output: [
|
|
(output / name).write_text("recovered")
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
|
|
],
|
|
).run(("introspect", "lsh"), resume_run_id=failed.run_id)
|
|
assert resumed.status == "succeeded"
|
|
|
|
|
|
def test_unsafe_lsh_filename_is_rejected(tmp_path):
|
|
import pytest
|
|
|
|
with pytest.raises(ValueError, match="flat safe"):
|
|
DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: None, build_lsh=lambda physical, output: None,
|
|
lsh_filenames=("../escape.pkl", "ok.pkl", "meta.json"),
|
|
)
|
|
|
|
|
|
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"):
|
|
(output / name).write_text(name)
|
|
|
|
first_pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("old"), build_lsh=build,
|
|
)
|
|
first = first_pipeline.run()
|
|
root = tmp_path / ".tht-dwh"
|
|
root_identity = (root.stat().st_dev, root.stat().st_ino)
|
|
original_fsync = module.os.fsync
|
|
failed_once = False
|
|
|
|
def fail_active_once(fd):
|
|
nonlocal failed_once
|
|
info = os.fstat(fd)
|
|
if (
|
|
(info.st_dev, info.st_ino) == root_identity
|
|
and "ACTIVE" in os.listdir(fd)
|
|
and not failed_once
|
|
):
|
|
failed_once = True
|
|
raise OSError("injected directory fsync failure")
|
|
original_fsync(fd)
|
|
|
|
second = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("new"), build_lsh=build,
|
|
)
|
|
monkeypatch.setattr(module.os, "fsync", fail_active_once)
|
|
failed = second.run()
|
|
assert failed.status == "failed"
|
|
assert (tmp_path / ".tht-dwh" / "ACTIVE").read_text().strip() == first.run_id
|
|
|
|
|
|
def test_snapshot_root_swap_after_lease_never_reads_replacement(monkeypatch, tmp_path):
|
|
import tht.jobs.dwh_pipeline as module
|
|
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("trusted"),
|
|
build_lsh=lambda physical, output: [
|
|
(output / name).write_text("trusted")
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
|
|
],
|
|
)
|
|
first = pipeline.run()
|
|
assert first.status == "succeeded"
|
|
root = tmp_path / ".tht-dwh"
|
|
moved = tmp_path / "moved-read-root"
|
|
replacement = root
|
|
real_read = module._read_owned_at
|
|
swapped = False
|
|
|
|
def swapping_read(directory_fd, name, *, readonly):
|
|
nonlocal swapped
|
|
if name == "ACTIVE" and not swapped:
|
|
swapped = True
|
|
replacement.rename(moved)
|
|
replacement.mkdir(mode=0o700)
|
|
(replacement / "sentinel").write_text("replacement-secret")
|
|
return real_read(directory_fd, name, readonly=readonly)
|
|
|
|
monkeypatch.setattr(module, "_read_owned_at", swapping_read)
|
|
try:
|
|
with lease_dwh_snapshot(snapshot_config(tmp_path)) as snapshot:
|
|
assert snapshot.physical.read_text() == "trusted"
|
|
except Exception as error:
|
|
assert "ACTIVE" in str(error) or "root" in str(error)
|
|
assert swapped
|
|
assert (replacement / "sentinel").read_text() == "replacement-secret"
|
|
|
|
|
|
def test_snapshot_copies_each_validated_artifact_once_without_reopen(monkeypatch, tmp_path):
|
|
import tht.jobs.dwh_pipeline as module
|
|
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("trusted"),
|
|
build_lsh=lambda physical, output: [
|
|
(output / name).write_text("trusted")
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
|
|
],
|
|
)
|
|
assert pipeline.run().status == "succeeded"
|
|
real_read = module._read_owned_at
|
|
reads = {}
|
|
|
|
def mutate_on_reopen(directory_fd, name, *, readonly):
|
|
reads[name] = reads.get(name, 0) + 1
|
|
if name.endswith(".pkl") and reads[name] > 1:
|
|
return b"MALICIOUS_PICKLE"
|
|
return real_read(directory_fd, name, readonly=readonly)
|
|
|
|
monkeypatch.setattr(module, "_read_owned_at", mutate_on_reopen)
|
|
with lease_dwh_snapshot(snapshot_config(tmp_path)) as snapshot:
|
|
assert (snapshot.lsh_dir / "demo_lsh.pkl").read_text() == "trusted"
|
|
assert "MALICIOUS" not in (snapshot.lsh_dir / "demo_lsh.pkl").read_text()
|
|
assert all(count == 1 for count in reads.values())
|
|
|
|
|
|
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(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("catalog"),
|
|
build_lsh=lambda physical, output: _write_lsh([], physical, output),
|
|
)
|
|
report = pipeline.run()
|
|
run_dir = tmp_path / ".tht-jobs" / "dwh" / "runs" / report.run_id
|
|
real_active = module._active_generation_fd
|
|
|
|
def mismatched_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)
|
|
source = SimpleNamespace(
|
|
run_id=report.run_id,
|
|
stages=(SimpleNamespace(
|
|
status="running", effect_state="intent", name="lsh",
|
|
artifact_files=("physical.yaml", "demo_lsh.pkl", "demo_minhashes.pkl",
|
|
"demo_meta.json"),
|
|
),),
|
|
)
|
|
before = len(os.listdir("/dev/fd"))
|
|
with pytest.raises(Exception, match="not ACTIVE"):
|
|
pipeline._reconcile_effects(source, run_dir)
|
|
assert len(os.listdir("/dev/fd")) == before
|
|
|
|
|
|
def test_pipeline_releases_materialized_snapshot_after_every_run(tmp_path):
|
|
import tht.jobs.dwh_pipeline as module
|
|
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("catalog"),
|
|
build_lsh=lambda physical, output: _write_lsh([], physical, output),
|
|
)
|
|
baseline = set(module._SNAPSHOT_DIRS)
|
|
for _ in range(3):
|
|
assert pipeline.run().status == "succeeded"
|
|
assert set(module._SNAPSHOT_DIRS) == baseline
|
|
assert pipeline._snapshot_holder is None
|
|
pipeline.introspect = lambda output: (_ for _ in ()).throw(RuntimeError("injected"))
|
|
assert pipeline.run().status == "failed"
|
|
assert set(module._SNAPSHOT_DIRS) == baseline
|
|
assert pipeline._snapshot_holder is None
|
|
|
|
|
|
def test_corrupt_resume_checkpoint_releases_materialized_snapshot(tmp_path):
|
|
import tht.jobs.dwh_pipeline as module
|
|
import pytest
|
|
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("catalog"),
|
|
build_lsh=lambda physical, output: _write_lsh([], physical, output),
|
|
)
|
|
assert pipeline.run().status == "succeeded"
|
|
baseline = set(module._SNAPSHOT_DIRS)
|
|
run_id = "e" * 32
|
|
run_dir = tmp_path / ".tht-jobs" / "dwh" / "runs" / run_id
|
|
run_dir.mkdir(parents=True)
|
|
(run_dir / "checkpoint.json").write_text("not-json")
|
|
|
|
with pytest.raises(Exception, match="checkpoint is invalid"):
|
|
pipeline.run(resume_run_id=run_id)
|
|
assert pipeline._snapshot_holder is None
|
|
assert set(module._SNAPSHOT_DIRS) == baseline
|
|
|
|
|
|
def test_job_spec_construction_failure_releases_materialized_snapshot(monkeypatch, tmp_path):
|
|
import tht.jobs.dwh_pipeline as module
|
|
import pytest
|
|
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("catalog"),
|
|
build_lsh=lambda physical, output: _write_lsh([], physical, output),
|
|
)
|
|
assert pipeline.run().status == "succeeded"
|
|
baseline = set(module._SNAPSHOT_DIRS)
|
|
captured = []
|
|
real_materialize = module._materialize_generation_fd
|
|
|
|
def capture(*args, **kwargs):
|
|
holder, root = real_materialize(*args, **kwargs)
|
|
captured.append(holder)
|
|
return holder, root
|
|
|
|
monkeypatch.setattr(module, "_materialize_generation_fd", capture)
|
|
monkeypatch.setattr(
|
|
module, "JobSpec",
|
|
lambda **kwargs: (_ for _ in ()).throw(RuntimeError("job spec injected")),
|
|
)
|
|
with pytest.raises(RuntimeError, match="job spec injected"):
|
|
pipeline.run()
|
|
assert pipeline._snapshot_holder is None
|
|
assert set(module._SNAPSHOT_DIRS) == baseline
|
|
assert captured and all(not path.exists() for path in captured)
|
|
|
|
|
|
def test_publish_root_swap_after_lease_never_writes_replacement(monkeypatch, tmp_path):
|
|
import tht.jobs.dwh_pipeline as module
|
|
|
|
def make(content):
|
|
return DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text(content),
|
|
build_lsh=lambda physical, output: [
|
|
(output / name).write_text(content)
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
|
|
],
|
|
)
|
|
|
|
first = make("old").run()
|
|
assert first.status == "succeeded", first
|
|
root = tmp_path / ".tht-dwh"
|
|
moved = tmp_path / "moved-publish-root"
|
|
real_replace = module.os.replace
|
|
swapped = False
|
|
|
|
def swapping_replace(source, destination, *args, **kwargs):
|
|
nonlocal swapped
|
|
if destination == "ACTIVE" and kwargs.get("dst_dir_fd") is not None and not swapped:
|
|
swapped = True
|
|
root.rename(moved)
|
|
root.mkdir(mode=0o700)
|
|
(root / "sentinel").write_text("replacement-safe")
|
|
return real_replace(source, destination, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(module.os, "replace", swapping_replace)
|
|
result = make("new").run()
|
|
assert result.status in {"succeeded", "failed"}
|
|
assert swapped
|
|
assert (root / "sentinel").read_text() == "replacement-safe"
|
|
moved_active = (moved / "ACTIVE").read_text().strip()
|
|
assert len(moved_active) == 32
|
|
assert (moved / "generations" / moved_active).is_dir()
|
|
|
|
|
|
def test_cleanup_root_swap_after_lease_never_deletes_replacement(monkeypatch, tmp_path):
|
|
import tht.jobs.dwh_pipeline as module
|
|
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("trusted"),
|
|
build_lsh=lambda physical, output: [
|
|
(output / name).write_text("trusted")
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
|
|
],
|
|
retain_generations=1,
|
|
)
|
|
pipeline.run()
|
|
root = tmp_path / ".tht-dwh"
|
|
moved = tmp_path / "moved-cleanup-root"
|
|
real_open = module.os.open
|
|
swapped = False
|
|
|
|
def swapping_open(path, flags, *args, **kwargs):
|
|
nonlocal swapped
|
|
if path == "generations" and kwargs.get("dir_fd") is not None and not swapped:
|
|
swapped = True
|
|
root.rename(moved)
|
|
root.mkdir(mode=0o700)
|
|
(root / "sentinel").write_text("replacement-safe")
|
|
return real_open(path, flags, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(module.os, "open", swapping_open)
|
|
pipeline._cleanup_generations()
|
|
assert swapped
|
|
assert (root / "sentinel").read_text() == "replacement-safe"
|
|
|
|
|
|
def test_snapshot_stays_on_one_generation_across_publish(tmp_path):
|
|
def pipeline(content):
|
|
return DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text(content),
|
|
build_lsh=lambda physical, output: [
|
|
(output / name).write_text(content)
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
|
|
],
|
|
)
|
|
|
|
first = pipeline("old").run()
|
|
cfg = snapshot_config(tmp_path)
|
|
snapshot = resolve_dwh_snapshot(cfg)
|
|
pipeline("new").run()
|
|
assert snapshot.generation == first.run_id
|
|
assert snapshot.physical.read_text() == "old"
|
|
assert (snapshot.lsh_dir / "demo_meta.json").read_text() == "old"
|
|
|
|
|
|
def test_generation_retention_keeps_active_and_one_rollback(tmp_path):
|
|
run_ids = []
|
|
for index in range(5):
|
|
report = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output, i=index: output.write_text(str(i)),
|
|
build_lsh=lambda physical, output, i=index: [
|
|
(output / name).write_text(str(i))
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
|
|
],
|
|
retain_generations=2,
|
|
).run()
|
|
run_ids.append(report.run_id)
|
|
remaining = {path.name for path in (tmp_path / ".tht-dwh" / "generations").iterdir()}
|
|
assert remaining == set(run_ids[-2:])
|
|
|
|
|
|
def test_corrupt_newer_directory_does_not_consume_rollback_slot(tmp_path):
|
|
run_ids = []
|
|
pipeline = None
|
|
for index in range(3):
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output, i=index: output.write_text(str(i)),
|
|
build_lsh=lambda physical, output, i=index: [
|
|
(output / name).write_text(str(i))
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
|
|
],
|
|
retain_generations=3,
|
|
)
|
|
run_ids.append(pipeline.run().run_id)
|
|
generations = tmp_path / ".tht-dwh" / "generations"
|
|
corrupt = generations / ("f" * 32)
|
|
corrupt.mkdir(mode=0o700)
|
|
(corrupt / "junk").write_text("not a published generation")
|
|
|
|
pipeline.retain_generations = 2
|
|
pipeline._cleanup_generations()
|
|
|
|
assert (generations / run_ids[-1]).is_dir()
|
|
assert (generations / run_ids[-2]).is_dir()
|
|
assert not (generations / run_ids[0]).exists()
|
|
assert corrupt.is_dir()
|
|
|
|
|
|
def test_retention_n_counts_active_plus_n_minus_one_rollbacks_even_if_active_is_old(tmp_path):
|
|
import os
|
|
|
|
run_ids = []
|
|
pipeline = None
|
|
for index in range(3):
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output, i=index: output.write_text(str(i)),
|
|
build_lsh=lambda physical, output, i=index: [
|
|
(output / name).write_text(str(i))
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
|
|
], retain_generations=3,
|
|
)
|
|
run_ids.append(pipeline.run().run_id)
|
|
generations = tmp_path / ".tht-dwh" / "generations"
|
|
os.utime(generations / run_ids[-1], ns=(1, 1))
|
|
|
|
pipeline.retain_generations = 2
|
|
pipeline._cleanup_generations()
|
|
|
|
remaining = {path.name for path in generations.iterdir() if path.is_dir()}
|
|
assert remaining == {run_ids[-1], run_ids[-2]}
|
|
|
|
|
|
def test_retention_candidate_swap_to_symlink_is_never_followed(monkeypatch, tmp_path):
|
|
import tht.jobs.dwh_pipeline as module
|
|
|
|
pipeline = DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text("active"),
|
|
build_lsh=lambda physical, output: [
|
|
(output / name).write_text("active")
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
|
|
], retain_generations=1,
|
|
)
|
|
pipeline.run()
|
|
generations = tmp_path / ".tht-dwh" / "generations"
|
|
candidate_name = "e" * 32
|
|
candidate = generations / candidate_name
|
|
candidate.mkdir(mode=0o700)
|
|
external = tmp_path / "external-crafted"
|
|
external.mkdir()
|
|
sentinel = external / "sentinel"
|
|
sentinel.write_text("must-not-read-or-mutate")
|
|
real_open = module.os.open
|
|
swapped = False
|
|
|
|
def swapping_open(path, flags, *args, **kwargs):
|
|
nonlocal swapped
|
|
if path == candidate_name and kwargs.get("dir_fd") is not None and not swapped:
|
|
swapped = True
|
|
candidate.rmdir()
|
|
candidate.symlink_to(external, target_is_directory=True)
|
|
return real_open(path, flags, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(module.os, "open", swapping_open)
|
|
pipeline._cleanup_generations()
|
|
assert swapped
|
|
assert sentinel.read_text() == "must-not-read-or-mutate"
|
|
assert candidate.is_symlink()
|
|
|
|
|
|
def test_reader_lease_blocks_retain_one_publisher_until_file_reads_finish(tmp_path):
|
|
import threading
|
|
import time
|
|
def make(content, retain=1):
|
|
return DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text(content),
|
|
build_lsh=lambda physical, output: [
|
|
(output / name).write_text(content)
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
|
|
], retain_generations=retain,
|
|
)
|
|
|
|
first = make("old").run()
|
|
cfg = snapshot_config(tmp_path)
|
|
completed = threading.Event()
|
|
with lease_dwh_snapshot(cfg) as snapshot:
|
|
thread = threading.Thread(target=lambda: (make("new").run(), completed.set()))
|
|
thread.start()
|
|
time.sleep(0.05)
|
|
assert not completed.is_set()
|
|
assert snapshot.physical.read_text() == "old"
|
|
assert snapshot.generation == first.run_id
|
|
thread.join(timeout=2)
|
|
assert completed.is_set()
|
|
assert not (tmp_path / ".tht-dwh" / "generations" / first.run_id).exists()
|
|
|
|
|
|
def test_cleanup_never_follows_top_level_or_child_symlinks(tmp_path):
|
|
external = tmp_path / "external"
|
|
external.mkdir()
|
|
victim = external / "victim"
|
|
victim.write_text("safe")
|
|
|
|
def make(content, retain=1):
|
|
return DwhPreprocessPipeline(
|
|
workspace_id="demo", workspace_root=tmp_path,
|
|
config_fingerprint=FP, input_fingerprint=FP,
|
|
introspect=lambda output: output.write_text(content),
|
|
build_lsh=lambda physical, output: [
|
|
(output / name).write_text(content)
|
|
for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
|
|
], retain_generations=retain,
|
|
)
|
|
|
|
first = make("one", retain=2).run()
|
|
generations = tmp_path / ".tht-dwh" / "generations"
|
|
(generations / ("a" * 32)).symlink_to(external, target_is_directory=True)
|
|
make("two", retain=2).run()
|
|
old = generations / first.run_id
|
|
old.chmod(0o700)
|
|
(old / "hostile-link").symlink_to(victim)
|
|
make("three").run()
|
|
assert victim.read_text() == "safe"
|
|
assert victim.stat().st_mode & 0o200
|