feat(api_hook_client): add wait_for_project_switch for deterministic test waits
Adds a polling helper that blocks until the project switch completes, errors out, or times out. Replaces the fragile 10x1s blind poll in test_full_live_workflow with a condition-based wait on the /api/project_switch_status endpoint. Features: - Polls /api/project_switch_status every 200ms (configurable) - Returns immediately on error (with the error in the result) - Path matching: exact match OR basename match (handles absolute vs relative) - Times out with a clear 'timeout' flag instead of a generic assertion - Optional expected_path: if None, returns on any in_progress=False - src/api_hook_client.py: new wait_for_project_switch method (37 lines) - tests/test_api_hook_client_wait_for_project_switch.py: 6 unit tests with mocked _make_request covering all paths
This commit is contained in:
@@ -373,6 +373,35 @@ class ApiHookClient:
|
||||
return {"in_progress": False, "path": None, "error": None}
|
||||
return result
|
||||
|
||||
def wait_for_project_switch(self, expected_path: str = None, timeout: float = 30.0, poll_interval: float = 0.2) -> dict[str, Any]:
|
||||
"""
|
||||
Blocks until the project switch completes (or fails). Returns the final
|
||||
status dict. If expected_path is provided, also waits until the reported
|
||||
path matches it (or its basename matches).
|
||||
Fails fast (within timeout seconds) with a clear error if:
|
||||
- the controller reports an error during the switch
|
||||
- the path doesn't reach expected_path within the timeout
|
||||
- the controller is hung (in_progress stays True)
|
||||
[C: tests/test_live_workflow.py:test_full_live_workflow]
|
||||
"""
|
||||
import os
|
||||
start = time.time()
|
||||
last_status = {"in_progress": True, "path": None, "error": None}
|
||||
while time.time() - start < timeout:
|
||||
last_status = self.get_project_switch_status()
|
||||
if last_status.get("error"):
|
||||
return last_status
|
||||
if not last_status.get("in_progress"):
|
||||
if expected_path is None:
|
||||
return last_status
|
||||
if last_status.get("path") == expected_path:
|
||||
return last_status
|
||||
if last_status.get("path") and os.path.basename(str(last_status.get("path"))) == os.path.basename(expected_path):
|
||||
return last_status
|
||||
time.sleep(poll_interval)
|
||||
last_status["timeout"] = True
|
||||
return last_status
|
||||
|
||||
def post_project(self, project_data: dict) -> dict[str, Any]:
|
||||
"""
|
||||
Updates the current project configuration.
|
||||
|
||||
Reference in New Issue
Block a user