fix(gisogd66): битый 500-сегмент портала — no-retry карточек, circuit breaker, периодический commit (#2371)
This commit is contained in:
parent
282c92747e
commit
ae3278f92f
2 changed files with 227 additions and 28 deletions
|
|
@ -13,15 +13,21 @@
|
|||
WGS84 lon/lat, может отсутствовать — hasGeometry=false), approvedOrganization, docNum,
|
||||
dateDoc, dateReg.
|
||||
|
||||
Две схемы (тянем обе): ``agate_ekbgo`` (ЕКБ, свежая) и ``agate_sverdregion`` (регион-
|
||||
агрегат, отстаёт ~неделю). Один документ может встретиться в обеих → бизнес-ключ
|
||||
``(doc_group, doc_num)``; при конфликте предпочитаем запись с более поздним ``date_reg``.
|
||||
Две схемы (тянем обе, sverdregion первым): ``agate_sverdregion`` (регион-агрегат,
|
||||
стабильные карточки, покрывает и ЕКБ) и ``agate_ekbgo`` (ЕКБ, свежее на ~неделю, но
|
||||
часть старых ключей битая на портале). Один документ может встретиться в обеих →
|
||||
бизнес-ключ ``(doc_group, doc_num)``; при конфликте предпочитаем запись с более
|
||||
поздним ``date_reg``.
|
||||
|
||||
Инкрементальность: список группы сверяется с БД по (source_schema, source_key, date_reg);
|
||||
карточки (второй запрос — дорогой) тянутся ТОЛЬКО для новых / изменившихся документов.
|
||||
|
||||
Politeness: единый httpx-клиент с limits(max_connections=5), timeout=20с, 2 ретрая на 5xx,
|
||||
малый sleep между карточками. UPSERT — per-row SAVEPOINT (битая строка не валит батч).
|
||||
Politeness: единый httpx-клиент с limits(max_connections=5), timeout=20с, 2 ретрая на
|
||||
сетевых ошибках. Карточки (5xx-битые записи портала) НЕ ретраятся — deterministically
|
||||
broken, не транзиент. Circuit breaker: 50 карточек подряд с ошибкой → прерываем группу
|
||||
(битый сегмент). UPSERT — per-row SAVEPOINT (битая строка не валит батч). Периодический
|
||||
db.commit() (каждые 200 записей + после каждой schema/group), чтобы партиал пережил
|
||||
soft_time_limit-kill.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -46,17 +52,26 @@ logger = logging.getLogger(__name__)
|
|||
GISOGD_BASE = "https://gisogd66.midural.ru/api/v1"
|
||||
_SECTION = "razdel13"
|
||||
|
||||
# Схемы источника — обе тянем (ekbgo свежее, sverdregion — регион-агрегат).
|
||||
SCHEMAS: tuple[str, ...] = ("agate_ekbgo", "agate_sverdregion")
|
||||
# Схемы источника — обе тянем. Порядок ВАЖЕН: sverdregion первым.
|
||||
# sverdregion (регион-агрегат) отдаёт карточки стабильно и покрывает в т.ч. ЕКБ-доки
|
||||
# через агрегацию, поэтому он — базовый источник. ekbgo вторым как freshness-энхансер
|
||||
# (свежее на ~неделю), НО у него большой сегмент старых ключей (префикс ~100001007xxx)
|
||||
# детерминированно отдаёт 500 на карточках (сломано на стороне портала, 2026-07-04) —
|
||||
# гоняем его после того, как регион уже покрыл основную массу (см. circuit breaker ниже).
|
||||
SCHEMAS: tuple[str, ...] = ("agate_sverdregion", "agate_ekbgo")
|
||||
|
||||
# Группа источника (doc group key) → наш doc_group-код в БД (CHECK IN ('RS','RV')).
|
||||
GROUP_CODE: dict[str, str] = {"DocRS": "RS", "DocRV": "RV"}
|
||||
|
||||
_HTTP_TIMEOUT = 20.0
|
||||
_MAX_CONNECTIONS = 5
|
||||
_MAX_RETRIES = 2 # ретраи на 5xx (helper делает суммарно до 1 + _MAX_RETRIES попыток)
|
||||
_MAX_RETRIES = 2 # ретраи (helper делает суммарно до 1 + _MAX_RETRIES попыток)
|
||||
_RETRY_BACKOFF_S = 1.5
|
||||
_CARD_SLEEP_S = 0.05 # вежливая пауза между карточками (второй, «дорогой» запрос)
|
||||
_COMMIT_EVERY = 200 # периодический commit внутри группы (партиал переживает time-limit)
|
||||
# Circuit breaker: столько карточек ПОДРЯД с ошибкой fetch → прерываем группу (портал
|
||||
# отдаёт битый сегмент старых ключей детерминированным 5xx; остаток догоним будущими ранами).
|
||||
_CARD_FAIL_BREAKER = 50
|
||||
|
||||
USER_AGENT = "GenDesign/1.0 (+https://gendsgn.ru) Site Finder gisogd permits loader"
|
||||
|
||||
|
|
@ -78,20 +93,25 @@ def _make_client() -> httpx.Client:
|
|||
)
|
||||
|
||||
|
||||
def _get_json(client: httpx.Client, path: str) -> Any:
|
||||
def _get_json(client: httpx.Client, path: str, *, retries_5xx: int = _MAX_RETRIES) -> Any:
|
||||
"""GET path → JSON с ретраями на 5xx / сетевых ошибках.
|
||||
|
||||
До (1 + _MAX_RETRIES) попыток с линейным backoff. 4xx → сразу raise (не ретраим).
|
||||
Сетевые Timeout/TransportError ретраятся всегда (до 1 + _MAX_RETRIES попыток) —
|
||||
это транзиент. 5xx ретраятся ``retries_5xx`` раз: списки группы — с дефолтом
|
||||
(транзиентная нагрузка портала), карточки — с ``retries_5xx=0`` (5xx карточки
|
||||
детерминированно битые записи сегмента портала, не транзиент). 4xx → сразу raise.
|
||||
"""
|
||||
last_exc: Exception | None = None
|
||||
for attempt in range(_MAX_RETRIES + 1):
|
||||
try:
|
||||
resp = client.get(path)
|
||||
if resp.status_code >= 500:
|
||||
# 5xx — транзиентная ошибка сервера, ретраим.
|
||||
last_exc = httpx.HTTPStatusError(
|
||||
f"{resp.status_code} on {path}", request=resp.request, response=resp
|
||||
)
|
||||
# 5xx: ретраим только пока attempt в пределах retries_5xx (0 → сразу raise).
|
||||
if attempt >= retries_5xx:
|
||||
raise last_exc
|
||||
logger.warning(
|
||||
"gisogd66: %s (попытка %d) на %s", resp.status_code, attempt + 1, path
|
||||
)
|
||||
|
|
@ -128,8 +148,11 @@ def fetch_doc_card(client: httpx.Client, schema: str, key: str | int) -> dict[st
|
|||
|
||||
Поля: spatialUnit, infosetGeoJson, hasGeometry, approvedOrganization, docNum,
|
||||
dateDoc, dateReg, name.
|
||||
|
||||
5xx карточки НЕ ретраятся (retries_5xx=0): битый сегмент старых ключей портала
|
||||
отдаёт детерминированный 500 — 3 попытки с backoff = ~12с потерянного времени/карточку.
|
||||
"""
|
||||
data = _get_json(client, f"/{schema}/gisogddocs/{_SECTION}/{key}")
|
||||
data = _get_json(client, f"/{schema}/gisogddocs/{_SECTION}/{key}", retries_5xx=0)
|
||||
if not isinstance(data, dict):
|
||||
logger.warning("gisogd66: карточка %s/%s не-dict (%s)", schema, key, type(data).__name__)
|
||||
return None
|
||||
|
|
@ -304,7 +327,7 @@ def load_gisogd_permits(db: Session | None = None) -> dict[str, int]:
|
|||
SessionLocal.
|
||||
|
||||
Returns: счётчики {listed, fetched_details, inserted, updated, skipped_unchanged,
|
||||
no_cad, no_geom}.
|
||||
no_cad, no_geom, card_failed, group_aborted}.
|
||||
"""
|
||||
owns_session = db is None
|
||||
if db is None:
|
||||
|
|
@ -318,6 +341,8 @@ def load_gisogd_permits(db: Session | None = None) -> dict[str, int]:
|
|||
"skipped_unchanged": 0,
|
||||
"no_cad": 0,
|
||||
"no_geom": 0,
|
||||
"card_failed": 0, # каждая упавшая карточка (5xx / сеть / не-dict)
|
||||
"group_aborted": 0, # групп прервано circuit breaker'ом (битый сегмент портала)
|
||||
}
|
||||
client = _make_client()
|
||||
try:
|
||||
|
|
@ -331,7 +356,9 @@ def load_gisogd_permits(db: Session | None = None) -> dict[str, int]:
|
|||
continue
|
||||
counts["listed"] += len(items)
|
||||
_sync_group(db, client, schema, doc_group, items, existing, counts)
|
||||
db.commit()
|
||||
# Commit после КАЖДОЙ (schema, group), а не один раз в конце: партиал
|
||||
# должен пережить soft_time_limit-kill (иначе теряем весь прогон).
|
||||
db.commit()
|
||||
except Exception as exc:
|
||||
db.rollback()
|
||||
logger.exception("gisogd66: outer tx rolled back: %s", exc)
|
||||
|
|
@ -354,7 +381,15 @@ def _sync_group(
|
|||
existing: dict[str, date | None],
|
||||
counts: dict[str, int],
|
||||
) -> None:
|
||||
"""Обработать одну (schema, group): инкрементальная сверка + карточки + UPSERT."""
|
||||
"""Обработать одну (schema, group): инкрементальная сверка + карточки + UPSERT.
|
||||
|
||||
Периодический commit каждые ``_COMMIT_EVERY`` обработанных записей (партиал переживёт
|
||||
time-limit kill). Circuit breaker: ``_CARD_FAIL_BREAKER`` карточек ПОДРЯД с ошибкой
|
||||
fetch → прерываем группу (битый сегмент портала — остаток догоним будущими ранами).
|
||||
Любая успешная карточка сбрасывает счётчик подряд-фейлов.
|
||||
"""
|
||||
processed = 0 # успешно обработанных (upsert/skip) с последнего commit
|
||||
consecutive_fail = 0 # карточек подряд с ошибкой fetch (для circuit breaker)
|
||||
for item in items:
|
||||
key = item.get("key")
|
||||
if key is None:
|
||||
|
|
@ -365,17 +400,36 @@ def _sync_group(
|
|||
# Инкрементальность: пропускаем карточку, если key уже есть И date_reg совпал.
|
||||
if source_key in existing and existing[source_key] == list_reg:
|
||||
counts["skipped_unchanged"] += 1
|
||||
consecutive_fail = 0 # skip — не фейл карточки, сбрасываем breaker
|
||||
processed += 1
|
||||
if processed % _COMMIT_EVERY == 0:
|
||||
db.commit()
|
||||
continue
|
||||
|
||||
card: dict[str, Any] | None
|
||||
try:
|
||||
card = fetch_doc_card(client, schema, source_key)
|
||||
except Exception as exc:
|
||||
logger.warning("gisogd66: карточка %s/%s упала: %s", schema, source_key, exc)
|
||||
continue
|
||||
card = None
|
||||
finally:
|
||||
time.sleep(_CARD_SLEEP_S)
|
||||
if card is None:
|
||||
# Неуспех fetch (5xx / сеть / не-dict): считаем и двигаем circuit breaker.
|
||||
counts["card_failed"] += 1
|
||||
consecutive_fail += 1
|
||||
if consecutive_fail >= _CARD_FAIL_BREAKER:
|
||||
logger.error(
|
||||
"gisogd66: %s/%s: %d карточек подряд 5xx — прерываю группу "
|
||||
"(битый сегмент портала)",
|
||||
schema,
|
||||
doc_group,
|
||||
_CARD_FAIL_BREAKER,
|
||||
)
|
||||
counts["group_aborted"] += 1
|
||||
return
|
||||
continue
|
||||
consecutive_fail = 0 # успешная карточка сбрасывает breaker
|
||||
counts["fetched_details"] += 1
|
||||
|
||||
rec = _build_record(item, card, schema, doc_group, source_key, list_reg)
|
||||
|
|
@ -393,6 +447,9 @@ def _sync_group(
|
|||
logger.warning("gisogd66: upsert %s %r упал: %s", doc_group, rec["doc_num"], exc)
|
||||
continue
|
||||
counts[outcome] += 1
|
||||
processed += 1
|
||||
if processed % _COMMIT_EVERY == 0:
|
||||
db.commit()
|
||||
|
||||
|
||||
def _build_record(
|
||||
|
|
|
|||
|
|
@ -18,6 +18,8 @@ from contextlib import contextmanager
|
|||
from datetime import date
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
from app.services.scrapers import gisogd66 as g
|
||||
|
||||
# ── Реальные фрагменты API (разведка 2026-07-04) ──────────────────────────────
|
||||
|
|
@ -221,6 +223,7 @@ class _FakeSession:
|
|||
self._scalar = scalar_value
|
||||
self._select_rows = select_rows or []
|
||||
self.committed = False
|
||||
self.commit_count = 0
|
||||
|
||||
@contextmanager
|
||||
def begin_nested(self): # type: ignore[no-untyped-def]
|
||||
|
|
@ -235,6 +238,7 @@ class _FakeSession:
|
|||
|
||||
def commit(self) -> None:
|
||||
self.committed = True
|
||||
self.commit_count += 1
|
||||
|
||||
def rollback(self) -> None:
|
||||
pass
|
||||
|
|
@ -284,6 +288,23 @@ def test_upsert_permit_no_geom_passes_null() -> None:
|
|||
# ── Инкрементальность через _sync_group ───────────────────────────────────────
|
||||
|
||||
|
||||
_COUNT_KEYS = (
|
||||
"inserted",
|
||||
"updated",
|
||||
"skipped_unchanged",
|
||||
"fetched_details",
|
||||
"no_cad",
|
||||
"no_geom",
|
||||
"card_failed",
|
||||
"group_aborted",
|
||||
)
|
||||
|
||||
|
||||
def _fresh_counts() -> dict[str, int]:
|
||||
"""Свежий словарь счётчиков _sync_group (все ключи, что лоадер инкрементит)."""
|
||||
return dict.fromkeys(_COUNT_KEYS, 0)
|
||||
|
||||
|
||||
class _RecordingClient:
|
||||
"""Заглушка httpx-клиента: считает, сколько карточек запрошено (fetch_doc_card)."""
|
||||
|
||||
|
|
@ -302,9 +323,7 @@ def test_sync_group_incremental_skip(monkeypatch: Any) -> None:
|
|||
_patch_card_fetch(monkeypatch, fetched)
|
||||
db = _FakeSession(scalar_value=True)
|
||||
existing = {"1000210001947595": date(2026, 6, 19)} # тот же dateReg, что в списке
|
||||
counts = dict.fromkeys(
|
||||
("inserted", "updated", "skipped_unchanged", "fetched_details", "no_cad", "no_geom"), 0
|
||||
)
|
||||
counts = _fresh_counts()
|
||||
g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", [dict(_LIST_ITEM)], existing, counts) # type: ignore[arg-type]
|
||||
assert fetched == [] # карточку не тянули
|
||||
assert counts["skipped_unchanged"] == 1
|
||||
|
|
@ -317,9 +336,7 @@ def test_sync_group_changed_datereg_fetches(monkeypatch: Any) -> None:
|
|||
_patch_card_fetch(monkeypatch, fetched)
|
||||
db = _FakeSession(scalar_value=False) # update
|
||||
existing = {"1000210001947595": date(2026, 6, 10)} # старее списочного 06-19
|
||||
counts = dict.fromkeys(
|
||||
("inserted", "updated", "skipped_unchanged", "fetched_details", "no_cad", "no_geom"), 0
|
||||
)
|
||||
counts = _fresh_counts()
|
||||
g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", [dict(_LIST_ITEM)], existing, counts) # type: ignore[arg-type]
|
||||
assert fetched == ["1000210001947595"] # карточку тянули
|
||||
assert counts["fetched_details"] == 1
|
||||
|
|
@ -331,9 +348,7 @@ def test_sync_group_new_key_inserts_and_counts_geom(monkeypatch: Any) -> None:
|
|||
fetched: list[Any] = []
|
||||
_patch_card_fetch(monkeypatch, fetched)
|
||||
db = _FakeSession(scalar_value=True) # insert
|
||||
counts = dict.fromkeys(
|
||||
("inserted", "updated", "skipped_unchanged", "fetched_details", "no_cad", "no_geom"), 0
|
||||
)
|
||||
counts = _fresh_counts()
|
||||
g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", [dict(_LIST_ITEM)], {}, counts) # type: ignore[arg-type]
|
||||
assert counts["inserted"] == 1
|
||||
assert counts["fetched_details"] == 1
|
||||
|
|
@ -350,15 +365,142 @@ def test_sync_group_no_cad_no_geom_counts(monkeypatch: Any) -> None:
|
|||
monkeypatch.setattr(g, "fetch_doc_card", _fake)
|
||||
db = _FakeSession(scalar_value=True)
|
||||
item = {**_LIST_ITEM, "key": 42, "dateReg": "2026-06-01T10:00:00"}
|
||||
counts = dict.fromkeys(
|
||||
("inserted", "updated", "skipped_unchanged", "fetched_details", "no_cad", "no_geom"), 0
|
||||
)
|
||||
counts = _fresh_counts()
|
||||
g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RV", [item], {}, counts) # type: ignore[arg-type]
|
||||
assert counts["no_cad"] == 1
|
||||
assert counts["no_geom"] == 1
|
||||
assert counts["inserted"] == 1
|
||||
|
||||
|
||||
# ── Хотфикс #2370: 5xx-карточки без ретраев, circuit breaker, периодический commit ──
|
||||
|
||||
|
||||
class _FakeResponse:
|
||||
"""Мини-заглушка httpx.Response: только status_code + json/raise_for_status/request."""
|
||||
|
||||
def __init__(self, status_code: int, payload: Any = None) -> None:
|
||||
self.status_code = status_code
|
||||
self._payload = payload
|
||||
self.request = httpx.Request("GET", "https://example/x")
|
||||
|
||||
def json(self) -> Any:
|
||||
return self._payload
|
||||
|
||||
def raise_for_status(self) -> None:
|
||||
if self.status_code >= 400:
|
||||
raise httpx.HTTPStatusError(
|
||||
f"{self.status_code}",
|
||||
request=self.request,
|
||||
response=None, # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
|
||||
class _CountingClient:
|
||||
"""Заглушка httpx.Client: считает GET-и, всегда отдаёт заданный status_code."""
|
||||
|
||||
def __init__(self, status_code: int, payload: Any = None) -> None:
|
||||
self.status_code = status_code
|
||||
self.payload = payload
|
||||
self.get_calls: list[str] = []
|
||||
|
||||
def get(self, path: str) -> _FakeResponse:
|
||||
self.get_calls.append(path)
|
||||
return _FakeResponse(self.status_code, self.payload)
|
||||
|
||||
|
||||
def test_fetch_doc_card_500_single_attempt(monkeypatch: Any) -> None:
|
||||
"""Хотфикс: карточка отдаёт 500 → ровно 1 HTTP-попытка (retries_5xx=0), затем raise.
|
||||
|
||||
Битый сегмент старых ключей портала детерминированно 5xx — ретраи = ~12с/карточку
|
||||
впустую. Убираем sleep, чтоб тест был мгновенным.
|
||||
"""
|
||||
monkeypatch.setattr(g.time, "sleep", lambda _s: None)
|
||||
client = _CountingClient(status_code=500)
|
||||
raised = False
|
||||
try:
|
||||
g.fetch_doc_card(client, "agate_ekbgo", "100001007000123") # type: ignore[arg-type]
|
||||
except httpx.HTTPStatusError:
|
||||
raised = True
|
||||
assert raised is True
|
||||
assert len(client.get_calls) == 1 # НИ одного ретрая на 5xx карточки
|
||||
|
||||
|
||||
def test_fetch_group_list_500_retries(monkeypatch: Any) -> None:
|
||||
"""Список группы 5xx ретраится с дефолтом (1 + _MAX_RETRIES попыток) — контраст с карточкой."""
|
||||
monkeypatch.setattr(g.time, "sleep", lambda _s: None)
|
||||
client = _CountingClient(status_code=500)
|
||||
raised = False
|
||||
try:
|
||||
g.fetch_group_list(client, "agate_ekbgo", "DocRS") # type: ignore[arg-type]
|
||||
except httpx.HTTPStatusError:
|
||||
raised = True
|
||||
assert raised is True
|
||||
assert len(client.get_calls) == 1 + g._MAX_RETRIES # 3 попытки
|
||||
|
||||
|
||||
def test_sync_group_circuit_breaker_aborts(monkeypatch: Any) -> None:
|
||||
"""50 карточек ПОДРЯД с 5xx → группа прервана: group_aborted=1, остаток НЕ запрошен."""
|
||||
monkeypatch.setattr(g.time, "sleep", lambda _s: None)
|
||||
requested: list[Any] = []
|
||||
|
||||
def _always_fail(client: Any, schema: str, key: Any) -> dict[str, Any] | None:
|
||||
requested.append(key)
|
||||
raise httpx.HTTPStatusError("500", request=httpx.Request("GET", "x"), response=None) # type: ignore[arg-type]
|
||||
|
||||
monkeypatch.setattr(g, "fetch_doc_card", _always_fail)
|
||||
db = _FakeSession(scalar_value=True)
|
||||
# 60 новых ключей — breaker (50) должен сработать раньше, чем дойдём до 60-го.
|
||||
items = [{**_LIST_ITEM, "key": 100001007000000 + i} for i in range(60)]
|
||||
counts = _fresh_counts()
|
||||
g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", items, {}, counts) # type: ignore[arg-type]
|
||||
assert counts["group_aborted"] == 1
|
||||
assert counts["card_failed"] == g._CARD_FAIL_BREAKER # ровно 50 фейлов до аборта
|
||||
assert len(requested) == g._CARD_FAIL_BREAKER # 51-й ключ НЕ запрашивался
|
||||
assert counts["inserted"] == 0
|
||||
|
||||
|
||||
def test_sync_group_success_resets_breaker(monkeypatch: Any) -> None:
|
||||
"""Успешная карточка между фейлами сбрасывает breaker → группа НЕ прерывается.
|
||||
|
||||
49 фейлов, потом успех (reset), потом ещё 49 фейлов — суммарно 98 фейлов, но НИ разу
|
||||
50 подряд → group_aborted=0, вся группа обработана.
|
||||
"""
|
||||
monkeypatch.setattr(g.time, "sleep", lambda _s: None)
|
||||
# Индексы 49 и 99 (0-based) — успех; остальные — фейл. Между ними ≤49 подряд.
|
||||
good_idx = {49, 99}
|
||||
seq: list[int] = []
|
||||
|
||||
def _mostly_fail(client: Any, schema: str, key: Any) -> dict[str, Any] | None:
|
||||
idx = len(seq)
|
||||
seq.append(idx)
|
||||
if idx in good_idx:
|
||||
return dict(_CARD)
|
||||
raise httpx.HTTPStatusError("500", request=httpx.Request("GET", "x"), response=None) # type: ignore[arg-type]
|
||||
|
||||
monkeypatch.setattr(g, "fetch_doc_card", _mostly_fail)
|
||||
db = _FakeSession(scalar_value=True)
|
||||
items = [{**_LIST_ITEM, "key": 200000000 + i} for i in range(100)]
|
||||
counts = _fresh_counts()
|
||||
g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", items, {}, counts) # type: ignore[arg-type]
|
||||
assert counts["group_aborted"] == 0 # breaker сбрасывался — не прервались
|
||||
assert counts["card_failed"] == 98 # 100 − 2 успешных
|
||||
assert counts["fetched_details"] == 2 # 2 успешные карточки
|
||||
assert len(seq) == 100 # прошли ВСЮ группу
|
||||
|
||||
|
||||
def test_sync_group_periodic_commit(monkeypatch: Any) -> None:
|
||||
""">200 обработанных записей → db.commit() вызывается периодически (каждые 200)."""
|
||||
monkeypatch.setattr(g.time, "sleep", lambda _s: None)
|
||||
_patch_card_fetch(monkeypatch, [])
|
||||
db = _FakeSession(scalar_value=True) # каждый upsert → inserted
|
||||
# 450 новых ключей → 2 периодических commit (на 200-й и 400-й записи).
|
||||
items = [{**_LIST_ITEM, "key": 300000000 + i} for i in range(450)]
|
||||
counts = _fresh_counts()
|
||||
g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", items, {}, counts) # type: ignore[arg-type]
|
||||
assert counts["inserted"] == 450
|
||||
assert db.commit_count == 2 # 450 // 200 = 2 периодических commit
|
||||
|
||||
|
||||
# ── Конфликт двух схем: поздний date_reg параметризуется в UPSERT ──────────────
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue