tradein-mvp/backend/app/api/v1/admin.py

@@ -2110,3 +2110,203 @@ async def rotate_proxy_ip(
         new_ip = data.get("new_ip") or data.get("ip") or data.get("proxy_ip")
     logger.info("rotate-ip: source=%s new_ip=%s", source, new_ip)
     return RotateIpResponse(ok=True, new_ip=str(new_ip) if new_ip else None)
+
+
+# ── Pacing live-регулятор (GET/PUT /scraper/pacing) ──────────────────────────
+
+
+class PacingProvider(BaseModel):
+    source: str
+    interval_s: float
+    env_default_s: float
+
+
+class PacingResponse(BaseModel):
+    providers: list[PacingProvider]
+
+
+class PacingUpdateRequest(BaseModel):
+    interval_s: float = Field(ge=0, le=120)
+
+
+async def _proxy_browser_pacing_get() -> dict:
+    """GET tradein-browser /pacing (timeout 5с). Пробрасывает ответ или кидает HTTPException."""
+    url = f"{settings.browser_http_endpoint.rstrip('/')}/pacing"
+    try:
+        async with httpx.AsyncClient(timeout=5.0) as client:
+            resp = await client.get(url)
+            resp.raise_for_status()
+            return resp.json()  # type: ignore[no-any-return]
+    except httpx.HTTPStatusError as exc:
+        raise HTTPException(
+            status_code=502,
+            detail=f"browser /pacing returned {exc.response.status_code}",
+        ) from exc
+    except Exception as exc:
+        raise HTTPException(
+            status_code=503,
+            detail=f"browser unreachable: {type(exc).__name__}",
+        ) from exc
+
+
+@router.get("/scraper/pacing", response_model=PacingResponse)
+async def get_scraper_pacing() -> PacingResponse:
+    """Текущие live-интервалы пейсинга из tradein-browser (in-memory, per-provider).
+
+    interval_s    — живой интервал (может быть изменён PUT).
+    env_default_s — значение из env (к чему сбросится на рестарте).
+    502/503 при недоступности browser-сервиса.
+    """
+    data = await _proxy_browser_pacing_get()
+    providers = [
+        PacingProvider(
+            source=p["source"],
+            interval_s=float(p["interval_s"]),
+            env_default_s=float(p["env_default_s"]),
+        )
+        for p in data.get("providers", [])
+    ]
+    return PacingResponse(providers=providers)
+
+
+@router.put("/scraper/pacing/{source}", response_model=dict)
+async def update_scraper_pacing(
+    source: Literal["avito", "cian", "yandex", "generic"],
+    payload: PacingUpdateRequest,
+) -> dict:
+    """Обновить live-интервал пейсинга для провайдера в tradein-browser (in-memory).
+
+    Значение сбрасывается к env-дефолту при рестарте контейнера — by design.
+    interval_s: 0..120 (сек). Возвращает {ok, source, interval_s}.
+    502/503 при недоступности browser-сервиса.
+    """
+    url = f"{settings.browser_http_endpoint.rstrip('/')}/pacing"
+    try:
+        async with httpx.AsyncClient(timeout=5.0) as client:
+            resp = await client.put(
+                url,
+                json={"source": source, "interval_s": payload.interval_s},
+            )
+            resp.raise_for_status()
+            return resp.json()  # type: ignore[no-any-return]
+    except httpx.HTTPStatusError as exc:
+        raise HTTPException(
+            status_code=502,
+            detail=f"browser /pacing returned {exc.response.status_code}",
+        ) from exc
+    except Exception as exc:
+        raise HTTPException(
+            status_code=503,
+            detail=f"browser unreachable: {type(exc).__name__}",
+        ) from exc
+
+
+# ── Data-quality coverage (GET /scraper/data-quality) ────────────────────────
+
+
+class SourceCoverage(BaseModel):
+    source: str
+    active_count: int
+    fields: dict[str, float]  # field_name -> fill% (0..100, round 1)
+
+
+class HousesCoverage(BaseModel):
+    total: int
+    validated_pct: float  # avito_validated_at IS NOT NULL %
+    rating_pct: float  # rating_score IS NOT NULL %
+    house_type_pct: float  # house_type IS NOT NULL %
+    reviews_count: int
+
+
+class DataQualityResponse(BaseModel):
+    sources: list[SourceCoverage]
+    houses: HousesCoverage
+
+
+# Поля listings для fill%-аудита. Каждый кортеж: (имя_поля, SQL-выражение IS NOT NULL).
+# Для photo_urls отдельная логика (jsonb не NULL и не '[]').
+_DQ_LISTING_FIELDS: list[tuple[str, str]] = [
+    ("description", "description IS NOT NULL AND description <> ''"),
+    ("photo_urls", "photo_urls IS NOT NULL AND photo_urls <> '[]'::jsonb"),
+    ("address", "address IS NOT NULL AND address <> ''"),
+    ("lat", "lat IS NOT NULL"),
+    ("lon", "lon IS NOT NULL"),
+    ("kitchen_area_m2", "kitchen_area_m2 IS NOT NULL"),
+    ("living_area_m2", "living_area_m2 IS NOT NULL"),
+    ("ceiling_height", "ceiling_height IS NOT NULL"),
+    ("ceiling_height_m", "ceiling_height_m IS NOT NULL"),
+    ("metro_stations", "metro_stations IS NOT NULL AND metro_stations <> '[]'::jsonb"),
+]
+
+
+@router.get("/scraper/data-quality", response_model=DataQualityResponse)
+def get_data_quality(
+    db: Annotated[Session, Depends(get_db)],
+) -> DataQualityResponse:
+    """Fill%-аудит detail-полей listings по source + дом-статистика.
+
+    Один проход per source через COUNT(*)...FILTER — не N запросов.
+    Поля listings: description, photo_urls, address, lat/lon, kitchen_area_m2,
+    living_area_m2, ceiling_height (cian), ceiling_height_m (avito), metro_stations.
+    houses: total, avito_validated_at%, rating_score%, house_type%.
+    house_reviews: общий count.
+    """
+    # Строим single-pass SELECT для listings полей через FILTER-агрегаты.
+    # Структура: COUNT(*) FILTER (WHERE <expr>) / NULLIF(COUNT(*), 0) * 100
+    # Одним запросом получаем active_count + все fill-counts по каждому source.
+    filter_exprs = ", ".join(
+        f"COUNT(*) FILTER (WHERE {expr}) AS f_{name}" for name, expr in _DQ_LISTING_FIELDS
+    )
+    sql_listings = text(f"""
+        SELECT
+            source,
+            COUNT(*) AS active_count,
+            {filter_exprs}
+        FROM listings
+        WHERE is_active = true
+        GROUP BY source
+        ORDER BY source
+    """)
+
+    rows = db.execute(sql_listings).mappings().all()
+
+    sources: list[SourceCoverage] = []
+    for row in rows:
+        cnt = int(row["active_count"]) or 1  # защита от деления на ноль
+        fields: dict[str, float] = {}
+        for name, _ in _DQ_LISTING_FIELDS:
+            raw = row[f"f_{name}"]
+            fill_pct = round(float(raw or 0) / cnt * 100, 1)
+            fields[name] = fill_pct
+        sources.append(
+            SourceCoverage(
+                source=row["source"],
+                active_count=int(row["active_count"]),
+                fields=fields,
+            )
+        )
+
+    # Houses: один агрегат
+    sql_houses = text("""
+        SELECT
+            COUNT(*) AS total,
+            COUNT(*) FILTER (WHERE avito_validated_at IS NOT NULL) AS validated_cnt,
+            COUNT(*) FILTER (WHERE rating_score IS NOT NULL) AS rating_cnt,
+            COUNT(*) FILTER (WHERE house_type IS NOT NULL) AS house_type_cnt
+        FROM houses
+    """)
+    h = db.execute(sql_houses).mappings().one()
+    total_houses = int(h["total"]) or 1  # guard против деления на ноль в %
+
+    reviews_count_row = db.execute(text("SELECT COUNT(*) AS cnt FROM house_reviews")).scalar()
+    reviews_count = int(reviews_count_row or 0)
+
+    houses = HousesCoverage(
+        total=int(h["total"]),
+        validated_pct=round(float(h["validated_cnt"] or 0) / total_houses * 100, 1),
+        rating_pct=round(float(h["rating_cnt"] or 0) / total_houses * 100, 1),
+        house_type_pct=round(float(h["house_type_cnt"] or 0) / total_houses * 100, 1),
+        reviews_count=reviews_count,
+    )
+
+    return DataQualityResponse(sources=sources, houses=houses)

tradein-mvp/backend/tests/test_scraper_admin_apis.py

@@ -306,3 +306,261 @@ def test_rotate_ip_changeip_error(client: TestClient) -> None:
     body = r.json()
     assert body["ok"] is False
     assert "changeip error" in body["reason"]
+
+
+# ── API 4: GET /scraper/pacing ───────────────────────────────────────────────
+
+
+def test_pacing_get_proxies_browser_response(client: TestClient) -> None:
+    """GET /scraper/pacing → парсит browser /pacing и возвращает PacingResponse."""
+    from app.api.v1 import admin as admin_module
+
+    fake_browser_resp = {
+        "providers": [
+            {"source": "avito", "interval_s": 3.0, "env_default_s": 2.0},
+            {"source": "cian", "interval_s": 18.0, "env_default_s": 2.0},
+            {"source": "yandex", "interval_s": 2.0, "env_default_s": 2.0},
+            {"source": "generic", "interval_s": 2.0, "env_default_s": 2.0},
+        ]
+    }
+
+    async def _fake_proxy() -> dict:
+        return fake_browser_resp
+
+    with patch.object(admin_module, "_proxy_browser_pacing_get", _fake_proxy):
+        r = client.get("/api/v1/admin/scraper/pacing")
+
+    assert r.status_code == 200
+    body = r.json()
+    assert "providers" in body
+    by_source = {p["source"]: p for p in body["providers"]}
+    assert set(by_source) == {"avito", "cian", "yandex", "generic"}
+    assert by_source["cian"]["interval_s"] == 18.0
+    assert by_source["avito"]["env_default_s"] == 2.0
+
+
+def test_pacing_get_browser_unreachable_503(client: TestClient) -> None:
+    """GET /scraper/pacing при недоступном browser → 503."""
+    from fastapi import HTTPException
+
+    from app.api.v1 import admin as admin_module
+
+    async def _boom() -> dict:
+        raise HTTPException(status_code=503, detail="browser unreachable: ConnectError")
+
+    with patch.object(admin_module, "_proxy_browser_pacing_get", _boom):
+        r = client.get("/api/v1/admin/scraper/pacing")
+
+    assert r.status_code == 503
+
+
+# ── API 5: PUT /scraper/pacing/{source} ─────────────────────────────────────
+
+
+def test_pacing_put_proxies_to_browser(client: TestClient) -> None:
+    """PUT /scraper/pacing/avito → проксирует на browser и отдаёт {ok,source,interval_s}."""
+    from app.api.v1 import admin as admin_module
+
+    captured: dict[str, Any] = {}
+
+    class _FakeResp:
+        def raise_for_status(self) -> None:
+            return None
+
+        def json(self) -> dict:
+            return {"ok": True, "source": "avito", "interval_s": 7.5}
+
+    class _FakeClient:
+        def __init__(self, *a: Any, **k: Any) -> None:
+            pass
+
+        async def __aenter__(self) -> _FakeClient:
+            return self
+
+        async def __aexit__(self, *a: Any) -> None:
+            return None
+
+        async def put(self, url: str, json: Any = None, **k: Any) -> _FakeResp:
+            captured["url"] = url
+            captured["json"] = json
+            return _FakeResp()
+
+    with patch.object(admin_module.httpx, "AsyncClient", _FakeClient):
+        r = client.put(
+            "/api/v1/admin/scraper/pacing/avito",
+            json={"interval_s": 7.5},
+        )
+
+    assert r.status_code == 200
+    body = r.json()
+    assert body["ok"] is True
+    assert body["source"] == "avito"
+    assert body["interval_s"] == 7.5
+    assert captured["json"]["source"] == "avito"
+    assert captured["json"]["interval_s"] == 7.5
+
+
+def test_pacing_put_invalid_source_422(client: TestClient) -> None:
+    """PUT /scraper/pacing/domclick → 422 (Literal-валидация)."""
+    r = client.put("/api/v1/admin/scraper/pacing/domclick", json={"interval_s": 5.0})
+    assert r.status_code == 422
+
+
+def test_pacing_put_negative_interval_422(client: TestClient) -> None:
+    """PUT /scraper/pacing/cian с interval_s < 0 → 422 (ge=0 валидация)."""
+    r = client.put("/api/v1/admin/scraper/pacing/cian", json={"interval_s": -1.0})
+    assert r.status_code == 422
+
+
+def test_pacing_put_over_max_interval_422(client: TestClient) -> None:
+    """PUT /scraper/pacing/yandex с interval_s > 120 → 422 (le=120 валидация)."""
+    r = client.put("/api/v1/admin/scraper/pacing/yandex", json={"interval_s": 200.0})
+    assert r.status_code == 422
+
+
+def test_pacing_put_browser_unreachable_503(client: TestClient) -> None:
+    """PUT /scraper/pacing/avito при недоступном browser → 503."""
+    from app.api.v1 import admin as admin_module
+
+    class _BoomClient:
+        def __init__(self, *a: Any, **k: Any) -> None:
+            pass
+
+        async def __aenter__(self) -> _BoomClient:
+            return self
+
+        async def __aexit__(self, *a: Any) -> None:
+            return None
+
+        async def put(self, *a: Any, **k: Any) -> Any:
+            raise RuntimeError("connection refused")
+
+    with patch.object(admin_module.httpx, "AsyncClient", _BoomClient):
+        r = client.put(
+            "/api/v1/admin/scraper/pacing/avito",
+            json={"interval_s": 5.0},
+        )
+
+    assert r.status_code == 503
+
+
+# ── API 6: GET /scraper/data-quality ────────────────────────────────────────
+
+
+def _make_dq_db_mock() -> MagicMock:
+    """Мок db-сессии для GET /scraper/data-quality."""
+    db = MagicMock()
+
+    # listings rows: avito=1000 active, cian=500 active
+    avito_row = {
+        "source": "avito",
+        "active_count": 1000,
+        "f_description": 900,
+        "f_photo_urls": 980,
+        "f_address": 1000,
+        "f_lat": 950,
+        "f_lon": 950,
+        "f_kitchen_area_m2": 600,
+        "f_living_area_m2": 100,
+        "f_ceiling_height": 50,
+        "f_ceiling_height_m": 300,
+        "f_metro_stations": 700,
+    }
+    cian_row = {
+        "source": "cian",
+        "active_count": 500,
+        "f_description": 450,
+        "f_photo_urls": 490,
+        "f_address": 500,
+        "f_lat": 480,
+        "f_lon": 480,
+        "f_kitchen_area_m2": 400,
+        "f_living_area_m2": 400,
+        "f_ceiling_height": 350,
+        "f_ceiling_height_m": 0,
+        "f_metro_stations": 100,
+    }
+
+    houses_row = {
+        "total": 2000,
+        "validated_cnt": 1800,
+        "rating_cnt": 1200,
+        "house_type_cnt": 1900,
+    }
+
+    listing_mappings = MagicMock()
+    listing_mappings.all.return_value = [avito_row, cian_row]
+
+    houses_mappings = MagicMock()
+    houses_mappings.one.return_value = houses_row
+
+    reviews_scalar = 4500
+
+    # db.execute() вызывается 3 раза: listings SQL, houses SQL, house_reviews COUNT
+    execute_results = iter(
+        [
+            MagicMock(mappings=lambda: listing_mappings),
+            MagicMock(mappings=lambda: houses_mappings),
+            MagicMock(scalar=lambda: reviews_scalar),
+        ]
+    )
+    db.execute.side_effect = lambda *a, **k: next(execute_results)
+    return db
+
+
+def test_data_quality_shape(client: TestClient) -> None:
+    """GET /scraper/data-quality → StructureCheck: sources + houses + reviews_count."""
+    from app.api.v1 import admin as admin_module  # noqa: F401
+    from app.core.db import get_db
+
+    app = client.app
+    app.dependency_overrides[get_db] = lambda: _make_dq_db_mock()
+
+    r = client.get("/api/v1/admin/scraper/data-quality")
+    assert r.status_code == 200
+    body = r.json()
+
+    assert "sources" in body
+    assert "houses" in body
+
+    # two sources
+    sources = {s["source"]: s for s in body["sources"]}
+    assert set(sources) >= {"avito", "cian"}
+
+    avito = sources["avito"]
+    assert avito["active_count"] == 1000
+    fields = avito["fields"]
+    assert 0.0 <= fields["description"] <= 100.0
+    assert fields["description"] == pytest.approx(90.0, abs=0.1)
+    assert fields["photo_urls"] == pytest.approx(98.0, abs=0.1)
+    assert fields["kitchen_area_m2"] == pytest.approx(60.0, abs=0.1)
+
+    cian = sources["cian"]
+    assert cian["active_count"] == 500
+    assert cian["fields"]["ceiling_height"] == pytest.approx(70.0, abs=0.1)
+    assert cian["fields"]["ceiling_height_m"] == pytest.approx(0.0, abs=0.1)
+
+    houses = body["houses"]
+    assert houses["total"] == 2000
+    assert houses["validated_pct"] == pytest.approx(90.0, abs=0.1)
+    assert houses["rating_pct"] == pytest.approx(60.0, abs=0.1)
+    assert houses["house_type_pct"] == pytest.approx(95.0, abs=0.1)
+    assert houses["reviews_count"] == 4500
+
+
+def test_data_quality_pct_in_range(client: TestClient) -> None:
+    """Все fill% в [0,100]."""
+    from app.core.db import get_db
+
+    app = client.app
+    app.dependency_overrides[get_db] = lambda: _make_dq_db_mock()
+
+    r = client.get("/api/v1/admin/scraper/data-quality")
+    assert r.status_code == 200
+    body = r.json()
+
+    for src in body["sources"]:
+        for field_name, pct in src["fields"].items():
+            assert (
+                0.0 <= pct <= 100.0
+            ), f"source={src['source']} field={field_name} pct={pct} вне [0,100]"

tradein-mvp/browser/server.py

@@ -710,6 +710,63 @@ class LoginError(Exception):
         self.payload = payload
 
 
+async def pacing_get_handler(request: web.Request) -> web.Response:
+    """GET /pacing → текущие live-интервалы + env-дефолты для всех провайдеров.
+
+    interval_s    — живое значение из _MIN_PAGE_INTERVAL_BY_PROVIDER (изменяется PUT).
+    env_default_s — что вернёт _resolve_min_interval(p) из os.environ (reset при рестарте).
+    Всегда 200; не требует browser-инстансов (memory-only).
+    """
+    providers = []
+    for p in PROVIDERS:
+        providers.append({
+            "source": p,
+            "interval_s": _MIN_PAGE_INTERVAL_BY_PROVIDER.get(p, BROWSER_MIN_PAGE_INTERVAL_S),
+            "env_default_s": _resolve_min_interval(p),
+        })
+    return web.json_response({"providers": providers})
+
+
+async def pacing_put_handler(request: web.Request) -> web.Response:
+    """PUT /pacing {"source": <p>, "interval_s": <float>} → обновляет in-memory интервал.
+
+    Валидация: source ∈ {avito,cian,yandex,generic}, interval_s ≥ 0.
+    Сброс к env-дефолту происходит при рестарте контейнера (by design).
+    """
+    try:
+        body = await request.json()
+    except Exception:
+        return web.json_response({"error": "invalid JSON body"}, status=400)
+
+    source = body.get("source")
+    if not isinstance(source, str) or source not in PROVIDERS:
+        return web.json_response(
+            {"error": f"source must be one of {list(PROVIDERS)}"},
+            status=400,
+        )
+
+    interval_s = body.get("interval_s")
+    if not isinstance(interval_s, (int, float)):
+        return web.json_response(
+            {"error": "interval_s must be a non-negative number"},
+            status=422,
+        )
+    interval_s = float(interval_s)
+    if interval_s < 0:
+        return web.json_response(
+            {"error": "interval_s must be >= 0"},
+            status=422,
+        )
+
+    _MIN_PAGE_INTERVAL_BY_PROVIDER[source] = interval_s
+    logger.info(
+        "tradein-browser: pacing updated source=%s interval_s=%.2f (resets on restart)",
+        source,
+        interval_s,
+    )
+    return web.json_response({"ok": True, "source": source, "interval_s": interval_s})
+
+
 async def login_handler(request: web.Request) -> web.Response:
     """POST /login {...} → {"cookies": [...]}
 
@@ -919,6 +976,8 @@ def build_app() -> web.Application:
     app.router.add_get("/health", health_handler)
     app.router.add_post("/fetch", fetch_handler)
     app.router.add_post("/login", login_handler)
+    app.router.add_get("/pacing", pacing_get_handler)
+    app.router.add_put("/pacing", pacing_put_handler)
     return app
 
 

tradein-mvp/browser/test_server_pacing.py

@@ -1,12 +1,14 @@
-"""test_server_pacing.py — юниты для per-provider pacing-интервалов (#1812 follow-up).
+"""test_server_pacing.py — юниты для per-provider pacing-интервалов + HTTP-эндпоинтов.
 
 Проверяет:
   1. per-provider env override переопределяет глобал для конкретного провайдера;
   2. отсутствие per-provider env → фолбэк на глобальный BROWSER_MIN_PAGE_INTERVAL_S;
   3. невалидный per-provider env (не float) → фолбэк на глобал, без исключения;
   4. пустая строка как per-provider env → фолбэк на глобал (ValueError при float("")).
+  5. GET /pacing — возвращает текущие live-значения + env-дефолты для всех провайдеров;
+  6. PUT /pacing — обновляет _MIN_PAGE_INTERVAL_BY_PROVIDER, валидирует source/interval_s.
 
-camoufox/Playwright НЕ поднимается: _resolve_min_interval вызывается с фейковым environ.
+camoufox/Playwright НЕ поднимается: тесты работают с чистым dict'ом и aiohttp mock.
 
 Запуск (из tradein-mvp/browser/)::
 
@@ -15,8 +17,14 @@ camoufox/Playwright НЕ поднимается: _resolve_min_interval вызы
 
 from __future__ import annotations
 
+import asyncio
 import importlib.util
+import json
 from pathlib import Path
+from typing import Any
+
+import pytest
+from aiohttp.test_utils import make_mocked_request
 
 # server.py — не пакет (отдельный сервис без __init__/pyproject). Грузим по пути.
 _SERVER_PATH = Path(__file__).resolve().parent / "server.py"
@@ -102,3 +110,131 @@ def test_all_providers_covered() -> None:
     for provider in ("avito", "cian", "yandex", "generic"):
         result = server._resolve_min_interval(provider, {})
         assert isinstance(result, float), f"{provider}: ожидали float, получили {type(result)}"
+
+
+# ── GET /pacing ─────────────────────────────────────────────────────────────────
+
+
+def _json_body(response: Any) -> dict[str, Any]:
+    return json.loads(response.body.decode())
+
+
+@pytest.fixture(autouse=True)
+def _reset_pacing_state(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Восстанавливаем _MIN_PAGE_INTERVAL_BY_PROVIDER после каждого теста."""
+    original = dict(server._MIN_PAGE_INTERVAL_BY_PROVIDER)
+    yield
+    server._MIN_PAGE_INTERVAL_BY_PROVIDER.clear()
+    server._MIN_PAGE_INTERVAL_BY_PROVIDER.update(original)
+
+
+def test_pacing_get_returns_all_providers() -> None:
+    """GET /pacing возвращает providers для всех 4 источников."""
+    response = asyncio.run(server.pacing_get_handler(make_mocked_request("GET", "/pacing")))
+    body = _json_body(response)
+    assert response.status == 200
+    providers = {p["source"]: p for p in body["providers"]}
+    assert set(providers) == {"avito", "cian", "yandex", "generic"}
+
+
+def test_pacing_get_reflects_live_value(monkeypatch: pytest.MonkeyPatch) -> None:
+    """GET /pacing.interval_s отражает живое значение из _MIN_PAGE_INTERVAL_BY_PROVIDER."""
+    server._MIN_PAGE_INTERVAL_BY_PROVIDER["cian"] = 99.0
+    response = asyncio.run(server.pacing_get_handler(make_mocked_request("GET", "/pacing")))
+    body = _json_body(response)
+    providers = {p["source"]: p for p in body["providers"]}
+    assert providers["cian"]["interval_s"] == 99.0
+
+
+def test_pacing_get_has_env_default() -> None:
+    """GET /pacing.env_default_s — это _resolve_min_interval(provider) из os.environ."""
+    response = asyncio.run(server.pacing_get_handler(make_mocked_request("GET", "/pacing")))
+    body = _json_body(response)
+    for p in body["providers"]:
+        assert isinstance(p["env_default_s"], float), (
+            f"{p['source']}: env_default_s должен быть float"
+        )
+
+
+# ── PUT /pacing ─────────────────────────────────────────────────────────────────
+
+
+def _put_request(body_dict: dict) -> Any:
+    req = make_mocked_request("PUT", "/pacing")
+    req.json = lambda: _coro_value(body_dict)  # type: ignore[method-assign]
+    return req
+
+
+def _coro_value(value: Any) -> Any:
+    async def _inner() -> Any:
+        return value
+    return _inner()
+
+
+def test_pacing_put_updates_dict() -> None:
+    """PUT /pacing с валидным source+interval_s обновляет _MIN_PAGE_INTERVAL_BY_PROVIDER."""
+    original = server._MIN_PAGE_INTERVAL_BY_PROVIDER.get("avito")
+    asyncio.run(server.pacing_put_handler(_put_request({"source": "avito", "interval_s": 42.0})))
+    assert server._MIN_PAGE_INTERVAL_BY_PROVIDER["avito"] == 42.0
+    # Восстановить (на случай если фикстура autouse пропустит)
+    if original is not None:
+        server._MIN_PAGE_INTERVAL_BY_PROVIDER["avito"] = original
+
+
+def test_pacing_put_returns_ok() -> None:
+    """PUT /pacing возвращает {ok:true, source, interval_s}."""
+    response = asyncio.run(
+        server.pacing_put_handler(_put_request({"source": "yandex", "interval_s": 15.5}))
+    )
+    assert response.status == 200
+    body = _json_body(response)
+    assert body["ok"] is True
+    assert body["source"] == "yandex"
+    assert body["interval_s"] == 15.5
+
+
+def test_pacing_put_zero_disables_pacing() -> None:
+    """PUT /pacing с interval_s=0 разрешён (выключение пейсинга)."""
+    response = asyncio.run(
+        server.pacing_put_handler(_put_request({"source": "cian", "interval_s": 0}))
+    )
+    assert response.status == 200
+    assert server._MIN_PAGE_INTERVAL_BY_PROVIDER["cian"] == 0.0
+
+
+def test_pacing_put_bad_source_400() -> None:
+    """PUT /pacing с невалидным source → 400."""
+    response = asyncio.run(
+        server.pacing_put_handler(_put_request({"source": "domclick", "interval_s": 5.0}))
+    )
+    assert response.status == 400
+    body = _json_body(response)
+    assert "source" in body["error"]
+
+
+def test_pacing_put_negative_interval_422() -> None:
+    """PUT /pacing с interval_s < 0 → 422."""
+    response = asyncio.run(
+        server.pacing_put_handler(_put_request({"source": "avito", "interval_s": -1.0}))
+    )
+    assert response.status == 422
+
+
+def test_pacing_put_missing_interval_422() -> None:
+    """PUT /pacing без interval_s → 422 (не число)."""
+    response = asyncio.run(
+        server.pacing_put_handler(_put_request({"source": "avito", "interval_s": "fast"}))
+    )
+    assert response.status == 422
+
+
+def test_pacing_put_invalid_json_400() -> None:
+    """PUT /pacing с битым JSON → 400."""
+    req = make_mocked_request("PUT", "/pacing")
+
+    async def _boom() -> Any:
+        raise ValueError("not json")
+
+    req.json = _boom  # type: ignore[method-assign]
+    response = asyncio.run(server.pacing_put_handler(req))
+    assert response.status == 400