javis_bot/tests/test_realtime_gemini_cli.py

"""Behaviour tests for the Gemini CLI (OAuth login) real-time search backend.

These verify the observable contract of ``gemini_cli_search``:
- a fenced answer when the CLI returns a JSON ``response``,
- fail-open (``None``) when the CLI is missing, times out, errors, or returns
  empty/garbage output,
- the child process never inherits ``GEMINI_API_KEY`` / ``GOOGLE_API_KEY`` so
  the CLI uses the account OAuth login rather than API-key auth.
"""

import subprocess
import types

import jarvis.tools.builtin.realtime_search as rs


def _fake_proc(stdout: str) -> types.SimpleNamespace:
    return types.SimpleNamespace(stdout=stdout, stderr="", returncode=0)


def test_returns_fenced_answer_on_success(monkeypatch):
    monkeypatch.setattr(rs, "_gemini_bin", lambda: "/usr/bin/gemini")
    monkeypatch.setattr(
        rs.subprocess,
        "run",
        lambda *a, **k: _fake_proc('{"response": "Seoul is sunny, 21C.", "stats": {}}'),
    )

    out = rs.gemini_cli_search("weather in Seoul")

    assert out is not None
    assert "Seoul is sunny, 21C." in out
    assert "UNTRUSTED WEB EXTRACT" in out  # untrusted-data fence preserved


def test_none_when_cli_not_installed(monkeypatch):
    monkeypatch.setattr(rs, "_gemini_bin", lambda: None)
    # subprocess.run must not even be reached
    monkeypatch.setattr(rs.subprocess, "run", lambda *a, **k: (_ for _ in ()).throw(AssertionError("should not run")))

    assert rs.gemini_cli_search("anything") is None


def test_none_on_empty_query(monkeypatch):
    monkeypatch.setattr(rs, "_gemini_bin", lambda: "/usr/bin/gemini")
    assert rs.gemini_cli_search("") is None


def test_none_on_empty_response(monkeypatch):
    monkeypatch.setattr(rs, "_gemini_bin", lambda: "/usr/bin/gemini")
    monkeypatch.setattr(rs.subprocess, "run", lambda *a, **k: _fake_proc('{"response": "", "stats": {}}'))
    assert rs.gemini_cli_search("q") is None


def test_none_on_garbage_output(monkeypatch):
    monkeypatch.setattr(rs, "_gemini_bin", lambda: "/usr/bin/gemini")
    monkeypatch.setattr(rs.subprocess, "run", lambda *a, **k: _fake_proc("not json at all"))
    assert rs.gemini_cli_search("q") is None


def test_none_on_timeout(monkeypatch):
    monkeypatch.setattr(rs, "_gemini_bin", lambda: "/usr/bin/gemini")

    def _boom(*a, **k):
        raise subprocess.TimeoutExpired(cmd="gemini", timeout=30)

    monkeypatch.setattr(rs.subprocess, "run", _boom)
    assert rs.gemini_cli_search("q") is None


def test_api_key_stripped_from_child_env(monkeypatch):
    monkeypatch.setattr(rs, "_gemini_bin", lambda: "/usr/bin/gemini")
    monkeypatch.setenv("GEMINI_API_KEY", "AIza-secret")
    monkeypatch.setenv("GOOGLE_API_KEY", "another-secret")
    captured = {}

    def _capture(*a, **k):
        captured["env"] = k.get("env", {})
        captured["cmd"] = a[0] if a else k.get("args")
        return _fake_proc('{"response": "ok"}')

    monkeypatch.setattr(rs.subprocess, "run", _capture)
    rs.gemini_cli_search("q")

    assert "GEMINI_API_KEY" not in captured["env"]
    assert "GOOGLE_API_KEY" not in captured["env"]
    # Google-account (OAuth) auth method is selected explicitly so a missing
    # login fails fast non-interactively instead of erroring on "no auth method".
    assert captured["env"].get("GOOGLE_GENAI_USE_GCA") == "true"
    # invoked headless with JSON output
    assert "-p" in captured["cmd"] and "-o" in captured["cmd"] and "json" in captured["cmd"]
    # never auto-approve all tools: a search query must not be able to trigger
    # write/shell tool execution.
    assert "yolo" not in captured["cmd"]
    assert "--yolo" not in captured["cmd"]


def test_oauth_ready_reflects_creds_file(monkeypatch, tmp_path):
    """``gemini_oauth_ready`` is the seeded-login signal: false until the CLI's
    ``~/.gemini/oauth_creds.json`` exists, true once it does."""
    monkeypatch.setenv("HOME", str(tmp_path))
    assert rs.gemini_oauth_ready() is False
    gdir = tmp_path / ".gemini"
    gdir.mkdir()
    (gdir / "oauth_creds.json").write_text("{}")
    assert rs.gemini_oauth_ready() is True
    assert rs.gemini_oauth_dir() == gdir


def test_hint_logged_once_when_oauth_not_seeded(monkeypatch):
    """When OAuth is selected but no login is seeded, the path still attempts the
    CLI (behaviour unchanged) but logs a single actionable hint so the silent
    DDG/Brave fallback is diagnosable."""
    monkeypatch.setattr(rs, "_gemini_bin", lambda: "/usr/bin/gemini")
    monkeypatch.setattr(rs, "gemini_oauth_ready", lambda: False)
    monkeypatch.setattr(rs.subprocess, "run", lambda *a, **k: _fake_proc('{"response": "ok"}'))
    logs: list[str] = []
    monkeypatch.setattr(rs, "debug_log", lambda msg, *a, **k: logs.append(msg))
    monkeypatch.setattr(rs, "_oauth_hint_shown", False)

    assert rs.gemini_cli_search("q") is not None  # still attempts, behaviour unchanged
    rs.gemini_cli_search("q again")  # second call must not re-log

    hints = [m for m in logs if "no Gemini login" in m]
    assert len(hints) == 1


def test_no_hint_when_oauth_seeded(monkeypatch):
    """A seeded login produces no fallback hint."""
    monkeypatch.setattr(rs, "_gemini_bin", lambda: "/usr/bin/gemini")
    monkeypatch.setattr(rs, "gemini_oauth_ready", lambda: True)
    monkeypatch.setattr(rs.subprocess, "run", lambda *a, **k: _fake_proc('{"response": "ok"}'))
    logs: list[str] = []
    monkeypatch.setattr(rs, "debug_log", lambda msg, *a, **k: logs.append(msg))
    monkeypatch.setattr(rs, "_oauth_hint_shown", False)

    rs.gemini_cli_search("q")

    assert not [m for m in logs if "no Gemini login" in m]