ThothII/harness/tests/test_corpus_publish.py

import os

import pytest

from tht.evidence.corpus.models import CorpusManifest
from tht.evidence.corpus.store import CorpusStore, UnsafeCorpusPath


def test_publish_switches_active_atomically_and_resolves_materialized_files(tmp_path):
    store = CorpusStore(tmp_path / "corpus")
    generation = store.stage(CorpusManifest(), {})
    seen = []
    store._replace = lambda source, target: (seen.append(source.read_text()), source.replace(target))
    published = store.publish(generation)
    assert published == generation
    assert store.active_generation() == generation
    assert seen == [generation + "\n"]


def test_active_manifest_is_a_consistent_reader_snapshot(tmp_path):
    store = CorpusStore(tmp_path / "corpus")
    first = store.stage(CorpusManifest(metadata={"name": "first"}), {})
    second = store.stage(CorpusManifest(metadata={"name": "second"}), {})
    store.publish(first)
    snapshot = store.active_manifest()
    store.publish(second)
    assert snapshot.metadata["name"] == "first"
    assert store.active_manifest().metadata["name"] == "second"


def test_store_rejects_symlinked_generation_root(tmp_path):
    outside = tmp_path / "outside"
    outside.mkdir()
    root = tmp_path / "corpus"
    root.symlink_to(outside, target_is_directory=True)
    with pytest.raises(UnsafeCorpusPath):
        CorpusStore(root)


def test_active_pointer_cannot_escape_generation_root(tmp_path):
    store = CorpusStore(tmp_path / "corpus")
    store.root.mkdir(parents=True, exist_ok=True)
    store.active_path.write_text("../outside\n")
    with pytest.raises(UnsafeCorpusPath):
        store.active_manifest()


def test_publish_restores_previous_active_when_directory_fsync_fails_after_replace(tmp_path, monkeypatch):
    store = CorpusStore(tmp_path / "corpus")
    first = store.stage(CorpusManifest(), {})
    second = store.stage(CorpusManifest(), {})
    store.publish(first)
    def fail_once():
        store._fsync_directory = store._sync_root
        raise OSError("post replace crash")

    store._fsync_directory = fail_once
    with pytest.raises(OSError, match="post replace"):
        store.publish(second)
    assert store.active_generation() == first


def test_read_document_rejects_symlink_hardlink_and_hash_mismatch(tmp_path):
    from tht.evidence.corpus.models import CanonicalDocument

    content = "trusted"
    digest = "sha256:" + __import__("hashlib").sha256(content.encode()).hexdigest()
    document = CanonicalDocument(
        document_id="doc:" + "a" * 64, source_id="fs:one", source_uri="file:///one",
        source_fingerprint="sha256:" + "b" * 64, content_hash=digest, content=content,
        pipeline_version="evidence-v1",
    )
    store = CorpusStore(tmp_path / "corpus")
    generation = store.stage(CorpusManifest(documents=(document,)), {document.document_id: content})
    path = store.resolve_document(document.document_id, generation)
    assert store.read_document(document.document_id, generation) == content

    path.unlink()
    path.symlink_to(tmp_path / "outside")
    (tmp_path / "outside").write_text(content)
    with pytest.raises(UnsafeCorpusPath):
        store.read_document(document.document_id, generation)

    path.unlink()
    os.link(tmp_path / "outside", path)
    with pytest.raises(UnsafeCorpusPath):
        store.read_document(document.document_id, generation)

    path.unlink()
    path.write_text("tampered")
    with pytest.raises(UnsafeCorpusPath):
        store.read_document(document.document_id, generation)


def test_generation_inventory_is_validated_and_sorted(tmp_path):
    store = CorpusStore(tmp_path / "corpus")
    first = store.stage(CorpusManifest(), {}, generation="gen:" + "1" * 32)
    second = store.stage(CorpusManifest(), {}, generation="gen:" + "2" * 32)
    (store.root / "unrelated").mkdir()
    assert store.list_generations() == [first, second]


def test_published_inventory_excludes_staged_and_invalid_newer_directories(tmp_path):
    store = CorpusStore(tmp_path / "corpus")
    first = store.stage(CorpusManifest(), {}, generation="gen:" + "1" * 32)
    store.publish(first)
    store.stage(CorpusManifest(), {}, generation="gen:" + "2" * 32)
    invalid = store.generation_path("gen:" + "3" * 32)
    invalid.mkdir()
    (invalid / "PUBLISHED").write_text("2026-01-01T00:00:00Z\n")
    assert store.published_generations() == [first]


def test_owned_copy_uses_validated_descriptor_bytes_when_source_is_replaced(tmp_path, monkeypatch):
    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",
        source_fingerprint="sha256:" + "d" * 64,
        content_hash="sha256:" + hashlib.sha256(content.encode()).hexdigest(),
        content=content, pipeline_version="evidence-v1",
    )
    store = CorpusStore(tmp_path / "corpus")
    generation = store.stage(CorpusManifest(documents=(document,)), {document.document_id: content})
    store.publish(generation)
    source = store.resolve_document(document.document_id)
    real_read = os.read

    def replace_after_read(fd, size):
        payload = real_read(fd, size)
        source.unlink()
        source.write_text("replacement")
        return payload

    monkeypatch.setattr(os, "read", replace_after_read)
    owned = store.materialize_document(document.document_id, tmp_path / "session" / "evidence.md")
    assert owned.read_text() == content
    assert hashlib.sha256(owned.read_bytes()).hexdigest() == document.content_hash.removeprefix("sha256:")


def test_materialized_snapshot_uses_identified_manifest_when_active_changes(tmp_path):
    import hashlib

    from tht.evidence.corpus.models import CanonicalDocument

    def doc(content, fingerprint):
        return CanonicalDocument(
            document_id="doc:" + hashlib.sha256(content.encode()).hexdigest(),
            source_id="fs:item", source_uri="file:///item",
            source_fingerprint="sha256:" + fingerprint * 64,
            content_hash="sha256:" + hashlib.sha256(content.encode()).hexdigest(),
            content=content, pipeline_version="evidence-v1",
        )

    store = CorpusStore(tmp_path / "corpus")
    old = doc("old", "a")
    old_generation = store.stage(CorpusManifest(documents=(old,)), {old.document_id: old.content})
    store.publish(old_generation)
    snapshot = store.active_manifest()
    new = doc("new", "b")
    new_generation = store.stage(CorpusManifest(documents=(new,)), {new.document_id: new.content})
    store.publish(new_generation)
    path = store.materialize_document(
        snapshot.documents[0].document_id, tmp_path / "owned.md", generation=snapshot.manifest_id,
    )
    assert path.read_text() == "old"