ed
69b7ab670d
The live_gui subprocess spawns the desktop GUI, which creates AppController with defer_warmup=True (src/gui_2.py:318). Warmup is deferred until the first frame is painted (src/gui_2.py:1076). The previous test queried /api/warmup_canaries immediately after wait_for_server, racing against the first frame - canary list was empty until start_warmup() ran. Replace the immediate assert with a poll-with-retry loop (15s deadline, 0.5s interval) per workflow.md 'Async Setters Need Poll-For-State' rule.
140 lines
5.3 KiB
Python
140 lines
5.3 KiB
Python
import pytest
|
|
import sys
|
|
import os
|
|
import time
|
|
from unittest.mock import patch
|
|
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
|
|
from src.api_hook_client import ApiHookClient
|
|
|
|
|
|
def test_get_warmup_status_calls_correct_endpoint() -> None:
|
|
"""get_warmup_status() hits GET /api/warmup_status."""
|
|
client = ApiHookClient()
|
|
with patch.object(client, "_make_request") as mock_make:
|
|
mock_make.return_value = {"pending": [], "completed": ["a", "b"], "failed": []}
|
|
result = client.get_warmup_status()
|
|
assert "pending" in result
|
|
assert "completed" in result
|
|
assert "failed" in result
|
|
mock_make.assert_called_once_with("GET", "/api/warmup_status")
|
|
|
|
|
|
def test_get_warmup_status_handles_empty_response() -> None:
|
|
"""get_warmup_status() returns empty dict when server returns None/empty."""
|
|
client = ApiHookClient()
|
|
with patch.object(client, "_make_request") as mock_make:
|
|
mock_make.return_value = None
|
|
result = client.get_warmup_status()
|
|
assert result == {}
|
|
|
|
|
|
def test_get_warmup_wait_passes_timeout_as_query_string() -> None:
|
|
"""get_warmup_wait(timeout=N) hits GET /api/warmup_wait?timeout=N."""
|
|
client = ApiHookClient()
|
|
with patch.object(client, "_make_request") as mock_make:
|
|
mock_make.return_value = {"pending": [], "completed": ["x"], "failed": []}
|
|
result = client.get_warmup_wait(timeout=12.5)
|
|
assert "completed" in result
|
|
mock_make.assert_called_once_with("GET", "/api/warmup_wait?timeout=12.5")
|
|
|
|
|
|
def test_get_warmup_wait_uses_default_timeout_when_unspecified() -> None:
|
|
"""get_warmup_wait() with no args uses the default timeout=30.0."""
|
|
client = ApiHookClient()
|
|
with patch.object(client, "_make_request") as mock_make:
|
|
mock_make.return_value = {"pending": [], "completed": [], "failed": []}
|
|
client.get_warmup_wait()
|
|
args, _ = mock_make.call_args
|
|
assert args[0] == "GET"
|
|
assert args[1].startswith("/api/warmup_wait?timeout=")
|
|
assert "30" in args[1]
|
|
|
|
|
|
def test_get_warmup_wait_handles_empty_response() -> None:
|
|
"""get_warmup_wait() returns empty dict when server returns None."""
|
|
client = ApiHookClient()
|
|
with patch.object(client, "_make_request") as mock_make:
|
|
mock_make.return_value = None
|
|
result = client.get_warmup_wait(timeout=5.0)
|
|
assert result == {}
|
|
|
|
|
|
def test_live_warmup_status_endpoint(live_gui) -> None:
|
|
"""Live: GET /api/warmup_status returns 200 + warmup dict (sub-track 3)."""
|
|
client = ApiHookClient()
|
|
assert client.wait_for_server(timeout=10)
|
|
status = client.get_warmup_status()
|
|
assert "pending" in status
|
|
assert "completed" in status
|
|
assert "failed" in status
|
|
assert isinstance(status["pending"], list)
|
|
assert isinstance(status["completed"], list)
|
|
assert isinstance(status["failed"], list)
|
|
|
|
|
|
def test_live_warmup_wait_endpoint_completes(live_gui) -> None:
|
|
"""Live: GET /api/warmup_wait?timeout=2.0 returns the (likely-completed) status."""
|
|
client = ApiHookClient()
|
|
assert client.wait_for_server(timeout=10)
|
|
result = client.get_warmup_wait(timeout=2.0)
|
|
assert "pending" in result
|
|
assert "completed" in result
|
|
assert "failed" in result
|
|
# In a live session the warmup either already finished (no pending) or
|
|
# completed within the 2s window. Either way the response is well-formed.
|
|
assert isinstance(result["pending"], list)
|
|
|
|
def test_get_warmup_canaries_calls_correct_endpoint() -> None:
|
|
"""get_warmup_canaries() hits GET /api/warmup_canaries and unwraps the list."""
|
|
client = ApiHookClient()
|
|
with patch.object(client, "_make_request") as mock_make:
|
|
mock_make.return_value = {"canaries": [{"canary_id": 1, "module": "os", "thread_name": "controller-io-0", "thread_id": 12345, "status": "completed"}]}
|
|
result = client.get_warmup_canaries()
|
|
assert isinstance(result, list)
|
|
assert len(result) == 1
|
|
assert result[0]["module"] == "os"
|
|
assert result[0]["status"] == "completed"
|
|
mock_make.assert_called_once_with("GET", "/api/warmup_canaries")
|
|
|
|
|
|
def test_get_warmup_canaries_handles_empty_response() -> None:
|
|
"""get_warmup_canaries() returns [] when server returns None or empty payload."""
|
|
client = ApiHookClient()
|
|
with patch.object(client, "_make_request") as mock_make:
|
|
mock_make.return_value = None
|
|
result = client.get_warmup_canaries()
|
|
assert result == []
|
|
|
|
|
|
def test_live_warmup_canaries_endpoint(live_gui) -> None:
|
|
"""Live: GET /api/warmup_canaries returns canary records with thread + status info.
|
|
|
|
The canary list is populated by WarmupManager.submit() which is called
|
|
from AppController.start_warmup(). For the live_gui subprocess the warmup
|
|
is deferred until the first frame is painted (src/gui_2.py:1076). This
|
|
test polls until at least one canary appears, with a 15s deadline, to
|
|
handle the race between server-ready and first-frame-painted.
|
|
"""
|
|
client = ApiHookClient()
|
|
assert client.wait_for_server(timeout=10)
|
|
canaries: list = []
|
|
deadline = time.time() + 15.0
|
|
while time.time() < deadline:
|
|
canaries = client.get_warmup_canaries()
|
|
if canaries:
|
|
break
|
|
time.sleep(0.5)
|
|
assert isinstance(canaries, list)
|
|
assert len(canaries) >= 1, "expected at least one canary record from live warmup"
|
|
for c in canaries:
|
|
assert "canary_id" in c
|
|
assert "module" in c
|
|
assert "thread_name" in c
|
|
assert "thread_id" in c
|
|
assert "status" in c
|
|
assert c["status"] in ("running", "completed", "failed", "cancelled")
|
|
completed = [c for c in canaries if c["status"] == "completed"]
|
|
assert len(completed) >= 1, "at least one canary should be completed"
|