graphify/tests/test_claude_cli_backend.py

"""Tests for the `claude-cli` backend (#855/#856).

Mocks subprocess.run + shutil.which so the suite runs on CI without
the `claude` binary or a live network call.
"""
from __future__ import annotations

import json
from unittest.mock import MagicMock, patch

import pytest

from graphify import llm

_ENVELOPE = {
    "type": "result",
    "subtype": "success",
    "is_error": False,
    "result": json.dumps({
        "nodes": [
            {"id": "foo_module", "label": "Foo", "file_type": "document", "source_file": "foo.md"},
            {"id": "foo_greet", "label": "greet", "file_type": "code", "source_file": "foo.md"},
        ],
        "edges": [
            {"source": "foo_module", "target": "foo_greet",
             "relation": "references", "confidence": "EXTRACTED", "confidence_score": 1.0},
        ],
        "hyperedges": [],
        "input_tokens": 0,
        "output_tokens": 0,
    }),
    "stop_reason": "end_turn",
    "usage": {
        "input_tokens": 6,
        "output_tokens": 11,
        "cache_read_input_tokens": 17837,
        "cache_creation_input_tokens": 30800,
    },
    "modelUsage": {"claude-opus-4-7[1m]": {"inputTokens": 6, "outputTokens": 11}},
}


@pytest.fixture
def fake_claude(monkeypatch):
    completed = MagicMock(returncode=0, stdout=json.dumps(_ENVELOPE), stderr="")
    monkeypatch.setattr(llm, "_response_is_hollow", lambda raw, parsed: False)
    with patch("shutil.which", return_value="/fake/bin/claude"), \
         patch("subprocess.run", return_value=completed) as run:
        yield run


def test_returns_parsed_nodes_and_edges(fake_claude):
    result = llm._call_claude_cli("dummy", max_tokens=8192)
    assert len(result["nodes"]) == 2
    assert len(result["edges"]) == 1


def test_token_accounting_includes_cache(fake_claude):
    result = llm._call_claude_cli("dummy", max_tokens=8192)
    assert result["input_tokens"] == 6 + 17837 + 30800
    assert result["output_tokens"] == 11
    assert result["model"] == "claude-opus-4-7[1m]"
    assert result["finish_reason"] == "stop"


def test_finish_reason_length_on_max_tokens(monkeypatch):
    envelope = dict(_ENVELOPE, stop_reason="max_tokens")
    completed = MagicMock(returncode=0, stdout=json.dumps(envelope), stderr="")
    monkeypatch.setattr(llm, "_response_is_hollow", lambda raw, parsed: False)
    with patch("shutil.which", return_value="/fake/bin/claude"), \
         patch("subprocess.run", return_value=completed):
        result = llm._call_claude_cli("dummy", max_tokens=8192)
    assert result["finish_reason"] == "length"


def test_raises_when_cli_missing():
    with patch("shutil.which", return_value=None):
        with pytest.raises(RuntimeError, match="Claude Code CLI not found"):
            llm._call_claude_cli("dummy", max_tokens=8192)


def test_raises_on_nonzero_exit():
    completed = MagicMock(returncode=2, stdout="", stderr="auth failed")
    with patch("shutil.which", return_value="/fake/bin/claude"), \
         patch("subprocess.run", return_value=completed):
        with pytest.raises(RuntimeError, match="exited 2"):
            llm._call_claude_cli("dummy", max_tokens=8192)


def test_raises_on_garbage_envelope():
    completed = MagicMock(returncode=0, stdout="not json", stderr="")
    with patch("shutil.which", return_value="/fake/bin/claude"), \
         patch("subprocess.run", return_value=completed):
        with pytest.raises(RuntimeError, match="unparseable JSON envelope"):
            llm._call_claude_cli("dummy", max_tokens=8192)


def test_extract_files_direct_dispatches_to_claude_cli(tmp_path, fake_claude):
    f = tmp_path / "foo.md"
    f.write_text("# Foo\n\nThe greet() helper formats a name.\n")
    result = llm.extract_files_direct(files=[f], backend="claude-cli", root=tmp_path)
    assert fake_claude.called
    assert len(result["nodes"]) == 2


def test_backend_registered_with_zero_cost():
    assert "claude-cli" in llm.BACKENDS
    pricing = llm.BACKENDS["claude-cli"]["pricing"]
    assert pricing["input"] == 0.0
    assert pricing["output"] == 0.0
    assert llm.estimate_cost("claude-cli", 1_000_000, 1_000_000) == 0.0


def test_no_session_persistence_flag_in_subprocess(fake_claude):
    llm._call_claude_cli("dummy", max_tokens=8192)
    call_args = fake_claude.call_args[0][0]
    assert "--no-session-persistence" in call_args


# ---------- extraction instructions delivered in the user turn ----------
# Newer Claude Code CLIs (>= ~2.1) do not honour a --system-prompt that asks
# for raw JSON: they keep their coding-agent context and reply conversationally
# to a bare file dump, which parses to zero nodes and gets bisected forever.
# The instructions must ride in the user turn instead. See the fix for the
# "hollow response" / infinite-bisection failure on Claude Code 2.1.x.


def test_no_system_prompt_flag_in_subprocess(fake_claude):
    """--system-prompt must NOT be used: the CLI ignores its 'raw JSON only'
    directive and replies with prose, breaking extraction."""
    llm._call_claude_cli("dummy source", max_tokens=8192)
    argv = fake_claude.call_args.args[0]
    assert "--system-prompt" not in argv


def test_extraction_instructions_ride_in_user_turn(fake_claude):
    """The full extraction schema, an explicit imperative, and the source must
    all be delivered via stdin (the user turn)."""
    llm._call_claude_cli("UNIQUE_SOURCE_MARKER", max_tokens=8192)
    sent = fake_claude.call_args.kwargs["input"]
    # schema text from _extraction_system
    assert "graphify semantic extraction agent" in sent
    # explicit imperative appended before the source
    assert "output ONLY the JSON object" in sent
    # the caller's source payload is preserved
    assert "UNIQUE_SOURCE_MARKER" in sent


def test_user_turn_preserves_untrusted_source_guardrails(fake_claude):
    """The <untrusted_source> guardrails from _extraction_system must survive
    the move into the user turn (prompt-injection defence is unchanged)."""
    llm._call_claude_cli("dummy", max_tokens=8192)
    sent = fake_claude.call_args.kwargs["input"]
    assert "untrusted_source" in sent


# ---------- structured output via --json-schema (#2076) ----------
# Newer Claude Code CLIs treat a bare file-dump prompt as an agentic task and
# REPORT the extraction in prose instead of returning JSON, so the graph comes
# out empty and adaptive-retry bisects forever. When the CLI supports
# `--json-schema`, graphify constrains the output shape structurally so the
# model must emit the object regardless of framing. Older CLIs that predate the
# flag fall back to the user-turn prompt, unchanged.


def test_json_schema_flag_added_when_cli_supports_it(monkeypatch, fake_claude):
    """When the CLI advertises --json-schema, it is passed with a schema that
    pins the top-level {nodes, edges} shape graphify parses."""
    monkeypatch.setattr(llm, "_claude_cli_supports_json_schema", lambda cmd: True)
    llm._call_claude_cli("dummy source", max_tokens=8192)
    argv = fake_claude.call_args.args[0]
    assert "--json-schema" in argv
    schema = json.loads(argv[argv.index("--json-schema") + 1])
    assert schema["type"] == "object"
    assert set(schema["required"]) == {"nodes", "edges"}


def test_json_schema_flag_absent_when_cli_lacks_it(monkeypatch, fake_claude):
    """Older CLIs without --json-schema must not receive the flag (it would be
    an unknown option) — extraction falls back to the user-turn prompt."""
    monkeypatch.setattr(llm, "_claude_cli_supports_json_schema", lambda cmd: False)
    result = llm._call_claude_cli("dummy source", max_tokens=8192)
    argv = fake_claude.call_args.args[0]
    assert "--json-schema" not in argv
    assert len(result["nodes"]) == 2  # result envelope still carries the JSON


def test_supports_json_schema_detects_flag_in_help():
    llm._JSON_SCHEMA_SUPPORT.clear()
    help_text = "Options:\n  --json-schema <schema>  JSON Schema for structured output\n"
    completed = MagicMock(returncode=0, stdout=help_text, stderr="")
    with patch("subprocess.run", return_value=completed):
        assert llm._claude_cli_supports_json_schema("/fake/claude-new") is True


def test_supports_json_schema_false_when_flag_absent():
    llm._JSON_SCHEMA_SUPPORT.clear()
    help_text = "Options:\n  --output-format <format>  text|json|stream-json\n"
    completed = MagicMock(returncode=0, stdout=help_text, stderr="")
    with patch("subprocess.run", return_value=completed):
        assert llm._claude_cli_supports_json_schema("/fake/claude-old") is False


def test_supports_json_schema_false_and_cached_on_probe_error():
    """A probe that fails to run is treated as unsupported (safe fallback) and
    cached so it is not re-probed for every chunk."""
    llm._JSON_SCHEMA_SUPPORT.clear()
    with patch("subprocess.run", side_effect=OSError("boom")) as run:
        assert llm._claude_cli_supports_json_schema("/fake/claude-broken") is False
        assert llm._claude_cli_supports_json_schema("/fake/claude-broken") is False
    assert run.call_count == 1


# ---------- Windows path resolution (#1072) ----------


def test_windows_prefers_claude_cmd_over_bare_claude(monkeypatch):
    """On Windows, npm installs `claude.ps1` alongside `claude.cmd`.
    `CreateProcess` cannot execute `.ps1` directly (raises WinError 2),
    so we must explicitly resolve `claude.cmd` and pass its full path
    to subprocess.run. See issue #1072."""
    completed = MagicMock(returncode=0, stdout=json.dumps(_ENVELOPE), stderr="")
    monkeypatch.setattr(llm, "_response_is_hollow", lambda raw, parsed: False)

    def fake_which(name):
        # Simulate Windows PATHEXT=.PS1;.CMD ordering: bare "claude"
        # resolves to the .ps1 (unexecutable by CreateProcess), while
        # "claude.cmd" resolves to the .cmd shim.
        return {
            "claude": r"C:\Users\u\AppData\Roaming\npm\claude.ps1",
            "claude.cmd": r"C:\Users\u\AppData\Roaming\npm\claude.cmd",
        }.get(name)

    with patch("platform.system", return_value="Windows"), \
         patch("shutil.which", side_effect=fake_which), \
         patch("subprocess.run", return_value=completed) as run:
        llm._call_claude_cli("dummy", max_tokens=8192)

    argv = run.call_args.args[0]
    assert argv[0] == r"C:\Users\u\AppData\Roaming\npm\claude.cmd", (
        f"Expected full path to claude.cmd on Windows, got {argv[0]!r}"
    )


def test_windows_falls_back_to_bare_claude_when_cmd_missing(monkeypatch):
    """If `claude.cmd` is somehow unavailable but `claude` resolves
    (e.g. WSL-style install), fall back to the bare name so the
    existing behaviour is preserved."""
    completed = MagicMock(returncode=0, stdout=json.dumps(_ENVELOPE), stderr="")
    monkeypatch.setattr(llm, "_response_is_hollow", lambda raw, parsed: False)

    def fake_which(name):
        if name == "claude.cmd":
            return None
        if name == "claude":
            return "/usr/local/bin/claude"
        return None

    with patch("platform.system", return_value="Windows"), \
         patch("shutil.which", side_effect=fake_which), \
         patch("subprocess.run", return_value=completed) as run:
        llm._call_claude_cli("dummy", max_tokens=8192)

    argv = run.call_args.args[0]
    assert argv[0] == "claude"


def test_windows_raises_when_neither_cmd_nor_bare_claude_present():
    """If neither `claude.cmd` nor `claude` are on PATH on Windows,
    raise the standard not-found error."""
    with patch("platform.system", return_value="Windows"), \
         patch("shutil.which", return_value=None):
        with pytest.raises(RuntimeError, match="Claude Code CLI not found"):
            llm._call_claude_cli("dummy", max_tokens=8192)


def test_non_windows_uses_bare_claude(monkeypatch):
    """On non-Windows platforms, behaviour is unchanged: bare `claude`
    is passed to subprocess.run (shell resolves it via PATH)."""
    completed = MagicMock(returncode=0, stdout=json.dumps(_ENVELOPE), stderr="")
    monkeypatch.setattr(llm, "_response_is_hollow", lambda raw, parsed: False)

    with patch("platform.system", return_value="Linux"), \
         patch("shutil.which", return_value="/usr/local/bin/claude"), \
         patch("subprocess.run", return_value=completed) as run:
        llm._call_claude_cli("dummy", max_tokens=8192)

    argv = run.call_args.args[0]
    assert argv[0] == "claude"


# ---------- GRAPHIFY_API_TIMEOUT honoured by all backends ----------


def test_resolve_api_timeout_default(monkeypatch):
    monkeypatch.delenv("GRAPHIFY_API_TIMEOUT", raising=False)
    assert llm._resolve_api_timeout() == 600.0


def test_resolve_api_timeout_env_override(monkeypatch):
    monkeypatch.setenv("GRAPHIFY_API_TIMEOUT", "45")
    assert llm._resolve_api_timeout() == 45.0


def test_resolve_api_timeout_ignores_invalid(monkeypatch):
    monkeypatch.setenv("GRAPHIFY_API_TIMEOUT", "not-a-number")
    assert llm._resolve_api_timeout() == 600.0


def test_resolve_api_timeout_ignores_nonpositive(monkeypatch):
    monkeypatch.setenv("GRAPHIFY_API_TIMEOUT", "0")
    assert llm._resolve_api_timeout() == 600.0


def test_claude_cli_extraction_honours_timeout(monkeypatch, fake_claude):
    monkeypatch.setenv("GRAPHIFY_API_TIMEOUT", "30")
    llm._call_claude_cli("dummy", max_tokens=8192)
    assert fake_claude.call_args.kwargs["timeout"] == 30.0


def test_call_llm_claude_cli_branch_honours_timeout(monkeypatch, fake_claude):
    monkeypatch.setenv("GRAPHIFY_API_TIMEOUT", "30")
    llm._call_llm(prompt="x", backend="claude-cli", max_tokens=10)
    assert fake_claude.call_args.kwargs["timeout"] == 30.0


def test_simple_completion_resolves_cmd_shim_on_windows(monkeypatch):
    """The label/_simple_completion path must spawn the resolved claude.cmd on
    Windows; a bare "claude" fails CreateProcess (WinError 2) under npm installs."""
    import json as _json
    from unittest.mock import patch, MagicMock

    captured = {}

    def fake_run(args, **kwargs):
        captured["argv0"] = args[0]
        proc = MagicMock()
        proc.returncode = 0
        proc.stdout = _json.dumps({"result": "ok"})
        return proc

    def fake_which(name):
        return r"C:\npm\claude.cmd" if name == "claude.cmd" else r"C:\npm\claude"

    with patch("platform.system", return_value="Windows"), \
         patch("shutil.which", side_effect=fake_which), \
         patch("subprocess.run", side_effect=fake_run):
        out = llm._call_llm("hi", backend="claude-cli")

    assert out == "ok"
    assert captured["argv0"] == r"C:\npm\claude.cmd"