fix: Multiple test fixes and improvements
- Fix mock_gemini_cli.py to use src/aggregate.py (moved to src directory) - Add wait_for_event method to ApiHookClient for simulation tests - Fix custom_callback path in app_controller to use absolute path - Fix test_gui2_parity.py to use correct callback file path
This commit is contained in:
@@ -72,6 +72,17 @@ class ApiHookClient:
|
|||||||
def clear_events(self) -> list[dict[str, Any]]:
|
def clear_events(self) -> list[dict[str, Any]]:
|
||||||
return self.get_events()
|
return self.get_events()
|
||||||
|
|
||||||
|
|
||||||
|
def wait_for_event(self, event_type: str, timeout: int = 5) -> dict[str, Any] | None:
|
||||||
|
start = time.time()
|
||||||
|
while time.time() - start < timeout:
|
||||||
|
events = self.get_events()
|
||||||
|
for ev in events:
|
||||||
|
if ev.get("type") == event_type or ev.get("event_type") == event_type:
|
||||||
|
return ev
|
||||||
|
time.sleep(0.2)
|
||||||
|
return None
|
||||||
|
|
||||||
def post_gui(self, payload: dict) -> dict[str, Any]:
|
def post_gui(self, payload: dict) -> dict[str, Any]:
|
||||||
"""Pushes an event to the GUI's SyncEventQueue via the /api/gui endpoint."""
|
"""Pushes an event to the GUI's SyncEventQueue via the /api/gui endpoint."""
|
||||||
return self._make_request('POST', '/api/gui', data=payload) or {}
|
return self._make_request('POST', '/api/gui', data=payload) or {}
|
||||||
|
|||||||
@@ -573,7 +573,9 @@ class AppController:
|
|||||||
|
|
||||||
def _test_callback_func_write_to_file(self, data: str) -> None:
|
def _test_callback_func_write_to_file(self, data: str) -> None:
|
||||||
"""A dummy function that a custom_callback would execute for testing."""
|
"""A dummy function that a custom_callback would execute for testing."""
|
||||||
with open("test_callback_output.txt", "w") as f:
|
callback_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tests", "artifacts", "temp_callback_output.txt")
|
||||||
|
os.makedirs(os.path.dirname(callback_path), exist_ok=True)
|
||||||
|
with open(callback_path, "w") as f:
|
||||||
f.write(data)
|
f.write(data)
|
||||||
|
|
||||||
def _handle_approve_script(self, user_data=None) -> None:
|
def _handle_approve_script(self, user_data=None) -> None:
|
||||||
|
|||||||
@@ -135,7 +135,7 @@ def main() -> None:
|
|||||||
"role": "assistant",
|
"role": "assistant",
|
||||||
"content": "I will read the first 10 lines of the file."
|
"content": "I will read the first 10 lines of the file."
|
||||||
}), flush=True)
|
}), flush=True)
|
||||||
file_path = "aggregate.py"
|
file_path = "src/aggregate.py"
|
||||||
print(json.dumps({
|
print(json.dumps({
|
||||||
"type": "tool_use",
|
"type": "tool_use",
|
||||||
"name": "read_file",
|
"name": "read_file",
|
||||||
|
|||||||
@@ -71,7 +71,7 @@ def test_gui2_custom_callback_hook_works(live_gui: Any) -> None:
|
|||||||
assert response == {'status': 'queued'}
|
assert response == {'status': 'queued'}
|
||||||
time.sleep(1.5) # Give gui_2.py time to process its task queue
|
time.sleep(1.5) # Give gui_2.py time to process its task queue
|
||||||
# Assert that the file WAS created and contains the correct data
|
# Assert that the file WAS created and contains the correct data
|
||||||
temp_workspace_file = Path('tests/artifacts/live_gui_workspace/tests/artifacts/temp_callback_output.txt')
|
temp_workspace_file = Path('tests/artifacts/temp_callback_output.txt')
|
||||||
assert temp_workspace_file.exists(), f"Custom callback was NOT executed, or file path is wrong! Expected: {temp_workspace_file}"
|
assert temp_workspace_file.exists(), f"Custom callback was NOT executed, or file path is wrong! Expected: {temp_workspace_file}"
|
||||||
with open(temp_workspace_file, "r") as f:
|
with open(temp_workspace_file, "r") as f:
|
||||||
content = f.read()
|
content = f.read()
|
||||||
|
|||||||
Reference in New Issue
Block a user