From 488874b796c475f065742f161b14d7840a21eee7 Mon Sep 17 00:00:00 2001 From: wcqqq1214 Date: Sun, 18 Oct 2026 19:40:47 +0800 Subject: [PATCH 1/2] test: fix core restart test timeout --- tests/unit/test_core_lifecycle.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_core_lifecycle.py b/tests/unit/test_core_lifecycle.py index 058f869fd61..1017568fa5d 100644 --- a/tests/unit/test_core_lifecycle.py +++ b/tests/unit/test_core_lifecycle.py @@ -9,6 +9,7 @@ from astrbot.core.core_lifecycle import AstrBotCoreLifecycle from astrbot.core.log import LogBroker +from astrbot.core.utils.trace_store import close_trace_store @pytest.fixture @@ -895,9 +896,17 @@ async def test_restart_terminates_managers_and_starts_thread( lifecycle.astrbot_updater = MagicMock() - with patch("astrbot.core.core_lifecycle.threading.Thread") as mock_thread: + # Mock to_thread as well: patching Thread also affects its executor workers. + with ( + patch("astrbot.core.core_lifecycle.threading.Thread") as mock_thread, + patch( + "astrbot.core.core_lifecycle.asyncio.to_thread", new_callable=AsyncMock + ) as mock_to_thread, + ): await lifecycle.restart() + mock_to_thread.assert_awaited_once_with(close_trace_store) + # Verify managers were terminated lifecycle.provider_manager.terminate.assert_awaited_once() lifecycle.platform_manager.terminate.assert_awaited_once() From 02539ce3869db92f55a6361a42e4e2cd3717eec1 Mon Sep 17 00:00:00 2001 From: wcqqq1214 Date: Sun, 18 Oct 2026 20:09:50 +0800 Subject: [PATCH 2/2] fix: preserve trace snapshot ordering and stabilize Windows tests --- astrbot/core/utils/trace.py | 7 ++++++- tests/test_fastapi_v1_dashboard.py | 8 ++++++-- tests/unit/test_trace_store.py | 29 ++++++++++++++++++++++++++--- 3 files changed, 38 insertions(+), 6 deletions(-) diff --git a/astrbot/core/utils/trace.py b/astrbot/core/utils/trace.py index c6de89ac061..135c04b4809 100644 --- a/astrbot/core/utils/trace.py +++ b/astrbot/core/utils/trace.py @@ -146,11 +146,16 @@ def _emit( event: Standard event associated with a span. closing: Whether completion must survive the trace payload limit. """ + # Preserve snapshot order on coarse clocks and across wall-clock corrections. + updated_ns = max( + time.time_ns(), + int(self._summary.get("updated_time_unix_nano", self._start_ns)) + 1, + ) payload = { "type": "trace", "schema_version": 3, "trace": self._summary, - "time_unix_nano": str(time.time_ns()), + "time_unix_nano": str(updated_ns), } self._summary["updated_time_unix_nano"] = payload["time_unix_nano"] if span is not None: diff --git a/tests/test_fastapi_v1_dashboard.py b/tests/test_fastapi_v1_dashboard.py index 7a22a0c4035..b2286bb11b4 100644 --- a/tests/test_fastapi_v1_dashboard.py +++ b/tests/test_fastapi_v1_dashboard.py @@ -4380,7 +4380,11 @@ async def test_v1_persistent_trace_routes(asgi_client, monkeypatch, tmp_path): }, } ) - assert store.flush() + # Allow bounded headroom for database initialization on slower CI disks. + assert await asyncio.to_thread(store.flush, timeout=30), ( + f"Trace flush failed: error={store.last_error!r}, " + f"writer_alive={store._thread.is_alive()}, ready={store._ready.is_set()}" + ) response = await asgi_client.get("/api/v1/trace/history") assert response.status_code == 401 token = jwt.encode( @@ -4420,7 +4424,7 @@ async def test_v1_persistent_trace_routes(asgi_client, monkeypatch, tmp_path): await asgi_client.get("/api/v1/trace/settings", headers=headers) ).status_code == 200 finally: - store.close() + await asyncio.to_thread(store.close) @pytest.mark.asyncio diff --git a/tests/unit/test_trace_store.py b/tests/unit/test_trace_store.py index 1e38fa37368..b5d25d5703a 100644 --- a/tests/unit/test_trace_store.py +++ b/tests/unit/test_trace_store.py @@ -10,6 +10,7 @@ import threading import time import weakref +from itertools import count from pathlib import Path import pytest @@ -416,7 +417,11 @@ def test_database_pages_filters_details_and_restart(persisted, capture): restarted.close() -@pytest.mark.parametrize("prompt", ["x" * 8192, "字" * 20000, "x" * 131072]) +@pytest.mark.parametrize( + "prompt", + ["x" * 8192, "字" * 20000, "x" * 131072], + ids=["ascii-8k", "unicode-20k", "ascii-128k"], +) def test_long_prompt_is_preserved_in_spans_events_and_database( persisted, capture, prompt ): @@ -530,11 +535,29 @@ def held_write(connection, batch): assert persisted._pending_bytes == 0 -def test_stale_updates_do_not_reopen_completed_trace(persisted, capture): +@pytest.mark.parametrize("clock_step", [None, 0, -1], ids=["real", "frozen", "backward"]) +@pytest.mark.parametrize("reverse", [False, True], ids=["ordered", "reversed"]) +def test_stale_updates_do_not_reopen_completed_trace( + persisted, capture, monkeypatch, clock_step, reverse +): + """Keep completed snapshots newer even when the wall clock repeats or regresses. + + Args: + persisted: Isolated trace database. + capture: Immutable trace snapshots. + monkeypatch: Pytest patch manager. + clock_step: Nanoseconds per clock read, or None for the real clock. + reverse: Whether to enqueue snapshots in reverse order. + """ + if clock_step is not None: + clock = count(time.time_ns(), clock_step) + monkeypatch.setattr(time, "time_ns", clock.__next__) trace = TraceSpan("message", message_outline="Literal % _") trace.record("event_received") trace.record("pipeline_end", status="success") - for frame in reversed(capture): + stamps = [int(frame["time_unix_nano"]) for frame in capture] + assert all(earlier < later for earlier, later in zip(stamps, stamps[1:])) + for frame in reversed(capture) if reverse else capture: assert persisted.append(frame) assert persisted.flush() detail = persisted.get_trace(trace.trace_id)