refactor(tests): re-architect tests + fix CI failures (#5946)
* refactor: re-architect tests to mirror the codebase
* Update tests.yml
* fix: add missing tool_error imports after registry refactor
* fix(tests): replace patch.dict with monkeypatch to prevent env var leaks under xdist
patch.dict(os.environ) can leak TERMINAL_ENV across xdist workers,
causing test_code_execution tests to hit the Modal remote path.
* fix(tests): fix update_check and telegram xdist failures
- test_update_check: replace patch("hermes_cli.banner.os.getenv") with
monkeypatch.setenv("HERMES_HOME") — banner.py no longer imports os
directly, it uses get_hermes_home() from hermes_constants.
- test_telegram_conflict/approval_buttons: provide real exception classes
for telegram.error mock (NetworkError, TimedOut, BadRequest) so the
except clause in connect() doesn't fail with "catching classes that do
not inherit from BaseException" when xdist pollutes sys.modules.
* fix(tests): accept unavailable_models kwarg in _prompt_model_selection mock
This commit is contained in:
Siddharth Balyan
194
tests/cron/test_codex_execution_paths.py
Normal file
194
tests/cron/test_codex_execution_paths.py
Normal file
@@ -0,0 +1,194 @@
|
||||
import asyncio
|
||||
import sys
|
||||
import types
|
||||
from types import SimpleNamespace
|
||||
|
||||
|
||||
sys.modules.setdefault("fire", types.SimpleNamespace(Fire=lambda *a, **k: None))
|
||||
sys.modules.setdefault("firecrawl", types.SimpleNamespace(Firecrawl=object))
|
||||
sys.modules.setdefault("fal_client", types.SimpleNamespace())
|
||||
|
||||
import cron.scheduler as cron_scheduler
|
||||
import gateway.run as gateway_run
|
||||
import run_agent
|
||||
from gateway.config import Platform
|
||||
from gateway.session import SessionSource
|
||||
|
||||
|
||||
def _patch_agent_bootstrap(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
run_agent,
|
||||
"get_tool_definitions",
|
||||
lambda **kwargs: [
|
||||
{
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "terminal",
|
||||
"description": "Run shell commands.",
|
||||
"parameters": {"type": "object", "properties": {}},
|
||||
},
|
||||
}
|
||||
],
|
||||
)
|
||||
monkeypatch.setattr(run_agent, "check_toolset_requirements", lambda: {})
|
||||
|
||||
|
||||
def _codex_message_response(text: str):
|
||||
return SimpleNamespace(
|
||||
output=[
|
||||
SimpleNamespace(
|
||||
type="message",
|
||||
content=[SimpleNamespace(type="output_text", text=text)],
|
||||
)
|
||||
],
|
||||
usage=SimpleNamespace(input_tokens=5, output_tokens=3, total_tokens=8),
|
||||
status="completed",
|
||||
model="gpt-5-codex",
|
||||
)
|
||||
|
||||
|
||||
class _UnauthorizedError(RuntimeError):
|
||||
def __init__(self):
|
||||
super().__init__("Error code: 401 - unauthorized")
|
||||
self.status_code = 401
|
||||
|
||||
|
||||
class _FakeOpenAI:
|
||||
def __init__(self, **kwargs):
|
||||
self.kwargs = kwargs
|
||||
|
||||
def close(self):
|
||||
return None
|
||||
|
||||
|
||||
class _Codex401ThenSuccessAgent(run_agent.AIAgent):
|
||||
refresh_attempts = 0
|
||||
last_init = {}
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
kwargs.setdefault("skip_context_files", True)
|
||||
kwargs.setdefault("skip_memory", True)
|
||||
kwargs.setdefault("max_iterations", 4)
|
||||
type(self).last_init = dict(kwargs)
|
||||
super().__init__(*args, **kwargs)
|
||||
self._cleanup_task_resources = lambda task_id: None
|
||||
self._persist_session = lambda messages, history=None: None
|
||||
self._save_trajectory = lambda messages, user_message, completed: None
|
||||
self._save_session_log = lambda messages: None
|
||||
|
||||
def _try_refresh_codex_client_credentials(self, *, force: bool = True) -> bool:
|
||||
type(self).refresh_attempts += 1
|
||||
return True
|
||||
|
||||
def run_conversation(self, user_message: str, conversation_history=None, task_id=None):
|
||||
calls = {"api": 0}
|
||||
|
||||
def _fake_api_call(api_kwargs):
|
||||
calls["api"] += 1
|
||||
if calls["api"] == 1:
|
||||
raise _UnauthorizedError()
|
||||
return _codex_message_response("Recovered via refresh")
|
||||
|
||||
self._interruptible_api_call = _fake_api_call
|
||||
return super().run_conversation(user_message, conversation_history=conversation_history, task_id=task_id)
|
||||
|
||||
|
||||
def test_cron_run_job_codex_path_handles_internal_401_refresh(monkeypatch):
|
||||
_patch_agent_bootstrap(monkeypatch)
|
||||
monkeypatch.setattr(run_agent, "OpenAI", _FakeOpenAI)
|
||||
monkeypatch.setattr(run_agent, "AIAgent", _Codex401ThenSuccessAgent)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.runtime_provider.resolve_runtime_provider",
|
||||
lambda requested=None: {
|
||||
"provider": "openai-codex",
|
||||
"api_mode": "codex_responses",
|
||||
"base_url": "https://chatgpt.com/backend-api/codex",
|
||||
"api_key": "codex-token",
|
||||
},
|
||||
)
|
||||
monkeypatch.setattr("hermes_cli.runtime_provider.format_runtime_provider_error", lambda exc: str(exc))
|
||||
|
||||
_Codex401ThenSuccessAgent.refresh_attempts = 0
|
||||
_Codex401ThenSuccessAgent.last_init = {}
|
||||
|
||||
success, output, final_response, error = cron_scheduler.run_job(
|
||||
{"id": "job-1", "name": "Codex Refresh Test", "prompt": "ping", "model": "gpt-5.3-codex"}
|
||||
)
|
||||
|
||||
assert success is True
|
||||
assert error is None
|
||||
assert final_response == "Recovered via refresh"
|
||||
assert "Recovered via refresh" in output
|
||||
assert _Codex401ThenSuccessAgent.refresh_attempts == 1
|
||||
assert _Codex401ThenSuccessAgent.last_init["provider"] == "openai-codex"
|
||||
assert _Codex401ThenSuccessAgent.last_init["api_mode"] == "codex_responses"
|
||||
|
||||
|
||||
def test_gateway_run_agent_codex_path_handles_internal_401_refresh(monkeypatch):
|
||||
_patch_agent_bootstrap(monkeypatch)
|
||||
monkeypatch.setattr(run_agent, "OpenAI", _FakeOpenAI)
|
||||
monkeypatch.setattr(run_agent, "AIAgent", _Codex401ThenSuccessAgent)
|
||||
monkeypatch.setattr(
|
||||
gateway_run,
|
||||
"_resolve_runtime_agent_kwargs",
|
||||
lambda: {
|
||||
"provider": "openai-codex",
|
||||
"api_mode": "codex_responses",
|
||||
"base_url": "https://chatgpt.com/backend-api/codex",
|
||||
"api_key": "codex-token",
|
||||
},
|
||||
)
|
||||
monkeypatch.setenv("HERMES_TOOL_PROGRESS", "false")
|
||||
monkeypatch.setenv("HERMES_MODEL", "gpt-5.3-codex")
|
||||
|
||||
_Codex401ThenSuccessAgent.refresh_attempts = 0
|
||||
_Codex401ThenSuccessAgent.last_init = {}
|
||||
|
||||
runner = gateway_run.GatewayRunner.__new__(gateway_run.GatewayRunner)
|
||||
runner.adapters = {}
|
||||
runner._ephemeral_system_prompt = ""
|
||||
runner._prefill_messages = []
|
||||
runner._reasoning_config = None
|
||||
runner._provider_routing = {}
|
||||
runner._fallback_model = None
|
||||
runner._running_agents = {}
|
||||
runner._smart_model_routing = {}
|
||||
from unittest.mock import MagicMock, AsyncMock
|
||||
runner.hooks = MagicMock()
|
||||
runner.hooks.emit = AsyncMock()
|
||||
runner.hooks.loaded_hooks = []
|
||||
runner._session_db = None
|
||||
# Ensure model resolution returns the codex model even if xdist
|
||||
# leaked env vars cleared HERMES_MODEL.
|
||||
monkeypatch.setattr(
|
||||
gateway_run.GatewayRunner,
|
||||
"_resolve_turn_agent_config",
|
||||
lambda self, msg, model, runtime: {
|
||||
"model": model or "gpt-5.3-codex",
|
||||
"runtime": runtime,
|
||||
},
|
||||
)
|
||||
|
||||
source = SessionSource(
|
||||
platform=Platform.LOCAL,
|
||||
chat_id="cli",
|
||||
chat_name="CLI",
|
||||
chat_type="dm",
|
||||
user_id="user-1",
|
||||
)
|
||||
|
||||
result = asyncio.run(
|
||||
runner._run_agent(
|
||||
message="ping",
|
||||
context_prompt="",
|
||||
history=[],
|
||||
source=source,
|
||||
session_id="session-1",
|
||||
session_key="agent:main:local:dm",
|
||||
)
|
||||
)
|
||||
|
||||
assert result["final_response"] == "Recovered via refresh"
|
||||
assert _Codex401ThenSuccessAgent.refresh_attempts == 1
|
||||
assert _Codex401ThenSuccessAgent.last_init["provider"] == "openai-codex"
|
||||
assert _Codex401ThenSuccessAgent.last_init["api_mode"] == "codex_responses"
|
||||
135
tests/cron/test_file_permissions.py
Normal file
135
tests/cron/test_file_permissions.py
Normal file
@@ -0,0 +1,135 @@
|
||||
"""Tests for file permissions hardening on sensitive files."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
class TestCronFilePermissions(unittest.TestCase):
|
||||
"""Verify cron files get secure permissions."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmpdir = tempfile.mkdtemp()
|
||||
self.cron_dir = Path(self.tmpdir) / "cron"
|
||||
self.output_dir = self.cron_dir / "output"
|
||||
|
||||
def tearDown(self):
|
||||
import shutil
|
||||
shutil.rmtree(self.tmpdir, ignore_errors=True)
|
||||
|
||||
@patch("cron.jobs.CRON_DIR")
|
||||
@patch("cron.jobs.OUTPUT_DIR")
|
||||
@patch("cron.jobs.JOBS_FILE")
|
||||
def test_ensure_dirs_sets_0700(self, mock_jobs_file, mock_output, mock_cron):
|
||||
mock_cron.__class__ = Path
|
||||
# Use real paths
|
||||
cron_dir = Path(self.tmpdir) / "cron"
|
||||
output_dir = cron_dir / "output"
|
||||
|
||||
with patch("cron.jobs.CRON_DIR", cron_dir), \
|
||||
patch("cron.jobs.OUTPUT_DIR", output_dir):
|
||||
from cron.jobs import ensure_dirs
|
||||
ensure_dirs()
|
||||
|
||||
cron_mode = stat.S_IMODE(os.stat(cron_dir).st_mode)
|
||||
output_mode = stat.S_IMODE(os.stat(output_dir).st_mode)
|
||||
self.assertEqual(cron_mode, 0o700)
|
||||
self.assertEqual(output_mode, 0o700)
|
||||
|
||||
@patch("cron.jobs.CRON_DIR")
|
||||
@patch("cron.jobs.OUTPUT_DIR")
|
||||
@patch("cron.jobs.JOBS_FILE")
|
||||
def test_save_jobs_sets_0600(self, mock_jobs_file, mock_output, mock_cron):
|
||||
cron_dir = Path(self.tmpdir) / "cron"
|
||||
output_dir = cron_dir / "output"
|
||||
jobs_file = cron_dir / "jobs.json"
|
||||
|
||||
with patch("cron.jobs.CRON_DIR", cron_dir), \
|
||||
patch("cron.jobs.OUTPUT_DIR", output_dir), \
|
||||
patch("cron.jobs.JOBS_FILE", jobs_file):
|
||||
from cron.jobs import save_jobs
|
||||
save_jobs([{"id": "test", "prompt": "hello"}])
|
||||
|
||||
file_mode = stat.S_IMODE(os.stat(jobs_file).st_mode)
|
||||
self.assertEqual(file_mode, 0o600)
|
||||
|
||||
def test_save_job_output_sets_0600(self):
|
||||
output_dir = Path(self.tmpdir) / "output"
|
||||
with patch("cron.jobs.OUTPUT_DIR", output_dir), \
|
||||
patch("cron.jobs.CRON_DIR", Path(self.tmpdir)), \
|
||||
patch("cron.jobs.ensure_dirs"):
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
from cron.jobs import save_job_output
|
||||
output_file = save_job_output("test-job", "test output content")
|
||||
|
||||
file_mode = stat.S_IMODE(os.stat(output_file).st_mode)
|
||||
self.assertEqual(file_mode, 0o600)
|
||||
|
||||
# Job output dir should also be 0700
|
||||
job_dir = output_dir / "test-job"
|
||||
dir_mode = stat.S_IMODE(os.stat(job_dir).st_mode)
|
||||
self.assertEqual(dir_mode, 0o700)
|
||||
|
||||
|
||||
class TestConfigFilePermissions(unittest.TestCase):
|
||||
"""Verify config files get secure permissions."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmpdir = tempfile.mkdtemp()
|
||||
|
||||
def tearDown(self):
|
||||
import shutil
|
||||
shutil.rmtree(self.tmpdir, ignore_errors=True)
|
||||
|
||||
def test_save_config_sets_0600(self):
|
||||
config_path = Path(self.tmpdir) / "config.yaml"
|
||||
with patch("hermes_cli.config.get_config_path", return_value=config_path), \
|
||||
patch("hermes_cli.config.ensure_hermes_home"):
|
||||
from hermes_cli.config import save_config
|
||||
save_config({"model": "test/model"})
|
||||
|
||||
file_mode = stat.S_IMODE(os.stat(config_path).st_mode)
|
||||
self.assertEqual(file_mode, 0o600)
|
||||
|
||||
def test_save_env_value_sets_0600(self):
|
||||
env_path = Path(self.tmpdir) / ".env"
|
||||
with patch("hermes_cli.config.get_env_path", return_value=env_path), \
|
||||
patch("hermes_cli.config.ensure_hermes_home"):
|
||||
from hermes_cli.config import save_env_value
|
||||
save_env_value("TEST_KEY", "test_value")
|
||||
|
||||
file_mode = stat.S_IMODE(os.stat(env_path).st_mode)
|
||||
self.assertEqual(file_mode, 0o600)
|
||||
|
||||
def test_ensure_hermes_home_sets_0700(self):
|
||||
home = Path(self.tmpdir) / ".hermes"
|
||||
with patch("hermes_cli.config.get_hermes_home", return_value=home):
|
||||
from hermes_cli.config import ensure_hermes_home
|
||||
ensure_hermes_home()
|
||||
|
||||
home_mode = stat.S_IMODE(os.stat(home).st_mode)
|
||||
self.assertEqual(home_mode, 0o700)
|
||||
|
||||
for subdir in ("cron", "sessions", "logs", "memories"):
|
||||
subdir_mode = stat.S_IMODE(os.stat(home / subdir).st_mode)
|
||||
self.assertEqual(subdir_mode, 0o700, f"{subdir} should be 0700")
|
||||
|
||||
|
||||
class TestSecureHelpers(unittest.TestCase):
|
||||
"""Test the _secure_file and _secure_dir helpers."""
|
||||
|
||||
def test_secure_file_nonexistent_no_error(self):
|
||||
from cron.jobs import _secure_file
|
||||
_secure_file(Path("/nonexistent/path/file.json")) # Should not raise
|
||||
|
||||
def test_secure_dir_nonexistent_no_error(self):
|
||||
from cron.jobs import _secure_dir
|
||||
_secure_dir(Path("/nonexistent/path")) # Should not raise
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user