Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 0 additions & 4 deletions tests/life/test_lifecycle_supervisor_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -522,10 +522,6 @@ def _with_preflight_pdf(tmp_path: Path) -> Path:
return project_root


def _journal_kinds(stub: _GateStub) -> list[str]:
return [getattr(e, "kind", None) for e in stub.journal_entries]


def test_gate_suppresses_premature_done_for_uncertified_emnlp(tmp_path: Path) -> None:
# main.pdf exists but reviewer has NOT certified → no DONE, dispatch
# proceeds, and the suppressed transition is never journaled/persisted.
Expand Down
79 changes: 29 additions & 50 deletions tests/test_codex_model_pricing_fallback.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,46 +37,37 @@ def _isolate_codex_config(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("ARGUS_SKILL_CODEX_CONFIG", raising=False)


# --- pure helper: model selection + traceable fallback source ---------------

def test_resolve_pricing_model_prefers_response_model() -> None:
assert resolve_pricing_model("gpt-5.5", "req", "def") == ("gpt-5.5", "")


def test_resolve_pricing_model_falls_back_to_request_when_response_empty() -> None:
assert resolve_pricing_model("", "req-model", "def") == ("req-model", "request")
# whitespace-only response is treated as empty
assert resolve_pricing_model(" ", "req-model", "def") == ("req-model", "request")

@pytest.fixture
def codex_config_dir(tmp_path, monkeypatch):
directory = tmp_path / "codex"
directory.mkdir()
monkeypatch.setenv("CODEX_HOME", str(directory))
return directory

def test_resolve_pricing_model_falls_back_to_configured_default() -> None:
assert resolve_pricing_model("", "", "gpt-5.5") == (
"gpt-5.5",
"configured_default",
)
assert resolve_pricing_model(None, None, "gpt-5.5") == (
"gpt-5.5",
"configured_default",
)

# --- pure helper: model selection + traceable fallback source ---------------

def test_resolve_pricing_model_empty_when_nothing_usable() -> None:
# No reliable fallback -> stay empty rather than invent a priced model.
assert resolve_pricing_model("", "", "") == ("", "none")
assert resolve_pricing_model(None, None, None) == ("", "none")
@pytest.mark.parametrize(("response", "requested", "configured", "expected"), [
pytest.param("gpt-5.5", "req", "def", ("gpt-5.5", ""), id="response"),
pytest.param("", "req-model", "def", ("req-model", "request"), id="request"),
pytest.param(" ", "req-model", "def", ("req-model", "request"), id="blank-response"),
pytest.param("", "", "gpt-5.5", ("gpt-5.5", "configured_default"), id="configured"),
pytest.param(None, None, "gpt-5.5", ("gpt-5.5", "configured_default"), id="unset-request"),
pytest.param("", "", "", ("", "none"), id="empty"),
pytest.param(None, None, None, ("", "none"), id="unset"),
])
def test_resolve_pricing_model_precedence(response, requested, configured, expected) -> None:
assert resolve_pricing_model(response, requested, configured) == expected


def test_configured_pricing_model_is_codex_only(
tmp_path,
codex_config_dir,
monkeypatch: pytest.MonkeyPatch,
) -> None:
codex_home = tmp_path / "codex"
codex_home.mkdir()
(codex_home / "config.toml").write_text(
(codex_config_dir / "config.toml").write_text(
'model = "gpt-5.6-sol"\n',
encoding="utf-8",
)
monkeypatch.setenv("CODEX_HOME", str(codex_home))
monkeypatch.setenv("ARGUS_SKILL_MODEL", "claude-sonnet-5")
backend = AgentCliBackend(backend="codex")
assert backend._configured_pricing_model() == "gpt-5.6-sol"
Expand All @@ -102,16 +93,12 @@ def test_codex_execution_model_honors_final_cli_override() -> None:


def test_codex_model_args_normalize_to_one_direct_flag(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
codex_config_dir,
) -> None:
codex_home = tmp_path / "codex"
codex_home.mkdir()
(codex_home / "config.toml").write_text(
(codex_config_dir / "config.toml").write_text(
'model = "gpt-5.6-sol"\n',
encoding="utf-8",
)
monkeypatch.setenv("CODEX_HOME", str(codex_home))
backend = AgentCliBackend(
backend="codex",
default_extra_args=["--model", "gpt-5.5"],
Expand All @@ -131,20 +118,16 @@ def test_codex_model_args_normalize_to_one_direct_flag(


def test_codex_profile_model_is_pinned_without_dropping_profile(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
codex_config_dir,
) -> None:
codex_home = tmp_path / "codex"
codex_home.mkdir()
(codex_home / "config.toml").write_text(
(codex_config_dir / "config.toml").write_text(
'model = "gpt-5.5"\n',
encoding="utf-8",
)
(codex_home / "research.config.toml").write_text(
(codex_config_dir / "research.config.toml").write_text(
'model = "gpt-5.6-sol"\n',
encoding="utf-8",
)
monkeypatch.setenv("CODEX_HOME", str(codex_home))
backend = AgentCliBackend(
backend="codex",
default_extra_args=["--profile", "research"],
Expand All @@ -161,24 +144,20 @@ def test_codex_profile_model_is_pinned_without_dropping_profile(


def test_call_profile_replaces_default_profile_once(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
codex_config_dir,
) -> None:
codex_home = tmp_path / "codex"
codex_home.mkdir()
(codex_home / "config.toml").write_text(
(codex_config_dir / "config.toml").write_text(
'model = "gpt-5.4"\n',
encoding="utf-8",
)
(codex_home / "default.config.toml").write_text(
(codex_config_dir / "default.config.toml").write_text(
'model = "gpt-5.5"\n',
encoding="utf-8",
)
(codex_home / "call.config.toml").write_text(
(codex_config_dir / "call.config.toml").write_text(
'model = "gpt-5.6-sol"\n',
encoding="utf-8",
)
monkeypatch.setenv("CODEX_HOME", str(codex_home))
backend = AgentCliBackend(
backend="codex",
default_extra_args=["--profile", "default", "--strict-config"],
Expand Down
150 changes: 30 additions & 120 deletions tests/test_dsh_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@

import os
from pathlib import Path
from types import SimpleNamespace

import pytest

Expand All @@ -34,26 +35,6 @@ def _overlay_path() -> Path:
return Path(_dsh_overlay_patch_path())


class _FakeProcess:
def __init__(
self,
*,
returncode: int,
stdout_lines: list[str],
stderr_lines: list[str],
) -> None:
self.returncode = returncode
self.stdout = iter(stdout_lines)
self.stderr = iter(stderr_lines)
self.stdin = None

def poll(self) -> int:
return self.returncode

def wait(self, timeout: float | None = None) -> int: # noqa: ARG002
return self.returncode


# ---------------------------------------------------------------- registration


Expand Down Expand Up @@ -242,112 +223,41 @@ def _state(
return state


def test_dsh_finalize_synthesizes_completion_from_stdout() -> None:
runner = _runner()
state = _state(
stdout_lines=["", " final answer ", ""],
stderr_lines=[],
@pytest.mark.parametrize(("stdout", "stderr", "exit_code", "completed", "messages", "error"), [
pytest.param(["", " final answer ", ""], [], 0, False, ["final answer"], None, id="stdout-completion"),
pytest.param([""], [], 0, False, [], "no assistant output", id="empty-output"),
pytest.param([], ["dsh: MISSING_CREDENTIAL: llm-deepseek: no API key"], 1, False, [], "MISSING_CREDENTIAL", id="nonzero-exit"),
pytest.param(["legacy"], [], 0, True, ["legacy"], None, id="already-completed"),
])
def test_dsh_finalize_terminal_receipts(stdout, stderr, exit_code, completed, messages, error) -> None:
state = _state(stdout_lines=stdout, stderr_lines=stderr)
if completed:
state.turn_completed = True
state.agent_messages = list(messages)
result = _runner()._finalize_turn_result(
process=SimpleNamespace(returncode=exit_code), command=["dsh"], options=RunnerOptions(), state=state,
)
process = _FakeProcess(returncode=0, stdout_lines=[], stderr_lines=[])

result = runner._finalize_turn_result(
process=process,
command=["dsh"],
options=RunnerOptions(),
state=state,
)

assert result.turn_completed is True
assert result.turn_failed is False
assert result.agent_messages == ["final answer"]
assert result.turn_completed is (error is None)
assert result.turn_failed is (error is not None)
assert result.agent_messages == messages
assert result.thread_id is None
if error:
assert error in (result.fatal_error or "")
else:
assert result.fatal_error is None


def test_dsh_finalize_empty_output_fails_closed() -> None:
runner = _runner()
state = _state(stdout_lines=[""], stderr_lines=[])
process = _FakeProcess(returncode=0, stdout_lines=[], stderr_lines=[])

result = runner._finalize_turn_result(
process=process,
command=["dsh"],
options=RunnerOptions(),
state=state,
)

assert result.turn_completed is False
assert result.turn_failed is True
assert "no assistant output" in (result.fatal_error or "")


def test_dsh_finalize_nonzero_exit_fails_closed_with_stderr() -> None:
runner = _runner()
state = _state(
stdout_lines=[],
stderr_lines=["dsh: MISSING_CREDENTIAL: llm-deepseek: no API key"],
)
process = _FakeProcess(returncode=1, stdout_lines=[], stderr_lines=[])

result = runner._finalize_turn_result(
process=process,
command=["dsh"],
options=RunnerOptions(),
state=state,
)

assert result.turn_completed is False
assert result.turn_failed is True
assert "MISSING_CREDENTIAL" in (result.fatal_error or "")


def test_dsh_readiness_accepts_key_from_dsh_home_env_file(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
@pytest.mark.parametrize("has_key", [True, False], ids=["home-env-key", "missing-key"])
def test_dsh_readiness_from_home_env_file(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, has_key) -> None:
from argus.core.backend_readiness import _probe_cli_auth

monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False)
monkeypatch.setenv("DSH_HOME", str(tmp_path))
(tmp_path / ".env").write_text(
"# comment\nDEEPSEEK_API_KEY=sk-from-env-file\n",
encoding="utf-8",
)

if has_key:
(tmp_path / ".env").write_text("# comment\nDEEPSEEK_API_KEY=sk-from-env-file\n", encoding="utf-8")
ready, detail = _probe_cli_auth("dsh", "dsh", timeout_s=5.0)

assert ready is True
assert detail == ""


def test_dsh_readiness_rejects_without_any_key(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
from argus.core.backend_readiness import _probe_cli_auth

monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False)
monkeypatch.setenv("DSH_HOME", str(tmp_path))

ready, detail = _probe_cli_auth("dsh", "dsh", timeout_s=5.0)

assert ready is False
assert "DEEPSEEK_API_KEY" in detail


def test_dsh_finalize_does_not_touch_completed_state() -> None:
"""A turn that already completed through the normal path is left alone."""
runner = _runner()
state = _state(stdout_lines=["legacy"], stderr_lines=[])
state.turn_completed = True
state.agent_messages = ["legacy"]
process = _FakeProcess(returncode=0, stdout_lines=[], stderr_lines=[])

result = runner._finalize_turn_result(
process=process,
command=["dsh"],
options=RunnerOptions(),
state=state,
)

assert result.turn_completed is True
assert result.agent_messages == ["legacy"]
assert ready is has_key
if has_key:
assert detail == ""
else:
assert "DEEPSEEK_API_KEY" in detail
Loading
Loading