feat(tests): Route VerificationLogger output to tests/logs

This commit is contained in:
2026-03-02 23:24:40 -05:00
parent 0140c5fd52
commit 51f7c2a772

View File

@@ -52,21 +52,25 @@ def app_instance() -> Generator[App, None, None]:
yield app yield app
# Cleanup: Ensure asyncio loop is stopped and tasks are cancelled # Cleanup: Ensure asyncio loop is stopped and tasks are cancelled
if hasattr(app, '_loop'): if hasattr(app, '_loop'):
# 1. Identify all pending tasks in app._loop. # 1. Stop the loop thread-safely first
tasks = [t for t in asyncio.all_tasks(app._loop) if not t.done()]
# 2. Cancel them using task.cancel().
for task in tasks:
task.cancel()
# Stop background thread to take control of the loop thread-safely
if app._loop.is_running(): if app._loop.is_running():
app._loop.call_soon_threadsafe(app._loop.stop) app._loop.call_soon_threadsafe(app._loop.stop)
if hasattr(app, '_loop_thread') and app._loop_thread.is_alive():
app._loop_thread.join(timeout=2.0) # 2. Join the loop thread
# 3. Wait for them to complete using loop.run_until_complete(asyncio.gather(*tasks, return_exceptions=True)). if hasattr(app, '_loop_thread') and app._loop_thread.is_alive():
if tasks: app._loop_thread.join(timeout=2.0)
app._loop.run_until_complete(asyncio.gather(*tasks, return_exceptions=True))
# 4. Then stop the loop. # 3. Check for pending tasks after thread is joined
app._loop.stop() if 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 @pytest.fixture
def mock_app(app_instance: App) -> App: def mock_app(app_instance: App) -> App:
@@ -81,7 +85,7 @@ class VerificationLogger:
def __init__(self, test_name: str, script_name: str): def __init__(self, test_name: str, script_name: str):
self.test_name = test_name self.test_name = test_name
self.script_name = script_name self.script_name = script_name
self.logs_dir = Path(f"logs/test/{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}") 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.logs_dir.mkdir(parents=True, exist_ok=True)
self.log_file = self.logs_dir / f"{script_name}.txt" self.log_file = self.logs_dir / f"{script_name}.txt"
self.entries = [] self.entries = []