test_determinism.py

"""Determinism harness: same input => byte-identical verdict, zero network."""

from __future__ import annotations

import json
import os
import socket
import subprocess
import sys
from pathlib import Path

import pytest

from taintline.cli import run_check
from taintline.detectors import run_all
from taintline.model import Span, Trace, Verdict
from tests._helpers import PAYLOAD, mk_span

FAIL_ON = frozenset({"taint"})


def _fixture() -> Trace:
    """A trace exercising a firing flow, a sanitized flow, and clean spans."""
    return Trace(
        spans=(
            mk_span("a-src", start_ns=10, tool_name="web_fetch", tool_output=f"intro\n{PAYLOAD}\n"),
            mk_span(
                "a-sink",
                start_ns=20,
                tool_name="transfer",
                tool_input={"instruction": PAYLOAD},
            ),
            mk_span("b-src", start_ns=30, tool_name="read_file", tool_output=f"note {PAYLOAD} x"),
            mk_span("b-san", start_ns=35, tool_name="sanitize_input"),
            mk_span("b-sink", start_ns=40, tool_name="send_email", tool_input={"body": PAYLOAD}),
            mk_span("c", start_ns=50, tool_name="web_search", tool_output="weather is fine"),
        )
    )


def test_verdict_byte_identical_over_10_runs() -> None:
    trace = _fixture()
    first_verdict: Verdict = run_all(trace)
    first_code, first_out = run_check(trace, FAIL_ON)
    for _ in range(10):
        assert run_all(trace) == first_verdict
        code, out = run_check(trace, FAIL_ON)
        assert code == first_code
        assert out == first_out
    # The firing flow (a-src -> a-sink) is present; the sanitized one is not.
    assert first_code == 1
    assert any(f.span_ids == ("a-src", "a-sink") for f in first_verdict.findings)
    assert all(f.span_ids != ("b-src", "b-sink") for f in first_verdict.findings)


def test_check_makes_zero_network_calls(monkeypatch: pytest.MonkeyPatch) -> None:
    def _boom(*_a: object, **_k: object) -> None:
        raise AssertionError("network access attempted in check path")

    monkeypatch.setattr(socket, "socket", _boom)
    monkeypatch.setattr(socket, "create_connection", _boom)
    code, out = run_check(_fixture(), FAIL_ON)
    assert code == 1
    assert out  # produced a verdict with the socket layer disabled


def _dump_otlp(trace: Trace) -> list[dict[str, object]]:
    """Serialize spans in the OTLP-ish shape the real ingest.load_trace parses."""

    def as_dict(s: Span) -> dict[str, object]:
        attrs: dict[str, object] = {}
        if s.tool_name is not None:
            attrs["gen_ai.tool.name"] = s.tool_name
        if s.tool_input is not None:
            attrs["gen_ai.tool.input"] = s.tool_input
        if s.tool_output is not None:
            attrs["gen_ai.tool.output"] = s.tool_output
        return {
            "spanId": s.span_id,
            "name": s.name,
            "startTimeUnixNano": s.start_ns,
            "attributes": attrs,
        }

    return [as_dict(s) for s in trace.spans]


def test_cli_byte_identical_across_hash_seeds(tmp_path: Path) -> None:
    # Subprocess runs pick a fresh PYTHONHASHSEED; an in-process loop cannot
    # catch a set-iteration-order leak because the seed is fixed per process.
    fixture = tmp_path / "trace.json"
    fixture.write_text(json.dumps({"spans": _dump_otlp(_fixture())}), encoding="utf-8")
    code = "from taintline.cli import main; raise SystemExit(main())"
    argv = [sys.executable, "-c", code, "check", str(fixture), "--fail-on", "taint"]

    outputs = set()
    for seed in ("0", "1", "42", "1000", "31337"):
        env = {**os.environ, "PYTHONHASHSEED": seed}
        proc = subprocess.run(argv, capture_output=True, text=True, env=env)
        assert proc.returncode == 1
        outputs.add(proc.stdout)
    assert len(outputs) == 1