arborist/tests/test_mesh_wire_e2e.py

"""Mesh wire end-to-end: two real HTTP peers exchange gossip.

Each peer binds an ephemeral port via host=127.0.0.1, port=0 and runs
ThreadingHTTPServer in a daemon thread. The other peer's MeshWireClient
talks to it over real httpx + TCP. No mocking — this exercises the
full request/response path including JSON serialization, HTTP framing,
and threaded handler dispatch.

Tests run under timeout via daemon threads; if a test hangs the worst
case is the test process exits and pytest times out the test, leaving
no zombie threads (daemon=True).
"""

from __future__ import annotations

from pathlib import Path

import pytest

from arborist.document import Document
from arborist.ingest import ingest_source
from arborist.mesh import init_identity
from arborist.mesh.members import add_member
from arborist.mesh.wire import (
    MeshWireClient,
    MeshWireServer,
    TYPE_ANNOUNCE_ROOT,
)
from arborist.store import connect


@pytest.fixture
def two_peers(tmp_path):
    """Spin up alice + bob with mutual rosters and HTTP servers running."""
    alice_db = tmp_path / "alice.db"
    bob_db = tmp_path / "bob.db"

    a_conn = connect(alice_db)
    try:
        alice = init_identity(a_conn, group_name="t", member_id="alice")
    finally:
        a_conn.close()

    b_conn = connect(bob_db)
    try:
        bob = init_identity(b_conn, group_name="t", member_id="bob")
    finally:
        b_conn.close()

    a_conn = connect(alice_db)
    try:
        add_member(a_conn, member_id="bob", sign_pub=bob.sign_pub, dh_pub=bob.dh_pub)
    finally:
        a_conn.close()

    b_conn = connect(bob_db)
    try:
        add_member(b_conn, member_id="alice", sign_pub=alice.sign_pub, dh_pub=alice.dh_pub)
    finally:
        b_conn.close()

    alice_srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
    bob_srv = MeshWireServer(bob_db, host="127.0.0.1", port=0)
    alice_srv.start_in_thread()
    bob_srv.start_in_thread()
    try:
        yield {
            "alice": {"db": alice_db, "srv": alice_srv, "id": alice},
            "bob": {"db": bob_db, "srv": bob_srv, "id": bob},
        }
    finally:
        alice_srv.stop()
        bob_srv.stop()


def test_info_endpoint_returns_member_id(two_peers):
    """GET /mesh/info returns the responder's identity (no signature required)."""
    with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c:
        info = c.info()
    assert info["member_id"] == "alice"
    assert info["group_name"] == "t"
    assert info["current_epoch"] == 1


def test_announce_root_round_trip(two_peers):
    """Bob announces a doc_root to Alice; Alice's audit chain gains a
    `mesh_received` event with the announce body."""
    with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c:
        resp = c.announce_root(
            document_root="ab" * 32,
            source_uri="https://en.wikipedia.org/wiki/Cloud_Strife",
            chunking_version="tok-512-v1",
            canonicalization_version="norm-v1",
            schema_version="v9.8.0",
        )
    assert resp["ok"] is True
    assert resp["wire_type"] == TYPE_ANNOUNCE_ROOT
    assert isinstance(resp["audit_event_hash"], str)
    assert len(resp["audit_event_hash"]) == 64

    a_conn = connect(two_peers["alice"]["db"])
    try:
        last = a_conn.execute(
            "SELECT event_type, body, subject_root FROM audit_events "
            "ORDER BY seq DESC LIMIT 1"
        ).fetchone()
    finally:
        a_conn.close()
    assert last["event_type"] == "mesh_received"
    assert last["subject_root"] == "ab" * 32
    import json
    body = json.loads(last["body"])
    assert body["wire_type"] == TYPE_ANNOUNCE_ROOT
    assert body["sender_id"] == "bob"
    assert body["envelope_body"]["source_uri"] == "https://en.wikipedia.org/wiki/Cloud_Strife"


class _OneDocSource:
    """One-shot source for tests — yields a single Document."""

    source_type = "wire_e2e"

    def __init__(self, uri: str, content: str, title: str):
        self._doc = Document(
            uri=uri, content=content, source_type=self.source_type, title=title
        )

    def iter_documents(self):
        yield self._doc


def _ingest_one(db_path, *, uri: str, content: str, title: str) -> str:
    """Ingest one document, return its document_root."""
    conn = connect(db_path)
    try:
        ingest_source(conn, _OneDocSource(uri, content, title))
        row = conn.execute(
            "SELECT document_root FROM documents WHERE document_uri=?", (uri,)
        ).fetchone()
    finally:
        conn.close()
    assert row is not None, f"document not found post-ingest for {uri!r}"
    return row["document_root"]


def test_request_body_pulls_doc_with_verified_merkle(two_peers):
    """Alice has a doc; Bob requests it. Client verifies the delivered
    Merkle root against the requested root before returning."""
    document_root = _ingest_one(
        two_peers["alice"]["db"],
        uri="https://en.wikipedia.org/wiki/test",
        content="Hello world. The quick brown fox jumps over the lazy dog.",
        title="Test",
    )
    with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c:
        delivered = c.request_body(root=document_root)

    assert delivered["root"] == document_root
    assert delivered["document_uri"] == "https://en.wikipedia.org/wiki/test"
    assert "Hello world" in delivered["text"]
    assert isinstance(delivered["leaves_hex"], list)
    assert len(delivered["leaves_hex"]) >= 1


def test_request_body_404_on_unknown_root(two_peers):
    """Asking for a root the responder doesn't have raises HTTPStatusError."""
    import httpx
    with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c:
        with pytest.raises(httpx.HTTPStatusError):
            c.request_body(root="ff" * 32)


def test_request_body_rejects_tampered_response(two_peers, monkeypatch):
    """If a peer returns a DELIVER_BODY whose leaves don't Merkle-derive
    to the claimed root, the client raises before returning."""
    document_root = _ingest_one(
        two_peers["alice"]["db"],
        uri="https://en.wikipedia.org/wiki/x",
        content="real content of the document",
        title="X",
    )
    with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c:
        from arborist.mesh import wire as _wire
        monkeypatch.setattr(_wire, "_merkle_root_matches", lambda *a, **k: False)
        with pytest.raises(ValueError, match="Merkle-derive"):
            c.request_body(root=document_root)