"""Tests for helpers.py — model info, bootstrap status, token tracking, system metrics.""" import asyncio import json from pathlib import Path from unittest.mock import AsyncMock, MagicMock import aiohttp import httpx import pytest from helpers import ( get_model_info, get_bootstrap_status, _update_lifetime_tokens, get_uptime, get_cpu_metrics, get_ram_metrics, check_service_health, get_all_services, get_llama_metrics, get_loaded_model, get_llama_context_size, get_disk_usage, dir_size_gb, invalidate_dir_size_cache, clear_dir_size_cache, _get_aio_session, set_services_cache, get_cached_services, _get_httpx_client, _get_lifetime_tokens, ) from models import BootstrapStatus, ServiceStatus, DiskUsage # --- get_model_info --- class TestGetModelInfo: def test_parses_32b_awq_model(self, install_dir): env_file = install_dir / ".env" env_file.write_text('LLM_MODEL=Qwen2.5-32B-Instruct-AWQ\n') info = get_model_info() assert info is not None assert info.name == "Qwen2.5-32B-Instruct-AWQ" assert info.size_gb == 16.0 assert info.quantization == "AWQ" def test_parses_7b_model(self, install_dir): env_file = install_dir / ".env" env_file.write_text('LLM_MODEL=Qwen2.5-7B-Instruct\n') info = get_model_info() assert info is not None assert info.size_gb == 4.0 assert info.quantization is None def test_parses_14b_gptq_model(self, install_dir): env_file = install_dir / ".env" env_file.write_text('LLM_MODEL=Qwen2.5-14B-Instruct-GPTQ\n') info = get_model_info() assert info is not None assert info.size_gb == 8.0 assert info.quantization == "GPTQ" def test_parses_70b_model(self, install_dir): env_file = install_dir / ".env" env_file.write_text('LLM_MODEL=Llama-3-70B-GGUF\n') info = get_model_info() assert info is not None assert info.size_gb == 35.0 assert info.quantization == "GGUF" def test_returns_none_when_no_env(self, install_dir): # No .env file created assert get_model_info() is None def test_returns_none_when_no_llm_model_line(self, install_dir): env_file = install_dir / ".env" env_file.write_text('SOME_OTHER_VAR=foo\n') assert get_model_info() is None def test_handles_quoted_value(self, install_dir): env_file = install_dir / ".env" env_file.write_text('LLM_MODEL="Qwen2.5-7B-Instruct"\n') info = get_model_info() assert info is not None assert info.name == "Qwen2.5-7B-Instruct" # --- get_bootstrap_status --- class TestGetBootstrapStatus: def test_inactive_when_no_file(self, data_dir): status = get_bootstrap_status() assert isinstance(status, BootstrapStatus) assert status.active is False def test_inactive_when_complete(self, data_dir): status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({"status": "complete"})) status = get_bootstrap_status() assert status.active is False def test_inactive_when_empty_status(self, data_dir): status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({"status": ""})) status = get_bootstrap_status() assert status.active is False def test_active_download(self, data_dir): status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({ "status": "downloading", "model": "Qwen2.5-32B", "percent": 42.5, "bytesDownloaded": 5 * 1024**3, "bytesTotal": 12 * 1024**3, "speedBytesPerSec": 50 * 1024**2, "eta": "3m 20s", })) status = get_bootstrap_status() assert status.active is True assert status.model_name == "Qwen2.5-32B" assert status.percent == 42.5 assert status.eta_seconds == 200 # 3*60 + 20 def test_eta_calculating(self, data_dir): status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({ "status": "downloading", "percent": 1.0, "eta": "calculating...", })) status = get_bootstrap_status() assert status.active is True assert status.eta_seconds is None def test_handles_malformed_json(self, data_dir): status_file = data_dir / "bootstrap-status.json" status_file.write_text("not json!") status = get_bootstrap_status() assert status.active is False def test_inactive_when_failed(self, data_dir): status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({"status": "failed", "model": "test.gguf"})) status = get_bootstrap_status() assert status.active is False def test_inactive_when_model_file_on_disk(self, data_dir): models_dir = data_dir / "models" models_dir.mkdir(exist_ok=True) (models_dir / "present.gguf").write_bytes(b"\x00" * 1024) status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({ "status": "downloading", "model": "present.gguf", "percent": 50, "bytesDownloaded": 500, "bytesTotal": 1024, })) status = get_bootstrap_status() assert status.active is False def test_active_during_verifying_even_if_file_exists(self, data_dir): models_dir = data_dir / "models" models_dir.mkdir(exist_ok=True) (models_dir / "verifying.gguf").write_bytes(b"\x00" * 1024) status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({ "status": "verifying", "model": "verifying.gguf", "percent": 100, "bytesDownloaded": 1024, "bytesTotal": 1024, })) status = get_bootstrap_status() assert status.active is True def test_active_during_swapping_even_if_file_exists(self, data_dir): models_dir = data_dir / "models" models_dir.mkdir(exist_ok=True) (models_dir / "swapping.gguf").write_bytes(b"\x00" * 1024) status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({ "status": "swapping", "model": "swapping.gguf", "percent": 100, "bytesDownloaded": 1024, "bytesTotal": 1024, })) status = get_bootstrap_status() assert status.active is True def test_path_traversal_rejected(self, data_dir): status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({ "status": "downloading", "model": "../../etc/passwd", "percent": 50, "bytesDownloaded": 500, "bytesTotal": 1000, })) status = get_bootstrap_status() assert status.active is True # --- _update_lifetime_tokens --- class TestUpdateLifetimeTokens: def test_fresh_start(self, data_dir): result = _update_lifetime_tokens(100.0) assert result == 100 def test_accumulates_across_calls(self, data_dir): _update_lifetime_tokens(100.0) result = _update_lifetime_tokens(250.0) assert result == 250 # 100 + (250 - 100) def test_handles_server_restart(self, data_dir): """When server_counter < prev, the counter has reset.""" _update_lifetime_tokens(500.0) # Server restarted, counter back to 50 result = _update_lifetime_tokens(50.0) # Should add 50 (treats reset counter as fresh delta) assert result == 550 # 500 + 50 def test_handles_corrupted_token_file(self, data_dir): """Corrupted JSON should log a warning and start fresh.""" token_file = data_dir / "token_counter.json" token_file.write_text("not valid json{{{") result = _update_lifetime_tokens(100.0) assert result == 100 def test_handles_unwritable_token_file(self, data_dir, monkeypatch): """When the token file cannot be written, should not raise.""" import helpers monkeypatch.setattr(helpers, "_TOKEN_FILE", data_dir / "readonly" / "token.json") # Parent dir doesn't exist, so write will fail result = _update_lifetime_tokens(50.0) assert result == 50 # --- System metrics (cross-platform) --- class TestGetUptime: def test_returns_int(self): result = get_uptime() assert isinstance(result, int) assert result >= 0 def test_returns_zero_on_unsupported_platform(self, monkeypatch): monkeypatch.setattr("helpers.platform.system", lambda: "UnknownOS") assert get_uptime() == 0 class TestGetCpuMetrics: def test_returns_expected_keys(self): result = get_cpu_metrics() assert "percent" in result assert "temp_c" in result assert isinstance(result["percent"], (int, float)) def test_returns_defaults_on_unsupported_platform(self, monkeypatch): monkeypatch.setattr("helpers.platform.system", lambda: "UnknownOS") result = get_cpu_metrics() assert result == {"percent": 0, "temp_c": None} class TestGetRamMetrics: def test_returns_expected_keys(self): result = get_ram_metrics() assert "used_gb" in result assert "total_gb" in result assert "percent" in result def test_returns_defaults_on_unsupported_platform(self, monkeypatch): monkeypatch.setattr("helpers.platform.system", lambda: "UnknownOS") result = get_ram_metrics() assert result == {"used_gb": 0, "total_gb": 0, "percent": 0} # --- check_service_health --- class TestCheckServiceHealth: _CONFIG = { "name": "test-svc", "port": 8080, "external_port": 8080, "health": "/health", "host": "localhost", } @pytest.mark.asyncio async def test_healthy_on_200(self, mock_aiohttp_session, monkeypatch): session = mock_aiohttp_session(status=200) monkeypatch.setattr("helpers._get_aio_session", AsyncMock(return_value=session)) result = await check_service_health("test-svc", self._CONFIG) assert result.status == "healthy" assert result.id == "test-svc" assert result.port == 8080 @pytest.mark.asyncio async def test_sends_host_localhost_header(self, mock_aiohttp_session, monkeypatch): session = mock_aiohttp_session(status=200) monkeypatch.setattr("helpers._get_aio_session", AsyncMock(return_value=session)) await check_service_health("test-svc", self._CONFIG) session.get.assert_called_once() _, kwargs = session.get.call_args assert kwargs.get("headers", {}).get("Host") == "localhost" @pytest.mark.asyncio async def test_unhealthy_on_500(self, mock_aiohttp_session, monkeypatch): session = mock_aiohttp_session(status=500) monkeypatch.setattr("helpers._get_aio_session", AsyncMock(return_value=session)) result = await check_service_health("test-svc", self._CONFIG) assert result.status == "unhealthy" @pytest.mark.asyncio async def test_degraded_on_timeout(self, monkeypatch): session = MagicMock() session.get = MagicMock(side_effect=asyncio.TimeoutError()) monkeypatch.setattr("helpers._get_aio_session", AsyncMock(return_value=session)) result = await check_service_health("test-svc", self._CONFIG) assert result.status == "degraded" @pytest.mark.asyncio async def test_not_deployed_on_dns_failure(self, monkeypatch): from collections import namedtuple ConnKey = namedtuple('ConnectionKey', ['host', 'port', 'is_ssl', 'ssl', 'proxy', 'proxy_auth', 'proxy_headers_hash']) conn_key = ConnKey('test-svc', 8080, False, None, None, None, None) os_err = OSError("Name or service not known") os_err.strerror = "Name or service not known" exc = aiohttp.ClientConnectorError(conn_key, os_err) session = MagicMock() session.get = MagicMock(side_effect=exc) monkeypatch.setattr("helpers._get_aio_session", AsyncMock(return_value=session)) result = await check_service_health("test-svc", self._CONFIG) assert result.status == "not_deployed" @pytest.mark.asyncio async def test_down_on_connection_refused(self, monkeypatch): conn_key = MagicMock() exc = aiohttp.ClientConnectorError(conn_key, OSError("Connection refused")) session = MagicMock() session.get = MagicMock(side_effect=exc) monkeypatch.setattr("helpers._get_aio_session", AsyncMock(return_value=session)) result = await check_service_health("test-svc", self._CONFIG) assert result.status == "down" @pytest.mark.asyncio async def test_down_on_os_error(self, monkeypatch): session = MagicMock() session.get = MagicMock(side_effect=OSError("connection refused")) monkeypatch.setattr("helpers._get_aio_session", AsyncMock(return_value=session)) result = await check_service_health("test-svc", self._CONFIG) assert result.status == "down" @pytest.mark.asyncio async def test_host_network_portless_service_is_not_deployed(self): config = { "name": "Portless", "port": 0, "external_port": 0, "health": "/health", "host": "localhost", "host_network": True, } result = await check_service_health("portless", config) assert result.status == "not_deployed" @pytest.mark.asyncio async def test_tailscale_not_running_is_not_deployed(self, monkeypatch): config = { "name": "Tailscale", "port": 0, "external_port": 0, "health": "/health", "host": "localhost", "host_network": True, } response = MagicMock() response.status_code = 200 response.json.return_value = {"running": False} client = MagicMock() client.get = AsyncMock(return_value=response) monkeypatch.setattr("helpers._get_httpx_client", AsyncMock(return_value=client)) result = await check_service_health("tailscale", config) assert result.status == "not_deployed" @pytest.mark.asyncio async def test_tailscale_authenticated_is_healthy(self, monkeypatch): config = { "name": "Tailscale", "port": 0, "external_port": 0, "health": "/health", "host": "localhost", "host_network": True, } response = MagicMock() response.status_code = 200 response.json.return_value = {"running": True, "authenticated": True} client = MagicMock() client.get = AsyncMock(return_value=response) monkeypatch.setattr("helpers._get_httpx_client", AsyncMock(return_value=client)) result = await check_service_health("tailscale", config) assert result.status == "healthy" # --- get_all_services --- class TestGetAllServices: @pytest.mark.asyncio async def test_returns_all_statuses(self, monkeypatch): fake_services = { "svc-a": {"name": "Service A", "port": 8001, "external_port": 8001, "health": "/health", "host": "localhost"}, "svc-b": {"name": "Service B", "port": 8002, "external_port": 8002, "health": "/health", "host": "localhost"}, } monkeypatch.setattr("helpers.SERVICES", fake_services) async def fake_health(sid, cfg): return ServiceStatus(id=sid, name=cfg["name"], port=cfg["port"], external_port=cfg["external_port"], status="healthy") monkeypatch.setattr("helpers.check_service_health", fake_health) result = await get_all_services() assert len(result) == 2 ids = {s.id for s in result} assert ids == {"svc-a", "svc-b"} @pytest.mark.asyncio async def test_exception_in_one_service_returns_down(self, monkeypatch): fake_services = { "ok-svc": {"name": "OK", "port": 8001, "external_port": 8001, "health": "/health", "host": "localhost"}, "bad-svc": {"name": "Bad", "port": 8002, "external_port": 8002, "health": "/health", "host": "localhost"}, } monkeypatch.setattr("helpers.SERVICES", fake_services) async def fake_health(sid, cfg): if sid == "bad-svc": raise RuntimeError("unexpected failure") return ServiceStatus(id=sid, name=cfg["name"], port=cfg["port"], external_port=cfg["external_port"], status="healthy") monkeypatch.setattr("helpers.check_service_health", fake_health) result = await get_all_services() assert len(result) == 2 bad = next(s for s in result if s.id == "bad-svc") assert bad.status == "down" ok = next(s for s in result if s.id == "ok-svc") assert ok.status == "healthy" @pytest.mark.asyncio async def test_empty_services_returns_empty(self, monkeypatch): monkeypatch.setattr("helpers.SERVICES", {}) result = await get_all_services() assert result == [] # --- get_llama_metrics --- class TestGetLlamaMetrics: @pytest.mark.asyncio async def test_parses_prometheus_metrics(self, monkeypatch): from conftest import load_golden_fixture prom_text = load_golden_fixture("prometheus_metrics.txt") fake_services = { "llama-server": {"host": "localhost", "port": 8080, "health": "/health", "name": "llama-server"}, } monkeypatch.setattr("helpers.SERVICES", fake_services) # Reset the previous token state so TPS calculation is fresh import helpers helpers._prev_tokens.update({"count": 0, "time": 0.0, "tps": 0.0}) mock_response = MagicMock() mock_response.text = prom_text mock_response.status_code = 200 mock_client = AsyncMock() mock_client.get = AsyncMock(return_value=mock_response) mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) monkeypatch.setattr("helpers.httpx.AsyncClient", lambda **kw: mock_client) result = await get_llama_metrics(model_hint="test-model") assert "tokens_per_second" in result assert "lifetime_tokens" in result assert isinstance(result["tokens_per_second"], (int, float)) @pytest.mark.asyncio async def test_returns_zero_on_failure(self, monkeypatch): fake_services = { "llama-server": {"host": "localhost", "port": 8080, "health": "/health", "name": "llama-server"}, } monkeypatch.setattr("helpers.SERVICES", fake_services) mock_client = AsyncMock() mock_client.get = AsyncMock(side_effect=OSError("connection refused")) mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) monkeypatch.setattr("helpers.httpx.AsyncClient", lambda **kw: mock_client) result = await get_llama_metrics(model_hint="test-model") assert result["tokens_per_second"] == 0 # --- get_loaded_model --- class TestGetLoadedModel: @pytest.mark.asyncio async def test_returns_model_with_loaded_status(self, monkeypatch): fake_services = { "llama-server": {"host": "localhost", "port": 8080, "health": "/health", "name": "llama-server"}, } monkeypatch.setattr("helpers.SERVICES", fake_services) mock_response = MagicMock() mock_response.json = MagicMock(return_value={ "data": [ {"id": "idle-model", "status": {"value": "idle"}}, {"id": "loaded-model", "status": {"value": "loaded"}}, ] }) mock_client = AsyncMock() mock_client.get = AsyncMock(return_value=mock_response) mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) monkeypatch.setattr("helpers.httpx.AsyncClient", lambda **kw: mock_client) result = await get_loaded_model() assert result == "loaded-model" @pytest.mark.asyncio async def test_returns_first_model_when_no_loaded(self, monkeypatch): fake_services = { "llama-server": {"host": "localhost", "port": 8080, "health": "/health", "name": "llama-server"}, } monkeypatch.setattr("helpers.SERVICES", fake_services) mock_response = MagicMock() mock_response.json = MagicMock(return_value={ "data": [ {"id": "only-model", "status": {"value": "idle"}}, ] }) mock_client = AsyncMock() mock_client.get = AsyncMock(return_value=mock_response) mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) monkeypatch.setattr("helpers.httpx.AsyncClient", lambda **kw: mock_client) result = await get_loaded_model() assert result == "only-model" @pytest.mark.asyncio async def test_returns_none_on_failure(self, monkeypatch): fake_services = { "llama-server": {"host": "localhost", "port": 8080, "health": "/health", "name": "llama-server"}, } monkeypatch.setattr("helpers.SERVICES", fake_services) mock_client = AsyncMock() mock_client.get = AsyncMock(side_effect=httpx.ConnectError("unreachable")) mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) monkeypatch.setattr("helpers.httpx.AsyncClient", lambda **kw: mock_client) result = await get_loaded_model() assert result is None # --- get_llama_context_size --- class TestGetLlamaContextSize: @pytest.mark.asyncio async def test_returns_n_ctx(self, monkeypatch): fake_services = { "llama-server": {"host": "localhost", "port": 8080, "health": "/health", "name": "llama-server"}, } monkeypatch.setattr("helpers.SERVICES", fake_services) mock_response = MagicMock() mock_response.json = MagicMock(return_value={ "default_generation_settings": {"n_ctx": 32768} }) mock_client = AsyncMock() mock_client.get = AsyncMock(return_value=mock_response) mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) monkeypatch.setattr("helpers.httpx.AsyncClient", lambda **kw: mock_client) result = await get_llama_context_size(model_hint="test-model") assert result == 32768 @pytest.mark.asyncio async def test_returns_none_on_failure(self, monkeypatch): fake_services = { "llama-server": {"host": "localhost", "port": 8080, "health": "/health", "name": "llama-server"}, } monkeypatch.setattr("helpers.SERVICES", fake_services) mock_client = AsyncMock() mock_client.get = AsyncMock(side_effect=httpx.ConnectError("unreachable")) mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) monkeypatch.setattr("helpers.httpx.AsyncClient", lambda **kw: mock_client) result = await get_llama_context_size(model_hint="test-model") assert result is None # --- get_disk_usage --- class TestGetDiskUsage: def test_returns_disk_usage(self, monkeypatch): monkeypatch.setattr("helpers.INSTALL_DIR", "/tmp") result = get_disk_usage() assert isinstance(result, DiskUsage) assert result.total_gb > 0 assert result.used_gb >= 0 assert 0 <= result.percent <= 100 def test_falls_back_to_home_dir(self, monkeypatch): monkeypatch.setattr("helpers.INSTALL_DIR", "/nonexistent/path/that/does/not/exist") import os result = get_disk_usage() assert isinstance(result, DiskUsage) assert result.path == os.path.expanduser("~") assert result.total_gb > 0 # --- _get_aio_session --- class TestGetAioSession: @pytest.mark.asyncio async def test_creates_session(self, monkeypatch): import helpers monkeypatch.setattr(helpers, "_aio_session", None) monkeypatch.setattr(helpers, "_aio_session_lock", None) session = await _get_aio_session() assert session is not None await session.close() @pytest.mark.asyncio async def test_reuses_session(self, monkeypatch): import helpers monkeypatch.setattr(helpers, "_aio_session", None) monkeypatch.setattr(helpers, "_aio_session_lock", None) s1 = await _get_aio_session() s2 = await _get_aio_session() assert s1 is s2 await s1.close() @pytest.mark.asyncio async def test_waits_for_singleton_lock_before_creating_session(self, monkeypatch): import helpers lock = asyncio.Lock() await lock.acquire() monkeypatch.setattr(helpers, "_aio_session", None) monkeypatch.setattr(helpers, "_aio_session_lock", lock) task = asyncio.create_task(_get_aio_session()) await asyncio.sleep(0) assert not task.done() lock.release() session = await task assert session is not None await session.close() # --- _get_httpx_client --- class TestGetHttpxClient: @pytest.mark.asyncio async def test_reuses_client(self, monkeypatch): import helpers monkeypatch.setattr(helpers, "_httpx_client", None) monkeypatch.setattr(helpers, "_httpx_client_lock", None) c1 = await _get_httpx_client() c2 = await _get_httpx_client() assert c1 is c2 await c1.aclose() @pytest.mark.asyncio async def test_waits_for_singleton_lock_before_creating_client(self, monkeypatch): import helpers lock = asyncio.Lock() await lock.acquire() monkeypatch.setattr(helpers, "_httpx_client", None) monkeypatch.setattr(helpers, "_httpx_client_lock", lock) task = asyncio.create_task(_get_httpx_client()) await asyncio.sleep(0) assert not task.done() lock.release() client = await task assert client is not None await client.aclose() # --- set_services_cache / get_cached_services --- class TestServicesCache: def test_set_and_get(self, monkeypatch): import helpers monkeypatch.setattr(helpers, "_services_cache", None) assert get_cached_services() is None fake = [ServiceStatus(id="s", name="S", port=80, external_port=80, status="healthy")] set_services_cache(fake) assert get_cached_services() == fake def test_optional_host_systemd_down_is_cached_as_not_deployed(self, monkeypatch): import helpers monkeypatch.setattr(helpers, "_services_cache", None) monkeypatch.setattr(helpers, "SERVICES", { "opencode": { "name": "OpenCode (IDE)", "port": 3003, "external_port": 3003, "health": "/", "host": "localhost", "type": "host-systemd", "category": "optional", }, "dashboard-api": { "name": "Dashboard API", "port": 3002, "external_port": 3002, "health": "/health", "host": "localhost", }, }) set_services_cache([ ServiceStatus( id="opencode", name="OpenCode (IDE)", port=3003, external_port=3003, status="down", ), ServiceStatus( id="dashboard-api", name="Dashboard API", port=3002, external_port=3002, status="down", ), ]) cached = {service.id: service for service in get_cached_services()} assert cached["opencode"].status == "not_deployed" assert cached["dashboard-api"].status == "down" # --- _get_lifetime_tokens --- class TestGetLifetimeTokens: def test_returns_zero_when_no_file(self, data_dir): assert _get_lifetime_tokens() == 0 def test_returns_lifetime_from_file(self, data_dir): token_file = data_dir / "token_counter.json" token_file.write_text(json.dumps({"lifetime": 42})) assert _get_lifetime_tokens() == 42 # --- check_service_health host-systemd --- class TestCheckServiceHealthSystemd: @pytest.mark.asyncio async def test_host_systemd_returns_healthy_when_host_agent_proves_port(self, monkeypatch): class FakeResponse: status_code = 200 def json(self): return {"reachable": True, "response_time_ms": 12.3} class FakeClient: async def get(self, url, *, params=None, headers=None): assert url.endswith("/v1/host/port") assert params == {"host": "127.0.0.1", "port": 3003} return FakeResponse() async def fake_client(): return FakeClient() monkeypatch.setattr("helpers._get_httpx_client", fake_client) config = { "name": "opencode", "port": 3003, "external_port": 3003, "health": "/health", "host": "localhost", "type": "host-systemd", } result = await check_service_health("opencode", config) assert result.status == "healthy" assert result.response_time_ms == 12.3 @pytest.mark.asyncio async def test_host_systemd_returns_not_deployed_when_host_port_closed(self, monkeypatch): class FakeResponse: status_code = 200 def json(self): return {"reachable": False, "response_time_ms": 2.0} class FakeClient: async def get(self, url, *, params=None, headers=None): return FakeResponse() async def fake_client(): return FakeClient() monkeypatch.setattr("helpers._get_httpx_client", fake_client) config = { "name": "opencode", "port": 3003, "external_port": 3003, "health": "/health", "host": "localhost", "type": "host-systemd", } result = await check_service_health("opencode", config) assert result.status == "not_deployed" assert result.response_time_ms == 2.0 # --- get_model_info error branch --- class TestGetModelInfoErrors: def test_returns_none_on_os_error(self, install_dir, monkeypatch): env_file = install_dir / ".env" env_file.write_text('LLM_MODEL=test\n') # Make the open fail after exists() returns True import builtins orig_open = builtins.open def failing_open(path, *a, **kw): if str(path).endswith(".env"): raise OSError("permission denied") return orig_open(path, *a, **kw) monkeypatch.setattr(builtins, "open", failing_open) assert get_model_info() is None # --- get_bootstrap_status eta/percent branches --- class TestGetBootstrapStatusEdgeCases: def test_eta_single_seconds_value(self, data_dir): status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({ "status": "downloading", "percent": 90, "eta": "45s", })) status = get_bootstrap_status() assert status.active is True assert status.eta_seconds == 45 def test_invalid_percent_type(self, data_dir): status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({ "status": "downloading", "percent": "not-a-number", "bytesDownloaded": 100, })) status = get_bootstrap_status() assert status.active is True assert status.percent is None def test_speed_and_sizes(self, data_dir): status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({ "status": "downloading", "bytesDownloaded": 2 * 1024**3, "bytesTotal": 10 * 1024**3, "speedBytesPerSec": 100 * 1024**2, })) status = get_bootstrap_status() assert status.active is True assert status.downloaded_gb is not None assert abs(status.downloaded_gb - 2.0) < 0.01 assert status.speed_mbps is not None def test_oversized_progress_is_clamped_for_active_download(self, data_dir): status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({ "status": "downloading", "model": "Full.gguf", "percent": 143.2, "bytesDownloaded": 150, "bytesTotal": 100, })) status = get_bootstrap_status() assert status.active is True assert status.percent == 100.0 assert status.downloaded_gb == 100 / (1024**3) assert status.total_gb == 100 / (1024**3) # --- get_uptime platform branches --- class TestGetUptimePlatforms: def test_linux_reads_proc_uptime(self, monkeypatch): monkeypatch.setattr("helpers.platform.system", lambda: "Linux") import builtins orig_open = builtins.open def fake_open(path, *a, **kw): if str(path) == "/proc/uptime": from io import StringIO return StringIO("12345.67 9876.54") return orig_open(path, *a, **kw) monkeypatch.setattr(builtins, "open", fake_open) assert get_uptime() == 12345 def test_darwin_branch(self, monkeypatch): monkeypatch.setattr("helpers.platform.system", lambda: "Darwin") import time mock_result = MagicMock() mock_result.returncode = 0 boot_time = int(time.time()) - 600 mock_result.stdout = f"{{ sec = {boot_time}, usec = 0 }} Mon Jan 1 00:00:00 2026" monkeypatch.setattr("subprocess.run", lambda *a, **kw: mock_result) result = get_uptime() assert 595 <= result <= 610 # --- get_llama_metrics TPS calculation branch --- class TestGetLlamaMetricsTPS: @pytest.mark.asyncio async def test_tps_calculated_on_second_call(self, monkeypatch): """TPS is calculated when previous token count and gen_secs are set.""" import helpers import time as _time fake_services = { "llama-server": {"host": "localhost", "port": 8080, "health": "/health", "name": "llama-server"}, } monkeypatch.setattr("helpers.SERVICES", fake_services) # Set up previous state helpers._prev_tokens.update({"count": 100, "time": _time.time() - 1, "tps": 0.0, "gen_secs": 5.0}) # Mock response with updated token counts mock_response = MagicMock() mock_response.text = ( "# HELP tokens_predicted_total\n" "tokens_predicted_total 200\n" "# HELP tokens_predicted_seconds_total\n" "tokens_predicted_seconds_total 10.0\n" ) mock_response.status_code = 200 mock_client = AsyncMock() mock_client.get = AsyncMock(return_value=mock_response) mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) monkeypatch.setattr("helpers.httpx.AsyncClient", lambda **kw: mock_client) result = await get_llama_metrics(model_hint="test") # 100 tokens / 5 seconds = 20.0 tps assert result["tokens_per_second"] == 20.0 # --- bootstrap status ETA edge cases --- class TestBootstrapStatusEtaEdge: def test_invalid_eta_string(self, data_dir): """ETA with unparseable content → eta_seconds is None.""" status_file = data_dir / "bootstrap-status.json" status_file.write_text(json.dumps({ "status": "downloading", "percent": 50, "eta": "not a number at all", })) status = get_bootstrap_status() assert status.active is True assert status.eta_seconds is None # --- dir_size_gb --- class TestDirSizeGb: @staticmethod def _symlink_or_skip(link: Path, target: Path): try: link.symlink_to(target) except OSError as exc: pytest.skip(f"symlink creation unavailable in this test environment: {exc}") def test_nonexistent_path_returns_zero(self, tmp_path): clear_dir_size_cache() assert dir_size_gb(tmp_path / "does-not-exist") == 0.0 def test_empty_directory_returns_zero(self, tmp_path): clear_dir_size_cache() empty = tmp_path / "empty" empty.mkdir() assert dir_size_gb(empty) == 0.0 def test_directory_with_files(self, tmp_path): clear_dir_size_cache() d = tmp_path / "data" d.mkdir() # Write 100 MiB (avoids allocating 1 GiB in CI) size = 1024 * 1024 * 100 (d / "bigfile.bin").write_bytes(b"\x00" * size) assert dir_size_gb(d) == 0.1 def test_symlinks_are_skipped(self, tmp_path): clear_dir_size_cache() d = tmp_path / "withlinks" d.mkdir() real = d / "real.bin" real.write_bytes(b"\x00" * 1024) link = d / "link.bin" self._symlink_or_skip(link, real) # Only real.bin should be counted (1024 B ≈ 0.0 GB when rounded to 2dp) result = dir_size_gb(d) assert result == 0.0 # 1024 bytes rounds to 0.0 GB def test_checks_symlink_before_is_file(self, tmp_path, monkeypatch): clear_dir_size_cache() d = tmp_path / "withlinks" d.mkdir() outside = tmp_path / "outside.bin" outside.write_bytes(b"\x00" * 1024) link = d / "outside-link.bin" self._symlink_or_skip(link, outside) original_is_file = Path.is_file def guarded_is_file(self): if self == link: raise AssertionError("dir_size_gb called is_file before skipping symlink") return original_is_file(self) monkeypatch.setattr(Path, "is_file", guarded_is_file) assert dir_size_gb(d) == 0.0 def test_uses_cached_value_until_invalidated(self, tmp_path, monkeypatch): clear_dir_size_cache() d = tmp_path / "cached" d.mkdir() (d / "data.bin").write_bytes(b"\x00" * 1024) assert dir_size_gb(d) == 0.0 def _unexpected_rglob(self, pattern): raise AssertionError("dir_size_gb unexpectedly walked the filesystem") monkeypatch.setattr(Path, "rglob", _unexpected_rglob) assert dir_size_gb(d) == 0.0 def test_invalidate_dir_size_cache_forces_refresh(self, tmp_path, monkeypatch): clear_dir_size_cache() d = tmp_path / "refresh" d.mkdir() (d / "data.bin").write_bytes(b"\x00" * 1024) assert dir_size_gb(d) == 0.0 original_rglob = Path.rglob calls = {"count": 0} def _tracking_rglob(self, pattern): calls["count"] += 1 return original_rglob(self, pattern) monkeypatch.setattr(Path, "rglob", _tracking_rglob) assert dir_size_gb(d) == 0.0 assert calls["count"] == 0 invalidate_dir_size_cache(d) assert dir_size_gb(d) == 0.0 assert calls["count"] == 1