diff --git a/tests/life/test_lifecycle_supervisor_integration.py b/tests/life/test_lifecycle_supervisor_integration.py index 370820249..a2107130d 100644 --- a/tests/life/test_lifecycle_supervisor_integration.py +++ b/tests/life/test_lifecycle_supervisor_integration.py @@ -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. diff --git a/tests/test_codex_model_pricing_fallback.py b/tests/test_codex_model_pricing_fallback.py index 4e67195b3..3ea87b1f5 100644 --- a/tests/test_codex_model_pricing_fallback.py +++ b/tests/test_codex_model_pricing_fallback.py @@ -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" @@ -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"], @@ -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"], @@ -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"], diff --git a/tests/test_dsh_backend.py b/tests/test_dsh_backend.py index c3dd620f3..5d878ab56 100644 --- a/tests/test_dsh_backend.py +++ b/tests/test_dsh_backend.py @@ -10,6 +10,7 @@ import os from pathlib import Path +from types import SimpleNamespace import pytest @@ -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 @@ -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 diff --git a/tests/test_grok_backend.py b/tests/test_grok_backend.py index 38260d1ed..6d5a040ca 100644 --- a/tests/test_grok_backend.py +++ b/tests/test_grok_backend.py @@ -86,96 +86,35 @@ def test_grok_prompt_uses_private_temporary_file() -> None: def test_grok_messages_stream_tracks_session_text_and_completion() -> None: runner = _runner() messages: list[str] = [] - state = runner._consume_event( - event={ - "type": "system", - "subtype": "init", - "session_id": "grok-session", - "model": "grok-build", - }, - thread_id=None, - agent_messages=messages, - turn_completed=False, - turn_failed=False, - fatal_error=None, - ) - state = runner._consume_event( - event={ - "type": "assistant", - "session_id": "grok-session", - "message": { - "role": "assistant", - "content": [{"type": "text", "text": "done"}], - }, - }, - thread_id=state[0], - agent_messages=messages, - turn_completed=state[1], - turn_failed=state[2], - fatal_error=state[3], - ) - state = runner._consume_event( - event={ - "type": "result", - "subtype": "success", - "is_error": False, - "session_id": "grok-session", - "result": "done", - "stop_reason": "end_turn", - }, - thread_id=state[0], - agent_messages=messages, - turn_completed=state[1], - turn_failed=state[2], - fatal_error=state[3], - ) - + state = (None, False, False, None) + events = [ + {"type": "system", "subtype": "init", "session_id": "grok-session", "model": "grok-build"}, + {"type": "assistant", "session_id": "grok-session", "message": { + "role": "assistant", "content": [{"type": "text", "text": "done"}]}}, + {"type": "result", "subtype": "success", "is_error": False, + "session_id": "grok-session", "result": "done", "stop_reason": "end_turn"}, + ] + for event in events: + state = runner._consume_event( + event=event, thread_id=state[0], agent_messages=messages, + turn_completed=state[1], turn_failed=state[2], fatal_error=state[3], + ) assert messages == ["done"] assert state == ("grok-session", True, False, None) -def test_grok_error_result_fails_closed() -> None: +@pytest.mark.parametrize(("fields", "error"), [ + pytest.param({"subtype": "error_during_execution", "is_error": True, + "result": "authentication failed"}, "authentication failed", id="provider-error"), + pytest.param({"subtype": "success", "is_error": False, "result": "partial", + "stop_reason": "max_tokens"}, "Grok Build stopped with max_tokens.", id="nonterminal-stop"), +]) +def test_grok_unsuccessful_result_fails_closed(fields, error) -> None: state = _runner()._consume_event( - event={ - "type": "result", - "subtype": "error_during_execution", - "is_error": True, - "session_id": "grok-session", - "result": "authentication failed", - }, - thread_id=None, - agent_messages=[], - turn_completed=False, - turn_failed=False, - fatal_error=None, - ) - - assert state == ("grok-session", False, True, "authentication failed") - - -def test_grok_non_terminal_stop_reason_fails_closed() -> None: - state = _runner()._consume_event( - event={ - "type": "result", - "subtype": "success", - "is_error": False, - "session_id": "grok-session", - "result": "partial", - "stop_reason": "max_tokens", - }, - thread_id=None, - agent_messages=[], - turn_completed=False, - turn_failed=False, - fatal_error=None, - ) - - assert state == ( - "grok-session", - False, - True, - "Grok Build stopped with max_tokens.", + event={"type": "result", "session_id": "grok-session", **fields}, + thread_id=None, agent_messages=[], turn_completed=False, turn_failed=False, fatal_error=None, ) + assert state == ("grok-session", False, True, error) def test_grok_result_usage_is_accounted() -> None: diff --git a/tests/test_rename_compatibility.py b/tests/test_rename_compatibility.py index dff179232..6d043c4d3 100644 --- a/tests/test_rename_compatibility.py +++ b/tests/test_rename_compatibility.py @@ -32,13 +32,6 @@ def _subprocess_env() -> dict[str, str]: return env -def test_the_alias_package_is_exactly_two_files() -> None: - shim = REPO_ROOT / "argus_skill" - assert sorted(p.name for p in shim.iterdir() if p.name != "__pycache__") == [ - "__init__.py", "__main__.py", - ] - - def test_legacy_import_name_is_the_same_module_object() -> None: import argus_skill import argus_skill.core.paths as legacy_paths diff --git a/tests/test_runner_binary_resolution.py b/tests/test_runner_binary_resolution.py index 58e3b2a8c..1b9a65da3 100644 --- a/tests/test_runner_binary_resolution.py +++ b/tests/test_runner_binary_resolution.py @@ -80,35 +80,23 @@ def test_persisted_runner_bin_stays_bound_to_its_backend() -> None: ) -def test_runner_resolves_user_local_bin_when_service_path_omits_it( - tmp_path: Path, - monkeypatch, -) -> None: - executable = tmp_path / ".local" / "bin" / "copilot" - executable.parent.mkdir(parents=True) - executable = _write_runner_executable(executable) +@pytest.mark.parametrize(("backend", "binary", "install_dir"), [ + pytest.param(BACKEND_COPILOT, "copilot", ".local/bin", id="copilot-user-local"), + pytest.param(BACKEND_OPENCODE, "opencode", ".opencode/bin", id="opencode-standard-install"), + pytest.param(BACKEND_OPENCODE, "opencode", "", id="opencode-path"), + pytest.param(BACKEND_PI, "pi", "", id="pi-path"), + pytest.param(BACKEND_GROK, "grok", "", id="grok-path"), + pytest.param(BACKEND_QODER, "qodercli", "", id="qoder-path"), +]) +def test_backend_binary_resolution(tmp_path: Path, monkeypatch, backend, binary, install_dir) -> None: + directory = tmp_path / install_dir + directory.mkdir(parents=True, exist_ok=True) + executable = _write_runner_executable(directory / binary) monkeypatch.setenv("HOME", str(tmp_path)) monkeypatch.setenv("USERPROFILE", str(tmp_path)) - monkeypatch.setenv("PATH", str(tmp_path / "service-bin")) - - _assert_same_path(resolve_runner_bin(BACKEND_COPILOT), executable) - _assert_same_path(AgentCliRunner(backend=BACKEND_COPILOT).agent_bin, executable) - - -def test_opencode_runner_uses_opencode_binary(tmp_path: Path, monkeypatch) -> None: - executable = _write_runner_executable(tmp_path / "opencode") - monkeypatch.setenv("PATH", str(tmp_path)) - - _assert_same_path(resolve_runner_bin(BACKEND_OPENCODE), executable) - _assert_same_path(AgentCliRunner(backend=BACKEND_OPENCODE).agent_bin, executable) - - -def test_pi_runner_uses_pi_binary(tmp_path: Path, monkeypatch) -> None: - executable = _write_runner_executable(tmp_path / "pi") - monkeypatch.setenv("PATH", str(tmp_path)) - - _assert_same_path(resolve_runner_bin(BACKEND_PI), executable) - _assert_same_path(AgentCliRunner(backend=BACKEND_PI).agent_bin, executable) + monkeypatch.setenv("PATH", str(tmp_path / "service-bin" if install_dir else tmp_path)) + _assert_same_path(resolve_runner_bin(backend), executable) + _assert_same_path(AgentCliRunner(backend=backend).agent_bin, executable) @pytest.mark.parametrize("suffix", [".CMD", ".EXE"]) @@ -153,22 +141,6 @@ def resolve_candidate(path: Path) -> str | None: assert inspected == [expected] -def test_grok_runner_uses_grok_binary(tmp_path: Path, monkeypatch) -> None: - executable = _write_runner_executable(tmp_path / "grok") - monkeypatch.setenv("PATH", str(tmp_path)) - - _assert_same_path(resolve_runner_bin(BACKEND_GROK), executable) - _assert_same_path(AgentCliRunner(backend=BACKEND_GROK).agent_bin, executable) - - -def test_qoder_runner_uses_qodercli_binary(tmp_path: Path, monkeypatch) -> None: - executable = _write_runner_executable(tmp_path / "qodercli") - monkeypatch.setenv("PATH", str(tmp_path)) - - _assert_same_path(resolve_runner_bin(BACKEND_QODER), executable) - _assert_same_path(AgentCliRunner(backend=BACKEND_QODER).agent_bin, executable) - - def test_runner_skips_inaccessible_path_candidate( tmp_path: Path, monkeypatch, @@ -248,21 +220,6 @@ def inspect(path: Path) -> bool: assert inspected == [candidate, matching_directory] -def test_opencode_runner_resolves_standard_install_directory( - tmp_path: Path, - monkeypatch, -) -> None: - executable = tmp_path / ".opencode" / "bin" / "opencode" - executable.parent.mkdir(parents=True) - executable = _write_runner_executable(executable) - monkeypatch.setenv("HOME", str(tmp_path)) - monkeypatch.setenv("USERPROFILE", str(tmp_path)) - monkeypatch.setenv("PATH", str(tmp_path / "service-bin")) - - _assert_same_path(resolve_runner_bin(BACKEND_OPENCODE), executable) - _assert_same_path(AgentCliRunner(backend=BACKEND_OPENCODE).agent_bin, executable) - - def test_missing_codex_falls_back_to_available_copilot( tmp_path: Path, monkeypatch,