test(tradein): гонка на sleep(0.05) вешала тест потолка оценок на 120с (#3270)

Тест занимал все слоты семафора четырьмя висящими запросами и ждал, что
они дошли до acquire, обычным сном на 50 мс. На нагруженном раннере этого
не хватало: пятый запрос заставал свободный слот, входил в подменённую
оценку и вставал на gate.wait() — а gate.set() стоит ниже по тому же
корутину. Дедлок до pytest-timeout, две минуты простоя job'а и красный CI
на постороннем PR (#3268).

Сон заменён счётным барьером: подменённая оценка отпускает семафор при
входе, тест дожидается ровно _CONCURRENCY входов. Вход означает, что слот
уже захвачен, — это то самое условие, которое сон угадывал по времени.

Проверено пробой с намеренно свободным слотом (держателей на одного
меньше): старая структура висит до таймаута, новая падает за 12с с
понятным сообщением.

tradein-mvp/backend/tests/test_3082_estimate_concurrency_cap.py

@@ -85,8 +85,11 @@ async def test_fifth_concurrent_estimate_gets_fast_429(
     )
 
     gate = asyncio.Event()
+    # Счётчик входов в оценку: слот к этому моменту уже захвачен эндпоинтом.
+    entered = asyncio.Semaphore(0)
 
     async def _slow_estimate(*args, **kwargs) -> AggregatedEstimate:
+        entered.release()
         await gate.wait()
         return _canned_estimate()
 
@@ -102,10 +105,24 @@ async def test_fifth_concurrent_estimate_gets_fast_429(
                 asyncio.create_task(client.post("/api/v1/trade-in/estimate", json=payload))
                 for _ in range(_CONCURRENCY)
             ]
-            # Дать держателям дойти до acquire и занять все слоты.
-            await asyncio.sleep(0.05)
+            # Дождаться, что все держатели ВОШЛИ в оценку, то есть слоты заняты.
+            # Сон вместо этого — гонка (#3270): на нагруженном раннере пятый
+            # запрос успевал захватить свободный слот и навсегда вис на
+            # gate.wait(), потому что gate.set() стоит ниже по этому же корутину.
+            for i in range(_CONCURRENCY):
+                try:
+                    await asyncio.wait_for(entered.acquire(), timeout=10)
+                except TimeoutError:  # pragma: no cover — диагностика
+                    gate.set()
+                    pytest.fail(f"до оценки дошли {i} держателей из {_CONCURRENCY}")
 
-            fifth = await client.post("/api/v1/trade-in/estimate", json=payload)
+            try:
+                fifth = await asyncio.wait_for(
+                    client.post("/api/v1/trade-in/estimate", json=payload), timeout=10
+                )
+            except TimeoutError:  # pragma: no cover — диагностика вместо 120с виса
+                gate.set()
+                pytest.fail("5-й запрос повис вместо быстрого 429 — слот оказался свободен")
             assert fifth.status_code == 429, (
                 f"5-й конкурентный запрос прошёл ({fifth.status_code}) — "
                 "потолка одновременности нет"