arborist/tests/test_versioned_ingest.py

"""Versioned re-ingest: same URI, changed content -> 'supersedes' edge."""

from __future__ import annotations

import time
from typing import Iterator

from arborist.document import Document
from arborist.ingest import ingest_source
from arborist.source import Source
from arborist.store import connect


class FakeSource(Source):
    source_type = "test"

    def __init__(self, docs: list[Document]):
        self.docs = docs

    def iter_documents(self) -> Iterator[Document]:
        yield from self.docs


def _doc(uri: str, content: str) -> Document:
    return Document(uri=uri, content=content, source_type="test", title=uri)


def test_unchanged_reingest_is_idempotent(tmp_path):
    db = tmp_path / "idem.db"
    conn = connect(db)
    try:
        d = _doc("test://stable", "same content same content same content " * 30)
        ingest_source(conn, FakeSource([d]))
        # Re-ingesting identical content = same root = idempotent skip.
        ingest_source(conn, FakeSource([d]))
        n = conn.execute(
            "SELECT COUNT(*) FROM documents WHERE document_uri='test://stable'"
        ).fetchone()[0]
        assert n == 1
        # No supersedes edge — content unchanged.
        n_super = conn.execute(
            "SELECT COUNT(*) FROM edges WHERE edge_type='supersedes'"
        ).fetchone()[0]
        assert n_super == 0
    finally:
        conn.close()


def test_changed_reingest_creates_supersedes_edge(tmp_path):
    db = tmp_path / "v.db"
    conn = connect(db)
    try:
        v1 = _doc("test://changing", "version one content " * 30)
        ingest_source(conn, FakeSource([v1]))
        time.sleep(1.05)  # cross integer second so ingest_ts differs

        v2 = _doc("test://changing", "version two content radically different " * 30)
        ingest_source(conn, FakeSource([v2]))

        # Two distinct documents share the URI.
        rows = conn.execute(
            "SELECT document_root FROM documents WHERE document_uri='test://changing' "
            "ORDER BY ingest_ts ASC"
        ).fetchall()
        assert len(rows) == 2
        old_root = rows[0]["document_root"]
        new_root = rows[1]["document_root"]
        assert old_root != new_root

        # Supersedes edge: new -> old.
        edge = conn.execute(
            "SELECT * FROM edges WHERE edge_type='supersedes'"
        ).fetchone()
        assert edge is not None
        assert edge["src_root"] == new_root
        assert edge["dst_root"] == old_root
        assert edge["dst_uri"] == "test://changing"

        # Audit chain records the supersedes link in the new ingest's body.
        from json import loads as _loads
        ev = conn.execute(
            "SELECT body FROM audit_events WHERE subject_root=?", (new_root,)
        ).fetchone()
        assert _loads(ev["body"])["supersedes"] == old_root
    finally:
        conn.close()


def test_three_version_chain(tmp_path):
    """v1 -> v2 -> v3 produces two supersedes edges in a chain."""
    db = tmp_path / "chain.db"
    conn = connect(db)
    try:
        for i, content in enumerate(
            ["alpha alpha alpha " * 30, "beta beta beta " * 30, "gamma gamma gamma " * 30]
        ):
            ingest_source(conn, FakeSource([_doc("test://multi", content)]))
            time.sleep(1.05)

        edges = conn.execute(
            "SELECT src_root, dst_root FROM edges WHERE edge_type='supersedes'"
        ).fetchall()
        assert len(edges) == 2
        # Walk the chain: v3 -> v2, v2 -> v1.
        roots = conn.execute(
            "SELECT document_root FROM documents WHERE document_uri='test://multi' "
            "ORDER BY ingest_ts ASC"
        ).fetchall()
        v1, v2, v3 = (r["document_root"] for r in roots)

        edge_pairs = {(e["src_root"], e["dst_root"]) for e in edges}
        assert (v2, v1) in edge_pairs
        assert (v3, v2) in edge_pairs
    finally:
        conn.close()