diff --git a/tests/test_agent_editor_routes.py b/tests/test_agent_editor_routes.py index f9f27e3e1..89641276b 100644 --- a/tests/test_agent_editor_routes.py +++ b/tests/test_agent_editor_routes.py @@ -1929,23 +1929,60 @@ def test_place_layer_reports_false_on_queued_response(monkeypatch): assert placed is False -def test_bridge_await_cancellation_handled_cleanly(): +def test_bridge_await_cancellation_handled_cleanly(monkeypatch): import asyncio + from contextlib import suppress from httpx import ASGITransport, AsyncClient app = _bridge_app() async def scenario(): + entered_wait = asyncio.Event() + real_wait_for = asyncio.wait_for + waits = [] + + async def observe_wait(awaitable, timeout): + # Observe the real wait; do not replace its result or cancellation. + waits.append(awaitable) + entered_wait.set() + return await real_wait_for(awaitable, timeout) + + monkeypatch.setattr(eaar.asyncio, "wait_for", observe_wait) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: cmd_resp = await c.post("/api/editor/bridge/command", json={"session": "cancel-sess", "op": "document.info", "wait": False}) + assert cmd_resp.status_code == 200 + assert cmd_resp.json()["completed"] is False cid = cmd_resp.json()["queued"][0] await_task = asyncio.create_task(c.post("/api/editor/bridge/await", json={"id": cid, "timeout": 30})) - await asyncio.sleep(0.05) - await_task.cancel() try: - await await_task - except asyncio.CancelledError: - pass + await real_wait_for(entered_wait.wait(), 5) + assert len(waits) == 1 + assert not waits[0].done() + assert not await_task.done(), "await returned without a queued result" + await_task.cancel() + response = await real_wait_for(await_task, 5) + assert response.status_code == 200 + assert response.json() == { + "ok": False, "error": "cancelled", "status": "cancelled", + "completed": False, "id": cid, + } + finally: + if not await_task.done(): + await_task.cancel() + with suppress(asyncio.CancelledError): + await await_task + + # Cancelling the waiter must not cancel the queued command itself. + payload = {"width": 137, "height": 251} + delivered = await c.post("/api/editor/bridge/result", json={"id": cid, "ok": True, "result": payload}) + assert delivered.status_code == 200 + completed = await c.post("/api/editor/bridge/await", json={"id": cid, "timeout": 1}) + assert completed.status_code == 200 + assert completed.json()["status"] == "completed" + assert completed.json()["completed"] is True + assert completed.json()["ok"] is True + assert completed.json()["result"] == payload + assert len(waits) == 2 asyncio.run(scenario()) diff --git a/tests/test_auth_config_lock_concurrency.py b/tests/test_auth_config_lock_concurrency.py index 34232b9e2..d1707e5b4 100644 --- a/tests/test_auth_config_lock_concurrency.py +++ b/tests/test_auth_config_lock_concurrency.py @@ -206,35 +206,50 @@ class TestDiskConsistency: @pytest.mark.slow def test_file_always_valid_json_during_concurrent_ops(self, tmp_path): mgr = _fresh_auth_manager(tmp_path) - mgr.create_user("admin", "adminpw", is_admin=True) + assert mgr.create_user("admin", "adminpw", is_admin=True) stop_event = threading.Event() + reader_ready = threading.Event() corruption_found = [] + observations = {"reads": 0, "with_writes": 0} def reader(): while not stop_event.is_set(): try: with open(mgr.auth_path, "r") as f: content = f.read() - json.loads(content) + data = json.loads(content) + observations["reads"] += 1 + if any(name.startswith("stress") for name in data["users"]): + observations["with_writes"] += 1 + reader_ready.set() except json.JSONDecodeError as e: corruption_found.append(str(e)) break - except FileNotFoundError: - pass time.sleep(0.001) def writer(): + assert reader_ready.wait(5), "reader never observed the initial file" + results = [] for i in range(50): - mgr.create_user(f"stress{i}", f"pw{i}") + results.append(mgr.create_user(f"stress{i}", f"pw{i}")) + return results - reader_thread = threading.Thread(target=reader) - writer_thread = threading.Thread(target=writer) - - reader_thread.start() - writer_thread.start() - writer_thread.join() - stop_event.set() - reader_thread.join() + # Futures propagate reader/writer exceptions to the test thread. + with ThreadPoolExecutor(max_workers=2) as pool: + reading = pool.submit(reader) + writing = pool.submit(writer) + try: + results = writing.result(timeout=120) + finally: + stop_event.set() + reading.result(timeout=5) assert not corruption_found, f"Corrupt JSON detected: {corruption_found[0]}" + assert len(results) == 50 and all(results), "writer operations did not succeed" + assert observations["reads"] > 0 + assert observations["with_writes"] > 0, "reader saw no persisted writer activity" + expected = {"admin"} | {f"stress{i}" for i in range(50)} + with open(mgr.auth_path) as persisted: + assert set(json.load(persisted)["users"]) == expected + assert set(_fresh_auth_manager(tmp_path).users) == expected diff --git a/tests/test_delete_user_invalidates_token_cache.py b/tests/test_delete_user_invalidates_token_cache.py index 91be50e93..2a534f89e 100644 --- a/tests/test_delete_user_invalidates_token_cache.py +++ b/tests/test_delete_user_invalidates_token_cache.py @@ -11,6 +11,8 @@ cache from the DB, where the rows are already gone, and the token is rejected. """ import asyncio import types +import pytest +from fastapi import HTTPException from routes.auth_routes import setup_auth_routes, DeleteUserRequest @@ -60,12 +62,9 @@ def test_refused_delete_does_not_invalidate_cache(): invalidations = [] router = setup_auth_routes(_auth_manager(delete_result=False)) handler = _handler(router) - try: + with pytest.raises(HTTPException) as refused: asyncio.run(handler(DeleteUserRequest(username="admin"), _fake_request(invalidations))) - raised = False - except Exception: - raised = True - assert raised, "a refused delete should raise (HTTP 400)" + assert refused.value.status_code == 400 assert invalidations == [], "a refused delete must not touch the token cache" diff --git a/tests/test_email_list_stale_while_revalidate.py b/tests/test_email_list_stale_while_revalidate.py index 4fe22a75d..13f347f5a 100644 --- a/tests/test_email_list_stale_while_revalidate.py +++ b/tests/test_email_list_stale_while_revalidate.py @@ -56,14 +56,35 @@ def test_cold_start_with_no_snapshot_and_no_index_is_not_marked_stale(monkeypatc # falls through to the real (blocking) path, so the response is not a # stale-marked snapshot. endpoint = _list_endpoint() + calls = [] + + class EmptyMailbox: + def select(self, folder, readonly=False): + calls.append(("select", folder, readonly)) + return "OK", [b"0"] + + def uid(self, command, *args): + assert command == "SEARCH" + calls.append(("search", args)) + return "OK", [b""] + + def logout(self): + pass + + def connect(*args, **kwargs): + calls.append(("connect", args, kwargs)) + return EmptyMailbox() + + monkeypatch.setattr(er, "_start_poller", lambda: None) monkeypatch.setattr(er, "_email_index_list", lambda *a, **k: ([], 0, None)) - monkeypatch.setattr(er, "_imap_connect", lambda *a, **k: (_ for _ in ()).throw( - RuntimeError("offline test"))) - try: - result = _call(endpoint) - except Exception: - return # the cold path legitimately attempted real IMAP + monkeypatch.setattr(er, "_imap_connect", connect) + result = _call(endpoint) + assert [call[0] for call in calls] == ["connect", "select", "search"] + assert calls[1][2] is True assert isinstance(result, dict) + assert result["emails"] == [] + assert result["total"] == 0 + assert not result.get("error") assert not (result.get("sync") or {}).get("stale")