feat(tests): Route VerificationLogger output to tests/logs
This commit is contained in:
@@ -52,21 +52,25 @@ def app_instance() -> Generator[App, None, None]:
|
||||
yield app
|
||||
# Cleanup: Ensure asyncio loop is stopped and tasks are cancelled
|
||||
if hasattr(app, '_loop'):
|
||||
# 1. Identify all pending tasks in app._loop.
|
||||
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
|
||||
# 1. Stop the loop thread-safely first
|
||||
if app._loop.is_running():
|
||||
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)
|
||||
# 3. Wait for them to complete using loop.run_until_complete(asyncio.gather(*tasks, return_exceptions=True)).
|
||||
if tasks:
|
||||
app._loop.run_until_complete(asyncio.gather(*tasks, return_exceptions=True))
|
||||
# 4. Then stop the loop.
|
||||
app._loop.stop()
|
||||
|
||||
# 2. Join the loop thread
|
||||
if hasattr(app, '_loop_thread') and app._loop_thread.is_alive():
|
||||
app._loop_thread.join(timeout=2.0)
|
||||
|
||||
# 3. Check for pending tasks after thread is joined
|
||||
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
|
||||
def mock_app(app_instance: App) -> App:
|
||||
@@ -81,7 +85,7 @@ class VerificationLogger:
|
||||
def __init__(self, test_name: str, script_name: str):
|
||||
self.test_name = test_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.log_file = self.logs_dir / f"{script_name}.txt"
|
||||
self.entries = []
|
||||
|
||||
Reference in New Issue
Block a user