test: require real cold reads, cancellation and auth writes

tests/TESTING_STANDARD.md

@@ -99,10 +99,8 @@ cleanup. Specifically:
   `os.environ[...] = ...` that outlives the test.
 - **Current working directory** - never `chdir` without restoring; never assert
   against cwd-relative paths like `./data`. Use a temp workspace helper instead.
-- **Database** - the root `conftest.py` defaults `DATABASE_URL` to a private,
-  worker-owned SQLite file for collection safety and consistent schema across
-  TestClient threads. This fallback is not a per-test data fixture. A test that
-  writes or depends on particular rows must opt
+- **Database** - the root `conftest.py` defaults `DATABASE_URL` to an in-memory
+  SQLite for collection safety. A test that needs a real file-backed DB must opt
   in explicitly via `tests.helpers.sqlite_db.make_temp_sqlite` and bind its
   `SessionLocal` onto the module under test. Do not rely on a persistent
   on-disk DB existing.

tests/conftest.py

@@ -12,17 +12,9 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 # File-backed databases remain fixture-owned. Cleanup restores the caller.
 # Keep the existing generated environment-reference locations stable.
 _database_environment = pytest.MonkeyPatch()
+_database_environment.setenv("DATABASE_URL", "sqlite:///:memory:")
 from tests.helpers.worker_runtime import bootstrap_runtime, configure_runtime
 _runtime_environment = bootstrap_runtime()
-# A file keeps the collection schema consistent across TestClient threads;
-# SingletonThreadPool can evict the populated in-memory main-thread database.
-# One patcher owns this variable, including restoration of an inherited URL.
-_database_environment.setenv(
-    "DATABASE_URL",
-    f"sqlite:///{_runtime_environment.root / 'collection.db'}"
-    if _runtime_environment is not None else "sqlite:///:memory:",
-)
-_collection_database_engine = None
 
 # Pre-import real heavy modules BEFORE any test file's module-level stubs can
 # replace them with MagicMock. Some test files (e.g. test_llm_core_sanitize_*)
@@ -34,7 +26,6 @@ try:
     import sqlalchemy.orm  # noqa: F401
     import core.database  # noqa: F401
     import src.database
-    _collection_database_engine = core.database.engine
 except ImportError:
     pass  # not installed - the stubs below will handle it
 
@@ -109,10 +100,6 @@ def pytest_configure(config):
     only registers marker names; it imports no production module.
     """
     config.add_cleanup(_database_environment.undo)
-    if _collection_database_engine is not None:
-        # Registered after runtime cleanup, so connections close before the
-        # owned file is removed (also required on Windows).
-        config.add_cleanup(_collection_database_engine.dispose)
 
     import pathlib
     from tests._taxonomy import discover_markers

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())
 

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

tests/test_database_test_isolation.py

@@ -85,9 +85,6 @@ def test_actual_tests_restore_process_state_and_ignore_inherited_database(tmp_pa
         import os
         import sys
         import pytest
-        import tempfile
-        from pathlib import Path
-        from sqlalchemy.engine import make_url
 
         def snapshot():
             import core
@@ -110,11 +107,7 @@ def test_actual_tests_restore_process_state_and_ignore_inherited_database(tmp_pa
         class StateGuard:
             def pytest_sessionstart(self):
                 self.initial = snapshot()
-                assert self.initial[0] != inherited_url
-                self.default_path = Path(make_url(self.initial[0]).database)
-                assert self.default_path.is_file()
-                assert self.default_path.name == "collection.db"
-                assert self.default_path.parent == Path(tempfile.gettempdir()).parent
+                assert self.initial[0] == "sqlite:///:memory:"
 
             def pytest_collection_finish(self):
                 assert snapshot() == self.initial, "collection changed database bindings"
@@ -125,11 +118,9 @@ def test_actual_tests_restore_process_state_and_ignore_inherited_database(tmp_pa
                 assert snapshot() == self.initial, "test leaked database or module state"
 
         inherited_url = os.environ["DATABASE_URL"]
-        guard = StateGuard()
         result = pytest.main(["-q", "-p", "no:cacheprovider", *sys.argv[1:]],
-                             plugins=[guard])
+                             plugins=[StateGuard()])
         assert os.environ["DATABASE_URL"] == inherited_url
-        assert not guard.default_path.exists(), "collection database outlived its runtime"
         raise SystemExit(result)
     ''')
     result = subprocess.run(

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"
 
 

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")
 
 

tests/test_worker_runtime.py

@@ -79,32 +79,3 @@ def test_tmp_path_under_private_runtime_fits_unix_socket(tmp_path):
     # witness. A nested pytest-of-<user> basetemp made it 110 bytes under xdist.
     with socket.socket(socket.AF_UNIX) as sock:
         sock.bind(str(tmp_path / "tmux.sock"))
-
-
-def test_collection_database_survives_thread_connection_churn():
-    """TestClient threads must not lose the collection database or its schema.
-
-    SingletonThreadPool's default five-connection limit used to close the main
-    thread's in-memory database after repeated route clients. Each new thread
-    also saw an empty database. Exercise more threads than that limit and then
-    verify both compatibility factories still see all registered tables.
-    """
-    from concurrent.futures import ThreadPoolExecutor
-    from sqlalchemy import inspect
-    import core.database as database
-    import src.database as compatibility
-
-    expected = set(database.Base.metadata.tables)
-
-    def table_names(factory):
-        with factory() as session:
-            return set(inspect(session.connection()).get_table_names())
-
-    assert {"documents", "model_endpoints", "mcp_servers", "calendars"} <= expected
-    assert expected <= table_names(database.SessionLocal)
-    for _ in range(8):
-        # A fresh executor creates a fresh thread rather than reusing one.
-        with ThreadPoolExecutor(max_workers=1) as executor:
-            assert expected <= executor.submit(table_names, database.SessionLocal).result()
-    assert expected <= table_names(database.SessionLocal)
-    assert expected <= table_names(compatibility.SessionLocal)