test: prove concurrency with barriers/witnesses instead of wall-clock bounds

Replace OS-scheduler-dependent elapsed-time ceilings with deterministic
concurrency proofs in three tests:

- test_context_refs_concurrent: asyncio.Barrier rendezvous — all 3 URL
  fetches must be in flight simultaneously before any returns.
- test_memory_boundary_commit: positive non-blocking witness — the
  provider call list must still be empty when the async commit returns.
- test_mem0_v3 slow-prefetch: threading.Event park/release — prefetch
  must return while the backend search is still parked.

The tests/tools/test_mcp_tool.py hunk from the original PR is dropped:
main no longer carries the 'elapsed < 2.5' assertion it targeted
(superseded by a delay-relative bound).

Salvaged from #71913.
This commit is contained in:
Kyzcreig 2026-07-29 20:13:00 -07:00 committed by Teknium
parent f04fd1e7ad
commit 9c28600e77
3 changed files with 101 additions and 22 deletions

View file

@ -1,15 +1,16 @@
"""Tests for concurrent @-reference expansion in context_references.
RED before the refactor: test_refs_expand_concurrently asserts that N URL refs
(each a ~0.2s fetch) complete in roughly one fetch-time, not N×. On the serial
`for ref in refs: await` loop this FAILS (takes ~N×0.2s); after switching to
asyncio.gather it passes. The output-contract test guards that concurrency does
NOT change ordering, warnings, blocks, or token accounting.
test_refs_expand_concurrently asserts that N URL refs are fetched CONCURRENTLY.
It proves this with an asyncio.Barrier rendezvous rather than a stopwatch: all
N fetches must be in flight at the same instant before any is allowed to
return. On the serial `for ref in refs: await` loop the first fetch waits for
partners that never arrive and the test fails; with asyncio.gather it passes.
The output-contract test guards that concurrency does NOT change ordering,
warnings, blocks, or token accounting.
"""
from __future__ import annotations
import asyncio
import time
import pytest
@ -26,13 +27,49 @@ async def _slow_fetcher(url: str) -> str:
async def test_refs_expand_concurrently(tmp_path):
# Three independent URL refs in one message.
msg = "see @url:https://a.example/x @url:https://b.example/y @url:https://c.example/z please"
t0 = time.perf_counter()
res = await preprocess_context_references_async(
msg, cwd=tmp_path, context_length=100_000, url_fetcher=_slow_fetcher,
)
elapsed = time.perf_counter() - t0
# Serial would be ~0.6s (3×0.2). Concurrent ~0.2s. Assert well under 2× one fetch.
assert elapsed < 0.4, f"expected concurrent (~0.2s), got {elapsed:.2f}s (serial?)"
# Concurrency is proven by construction, not by measuring elapsed time.
#
# The old form asserted `elapsed < 0.4` ("well under 2x one 0.2s fetch").
# That makes the event-loop scheduler and any fixed setup part of the
# assertion: under a loaded CI box the inequality can flip with nothing
# wrong in the code under test, and the margin shrinks silently if setup
# cost is ever added ahead of dispatch.
#
# A barrier asserts the invariant directly: all THREE fetches must be
# inside the fetcher AT THE SAME TIME before any is allowed to return. If
# expansion ever goes serial the first fetch blocks waiting for partners
# that will not arrive, the barrier times out, and the test fails with an
# explicit message. No wall-clock constant, no load sensitivity.
N_REFS = 3
rendezvous = asyncio.Barrier(N_REFS)
entered: list[str] = []
overlapped = asyncio.Event()
async def barrier_fetcher(url: str) -> str:
entered.append(url)
# Generous relative to real scheduling latency (a rendezvous between
# already-dispatched coroutines needs milliseconds), but finite so a
# serial regression fails fast instead of hanging the suite.
async with asyncio.timeout(10):
await rendezvous.wait()
overlapped.set()
return f"CONTENT[{url}]"
try:
res = await preprocess_context_references_async(
msg, cwd=tmp_path, context_length=100_000, url_fetcher=barrier_fetcher,
)
except (asyncio.BrokenBarrierError, TimeoutError): # pragma: no cover - serial regression
pytest.fail(
"references did not expand concurrently: a fetch reached the "
f"rendezvous alone, so expansion never had {N_REFS} fetches in "
f"flight at once (entered: {entered})"
)
# The barrier only clears when all three fetches are in flight together.
assert overlapped.is_set(), f"references never overlapped (entered: {entered})"
assert len(entered) == N_REFS, f"expected {N_REFS} fetches, got {entered}"
# All three blocks present, in order.
assert res.expanded
body = res.message

View file

@ -66,14 +66,27 @@ def test_boundary_commit_delivers_end_strictly_before_switch():
mm = _make_manager(provider)
msgs = [{"role": "user", "content": "old turn"}]
t0 = time.monotonic()
mm.commit_session_boundary_async(
msgs, new_session_id="new-sid", parent_session_id="old-sid"
)
# Caller returns immediately — the slow extraction must not block /new.
assert time.monotonic() - t0 < 0.1
# DETERMINISTIC non-blocking witness — replaces `assert elapsed < 0.1`.
#
# The old form timed `commit_session_boundary_async` and required it under
# 100ms, which makes the scheduler part of the assertion: thread startup
# alone can exceed that on a loaded box, flipping the inequality with
# nothing wrong in the code under test.
#
# The real contract is that the caller returns WITHOUT waiting for the slow
# extraction. Assert it directly: the background `on_session_end` sleeps
# 0.15s before recording anything, so if the caller had blocked on it, the
# provider would already have recorded the "end" call by the time we get
# here. An empty call list is a positive witness that /new was not gated.
assert provider.calls == [], (
"commit_session_boundary_async blocked on the slow extraction: "
f"provider already recorded {provider.calls} before the caller returned"
)
assert mm.flush_pending(timeout=5)
assert mm.flush_pending(timeout=30)
kinds = [c[0] for c in provider.calls]
assert kinds == ["end", "switch"], f"ordering violated: {provider.calls}"

View file

@ -1,6 +1,7 @@
"""Tests for Mem0 v3 API — new tool names, paginated responses, update/delete tools."""
import json
import threading
import time
import pytest
@ -186,19 +187,47 @@ class TestMem0Prefetch:
assert len([c for c in backend.captured if c[0] == "search"]) == 1
def test_slow_prefetch_returns_quickly(self, monkeypatch):
entered = threading.Event()
release = threading.Event()
search_returned = threading.Event()
class SlowBackend(FakeBackend):
def search(self, query, *, filters, top_k=10, rerank=True):
time.sleep(0.2)
return super().search(query, filters=filters, top_k=top_k, rerank=rerank)
entered.set()
try:
release.wait(30)
return super().search(
query, filters=filters, top_k=top_k, rerank=rerank
)
finally:
search_returned.set()
monkeypatch.setattr(mem0_plugin, "_PREFETCH_WAIT_SECS", 0.01)
provider = self._make_provider(
SlowBackend(search_results=[{"id": "m1", "memory": "lives in Berlin"}])
)
started = time.monotonic()
# DETERMINISTIC non-blocking witness — replaces `assert elapsed < 0.1`.
#
# The old form slept 0.2s in the backend and asserted prefetch returned
# in under 0.1s. That makes the OS scheduler part of the assertion: on
# a loaded box thread startup alone can eat the 100ms budget, so the
# inequality flips with nothing wrong in the code under test. Observed
# failing in a full-directory run of tests/plugins/memory.
#
# The real contract is that prefetch gives up on the slow backend
# instead of waiting for it. Assert it directly: the backend search is
# STILL PARKED (release unset, so `search_returned` cannot be set). If
# prefetch ever waited for the backend, the search would have returned
# first and this fails. No wall-clock constant.
assert provider.prefetch("where do I live?") == ""
assert time.monotonic() - started < 0.1
provider._prefetch_thread.join(timeout=1)
assert entered.wait(30), "prefetch never reached the backend"
assert not search_returned.is_set(), (
"prefetch blocked on the slow backend: the backend search had "
"already returned by the time prefetch did"
)
release.set()
provider._prefetch_thread.join(timeout=30)
assert "lives in Berlin" in provider.prefetch("where do I live?")