wow this ai messed up.
This commit is contained in:
@@ -7,95 +7,42 @@ import os
|
||||
import signal
|
||||
import sys
|
||||
import datetime
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from typing import Generator, Any
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
# Ensure project root is in path
|
||||
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
||||
|
||||
import ai_client
|
||||
# Import the App class after patching if necessary, but here we just need the type hint
|
||||
from gui_2 import App
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_ai_client() -> Generator[None, None, None]:
|
||||
"""Reset ai_client global state between every test to prevent state pollution."""
|
||||
ai_client.reset_session()
|
||||
# Default to a safe model
|
||||
ai_client.set_provider("gemini", "gemini-2.5-flash-lite")
|
||||
yield
|
||||
|
||||
@pytest.fixture
|
||||
def app_instance() -> Generator[App, None, None]:
|
||||
"""
|
||||
Centralized App instance with all external side effects mocked.
|
||||
Matches the pattern used in test_token_viz.py and test_gui_phase4.py.
|
||||
"""
|
||||
with (
|
||||
patch('gui_2.load_config', return_value={
|
||||
'ai': {'provider': 'gemini', 'model': 'gemini-2.5-flash-lite'},
|
||||
'projects': {'paths': [], 'active': ''},
|
||||
'gui': {'show_windows': {}}
|
||||
}),
|
||||
patch('gui_2.save_config'),
|
||||
patch('gui_2.project_manager'),
|
||||
patch('gui_2.session_logger'),
|
||||
patch('gui_2.immapp.run'),
|
||||
patch.object(App, '_load_active_project'),
|
||||
patch.object(App, '_fetch_models'),
|
||||
patch.object(App, '_load_fonts'),
|
||||
patch.object(App, '_post_init'),
|
||||
patch.object(App, '_prune_old_logs'),
|
||||
patch.object(App, '_init_ai_and_hooks')
|
||||
):
|
||||
app = App()
|
||||
yield app
|
||||
# Cleanup: Ensure background threads and asyncio loop are stopped
|
||||
app.shutdown()
|
||||
|
||||
if hasattr(app, '_loop') and not app._loop.is_closed():
|
||||
tasks = [t for t in asyncio.all_tasks(app._loop) if not t.done()]
|
||||
if tasks:
|
||||
# Cancel tasks so they can be gathered
|
||||
for task in tasks:
|
||||
task.cancel()
|
||||
app._loop.run_until_complete(asyncio.gather(*tasks, return_exceptions=True))
|
||||
|
||||
# 4. Finally close the loop
|
||||
app._loop.close()
|
||||
|
||||
@pytest.fixture
|
||||
def mock_app(app_instance: App) -> App:
|
||||
"""
|
||||
Simpler fixture returning a mocked App instance.
|
||||
Reuses app_instance for automatic cleanup and consistent mocking.
|
||||
"""
|
||||
return app_instance
|
||||
|
||||
class VerificationLogger:
|
||||
"""High-signal reporting for automated tests, inspired by Unreal Engine's diagnostic style."""
|
||||
def __init__(self, test_name: str, script_name: str):
|
||||
def __init__(self, test_name: str, script_name: str) -> None:
|
||||
self.test_name = test_name
|
||||
self.script_name = script_name
|
||||
self.entries = []
|
||||
self.start_time = time.time()
|
||||
# Route artifacts to tests/logs/
|
||||
self.logs_dir = Path(f"tests/logs/{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}")
|
||||
self.logs_dir.mkdir(parents=True, exist_ok=True)
|
||||
self.log_file = self.logs_dir / f"{script_name}.txt"
|
||||
self.entries = []
|
||||
|
||||
def log_state(self, field: str, before: Any, after: Any, delta: Any = None):
|
||||
def log_state(self, field: str, before: Any, after: Any) -> None:
|
||||
delta = ""
|
||||
if isinstance(before, (int, float)) and isinstance(after, (int, float)):
|
||||
diff = after - before
|
||||
delta = f"{'+' if diff > 0 else ''}{diff}"
|
||||
self.entries.append({
|
||||
"Field": field,
|
||||
"Before": str(before),
|
||||
"After": str(after),
|
||||
"Delta": str(delta) if delta is not None else ""
|
||||
"Delta": delta
|
||||
})
|
||||
# Also print to stdout for real-time visibility in CI
|
||||
print(f"[STATE] {field}: {before} -> {after}")
|
||||
|
||||
def finalize(self, description: str, status: str, result_msg: str):
|
||||
with open(self.log_file, "a", encoding="utf-8") as f:
|
||||
def finalize(self, title: str, status: str, result_msg: str) -> None:
|
||||
elapsed = round(time.time() - self.start_time, 2)
|
||||
log_file = self.logs_dir / f"{self.script_name}.txt"
|
||||
with open(log_file, "w", encoding="utf-8") as f:
|
||||
f.write(f"[ Test: {self.test_name} ]\n")
|
||||
f.write(f"({description})\n\n")
|
||||
f.write(f"({title})\n\n")
|
||||
f.write(f"{self.test_name}: before vs after\n")
|
||||
f.write(f"{'Field':<25} {'Before':<20} {'After':<20} {'Delta':<15}\n")
|
||||
f.write("-" * 80 + "\n")
|
||||
@@ -131,15 +78,98 @@ def kill_process_tree(pid: int | None) -> None:
|
||||
except Exception as e:
|
||||
print(f"[Fixture] Error killing process tree {pid}: {e}")
|
||||
|
||||
@pytest.fixture
|
||||
def mock_app() -> App:
|
||||
"""
|
||||
Mock version of the App for simple unit tests that don't need a loop.
|
||||
"""
|
||||
with (
|
||||
patch('gui_2.load_config', return_value={
|
||||
'ai': {'provider': 'gemini', 'model': 'gemini-2.5-flash-lite'},
|
||||
'projects': {'paths': [], 'active': ''},
|
||||
'gui': {'show_windows': {}}
|
||||
}),
|
||||
patch('gui_2.save_config'),
|
||||
patch('gui_2.project_manager'),
|
||||
patch('gui_2.session_logger'),
|
||||
patch('gui_2.immapp.run'),
|
||||
patch.object(App, '_load_active_project'),
|
||||
patch.object(App, '_fetch_models'),
|
||||
patch.object(App, '_load_fonts'),
|
||||
patch.object(App, '_post_init'),
|
||||
patch.object(App, '_prune_old_logs'),
|
||||
patch.object(App, '_init_ai_and_hooks')
|
||||
):
|
||||
return App()
|
||||
|
||||
@pytest.fixture
|
||||
def app_instance() -> Generator[App, None, None]:
|
||||
"""
|
||||
Centralized App instance with all external side effects mocked.
|
||||
Matches the pattern used in test_token_viz.py and test_gui_phase4.py.
|
||||
"""
|
||||
with (
|
||||
patch('gui_2.load_config', return_value={
|
||||
'ai': {'provider': 'gemini', 'model': 'gemini-2.5-flash-lite'},
|
||||
'projects': {'paths': [], 'active': ''},
|
||||
'gui': {'show_windows': {}}
|
||||
}),
|
||||
patch('gui_2.save_config'),
|
||||
patch('gui_2.project_manager'),
|
||||
patch('gui_2.session_logger'),
|
||||
patch('gui_2.immapp.run'),
|
||||
patch.object(App, '_load_active_project'),
|
||||
patch.object(App, '_fetch_models'),
|
||||
patch.object(App, '_load_fonts'),
|
||||
patch.object(App, '_post_init'),
|
||||
patch.object(App, '_prune_old_logs'),
|
||||
patch.object(App, '_init_ai_and_hooks')
|
||||
):
|
||||
app = App()
|
||||
yield app
|
||||
# Cleanup: Ensure background threads and asyncio loop are stopped
|
||||
if hasattr(app, 'shutdown'):
|
||||
app.shutdown()
|
||||
|
||||
if hasattr(app, '_loop') and not app._loop.is_closed():
|
||||
tasks = [t for t in asyncio.all_tasks(app._loop) if not t.done()]
|
||||
if tasks:
|
||||
# Cancel tasks so they can be gathered
|
||||
for task in tasks:
|
||||
task.cancel()
|
||||
# We can't really run the loop if it's already stopping or thread is dead,
|
||||
# but we try to be clean.
|
||||
try:
|
||||
if app._loop.is_running():
|
||||
app._loop.call_soon_threadsafe(app._loop.stop)
|
||||
except: pass
|
||||
|
||||
# Finally close the loop if we can
|
||||
try:
|
||||
if not app._loop.is_running():
|
||||
app._loop.close()
|
||||
except: pass
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def live_gui() -> Generator[tuple[subprocess.Popen, str], None, None]:
|
||||
"""
|
||||
Session-scoped fixture that starts gui_2.py with --enable-test-hooks.
|
||||
Includes high-signal environment telemetry.
|
||||
Includes high-signal environment telemetry and workspace isolation.
|
||||
"""
|
||||
gui_script = "gui_2.py"
|
||||
gui_script = os.path.abspath("gui_2.py")
|
||||
diag = VerificationLogger("live_gui_startup", "live_gui_diag")
|
||||
diag.log_state("GUI Script", "N/A", gui_script)
|
||||
diag.log_state("GUI Script", "N/A", "gui_2.py")
|
||||
|
||||
# 1. Create a isolated workspace for the live GUI
|
||||
temp_workspace = Path("tests/artifacts/live_gui_workspace")
|
||||
if temp_workspace.exists():
|
||||
shutil.rmtree(temp_workspace)
|
||||
temp_workspace.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Create dummy config and project files to avoid cluttering root
|
||||
(temp_workspace / "config.toml").write_text("[projects]\npaths = []\nactive = ''\n", encoding="utf-8")
|
||||
(temp_workspace / "manual_slop.toml").write_text("[project]\nname = 'TestProject'\n", encoding="utf-8")
|
||||
(temp_workspace / "conductor" / "tracks").mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Check if already running (shouldn't be)
|
||||
try:
|
||||
@@ -148,14 +178,22 @@ def live_gui() -> Generator[tuple[subprocess.Popen, str], None, None]:
|
||||
except: already_up = False
|
||||
diag.log_state("Hook Server Port 8999", "Down", "UP" if already_up else "Down")
|
||||
|
||||
print(f"\n[Fixture] Starting {gui_script} --enable-test-hooks...")
|
||||
print(f"\n[Fixture] Starting {gui_script} --enable-test-hooks in {temp_workspace}...")
|
||||
os.makedirs("logs", exist_ok=True)
|
||||
log_file = open(f"logs/{gui_script.replace('.', '_')}_test.log", "w", encoding="utf-8")
|
||||
|
||||
# Use environment variable to point to temp config if App supports it,
|
||||
# or just run from that CWD.
|
||||
env = os.environ.copy()
|
||||
env["PYTHONPATH"] = os.getcwd()
|
||||
|
||||
process = subprocess.Popen(
|
||||
["uv", "run", "python", "-u", gui_script, "--enable-test-hooks"],
|
||||
stdout=log_file,
|
||||
stderr=log_file,
|
||||
text=True,
|
||||
cwd=str(temp_workspace.absolute()),
|
||||
env=env,
|
||||
creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if os.name == 'nt' else 0
|
||||
)
|
||||
|
||||
@@ -202,3 +240,7 @@ def live_gui() -> Generator[tuple[subprocess.Popen, str], None, None]:
|
||||
except: pass
|
||||
kill_process_tree(process.pid)
|
||||
log_file.close()
|
||||
# Cleanup temp workspace
|
||||
try:
|
||||
shutil.rmtree(temp_workspace)
|
||||
except: pass
|
||||
|
||||
Reference in New Issue
Block a user