skills/vectorize-io__hindsight
1
0
Fork 0
mirror of https://github.com/vectorize-io/hindsight.git synced 2026-09-14 19:31:49 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
vectorize-io__hindsight/hindsight-embed/tests/test_daemon_client.py
T
Nicolò Boschi 31a8c52db5 refactor: replace httpx and sync HTTP with aiohttp across production code (#4318) 
Production code now talks HTTP through aiohttp only, and only asynchronously.
httpx's async client is 5-12x slower per request than aiohttp at the
concurrency we run embeddings/rerank/LLM calls with (local stub: 1.3k vs 6.8k
req/s at c=8, 527 vs 6.3k at c=64), and the sync paths needed a thread per
in-flight request.

- engine/aiohttp_session.py: LoopLocal / LoopLocalSession (one session per event
  loop, created lazily, closed-loop entries released), per_phase_timeout
  (httpx-style per-phase timeouts), raise_for_status -> UpstreamHTTPError
  (keeps body + status_code for remote_retry), close_loop_sessions() called from
  MemoryEngine.close().
- Embeddings interface is async (encode/encode_query/encode_documents); remote
  providers use aiohttp or the SDK's async client, bounded fan-out via a per-loop
  semaphore instead of a thread pool. Local models stay in a worker thread.
- Rerankers, Codex/Nous/xAI OAuth providers (async token refresh with a per-loop
  asyncio.Lock + non-blocking flock), Fireworks, Ollama-native, llama.cpp,
  LlamaParse and Iris parsers moved to aiohttp.
- Webhook SSRF guard re-implemented as an aiohttp resolver that only returns
  validated addresses (no DNS cache, no redirects, no env proxies), with IP
  literals checked before sending.
- hindsight-embed probes, supabase-tenant extension and the litellm
  integration's async hooks moved to aiohttp.
- ruff TID251 bans httpx/requests/urllib.request/urllib3/http.client in
  hindsight-api-slim and hindsight-embed production code; code-review skill
  documents the rule (tests exempt).
2026-09-11 11:36:28 +02:00

815 lines
37 KiB
Python
Raw Permalink Blame History

"""Tests for daemon_client module."""
from unittest.mock import MagicMock, Mock, patch
import pytest
from hindsight_embed import daemon_client
from hindsight_embed.daemon_embed_manager import DaemonEmbedManager, _parse_non_negative_int
from .http_stub import closed_port, replies, reply, serve
_HEALTHY = {"status": "healthy", "database": "connected"}
# What /proc/<pid>/cmdline reports for a daemon this manager would have started.
# Ownership is decided on the listener's command line, so tests that expect a
# kill have to present one that identifies the process as ours (#3520).
_DAEMON_CMDLINE = "/home/u/.venv/bin/hindsight-api --port 9555"
@pytest.fixture
def config():
"""Default config for tests."""
return {
"llm_api_key": "test-key",
"llm_provider": "openai",
"llm_model": "gpt-4o-mini",
"bank_id": "test-bank",
}
@pytest.fixture
def mock_cli_binary(tmp_path):
"""Create a mock CLI binary."""
cli_path = tmp_path / "hindsight"
cli_path.write_text("#!/bin/bash\nexit 0")
cli_path.chmod(0o755)
return cli_path
class TestRunCli:
"""Tests for run_cli function."""
def test_run_cli_with_external_api_url(self, config, mock_cli_binary, monkeypatch):
"""Test that external HINDSIGHT_EMBED_API_URL skips daemon startup."""
# Set up environment with external API URL
external_api_url = "http://external-api:8000"
monkeypatch.setenv("HINDSIGHT_EMBED_API_URL", external_api_url)
# Mock functions
mock_ensure_cli = Mock(return_value=True)
mock_find_cli = Mock(return_value=mock_cli_binary)
mock_ensure_daemon = Mock(return_value=True)
mock_subprocess_run = Mock(return_value=Mock(returncode=0))
with (
patch.object(daemon_client, "ensure_cli_installed", mock_ensure_cli),
patch.object(daemon_client, "find_cli_binary", mock_find_cli),
patch.object(daemon_client, "ensure_daemon_running", mock_ensure_daemon),
patch("subprocess.run", mock_subprocess_run),
):
# Run CLI
exit_code = daemon_client.run_cli(["memory", "recall", "test", "query"], config)
# Verify daemon was NOT started (since external API URL is set)
assert mock_ensure_daemon.call_count == 0
# Verify CLI was called
assert mock_subprocess_run.call_count == 1
call_args = mock_subprocess_run.call_args
# Verify environment contains the external API URL
assert call_args.kwargs["env"]["HINDSIGHT_API_URL"] == external_api_url
# Verify exit code
assert exit_code == 0
def test_run_cli_without_external_api_url(self, config, mock_cli_binary, monkeypatch):
"""Test that without external API URL, daemon is started."""
# Ensure HINDSIGHT_EMBED_API_URL is not set
monkeypatch.delenv("HINDSIGHT_EMBED_API_URL", raising=False)
# Mock functions
mock_ensure_cli = Mock(return_value=True)
mock_find_cli = Mock(return_value=mock_cli_binary)
mock_ensure_daemon = Mock(return_value=True)
mock_subprocess_run = Mock(return_value=Mock(returncode=0))
with (
patch.object(daemon_client, "ensure_cli_installed", mock_ensure_cli),
patch.object(daemon_client, "find_cli_binary", mock_find_cli),
patch.object(daemon_client, "ensure_daemon_running", mock_ensure_daemon),
patch("subprocess.run", mock_subprocess_run),
):
# Run CLI
exit_code = daemon_client.run_cli(["memory", "recall", "test", "query"], config)
# Verify daemon WAS started (since no external API URL)
assert mock_ensure_daemon.call_count == 1
assert mock_ensure_daemon.call_args[0][0] == config
# Verify CLI was called
assert mock_subprocess_run.call_count == 1
call_args = mock_subprocess_run.call_args
# Verify environment contains the local daemon URL
assert call_args.kwargs["env"]["HINDSIGHT_API_URL"] == daemon_client.get_daemon_url()
# Verify exit code
assert exit_code == 0
def test_run_cli_daemon_startup_failure(self, config, mock_cli_binary, monkeypatch):
"""Test that daemon startup failure is handled properly."""
# Ensure HINDSIGHT_EMBED_API_URL is not set
monkeypatch.delenv("HINDSIGHT_EMBED_API_URL", raising=False)
# Mock functions - daemon startup fails
mock_ensure_cli = Mock(return_value=True)
mock_find_cli = Mock(return_value=mock_cli_binary)
mock_ensure_daemon = Mock(return_value=False) # Daemon fails to start
with (
patch.object(daemon_client, "ensure_cli_installed", mock_ensure_cli),
patch.object(daemon_client, "find_cli_binary", mock_find_cli),
patch.object(daemon_client, "ensure_daemon_running", mock_ensure_daemon),
):
# Run CLI
exit_code = daemon_client.run_cli(["memory", "recall", "test", "query"], config)
# Verify daemon startup was attempted
assert mock_ensure_daemon.call_count == 1
# Verify exit code indicates failure
assert exit_code == 1
def test_run_cli_without_cli_binary(self, config, monkeypatch):
"""Test that missing CLI binary is handled properly."""
# Ensure HINDSIGHT_EMBED_API_URL is not set
monkeypatch.delenv("HINDSIGHT_EMBED_API_URL", raising=False)
# Mock functions - CLI not installed
mock_ensure_cli = Mock(return_value=True)
mock_find_cli = Mock(return_value=None) # CLI not found
with (
patch.object(daemon_client, "ensure_cli_installed", mock_ensure_cli),
patch.object(daemon_client, "find_cli_binary", mock_find_cli),
):
# Run CLI
exit_code = daemon_client.run_cli(["memory", "recall", "test", "query"], config)
# Verify exit code indicates failure
assert exit_code == 1
def test_run_cli_with_api_token(self, config, mock_cli_binary, monkeypatch):
"""Test that HINDSIGHT_EMBED_API_TOKEN is passed through to the CLI."""
# Set up environment with external API URL and token
external_api_url = "http://external-api:8000"
api_token = "test-bearer-token-12345"
monkeypatch.setenv("HINDSIGHT_EMBED_API_URL", external_api_url)
monkeypatch.setenv("HINDSIGHT_EMBED_API_TOKEN", api_token)
# Mock functions
mock_ensure_cli = Mock(return_value=True)
mock_find_cli = Mock(return_value=mock_cli_binary)
mock_ensure_daemon = Mock(return_value=True)
mock_subprocess_run = Mock(return_value=Mock(returncode=0))
with (
patch.object(daemon_client, "ensure_cli_installed", mock_ensure_cli),
patch.object(daemon_client, "find_cli_binary", mock_find_cli),
patch.object(daemon_client, "ensure_daemon_running", mock_ensure_daemon),
patch("subprocess.run", mock_subprocess_run),
):
# Run CLI
exit_code = daemon_client.run_cli(["memory", "recall", "test", "query"], config)
# Verify daemon was NOT started (since external API URL is set)
assert mock_ensure_daemon.call_count == 0
# Verify CLI was called
assert mock_subprocess_run.call_count == 1
call_args = mock_subprocess_run.call_args
# Verify environment contains both the API URL and the API key
assert call_args.kwargs["env"]["HINDSIGHT_API_URL"] == external_api_url
assert call_args.kwargs["env"]["HINDSIGHT_API_KEY"] == api_token
# Verify exit code
assert exit_code == 0
class TestClearPort:
"""Tests for DaemonEmbedManager._clear_port."""
def test_port_free(self):
"""Port not in use — returns True immediately."""
manager = DaemonEmbedManager()
with patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=False):
assert manager._clear_port(9555) is True
def test_port_occupied_by_healthy_hindsight_is_reused(self):
"""Port occupied by a healthy hindsight daemon — reused without killing.
Regression: previously _clear_port killed any hindsight daemon on the
target port, which meant two concurrent `start` calls would kill each
other's freshly-started daemons.
"""
manager = DaemonEmbedManager()
with (
serve(reply(200, _HEALTHY)) as stub,
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch.object(DaemonEmbedManager, "_listening_pids") as mock_find_pid,
patch.object(DaemonEmbedManager, "_kill_process") as mock_kill,
):
assert manager._clear_port(stub.port) is True
mock_find_pid.assert_not_called()
mock_kill.assert_not_called()
def test_port_occupied_by_non_hindsight_returns_false(self):
"""Port occupied by non-hindsight process — returns False."""
manager = DaemonEmbedManager()
with (
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 0.0),
):
# /health refuses the connection.
assert manager._clear_port(closed_port()) is False
def test_port_occupied_health_non_200_returns_false(self):
"""Port responds but not with 200 — treated as non-hindsight."""
manager = DaemonEmbedManager()
with (
serve(reply(404)) as stub,
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 0.0),
):
assert manager._clear_port(stub.port) is False
def test_unhealthy_daemon_pid_not_found_returns_false(self):
"""Unhealthy process on port, no PID — cannot reclaim, returns False."""
manager = DaemonEmbedManager()
with (
serve(reply(500)) as stub,
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[]),
patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 0.0),
):
assert manager._clear_port(stub.port) is False
def test_unhealthy_daemon_kill_fails_returns_false(self):
"""Unhealthy process on port, kill failed — returns False."""
manager = DaemonEmbedManager()
with (
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[12345]),
patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE),
patch.object(DaemonEmbedManager, "_kill_process", return_value=False),
patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 0.0),
):
# /health refuses the connection.
assert manager._clear_port(closed_port()) is False
def test_unhealthy_daemon_kill_succeeds_returns_true(self):
"""Unhealthy hindsight daemon (stale from version upgrade) reclaimed via kill."""
manager = DaemonEmbedManager()
with (
serve(reply(503)) as stub,
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[12345]),
patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE),
patch.object(DaemonEmbedManager, "_kill_process", return_value=True),
patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 0.0),
):
assert manager._clear_port(stub.port) is True
def test_port_occupied_by_warming_hindsight_is_reused(self):
"""Port bound before /health is ready — wait briefly and reuse when healthy."""
manager = DaemonEmbedManager()
with (
serve(replies(reply(503), reply(503), reply(200, _HEALTHY))) as stub,
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch.object(DaemonEmbedManager, "_listening_pids") as mock_find_pid,
patch.object(DaemonEmbedManager, "_kill_process") as mock_kill,
patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 5.0),
patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_CHECK_INTERVAL", 0.01),
):
assert manager._clear_port(stub.port) is True
assert stub.paths == ["/health"] * 3
mock_find_pid.assert_not_called()
mock_kill.assert_not_called()
def test_port_occupied_by_foreign_health_200_returns_false(self):
"""HTTP 200 alone is not enough to identify the listener as Hindsight."""
manager = DaemonEmbedManager()
with (
serve(reply(200, {"status": "ok"})) as stub,
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[]) as mock_find_pid,
patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 0.0),
):
assert manager._clear_port(stub.port) is False
mock_find_pid.assert_called_once_with(stub.port)
def test_port_cleared_during_grace_returns_true(self):
"""If a stale listener exits during the grace wait, the port is already clear."""
manager = DaemonEmbedManager()
with (
serve(reply(503)) as stub,
patch.object(DaemonEmbedManager, "_is_port_in_use", side_effect=[True, False, False]),
patch.object(DaemonEmbedManager, "_listening_pids") as mock_find_pid,
patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 1.0),
patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_CHECK_INTERVAL", 0.01),
):
assert manager._clear_port(stub.port) is True
mock_find_pid.assert_not_called()
def test_invalid_port_health_timeout_is_bounded(self):
"""Invalid grace timeout values must not create an unbounded wait."""
manager = DaemonEmbedManager()
with (
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch.object(DaemonEmbedManager, "_port_health_ok", return_value=False) as mock_health,
patch("hindsight_embed.daemon_embed_manager.time.sleep") as mock_sleep,
):
assert manager._wait_for_port_health(9555, timeout=float("nan")) is False
mock_health.assert_called_once_with(9555)
mock_sleep.assert_not_called()
class TestStartDaemonSerialization:
"""Tests that _start_daemon serializes concurrent starts via the profile lock.
Regression: two processes calling start() concurrently used to race into
_clear_port and kill each other's daemons. The per-profile flock
serializes them; the losing caller discovers the winner's daemon via
is_running() and short-circuits without spawning.
"""
def test_second_start_sees_daemon_up_and_skips_spawn(self, tmp_path, monkeypatch):
"""If is_running() is true inside the lock, _start_daemon returns without spawning."""
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
manager = DaemonEmbedManager()
with (
patch.object(DaemonEmbedManager, "is_running", return_value=True),
patch.object(DaemonEmbedManager, "_clear_port") as mock_clear,
patch.object(DaemonEmbedManager, "_start_daemon_locked") as mock_locked,
patch.object(DaemonEmbedManager, "_register_profile") as mock_register,
):
assert manager._start_daemon({}, "codex") is True
mock_clear.assert_not_called()
mock_locked.assert_not_called()
mock_register.assert_called_once()
def test_start_daemon_locked_skips_spawn_when_port_already_healthy(self, tmp_path, monkeypatch):
"""Inside _start_daemon_locked: healthy daemon appearing between checks skips spawn.
Simulates a healthy foreign-started daemon showing up after is_running()
first returned False. _clear_port keeps it; _start_daemon_locked's
second is_running() check returns True and we must NOT proceed to
subprocess.Popen.
"""
from hindsight_embed.profile_manager import ProfilePaths
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
manager = DaemonEmbedManager()
log_path = tmp_path / "daemon.log"
paths = ProfilePaths(
config=tmp_path / "embed",
lock=tmp_path / "daemon.lock",
log=log_path,
port=9600,
)
with (
patch.object(DaemonEmbedManager, "_clear_port", return_value=True),
patch.object(DaemonEmbedManager, "is_running", return_value=True),
patch.object(DaemonEmbedManager, "_register_profile"),
patch.object(DaemonEmbedManager, "_rotate_daemon_log") as mock_rotate,
patch("subprocess.Popen") as mock_popen,
):
assert manager._start_daemon_locked({}, "codex", paths) is True
mock_popen.assert_not_called()
mock_rotate.assert_not_called()
def test_rotation_runs_after_port_clear_and_before_spawn(self, tmp_path, monkeypatch):
from hindsight_embed.profile_manager import ProfilePaths
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.setattr("time.sleep", lambda _: None)
manager = DaemonEmbedManager()
paths = ProfilePaths(
config=tmp_path / "embed", lock=tmp_path / "daemon.lock", log=tmp_path / "daemon.log", port=9600
)
calls = []
with (
patch.object(DaemonEmbedManager, "_clear_port", side_effect=lambda _: calls.append("clear") or True),
patch.object(DaemonEmbedManager, "is_running", side_effect=[False, True, True]),
patch.object(
DaemonEmbedManager, "_rotate_daemon_log", side_effect=lambda *_args, **_kwargs: calls.append("rotate")
),
patch.object(DaemonEmbedManager, "_find_api_command", return_value=["hindsight-api"]),
patch.object(DaemonEmbedManager, "_component_version", return_value=None),
patch.object(DaemonEmbedManager, "_register_profile"),
patch("subprocess.Popen", side_effect=lambda *_args, **_kwargs: calls.append("spawn")),
patch("hindsight_embed.daemon_embed_manager.Live"),
):
assert manager._start_daemon_locked({}, "codex", paths) is True
assert calls == ["clear", "rotate", "spawn"]
class TestDaemonLogRotation:
"""Tests for restart-boundary daemon log rotation."""
def test_small_log_is_left_in_place(self, tmp_path):
log_path = tmp_path / "daemon.log"
log_path.write_bytes(b"1234")
DaemonEmbedManager._rotate_daemon_log(log_path, max_bytes=5, backup_count=2)
assert log_path.read_bytes() == b"1234"
assert not (tmp_path / "daemon.log.1").exists()
def test_full_log_rotates_and_retention_is_bounded(self, tmp_path):
log_path = tmp_path / "daemon.log"
log_path.write_bytes(b"current")
(tmp_path / "daemon.log.1").write_bytes(b"previous")
(tmp_path / "daemon.log.2").write_bytes(b"oldest")
DaemonEmbedManager._rotate_daemon_log(log_path, max_bytes=7, backup_count=2)
assert not log_path.exists()
assert (tmp_path / "daemon.log.1").read_bytes() == b"current"
assert (tmp_path / "daemon.log.2").read_bytes() == b"previous"
def test_zero_backups_truncates_full_log(self, tmp_path):
log_path = tmp_path / "daemon.log"
log_path.write_bytes(b"current")
DaemonEmbedManager._rotate_daemon_log(log_path, max_bytes=1, backup_count=0)
assert log_path.exists()
assert log_path.read_bytes() == b""
def test_zero_max_bytes_disables_rotation(self, tmp_path):
log_path = tmp_path / "daemon.log"
log_path.write_bytes(b"current")
DaemonEmbedManager._rotate_daemon_log(log_path, max_bytes=0, backup_count=2)
assert log_path.read_bytes() == b"current"
def test_non_negative_int_falls_back_for_invalid_values(self, caplog):
assert _parse_non_negative_int("10MB", 42, "LIMIT") == 42
assert _parse_non_negative_int("-1", 42, "LIMIT") == 42
assert "Invalid LIMIT" in caplog.text
class TestStop:
"""Tests for DaemonEmbedManager.stop() - regression coverage for #3169.
Every decision in stop() is based on port occupancy. The /health probe
reports responsiveness, not identity or liveness, so a busy daemon fails
it; using it as the already-stopped guard or as the success condition is
what made `daemon stop` report success without sending any signal.
"""
def _paths(self, tmp_path, port=9700):
from hindsight_embed.profile_manager import ProfilePaths
return ProfilePaths(
config=tmp_path / "embed",
lock=tmp_path / "daemon.lock",
log=tmp_path / "daemon.log",
port=port,
)
def test_busy_daemon_is_terminated(self, tmp_path):
"""A daemon that holds the port but fails /health is still stopped.
This is the #3169 scenario. Both health probes are patched to raise so
the test fails if any path in stop() consults responsiveness.
"""
manager = DaemonEmbedManager()
with (
patch.object(
manager._profile_manager,
"resolve_profile_paths",
return_value=self._paths(tmp_path),
),
patch.object(DaemonEmbedManager, "_is_port_in_use", side_effect=[True, False]),
patch.object(
DaemonEmbedManager,
"is_running",
side_effect=AssertionError("stop() must not consult /health"),
),
patch.object(
DaemonEmbedManager,
"_port_health_ok",
side_effect=AssertionError("stop() must not consult /health"),
),
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]),
patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE),
patch.object(DaemonEmbedManager, "_kill_process", return_value=True) as mock_kill,
):
assert manager.stop("default") is True
mock_kill.assert_called_once_with(4242)
def test_unresponsive_listener_is_reclaimed_like_clear_port(self, tmp_path):
"""stop() reclaims an occupied, unhealthy port the way _clear_port() does.
Responsiveness cannot distinguish "busy" from "foreign", so both paths
reclaim the profile's port from an unhealthy listener — but only once
the listener's command line identifies it as a hindsight daemon
(#3520). Refusing outright would leave a wedged daemon unstoppable,
which is the #3169 symptom.
"""
manager = DaemonEmbedManager()
with (
patch.object(
manager._profile_manager,
"resolve_profile_paths",
return_value=self._paths(tmp_path),
),
patch.object(DaemonEmbedManager, "_is_port_in_use", side_effect=[True, False]),
patch.object(DaemonEmbedManager, "_port_health_ok", return_value=False),
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[9999]),
patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE),
patch.object(DaemonEmbedManager, "_kill_process", return_value=True) as mock_kill,
):
assert manager.stop("default") is True
mock_kill.assert_called_once_with(9999)
with (
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch.object(DaemonEmbedManager, "_wait_for_port_health", return_value=False),
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[9999]),
patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE),
patch.object(DaemonEmbedManager, "_kill_process", return_value=True) as mock_kill,
):
assert manager._clear_port(9700) is True
mock_kill.assert_called_once_with(9999)
def test_failed_termination_returns_false(self, tmp_path):
"""_kill_process() returning False must not be converted into success."""
manager = DaemonEmbedManager()
with (
patch.object(
manager._profile_manager,
"resolve_profile_paths",
return_value=self._paths(tmp_path),
),
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]),
patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE),
patch.object(DaemonEmbedManager, "_kill_process", return_value=False),
):
assert manager.stop("default") is False
def test_bound_port_without_pid_returns_false(self, tmp_path):
"""A bound port whose PID can't be found is a failure, not a success."""
manager = DaemonEmbedManager()
with (
patch.object(
manager._profile_manager,
"resolve_profile_paths",
return_value=self._paths(tmp_path),
),
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[]),
patch.object(DaemonEmbedManager, "_kill_process") as mock_kill,
):
assert manager.stop("default") is False
mock_kill.assert_not_called()
def test_unbound_port_reports_already_stopped(self, tmp_path):
"""Nothing bound to the port means there is nothing to stop."""
manager = DaemonEmbedManager()
with (
patch.object(
manager._profile_manager,
"resolve_profile_paths",
return_value=self._paths(tmp_path),
),
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=False),
patch.object(DaemonEmbedManager, "_listening_pids") as mock_find,
):
assert manager.stop("default") is True
mock_find.assert_not_called()
def test_lingering_listener_after_kill_returns_false(self, tmp_path):
"""If the listener never disappears after the kill, stop() must not claim success."""
manager = DaemonEmbedManager()
with (
patch.object(
manager._profile_manager,
"resolve_profile_paths",
return_value=self._paths(tmp_path),
),
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]),
patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE),
patch.object(DaemonEmbedManager, "_kill_process", return_value=True),
patch("hindsight_embed.daemon_embed_manager.time.sleep"),
):
assert manager.stop("default") is False
class TestListeningPidDiscovery:
"""Regression coverage for #3517: PID lookup must not depend on lsof alone."""
def test_posix_falls_back_to_ss_when_lsof_is_missing(self, monkeypatch):
"""Hosts without lsof (minimal containers, Arch-based distros) still resolve a PID."""
commands = []
def fake_run(cmd, **kwargs):
commands.append(cmd)
if cmd[0] == "lsof":
raise FileNotFoundError("lsof")
return Mock(
returncode=0,
stdout=('LISTEN 0 4096 127.0.0.1:9177 0.0.0.0:* users:(("hindsight-api",pid=15774,fd=19))\n'),
)
monkeypatch.setattr("hindsight_embed.daemon_embed_manager.platform.system", lambda: "Linux")
monkeypatch.setattr("hindsight_embed.daemon_embed_manager.subprocess.run", fake_run)
assert DaemonEmbedManager._listening_pids(9177) == [15774]
assert [cmd[0] for cmd in commands] == ["lsof", "ss"]
def test_posix_prefers_lsof_and_returns_every_listener(self, monkeypatch):
"""lsof output can name several PIDs; all are candidates, not just the first."""
monkeypatch.setattr("hindsight_embed.daemon_embed_manager.platform.system", lambda: "Linux")
monkeypatch.setattr(
"hindsight_embed.daemon_embed_manager.subprocess.run",
lambda cmd, **kwargs: Mock(returncode=0, stdout="111\n222\n"),
)
assert DaemonEmbedManager._listening_pids(9177) == [111, 222]
def test_windows_matches_non_ipv4_local_addresses(self, monkeypatch):
"""The UI can bind 0.0.0.0 or [::1]; a 127.0.0.1-literal match missed it."""
netstat = (
" TCP 0.0.0.0:19177 0.0.0.0:0 LISTENING 4321\n"
" TCP [::1]:19177 [::]:0 LISTENING 4322\n"
" TCP 127.0.0.1:29177 0.0.0.0:0 LISTENING 9999\n"
)
monkeypatch.setattr("hindsight_embed.daemon_embed_manager.platform.system", lambda: "Windows")
monkeypatch.setattr(
"hindsight_embed.daemon_embed_manager.subprocess.run",
lambda cmd, **kwargs: Mock(returncode=0, stdout=netstat),
)
assert DaemonEmbedManager._listening_pids(19177) == [4321, 4322]
class TestProcessOwnership:
"""Regression coverage for #3520: never signal a process we cannot identify."""
def test_foreign_listener_is_not_selected(self):
"""A PID whose command line isn't ours is refused, not returned for killing."""
with (
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]),
patch.object(DaemonEmbedManager, "_process_command_line", return_value="/usr/bin/postgres -D /data"),
):
assert DaemonEmbedManager._owned_pid_on_port(9555, ("hindsight-api",), "hindsight daemon") is None
def test_unknown_command_line_is_refused(self):
"""When the command line cannot be read at all, we refuse rather than guess."""
with (
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]),
patch.object(DaemonEmbedManager, "_process_command_line", return_value=None),
):
assert DaemonEmbedManager._owned_pid_on_port(9555, ("hindsight-api",), "hindsight daemon") is None
def test_our_daemon_is_selected_among_several_listeners(self):
"""With more than one socket on the port, pick ours instead of the first PID."""
with (
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[111, 222]),
patch.object(
DaemonEmbedManager,
"_process_command_line",
side_effect=lambda pid: "/usr/bin/nginx" if pid == 111 else _DAEMON_CMDLINE,
),
):
assert DaemonEmbedManager._owned_pid_on_port(9555, ("hindsight-api",), "hindsight daemon") == 222
def test_clear_port_leaves_foreign_process_alive(self):
"""The #3520 scenario: an unrelated service on the port must survive."""
manager = DaemonEmbedManager()
with (
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch.object(DaemonEmbedManager, "_wait_for_port_health", return_value=False),
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]),
patch.object(DaemonEmbedManager, "_process_command_line", return_value="/usr/sbin/sshd -D"),
patch.object(DaemonEmbedManager, "_kill_process") as mock_kill,
):
assert manager._clear_port(9555) is False
mock_kill.assert_not_called()
def test_stop_leaves_foreign_process_alive(self, tmp_path):
"""stop() reclaims the port the same way, so it needs the same guard."""
from hindsight_embed.profile_manager import ProfilePaths
manager = DaemonEmbedManager()
paths = ProfilePaths(
config=tmp_path / "embed",
lock=tmp_path / "daemon.lock",
log=tmp_path / "daemon.log",
port=9700,
)
with (
patch.object(manager._profile_manager, "resolve_profile_paths", return_value=paths),
patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True),
patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]),
patch.object(DaemonEmbedManager, "_process_command_line", return_value="/usr/sbin/sshd -D"),
patch.object(DaemonEmbedManager, "_kill_process") as mock_kill,
):
assert manager.stop("default") is False
mock_kill.assert_not_called()
def test_dev_mode_and_windows_daemon_command_lines_are_recognized(self):
"""The daemon is spawned several ways; each must still be identifiable."""
for cmdline in (
"/home/u/.venv/bin/hindsight-api",
"C:\\venv\\Scripts\\pythonw.exe -m hindsight_api.main",
"/home/u/.cache/uv/archive/bin/hindsight-api",
):
with patch.object(DaemonEmbedManager, "_process_command_line", return_value=cmdline):
from hindsight_embed.daemon_embed_manager import DAEMON_PROCESS_MARKERS
assert DaemonEmbedManager._process_matches(1, DAEMON_PROCESS_MARKERS) is True
class TestStartDaemonLockTimeout:
"""A start that cannot get the profile lock reports failure, not an OSError."""
def test_lock_timeout_returns_false(self, tmp_path, monkeypatch):
from hindsight_embed.profile_manager import ProfileLockTimeout
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
manager = DaemonEmbedManager()
def boom(file_obj, timeout=None):
raise ProfileLockTimeout("held by PID 999")
with (
patch("hindsight_embed.daemon_embed_manager.lock_file", boom),
patch.object(DaemonEmbedManager, "_start_daemon_locked") as mock_locked,
patch.object(DaemonEmbedManager, "is_running", return_value=False),
):
assert manager._start_daemon({}, "codex") is False
mock_locked.assert_not_called()
class TestProcessCommandLine:
"""`_process_command_line` decides every kill, so it needs direct coverage.
The ownership tests above patch it out; this exercises the real lookup
against a process whose command line we already know.
"""
def test_reads_the_command_line_of_a_live_process(self):
import os
import sys
cmdline = DaemonEmbedManager._process_command_line(os.getpid())
assert cmdline is not None
# argv[0] is the interpreter running pytest.
assert os.path.basename(sys.executable).split(".")[0] in cmdline.lower()
def test_windows_probes_get_a_deadline_longer_than_powershell_start(self, monkeypatch):
"""The Windows lookup must not be capped by the POSIX probe deadline.
wmic is absent from current Windows images, so the answer comes from a
fresh PowerShell, and its cold start alone can pass five seconds on a
loaded machine. Timing out is not a harmless miss: the caller reads None
as "not one of our processes", so the manager refuses to stop a daemon it
owns — and it made test-embed-windows flaky, which is how it surfaced.
"""
from hindsight_embed import daemon_embed_manager as dem
timeouts: list[float] = []
def _record(cmd, timeout=dem._PROBE_TIMEOUT_S):
timeouts.append(timeout)
return None
monkeypatch.setattr(dem.platform, "system", lambda: "Windows")
monkeypatch.setattr(DaemonEmbedManager, "_run_probe", staticmethod(_record))
assert DaemonEmbedManager._process_command_line(1234) is None
assert timeouts, "expected the Windows branch to run at least one probe"
assert all(t > dem._PROBE_TIMEOUT_S for t in timeouts), (
f"every Windows probe needs its own longer deadline, got {timeouts}"
)
def test_returns_none_for_a_pid_that_does_not_exist(self):
# Above the default pid_max on Linux and unused elsewhere, so no live
# process can answer and both the procfs and `ps` lookups must fail.
assert DaemonEmbedManager._process_command_line(4_294_967_000) is None
def test_unknown_pid_is_never_treated_as_ours(self):
"""The refusal path depends on an unreadable command line meaning "not ours"."""
from hindsight_embed.daemon_embed_manager import DAEMON_PROCESS_MARKERS
assert DaemonEmbedManager._process_matches(4_294_967_000, DAEMON_PROCESS_MARKERS) is False
Reference in New Issue View Git Blame Copy Permalink