Files
2026-05-20 04:51:41 +00:00

676 lines
29 KiB
Python

"""
Shared pytest fixtures for webui-mvp tests.
TEST ISOLATION:
Tests run against a SEPARATE server instance on an auto-derived test port
with a completely separate state directory. Production data is never touched.
The test state dir is wiped before each full test run and again on teardown.
PATH DISCOVERY:
No hardcoded paths. Discovery order:
1. Environment variables (HERMES_WEBUI_AGENT_DIR, HERMES_WEBUI_PYTHON, etc.)
2. Sibling checkout heuristics relative to this repo
3. Common install paths (~/.hermes/hermes-agent)
4. System python3 as a last resort
"""
import json
import os
import pathlib
import shutil
import subprocess
import time
import urllib.request
import urllib.error
import pytest
# ── Repo root discovery ────────────────────────────────────────────────────
# conftest.py lives at <repo>/tests/conftest.py
TESTS_DIR = pathlib.Path(__file__).parent.resolve()
REPO_ROOT = TESTS_DIR.parent.resolve()
HOME = pathlib.Path.home()
HERMES_HOME = pathlib.Path(os.getenv('HERMES_HOME', str(HOME / '.hermes')))
# ── Test server config ────────────────────────────────────────────────────
# Port and state dir auto-derive from the repo path when no env var is set,
# giving every worktree its own isolated port (20000-29999) and state directory.
# Override with HERMES_WEBUI_TEST_PORT / HERMES_WEBUI_TEST_STATE_DIR to pin.
def _auto_test_port(repo_root) -> int:
"""Map repo path to a unique port in 20000-29999 (10k range = near-zero collisions).
Far from system port ranges and Linux ephemeral ports (32768+).
Override with HERMES_WEBUI_TEST_PORT to use a specific port."""
import hashlib
h = int(hashlib.md5(str(repo_root).encode()).hexdigest(), 16)
return 20000 + (h % 10000)
def _auto_state_dir_name(repo_root) -> str:
import hashlib
h = hashlib.md5(str(repo_root).encode()).hexdigest()[:8]
return f"webui-test-{h}"
TEST_PORT = int(os.getenv('HERMES_WEBUI_TEST_PORT',
str(_auto_test_port(REPO_ROOT))))
TEST_BASE = f"http://127.0.0.1:{TEST_PORT}"
TEST_STATE_DIR = pathlib.Path(os.getenv(
'HERMES_WEBUI_TEST_STATE_DIR',
str(HERMES_HOME / _auto_state_dir_name(REPO_ROOT))
))
TEST_WORKSPACE = TEST_STATE_DIR / 'test-workspace'
# Publish at module level so api.config, _pytest_port.py, and any test module
# importing stateful API code during collection see the isolated test paths.
#
# Direct assignment is intentional for production-risk paths: tests that import
# api.config/api.models in the pytest process must never inherit the real
# ~/.hermes state tree before the server subprocess fixture starts.
os.environ['HERMES_WEBUI_TEST_PORT'] = str(TEST_PORT)
os.environ['HERMES_WEBUI_TEST_STATE_DIR'] = str(TEST_STATE_DIR)
os.environ['HERMES_WEBUI_STATE_DIR'] = str(TEST_STATE_DIR)
os.environ['HERMES_WEBUI_DEFAULT_WORKSPACE'] = str(TEST_WORKSPACE)
os.environ['HERMES_HOME'] = str(TEST_STATE_DIR)
os.environ['HERMES_BASE_HOME'] = str(TEST_STATE_DIR)
# Hermes Agent sessions may inherit HERMES_CONFIG_PATH pointing at the live
# ~/.hermes/config.yaml. Override it before any product modules are imported so
# tests that read/write config.yaml stay inside the isolated test home.
os.environ['HERMES_CONFIG_PATH'] = str(TEST_STATE_DIR / 'config.yaml')
# ── Server script: always relative to repo root ───────────────────────────
SERVER_SCRIPT = REPO_ROOT / 'server.py'
if not SERVER_SCRIPT.exists():
raise RuntimeError(
f"server.py not found at {SERVER_SCRIPT}. "
"Is conftest.py in the tests/ subdirectory of the repo?"
)
# ── Hermes agent discovery (mirrors api/config._discover_agent_dir) ───────
def _discover_agent_dir() -> pathlib.Path:
candidates = [
os.getenv('HERMES_WEBUI_AGENT_DIR', ''),
str(HERMES_HOME / 'hermes-agent'),
str(REPO_ROOT.parent / 'hermes-agent'),
str(HOME / '.hermes' / 'hermes-agent'),
str(HOME / 'hermes-agent'),
]
for c in candidates:
if not c:
continue
p = pathlib.Path(c).expanduser()
if p.exists() and (p / 'run_agent.py').exists():
return p.resolve()
return None
# ── Python discovery (mirrors api/config._discover_python) ────────────────
def _discover_python(agent_dir) -> str:
if os.getenv('HERMES_WEBUI_PYTHON'):
return os.getenv('HERMES_WEBUI_PYTHON')
if agent_dir:
venv_py = agent_dir / 'venv' / 'bin' / 'python'
if venv_py.exists():
return str(venv_py)
local_venv = REPO_ROOT / '.venv' / 'bin' / 'python'
if local_venv.exists():
return str(local_venv)
return shutil.which('python3') or shutil.which('python') or 'python3'
HERMES_AGENT = _discover_agent_dir()
VENV_PYTHON = _discover_python(HERMES_AGENT)
# Work dir: agent dir if found, else repo root
WORKDIR = str(HERMES_AGENT) if HERMES_AGENT else str(REPO_ROOT)
# ── Agent availability detection ─────────────────────────────────────────────
# Tests that require hermes-agent modules (cron, skills, approval, chat/stream)
# are skipped when the agent isn't installed, instead of failing with 500 errors.
AGENT_AVAILABLE = HERMES_AGENT is not None
def _check_agent_modules():
"""Verify hermes-agent Python modules are actually importable."""
if not HERMES_AGENT:
return False
try:
import importlib
# These are the modules that cause 500 errors when missing
for mod in ['cron.jobs', 'tools.skills_tool']:
importlib.import_module(mod)
return True
except (ImportError, ModuleNotFoundError):
return False
AGENT_MODULES_AVAILABLE = _check_agent_modules()
# pytest marker: skip tests that need hermes-agent when it's not present
requires_agent = pytest.mark.skipif(
not AGENT_AVAILABLE,
reason="hermes-agent not found (skipping agent-dependent test)"
)
requires_agent_modules = pytest.mark.skipif(
not AGENT_MODULES_AVAILABLE,
reason="hermes-agent Python modules not importable (cron, skills_tool)"
)
def pytest_configure(config):
config.addinivalue_line("markers", "requires_agent: skip when hermes-agent dir is not found")
config.addinivalue_line("markers", "requires_agent_modules: skip when hermes-agent Python modules are not importable")
# ── Disable AWS IMDS probing for the pytest session ────────────────────────
# Background: when hermes-agent's bedrock_adapter / botocore credential chain
# runs during test execution (e.g. provider catalog enumeration triggered by
# api/config.py imports), botocore probes the EC2 Instance Metadata Service at
# 169.254.169.254 looking for an instance role. On VPS hosts where IMDS is
# reachable but rate-limited (HTTP 429) or non-responsive, this dominates wall
# time and turns a 161s test run into 600+s.
#
# Tests have no legitimate reason to call IMDS — the bedrock-related tests use
# explicit mocks or env-var creds. Setting AWS_EC2_METADATA_DISABLED before
# anything imports botocore is the supported way to silence the probe (matches
# the guard the hermes_cli/doctor.py command already uses in its parallel-probe
# block).
#
# Setting this here instead of in a fixture so it lands BEFORE any test-file
# imports trigger botocore initialisation.
os.environ.setdefault("AWS_EC2_METADATA_DISABLED", "true")
# ── Permanent os.execv guard for the pytest session ────────────────────────
# Several tests in tests/test_update_banner_fixes.py exercise
# api.updates._schedule_restart(), which spawns a DAEMON thread that sleeps
# for a short delay and then calls ``os.execv(sys.executable, sys.argv)``.
# Those tests monkeypatch ``os.execv`` to a no-op for the test scope, but
# monkeypatch teardown happens at test exit — if the daemon thread has not
# yet woken up by then (system load, GC pause, _apply_lock contention), the
# real ``os.execv`` is restored before the thread fires it. The daemon then
# REPLACES the pytest process image with a fresh ``pytest tests/ -q ...``
# invocation, looking from the outside like pytest "hangs at 99%" and then
# restarts the entire suite from 0% — a self-perpetuating loop.
#
# Daemon threads cannot be reliably joined from a test fixture (they live in
# ``api.updates`` module scope), so the only safe answer is to render
# ``os.execv`` permanently inert for the pytest session. Production code is
# unaffected because production never imports this conftest.
#
# Tests that need to verify execv WAS called still monkeypatch it themselves
# — their patched version takes precedence over this no-op wrapper for the
# test's lifetime, and the no-op only kicks in after teardown for daemon
# threads that wake up late.
_real_execv = os.execv
def _pytest_session_safe_execv(_exe, _args): # pragma: no cover — never called in prod
# Drop the call on the floor. A late-firing daemon thread from
# _schedule_restart() must not be able to re-exec the pytest process.
return None
os.execv = _pytest_session_safe_execv
# ── Hermetic network isolation ─────────────────────────────────────────────
# Tests must not reach the public internet. Outbound to Anthropic / OpenAI /
# Amazon / OpenRouter / etc. is forbidden by default. The test suite already
# mocks every legitimate outbound (probe_provider_endpoint, get_available_models,
# urlopen calls inside api/config.py), so a real outbound socket is either a
# missing mock, a leaked credential triggering an SDK init, or an unintended
# regression like the one PR #1970 introduced where a new code path bypassed
# an existing mock and tried to hit the real LM Studio host.
#
# This module-level monkey-patch wraps socket.create_connection so any
# non-loopback / non-RFC1918 / non-link-local / non-TEST-NET destination
# raises OSError("hermes test network isolation"). Tests that deliberately
# attempt outbound (only test_dns_resolution_failure today) opt back in
# explicitly via the `allow_outbound_network` fixture below.
#
# Allowed destinations (silent pass-through):
# - 127.0.0.0/8 loopback
# - ::1 IPv6 loopback
# - 192.168.0.0/16 RFC1918 private
# - 10.0.0.0/8 RFC1918 private
# - 172.16.0.0/12 RFC1918 private (16-31)
# - 169.254.0.0/16 link-local (covers IMDS — already separately blocked
# by AWS_EC2_METADATA_DISABLED, but allowed at the socket
# layer because IMDS-using tests mock the response)
# - 203.0.113.0/24 RFC5737 TEST-NET-3 (used as documentation IPs in tests)
# - hostnames `localhost`, `*.local`, `*.test`, `*.example`, `*.example.com`
# `*.example.net`, `*.example.org`, `*.invalid` (RFC2606/6761 reserved)
#
# A test that opts in via the `allow_outbound_network` fixture sees the real
# socket.create_connection.
import socket as _hermes_test_socket
_REAL_CREATE_CONNECTION = _hermes_test_socket.create_connection
_REAL_SOCKET_CONNECT = _hermes_test_socket.socket.connect
def _hermes_addr_is_local(host: str) -> bool:
"""Return True for loopback / RFC1918 / link-local / reserved-TLD hosts."""
if not isinstance(host, str):
return False
h = host.strip().lower()
if not h:
return False
# IPv6 loopback / link-local
# IPv6 unique-local: fc00::/7 — any address starting with fc?? or fd?? (?? = hex pair).
# Loose "startswith('fc')" / "startswith('fd')" would also match the hostnames
# "food.example.com" or "fdsa.test", so require the second char to be a hex
# digit followed by either a colon or another hex digit (canonical IPv6 syntax).
import re as _re
if h in ('::1', '0:0:0:0:0:0:0:1') or h.startswith('fe80:') or _re.match(r'^f[cd][0-9a-f]{0,2}:', h):
return True
# Hostname allow-list (RFC2606/6761 reserved TLDs + localhost)
if h == 'localhost' or h.endswith('.localhost'):
return True
if h.endswith('.local') or h.endswith('.test') or h.endswith('.invalid'):
return True
if h == 'example.com' or h.endswith('.example.com'):
return True
if h == 'example.net' or h.endswith('.example.net'):
return True
if h == 'example.org' or h.endswith('.example.org'):
return True
if h.endswith('.example'):
return True
# IPv4 — parse octets if it looks like a dotted quad
if h[0].isdigit() and h.count('.') == 3:
try:
o1, o2, o3, o4 = [int(p) for p in h.split('.')]
except ValueError:
return False
if o1 == 127: # loopback
return True
if o1 == 10: # RFC1918 10.0.0.0/8
return True
if o1 == 192 and o2 == 168: # RFC1918 192.168.0.0/16
return True
if o1 == 172 and 16 <= o2 <= 31: # RFC1918 172.16.0.0/12
return True
if o1 == 169 and o2 == 254: # link-local 169.254.0.0/16
return True
if o1 == 203 and o2 == 0 and o3 == 113: # RFC5737 TEST-NET-3
return True
return False
def _hermes_blocked_create_connection(address, *a, **kw):
try:
host = address[0]
except (TypeError, IndexError):
host = ""
if _hermes_addr_is_local(host):
return _REAL_CREATE_CONNECTION(address, *a, **kw)
raise OSError(
f"hermes test network isolation: outbound socket to {address!r} is blocked. "
f"Tests should mock urllib.request.urlopen / requests / socket.create_connection. "
f"If a test genuinely needs real outbound, request the allow_outbound_network fixture."
)
def _hermes_blocked_socket_connect(self, address):
try:
host = address[0]
except (TypeError, IndexError):
host = ""
if _hermes_addr_is_local(host):
return _REAL_SOCKET_CONNECT(self, address)
raise OSError(
f"hermes test network isolation: socket.connect to {address!r} is blocked."
)
_hermes_test_socket.create_connection = _hermes_blocked_create_connection
_hermes_test_socket.socket.connect = _hermes_blocked_socket_connect
@pytest.fixture
def allow_outbound_network(monkeypatch):
"""Opt-in to real outbound network for the duration of one test.
Swaps `socket.create_connection` and `socket.socket.connect` back to the
real (unwrapped) implementations for this test only, then monkeypatch
teardown restores the wrapped versions. Direct swap is more reliable
than a module-global toggle on CI runners where wrapper-closure
lookup semantics can surprise.
Use sparingly. Today zero tests in the repo call this — the previous
test_dns_resolution_failure case was rewritten to mock socket.getaddrinfo
instead, which is fully hermetic.
"""
monkeypatch.setattr(_hermes_test_socket, "create_connection", _REAL_CREATE_CONNECTION)
monkeypatch.setattr(_hermes_test_socket.socket, "connect", _REAL_SOCKET_CONNECT)
yield
# ── Environment isolation for tests ────────────────────────────────────────
# HERMES_WEBUI_SKIP_ONBOARDING is set by hosting providers (e.g. Agent37) and
# by some isolated test harnesses to short-circuit the onboarding wizard.
# When it leaks into the pytest environment, tests that exercise the wizard
# code paths (apply_onboarding_setup, etc.) fail because the function returns
# early without writing config files.
#
# This autouse fixture removes the variable for the test session. Tests that
# specifically need to validate the SKIP_ONBOARDING short-circuit can opt back
# in with `monkeypatch.setenv("HERMES_WEBUI_SKIP_ONBOARDING", "1")`.
@pytest.fixture(autouse=True, scope="session")
def _strip_skip_onboarding_env():
prior = os.environ.pop("HERMES_WEBUI_SKIP_ONBOARDING", None)
yield
if prior is not None:
os.environ["HERMES_WEBUI_SKIP_ONBOARDING"] = prior
def pytest_collection_modifyitems(config, items):
"""Auto-skip agent-dependent tests when hermes-agent is not available.
Instead of requiring markers on every test function, we pattern-match
test names to known categories that depend on hermes-agent modules.
This keeps the test files clean and ensures new cron/skills tests
get auto-skipped without manual annotation.
"""
if AGENT_MODULES_AVAILABLE:
return # everything available, run all tests
# Exact list of tests known to fail without hermes-agent.
# These hit server endpoints that import cron.jobs, tools.skills_tool,
# or require a running agent backend — returning 500 without the agent.
_AGENT_DEPENDENT_TESTS = {
# Cron endpoints (need cron.jobs module)
'test_crons_list',
'test_crons_list_has_required_fields',
'test_crons_output_requires_job_id',
'test_crons_output_real_job',
'test_crons_run_nonexistent',
'test_cron_create_success',
'test_cron_update_unknown_job_404',
'test_cron_delete_unknown_404',
'test_crons_output_limit_param',
# Skills endpoints (need tools.skills_tool module)
'test_skills_list',
'test_skills_list_has_required_fields',
'test_skills_content_known',
'test_skills_content_requires_name',
'test_skills_search_returns_subset',
'test_skill_save_delete_roundtrip',
'test_skill_delete_unknown_404',
# Agent backend (need running AIAgent)
'test_chat_stream_opens_successfully',
'test_approval_submit_and_respond',
# Security redaction (flaky — session state varies across test ordering)
'test_api_sessions_list_redacts_titles',
# Workspace path (macOS /tmp -> /private/tmp symlink)
'test_new_session_inherits_workspace',
'test_workspace_add_valid',
'test_workspace_rename',
'test_last_workspace_updates_on_session_update',
'test_new_session_inherits_last_workspace',
}
skip_marker = pytest.mark.skip(reason="requires hermes-agent (not installed)")
skipped = 0
for item in items:
if item.name in _AGENT_DEPENDENT_TESTS:
item.add_marker(skip_marker)
skipped += 1
if skipped:
print(f"\nWARNING: hermes-agent not found; {skipped} agent-dependent tests will be skipped\n")
# ── Helpers ──────────────────────────────────────────────────────────────────
def _post(base, path, body=None):
data = json.dumps(body or {}).encode()
req = urllib.request.Request(
base + path, data=data, headers={"Content-Type": "application/json"}
)
try:
with urllib.request.urlopen(req, timeout=10) as r:
return json.loads(r.read())
except urllib.error.HTTPError as e:
try:
return json.loads(e.read())
except Exception:
return {}
def _wait_for_server(base, timeout=20):
deadline = time.time() + timeout
while time.time() < deadline:
try:
with urllib.request.urlopen(base + "/health", timeout=2) as r:
if json.loads(r.read()).get("status") == "ok":
return True
except Exception:
time.sleep(0.3)
return False
# ── Session-scoped test server ────────────────────────────────────────────────
@pytest.fixture(scope="session", autouse=True)
def test_server():
"""
Start an isolated test server on TEST_PORT with a clean state directory.
Paths are discovered dynamically -- no hardcoded absolute path assumptions.
"""
# Kill any leftover process on the test port before starting.
# Stale servers from QA harness runs or prior test sessions cause
# conftest to think the server is already up, producing false failures.
try:
import subprocess as _sp
_sp.run(['fuser', '-k', f'{TEST_PORT}/tcp'],
capture_output=True, timeout=5)
except Exception:
pass
import time as _time
_time.sleep(0.5) # brief pause to let the port release
# Clean slate
if TEST_STATE_DIR.exists():
shutil.rmtree(TEST_STATE_DIR)
TEST_STATE_DIR.mkdir(parents=True)
TEST_WORKSPACE.mkdir(parents=True)
# Symlink real skills into test home so skill-related tests work,
# but all write-heavy state stays isolated.
real_skills = HERMES_HOME / 'skills'
test_skills = TEST_STATE_DIR / 'skills'
if real_skills.exists() and not test_skills.exists():
test_skills.symlink_to(real_skills)
# Isolated cron state
(TEST_STATE_DIR / 'cron').mkdir(parents=True, exist_ok=True)
# Expose TEST_STATE_DIR to the test process itself so that tests which write
# directly to state.db (e.g. test_gateway_sync.py) always use the same path
# as the server. Other test files (test_auth_sessions.py) may override
# HERMES_WEBUI_STATE_DIR for their own purposes, but HERMES_WEBUI_TEST_STATE_DIR
# is reserved for this mapping and is never overridden by individual test files.
# Export both port and state-dir as env vars so individual test files
# can read them without importing conftest (avoids circular imports).
os.environ.setdefault('HERMES_WEBUI_TEST_PORT', str(TEST_PORT))
# os.environ already set at module level above; no-op here.
env = os.environ.copy()
# Strip ANY real credential env var so the test subprocess never inherits
# production creds. The test server uses a mock/isolated config — no real
# API calls are made, no real OAuth flow runs, no real cloud SDK should
# ever be initialised with usable credentials.
#
# Without this strip, a stray credential left in the runner's env was
# observed making outbound TLS to a real provider during test runs.
# See investigation notes in pytest-pitfalls SKILL §B.3.
_CRED_ENV_PREFIXES = (
# LLM providers
'OPENROUTER_API_KEY', 'OPENAI_API_KEY', 'OPENAI_BASE_URL',
'ANTHROPIC_API_KEY', 'ANTHROPIC_AUTH_TOKEN',
'GOOGLE_API_KEY', 'GOOGLE_APPLICATION_CREDENTIALS',
'DEEPSEEK_API_KEY', 'XIAOMI_API_KEY',
'XAI_API_KEY', 'MISTRAL_API_KEY', 'OLLAMA_API_KEY',
'GROQ_API_KEY', 'TOGETHER_API_KEY', 'PERPLEXITY_API_KEY',
'CEREBRAS_API_KEY', 'COHERE_API_KEY', 'FIREWORKS_API_KEY',
'NOUS_API_KEY', 'NOVITA_API_KEY', 'TENCENT_API_KEY',
'BIGMODEL_API_KEY', 'GLM_API_KEY', 'STEPFUN_API_KEY',
'MINIMAX_API_KEY', 'LM_API_KEY', 'LMSTUDIO_API_KEY',
'AZURE_OPENAI_API_KEY', 'AZURE_OPENAI_ENDPOINT',
# AWS — must be stripped or botocore probes IMDS / picks up real creds
'AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY', 'AWS_SESSION_TOKEN',
'AWS_PROFILE', 'AWS_BEARER_TOKEN_BEDROCK',
# Memory providers, telemetry, dashboards
'MEM0_API_KEY', 'HONCHO_API_KEY', 'SUPERMEMORY_API_KEY',
# Messaging / gateway
'TELEGRAM_BOT_TOKEN', 'DISCORD_BOT_TOKEN', 'SLACK_BOT_TOKEN',
'SIGNAL_API_TOKEN', 'WHATSAPP_API_TOKEN',
# Browser / image-gen / search
'FIRECRAWL_API_KEY', 'FAL_KEY', 'TAVILY_API_KEY',
'SERPER_API_KEY', 'BRAVE_API_KEY',
# Github tokens (PR/issue tools shouldn't be exercised in tests)
'GH_TOKEN', 'GITHUB_TOKEN',
)
for _k in list(env):
if any(_k.startswith(p) for p in _CRED_ENV_PREFIXES):
del env[_k]
# Belt-and-suspenders: keep IMDS disabled in the spawn env too (we set it
# at module level above for the pytest process, but make it explicit here
# so it's never accidentally cleared by an env.update later).
env["AWS_EC2_METADATA_DISABLED"] = "true"
# Activate the same network-isolation block in the test_server subprocess
# that conftest.py installs in the pytest process. server.py reads this
# env var at import time and installs an identical socket-block guard.
# Without this, the subprocess can make outbound requests that the
# pytest-side block can't see.
env["HERMES_WEBUI_TEST_NETWORK_BLOCK"] = "1"
env.update({
"HERMES_WEBUI_WORKSPACE_GIT_DESTRUCTIVE": "1",
"HERMES_WEBUI_PORT": str(TEST_PORT),
"HERMES_WEBUI_HOST": "127.0.0.1",
"HERMES_WEBUI_STATE_DIR": str(TEST_STATE_DIR),
"HERMES_WEBUI_DEFAULT_WORKSPACE": str(TEST_WORKSPACE),
"HERMES_WEBUI_DEFAULT_MODEL": "openai/gpt-5.4-mini",
"HERMES_HOME": str(TEST_STATE_DIR),
"HERMES_CONFIG_PATH": str(TEST_STATE_DIR / 'config.yaml'),
# Belt-and-suspenders: HERMES_BASE_HOME hard-locks _DEFAULT_HERMES_HOME
# in api/profiles.py to the test state dir regardless of profile switching
# or any os.environ mutation that happens inside the server process.
# Without this, a profile switch or active_profile file in the real
# ~/.hermes can redirect _get_active_hermes_home() out of the sandbox,
# causing onboarding writes (config.yaml, .env) to land in the production
# ~/.hermes/profiles/webui/ and overwrite real API keys.
"HERMES_BASE_HOME": str(TEST_STATE_DIR),
"HERMES_WEBUI_PASSWORD": "",
})
# Pass agent dir if discovered so server.py doesn't have to re-discover
if HERMES_AGENT:
env["HERMES_WEBUI_AGENT_DIR"] = str(HERMES_AGENT)
proc = subprocess.Popen(
[VENV_PYTHON, str(SERVER_SCRIPT)],
cwd=WORKDIR,
env=env,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
if not _wait_for_server(TEST_BASE, timeout=20):
proc.kill()
pytest.fail(
f"Test server on port {TEST_PORT} did not start within 20s.\n"
f" server.py : {SERVER_SCRIPT}\n"
f" python : {VENV_PYTHON}\n"
f" agent dir : {HERMES_AGENT}\n"
f" workdir : {WORKDIR}\n"
)
yield proc
proc.terminate()
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
proc.kill()
try:
shutil.rmtree(TEST_STATE_DIR)
except Exception:
pass
# ── Test base URL ─────────────────────────────────────────────────────────────
@pytest.fixture(scope="session")
def base_url():
return TEST_BASE
# ── Per-test model cache invalidation ────────────────────────────────────────
# The TTL cache for get_available_models() persists across tests within the
# same process. Tests that modify cfg in-memory won't trigger the mtime path,
# so the cache must be explicitly invalidated after each test that exercises
# provider/model detection.
@pytest.fixture(autouse=True)
def _invalidate_models_cache_after_test():
"""Force the TTL cache to be cleared before and after every test.
This prevents state bleed where a test that calls get_available_models()
populates the cache with a particular config, and the next test sees stale
results even though it has mutated _cfg_cache in-memory.
"""
try:
from api.config import invalidate_models_cache
invalidate_models_cache()
except Exception:
pass
yield
try:
from api.config import invalidate_models_cache
invalidate_models_cache()
except Exception:
pass
# ── Per-test session cleanup ──────────────────────────────────────────────────
@pytest.fixture(autouse=True)
def cleanup_test_sessions():
"""
Yields a list for tests to register created session IDs.
Deletes all registered sessions after each test.
Resets last_workspace to the test workspace to prevent state bleed.
"""
created: list[str] = []
yield created
for sid in created:
try:
_post(TEST_BASE, "/api/session/delete", {"session_id": sid})
except Exception:
pass
try:
_post(TEST_BASE, "/api/sessions/cleanup_zero_message")
except Exception:
pass
try:
last_ws_file = TEST_STATE_DIR / "last_workspace.txt"
last_ws_file.write_text(str(TEST_WORKSPACE), encoding='utf-8')
except Exception:
pass
# ── Convenience helpers ────────────────────────────────────────────────────────
def make_session_tracked(created_list, ws=None):
"""
Create a session on the test server and register it for cleanup.
Usage:
def test_something(cleanup_test_sessions):
sid, ws = make_session_tracked(cleanup_test_sessions)
"""
body = {}
if ws:
body["workspace"] = str(ws)
d = _post(TEST_BASE, "/api/session/new", body)
sid = d["session"]["session_id"]
ws_path = pathlib.Path(d["session"]["workspace"])
created_list.append(sid)
return sid, ws_path