test(conductor): Add integration test for ApiHookClient usage in phase completion
This commit is contained in:
78
tests/test_conductor_api_hook_integration.py
Normal file
78
tests/test_conductor_api_hook_integration.py
Normal file
@@ -0,0 +1,78 @@
|
|||||||
|
import pytest
|
||||||
|
from unittest.mock import MagicMock, patch
|
||||||
|
import os
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
import json
|
||||||
|
|
||||||
|
from api_hooks import HookServer
|
||||||
|
from api_hook_client import ApiHookClient
|
||||||
|
|
||||||
|
@pytest.fixture(scope="module")
|
||||||
|
def hook_server_fixture_for_integration():
|
||||||
|
# Mock the 'app' object that HookServer expects
|
||||||
|
mock_app = MagicMock()
|
||||||
|
mock_app.test_hooks_enabled = True # Essential for the server to start
|
||||||
|
mock_app.project = {'name': 'test_project'}
|
||||||
|
mock_app.disc_entries = [{'role': 'user', 'content': 'hello'}]
|
||||||
|
mock_app._pending_gui_tasks = []
|
||||||
|
mock_app._pending_gui_tasks_lock = threading.Lock()
|
||||||
|
|
||||||
|
# Use an ephemeral port (0) to avoid conflicts
|
||||||
|
server = HookServer(mock_app, port=0)
|
||||||
|
server.start()
|
||||||
|
|
||||||
|
time.sleep(0.1) # Wait a moment for the server thread to start and bind
|
||||||
|
|
||||||
|
actual_port = server.server.server_address[1]
|
||||||
|
client_base_url = f"http://127.0.0.1:{actual_port}"
|
||||||
|
|
||||||
|
yield client_base_url, mock_app
|
||||||
|
|
||||||
|
server.stop()
|
||||||
|
|
||||||
|
|
||||||
|
def simulate_conductor_phase_completion(client_base_url: str, mock_app: MagicMock, plan_content: str):
|
||||||
|
"""
|
||||||
|
Simulates the Conductor agent's logic for phase completion.
|
||||||
|
This function, in the *actual* implementation, will be *my* (the agent's) code.
|
||||||
|
"""
|
||||||
|
print(f"Simulating Conductor phase completion. Client base URL: {client_base_url}")
|
||||||
|
# In the actual Conductor implementation, the agent would instantiate ApiHookClient
|
||||||
|
# and call its verification methods.
|
||||||
|
client = ApiHookClient(base_url=client_base_url)
|
||||||
|
|
||||||
|
# Perform a verification call, for example, get_status
|
||||||
|
try:
|
||||||
|
status = client.get_status()
|
||||||
|
print(f"API Hook Client status response: {status}")
|
||||||
|
if status.get('status') == 'ok':
|
||||||
|
mock_app.verification_successful = True # Simulate success flag
|
||||||
|
else:
|
||||||
|
mock_app.verification_successful = False
|
||||||
|
except Exception as e:
|
||||||
|
print(f"API Hook Client verification failed: {e}")
|
||||||
|
mock_app.verification_successful = False
|
||||||
|
|
||||||
|
def test_conductor_integrates_api_hook_client_for_verification(hook_server_fixture_for_integration):
|
||||||
|
"""
|
||||||
|
Verify that Conductor's simulated phase completion logic properly integrates
|
||||||
|
and uses the ApiHookClient for verification. This test *should* pass (Green Phase)
|
||||||
|
if the integration in `simulate_conductor_phase_completion` is correct.
|
||||||
|
"""
|
||||||
|
client_base_url, mock_app = hook_server_fixture_for_integration
|
||||||
|
|
||||||
|
dummy_plan_content = """
|
||||||
|
# Implementation Plan: Test Track
|
||||||
|
|
||||||
|
## Phase 1: Initial Setup [checkpoint: abcdefg]
|
||||||
|
- [x] Task: Dummy Task 1 [1234567]
|
||||||
|
- [ ] Task: Conductor - User Manual Verification 'Phase 1: Initial Setup' (Protocol in workflow.md)
|
||||||
|
"""
|
||||||
|
# Reset mock_app's success flag for this test run
|
||||||
|
mock_app.verification_successful = False
|
||||||
|
|
||||||
|
simulate_conductor_phase_completion(client_base_url, mock_app, dummy_plan_content)
|
||||||
|
|
||||||
|
# Assert that the verification was considered successful by the simulated Conductor
|
||||||
|
assert mock_app.verification_successful is True
|
||||||
Reference in New Issue
Block a user