fix(admin-jobs): UPSERT в PUT /admin/jobs/settings — не терять изменения (#1223)
UPDATE без проверки rowcount затрагивал 0 строк для несуществующего job_type; get_one возвращал hardcoded _fallback. PUT отвечал 200, админ считал настройку сохранённой — но она терялась (новый job_type до seed-миграции, опечатка в path). Patch: update() проверяет result.rowcount: при 0 делает INSERT с переданными колонками (непереданные → дефолты таблицы из м.81). INSERT использует ON CONFLICT (job_type) DO NOTHING + повторный UPDATE как защита от гонки. Поведение для существующих строк не изменилось. 8 новых тестов (6 service + 2 API). Closes #1223
This commit is contained in:
parent
f5e7fcdad9
commit
a5ff281bad
3 changed files with 409 additions and 12 deletions
|
|
@ -192,10 +192,30 @@ def update(
|
|||
extra_config: dict | None = None,
|
||||
updated_by: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""PATCH-style update — обновляет только переданные поля.
|
||||
"""PATCH-style upsert — обновляет только переданные поля.
|
||||
|
||||
Если строки с таким job_type ещё нет, она создаётся (с дефолтами таблицы из
|
||||
data/sql/81_job_settings.sql для непереданных колонок). Раньше UPDATE без
|
||||
проверки rowcount молча терял изменения для несуществующих job_type, а PUT
|
||||
отвечал 200 c fallback-дефолтами (issue #1223).
|
||||
|
||||
cron_schedule='' → NULL (manual-only). updated_at выставляется в NOW().
|
||||
"""
|
||||
import json
|
||||
|
||||
# Нормализация значений по той же логике, что и в исходном UPDATE.
|
||||
# cron_schedule: None = не трогать; '' = явный NULL (manual-only)
|
||||
cron_provided = cron_schedule is not None
|
||||
norm_cron: str | None = cron_schedule if cron_schedule else None
|
||||
|
||||
# rate_ms: None = не трогать; 0 = явно обнулить (no throttle)
|
||||
rate_provided = rate_ms is not None
|
||||
norm_rate: int | None = rate_ms if rate_provided and rate_ms and rate_ms > 0 else None
|
||||
|
||||
extra_provided = extra_config is not None
|
||||
norm_extra_json = json.dumps(extra_config, ensure_ascii=False) if extra_provided else None
|
||||
|
||||
# ── UPDATE для существующей строки (PATCH-семантика без изменений) ───────
|
||||
sets: list[str] = ["updated_at = NOW()"]
|
||||
params: dict[str, Any] = {"jt": job_type}
|
||||
|
||||
|
|
@ -210,26 +230,81 @@ def update(
|
|||
sets.append(f"{field} = :{field}")
|
||||
params[field] = value
|
||||
|
||||
# cron_schedule: None = не трогать; '' = явный NULL (manual-only)
|
||||
if cron_schedule is not None:
|
||||
if cron_provided:
|
||||
sets.append("cron_schedule = :cron_schedule")
|
||||
params["cron_schedule"] = cron_schedule if cron_schedule else None
|
||||
params["cron_schedule"] = norm_cron
|
||||
|
||||
# rate_ms: None = не трогать; 0 = явно обнулить (no throttle)
|
||||
if rate_ms is not None:
|
||||
if rate_provided:
|
||||
sets.append("rate_ms = :rate_ms")
|
||||
params["rate_ms"] = rate_ms if rate_ms > 0 else None
|
||||
params["rate_ms"] = norm_rate
|
||||
|
||||
if extra_config is not None:
|
||||
if extra_provided:
|
||||
sets.append("extra_config = CAST(:extra_config AS jsonb)")
|
||||
import json
|
||||
params["extra_config"] = norm_extra_json
|
||||
|
||||
params["extra_config"] = json.dumps(extra_config, ensure_ascii=False)
|
||||
|
||||
db.execute(
|
||||
result = db.execute(
|
||||
text(f"UPDATE job_settings SET {', '.join(sets)} WHERE job_type = :jt"),
|
||||
params,
|
||||
)
|
||||
|
||||
# Если строки не было — INSERT с явными значениями только для переданных колонок.
|
||||
# Остальные колонки получат defaults таблицы (enabled=true, queue_name='celery',
|
||||
# max_retries=2, max_concurrency=1, extra_config='{}'::jsonb, updated_at=now()).
|
||||
if (result.rowcount or 0) == 0:
|
||||
insert_cols: list[str] = ["job_type"]
|
||||
insert_vals: list[str] = [":jt"]
|
||||
insert_params: dict[str, Any] = {"jt": job_type}
|
||||
|
||||
for field, value in [
|
||||
("enabled", enabled),
|
||||
("queue_name", queue_name),
|
||||
("max_retries", max_retries),
|
||||
("max_concurrency", max_concurrency),
|
||||
("updated_by", updated_by),
|
||||
]:
|
||||
if value is not None:
|
||||
insert_cols.append(field)
|
||||
insert_vals.append(f":{field}")
|
||||
insert_params[field] = value
|
||||
|
||||
if cron_provided:
|
||||
insert_cols.append("cron_schedule")
|
||||
insert_vals.append(":cron_schedule")
|
||||
insert_params["cron_schedule"] = norm_cron
|
||||
|
||||
if rate_provided:
|
||||
insert_cols.append("rate_ms")
|
||||
insert_vals.append(":rate_ms")
|
||||
insert_params["rate_ms"] = norm_rate
|
||||
|
||||
if extra_provided:
|
||||
insert_cols.append("extra_config")
|
||||
insert_vals.append("CAST(:extra_config AS jsonb)")
|
||||
insert_params["extra_config"] = norm_extra_json
|
||||
|
||||
# ON CONFLICT DO NOTHING — защита от гонки: между нашими UPDATE и INSERT
|
||||
# параллельный PUT мог вставить строку. В этом случае повторим UPDATE,
|
||||
# чтобы наши изменения не потерялись.
|
||||
insert_result = db.execute(
|
||||
text(
|
||||
f"INSERT INTO job_settings ({', '.join(insert_cols)}) "
|
||||
f"VALUES ({', '.join(insert_vals)}) "
|
||||
"ON CONFLICT (job_type) DO NOTHING"
|
||||
),
|
||||
insert_params,
|
||||
)
|
||||
if (insert_result.rowcount or 0) == 0:
|
||||
logger.info(
|
||||
"job_settings.update: гонка по '%s' — повторный UPDATE после INSERT race",
|
||||
job_type,
|
||||
)
|
||||
db.execute(
|
||||
text(f"UPDATE job_settings SET {', '.join(sets)} WHERE job_type = :jt"),
|
||||
params,
|
||||
)
|
||||
else:
|
||||
logger.info("job_settings.update: вставлена новая строка '%s'", job_type)
|
||||
|
||||
db.commit()
|
||||
return get_one(job_type, db)
|
||||
|
||||
|
|
|
|||
136
backend/tests/api/v1/test_admin_jobs_settings.py
Normal file
136
backend/tests/api/v1/test_admin_jobs_settings.py
Normal file
|
|
@ -0,0 +1,136 @@
|
|||
"""API-тесты для PUT /api/v1/admin/jobs/settings/{job_type}.
|
||||
|
||||
Регрессия по issue #1223: PUT для отсутствующей строки молча терял изменения
|
||||
и отвечал 200 с fallback-дефолтами. После фикса update() выполняет INSERT,
|
||||
если UPDATE затронул 0 строк, и возвращает новую строку с сохранёнными полями.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import app
|
||||
|
||||
|
||||
def _make_db_recorder(update_rowcounts: list[int]) -> tuple[MagicMock, list[tuple[str, Any]]]:
|
||||
"""Db-mock, который:
|
||||
- возвращает заданные rowcount для UPDATE/INSERT (по списку),
|
||||
- имитирует поведение get_one(): SELECT возвращает строку с теми же
|
||||
значениями, что были последним INSERT (или None, если не вставлялась).
|
||||
"""
|
||||
db = MagicMock()
|
||||
calls: list[tuple[str, Any]] = []
|
||||
state: dict[str, Any] = {"inserted": None}
|
||||
plan = list(update_rowcounts)
|
||||
|
||||
def execute(stmt: Any, params: Any = None) -> Any:
|
||||
sql = str(stmt)
|
||||
calls.append((sql, params))
|
||||
result = MagicMock()
|
||||
if "UPDATE job_settings" in sql:
|
||||
result.rowcount = plan.pop(0) if plan else 0
|
||||
elif "INSERT INTO job_settings" in sql:
|
||||
result.rowcount = plan.pop(0) if plan else 0
|
||||
if result.rowcount > 0 and isinstance(params, dict):
|
||||
state["inserted"] = params
|
||||
elif "SELECT" in sql and "job_settings" in sql:
|
||||
# Имитация get_one после успешного INSERT
|
||||
row = state["inserted"]
|
||||
if row is None:
|
||||
result.mappings.return_value.first.return_value = None
|
||||
else:
|
||||
import datetime
|
||||
|
||||
result.mappings.return_value.first.return_value = {
|
||||
"job_type": row.get("jt"),
|
||||
"enabled": row.get("enabled", True),
|
||||
"queue_name": row.get("queue_name", "celery"),
|
||||
"cron_schedule": row.get("cron_schedule"),
|
||||
"rate_ms": row.get("rate_ms"),
|
||||
"max_retries": row.get("max_retries", 2),
|
||||
"max_concurrency": row.get("max_concurrency", 1),
|
||||
"extra_config": {}, # JSON-строка сериализована — для теста плоский dict ok
|
||||
"updated_at": datetime.datetime(2026, 6, 13, tzinfo=datetime.UTC),
|
||||
"updated_by": row.get("updated_by"),
|
||||
"description": None,
|
||||
}
|
||||
return result
|
||||
|
||||
db.execute = execute
|
||||
db.commit = MagicMock()
|
||||
return db, calls
|
||||
|
||||
|
||||
def _override_db(db: MagicMock) -> Any:
|
||||
def _factory():
|
||||
yield db
|
||||
|
||||
return _factory
|
||||
|
||||
|
||||
def test_put_job_settings_existing_row_updates_only() -> None:
|
||||
"""Существующая строка — UPDATE, без INSERT."""
|
||||
from app.core.db import get_db
|
||||
|
||||
db, calls = _make_db_recorder(update_rowcounts=[1])
|
||||
app.dependency_overrides[get_db] = _override_db(db)
|
||||
|
||||
try:
|
||||
client = TestClient(app)
|
||||
r = client.put(
|
||||
"/api/v1/admin/jobs/settings/scrape_kn",
|
||||
json={"rate_ms": 750},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert any("UPDATE job_settings" in sql for sql, _ in calls)
|
||||
assert not any("INSERT INTO job_settings" in sql for sql, _ in calls)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
def test_put_job_settings_missing_row_inserts_and_returns_persisted_values() -> None:
|
||||
"""Issue #1223: PUT для отсутствующей строки должен сохранить изменения.
|
||||
|
||||
После фикса: UPDATE → 0 строк, затем INSERT с переданными полями.
|
||||
Ответ должен содержать **переданные** значения, а не fallback-дефолты.
|
||||
"""
|
||||
from app.core.db import get_db
|
||||
|
||||
# UPDATE 0, INSERT 1 — типовой happy-path для несуществующей строки.
|
||||
db, calls = _make_db_recorder(update_rowcounts=[0, 1])
|
||||
app.dependency_overrides[get_db] = _override_db(db)
|
||||
|
||||
try:
|
||||
client = TestClient(app)
|
||||
r = client.put(
|
||||
"/api/v1/admin/jobs/settings/brand_new_job",
|
||||
json={
|
||||
"enabled": False,
|
||||
"rate_ms": 1234,
|
||||
"max_retries": 5,
|
||||
},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
body = r.json()
|
||||
# Ключевая проверка: ответ — НЕ fallback (_fallback ставит enabled=True).
|
||||
# Значит INSERT действительно произошёл, get_one вернул вставленную строку.
|
||||
assert body["job_type"] == "brand_new_job"
|
||||
assert body["enabled"] is False
|
||||
assert body["rate_ms"] == 1234
|
||||
assert body["max_retries"] == 5
|
||||
|
||||
# Подтверждаем, что INSERT реально был.
|
||||
assert any("INSERT INTO job_settings" in sql for sql, _ in calls)
|
||||
insert_call = next(
|
||||
(sql, params) for sql, params in calls if "INSERT INTO job_settings" in sql
|
||||
)
|
||||
assert "ON CONFLICT (job_type) DO NOTHING" in insert_call[0]
|
||||
assert insert_call[1]["jt"] == "brand_new_job"
|
||||
assert insert_call[1]["enabled"] is False
|
||||
assert insert_call[1]["rate_ms"] == 1234
|
||||
assert insert_call[1]["updated_by"] == "admin-ui"
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
186
backend/tests/services/test_job_settings.py
Normal file
186
backend/tests/services/test_job_settings.py
Normal file
|
|
@ -0,0 +1,186 @@
|
|||
"""Unit-тесты для app.services.job_settings.update (UPSERT поведение).
|
||||
|
||||
Покрывают issue #1223: PUT /admin/jobs/settings/{job_type} для несуществующей
|
||||
строки молча терял изменения. После фикса update() выполняет INSERT, если
|
||||
UPDATE затронул 0 строк.
|
||||
|
||||
DB здесь полностью замокана — мы проверяем, что service:
|
||||
1. Делает UPDATE сначала и фиксирует rowcount.
|
||||
2. При rowcount == 0 делает INSERT с переданными колонками.
|
||||
3. При rowcount > 0 НЕ делает INSERT (PATCH-семантика сохранена).
|
||||
4. Защищён от гонки UPDATE→INSERT через ON CONFLICT DO NOTHING.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from app.services import job_settings as svc
|
||||
|
||||
|
||||
class _ExecRecorder:
|
||||
"""Имитация db.execute() с задаваемым rowcount и записью всех вызовов."""
|
||||
|
||||
def __init__(self, plan: list[int]) -> None:
|
||||
# plan: список rowcount-ов для последовательных execute() вызовов.
|
||||
# get_one() в конце update() тоже бьёт execute() — этот вызов покрывается
|
||||
# последним элементом или дефолтным MagicMock (см. __call__).
|
||||
self._plan = list(plan)
|
||||
self.calls: list[tuple[str, dict[str, Any]]] = []
|
||||
|
||||
def __call__(self, stmt: Any, params: Any = None) -> Any:
|
||||
sql = str(stmt)
|
||||
self.calls.append((sql, dict(params) if isinstance(params, dict) else {}))
|
||||
result = MagicMock()
|
||||
# rowcount только для DML; для SELECT/get_one — не важен.
|
||||
result.rowcount = self._plan.pop(0) if self._plan else 0
|
||||
# get_one() ожидает .mappings().first() → None (→ fallback).
|
||||
result.mappings.return_value.first.return_value = None
|
||||
return result
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def _patched_get_one(monkeypatch: pytest.MonkeyPatch) -> MagicMock:
|
||||
"""get_one() имеет свою БД-логику; патчим, чтобы не мешала assert-ам на execute()."""
|
||||
mock = MagicMock(return_value={"job_type": "x", "stub": True})
|
||||
monkeypatch.setattr(svc, "get_one", mock)
|
||||
return mock
|
||||
|
||||
|
||||
def _count_statements(calls: list[tuple[str, dict[str, Any]]], keyword: str) -> int:
|
||||
return sum(1 for sql, _ in calls if keyword in sql)
|
||||
|
||||
|
||||
def test_update_existing_row_only_updates(_patched_get_one: MagicMock) -> None:
|
||||
"""Когда UPDATE затронул 1 строку — INSERT НЕ выполняется (regression guard)."""
|
||||
db = MagicMock()
|
||||
exec_rec = _ExecRecorder(plan=[1]) # UPDATE → 1 row affected
|
||||
db.execute = exec_rec
|
||||
|
||||
svc.update("scrape_kn", db, rate_ms=500, updated_by="admin-ui")
|
||||
|
||||
assert _count_statements(exec_rec.calls, "UPDATE job_settings") == 1
|
||||
assert _count_statements(exec_rec.calls, "INSERT INTO job_settings") == 0
|
||||
db.commit.assert_called_once()
|
||||
|
||||
|
||||
def test_update_missing_row_inserts_with_provided_fields(
|
||||
_patched_get_one: MagicMock,
|
||||
) -> None:
|
||||
"""Issue #1223: PUT для несуществующей строки должен сохранять изменения.
|
||||
|
||||
UPDATE возвращает rowcount=0, после чего service делает INSERT с теми же
|
||||
колонками, что были в SET-списке.
|
||||
"""
|
||||
db = MagicMock()
|
||||
exec_rec = _ExecRecorder(plan=[0, 1]) # UPDATE 0, INSERT 1
|
||||
db.execute = exec_rec
|
||||
|
||||
svc.update(
|
||||
"new_job_type",
|
||||
db,
|
||||
enabled=False,
|
||||
queue_name="celery",
|
||||
cron_schedule="*/5 * * * *",
|
||||
rate_ms=1000,
|
||||
max_retries=3,
|
||||
max_concurrency=2,
|
||||
extra_config={"key": "value"},
|
||||
updated_by="admin-ui",
|
||||
)
|
||||
|
||||
# Должен быть ровно один UPDATE и ровно один INSERT.
|
||||
assert _count_statements(exec_rec.calls, "UPDATE job_settings") == 1
|
||||
assert _count_statements(exec_rec.calls, "INSERT INTO job_settings") == 1
|
||||
|
||||
insert_sql, insert_params = next(
|
||||
(sql, params) for sql, params in exec_rec.calls if "INSERT" in sql
|
||||
)
|
||||
assert "ON CONFLICT (job_type) DO NOTHING" in insert_sql
|
||||
assert insert_params["jt"] == "new_job_type"
|
||||
assert insert_params["enabled"] is False
|
||||
assert insert_params["queue_name"] == "celery"
|
||||
assert insert_params["cron_schedule"] == "*/5 * * * *"
|
||||
assert insert_params["rate_ms"] == 1000
|
||||
assert insert_params["max_retries"] == 3
|
||||
assert insert_params["max_concurrency"] == 2
|
||||
assert insert_params["updated_by"] == "admin-ui"
|
||||
# extra_config сериализуется в JSON-строку — приводится через CAST.
|
||||
assert "CAST(:extra_config AS jsonb)" in insert_sql
|
||||
assert '"key": "value"' in insert_params["extra_config"]
|
||||
|
||||
db.commit.assert_called_once()
|
||||
|
||||
|
||||
def test_update_missing_row_minimal_payload(_patched_get_one: MagicMock) -> None:
|
||||
"""INSERT для пустого payload (только updated_by из роута) — лишь job_type/updated_by.
|
||||
|
||||
Остальные колонки должны получить таблицы-дефолты — мы их не упоминаем в
|
||||
INSERT, проверяем что они НЕ попали в params.
|
||||
"""
|
||||
db = MagicMock()
|
||||
exec_rec = _ExecRecorder(plan=[0, 1])
|
||||
db.execute = exec_rec
|
||||
|
||||
svc.update("brand_new", db, updated_by="admin-ui")
|
||||
|
||||
insert_sql, insert_params = next(
|
||||
(sql, params) for sql, params in exec_rec.calls if "INSERT" in sql
|
||||
)
|
||||
assert insert_params == {"jt": "brand_new", "updated_by": "admin-ui"}
|
||||
# В INSERT не должно быть колонок, которые мы не передали.
|
||||
assert "enabled" not in insert_sql
|
||||
assert "queue_name" not in insert_sql
|
||||
assert "extra_config" not in insert_sql
|
||||
|
||||
|
||||
def test_update_insert_race_falls_back_to_second_update(
|
||||
_patched_get_one: MagicMock,
|
||||
) -> None:
|
||||
"""Гонка UPDATE→INSERT: параллельный PUT вставил строку между нашими запросами.
|
||||
|
||||
INSERT ... ON CONFLICT DO NOTHING вернёт rowcount=0; service должен повторить
|
||||
UPDATE, чтобы наши изменения не потерялись.
|
||||
"""
|
||||
db = MagicMock()
|
||||
# UPDATE→0, INSERT→0 (конфликт), second UPDATE→1.
|
||||
exec_rec = _ExecRecorder(plan=[0, 0, 1])
|
||||
db.execute = exec_rec
|
||||
|
||||
svc.update("racing_jt", db, enabled=True, updated_by="admin-ui")
|
||||
|
||||
assert _count_statements(exec_rec.calls, "UPDATE job_settings") == 2
|
||||
assert _count_statements(exec_rec.calls, "INSERT INTO job_settings") == 1
|
||||
db.commit.assert_called_once()
|
||||
|
||||
|
||||
def test_update_cron_empty_string_becomes_null(_patched_get_one: MagicMock) -> None:
|
||||
"""cron_schedule='' → NULL и при UPDATE, и при INSERT (manual-only режим)."""
|
||||
db = MagicMock()
|
||||
# UPDATE→0 → попадаем в INSERT-ветку.
|
||||
exec_rec = _ExecRecorder(plan=[0, 1])
|
||||
db.execute = exec_rec
|
||||
|
||||
svc.update("nspd_geo", db, cron_schedule="", updated_by="admin-ui")
|
||||
|
||||
update_params = exec_rec.calls[0][1]
|
||||
insert_params = exec_rec.calls[1][1]
|
||||
assert update_params["cron_schedule"] is None
|
||||
assert insert_params["cron_schedule"] is None
|
||||
|
||||
|
||||
def test_update_rate_ms_zero_becomes_null(_patched_get_one: MagicMock) -> None:
|
||||
"""rate_ms=0 → NULL (no throttle)."""
|
||||
db = MagicMock()
|
||||
exec_rec = _ExecRecorder(plan=[0, 1])
|
||||
db.execute = exec_rec
|
||||
|
||||
svc.update("nspd_geo", db, rate_ms=0, updated_by="admin-ui")
|
||||
|
||||
update_params = exec_rec.calls[0][1]
|
||||
insert_params = exec_rec.calls[1][1]
|
||||
assert update_params["rate_ms"] is None
|
||||
assert insert_params["rate_ms"] is None
|
||||
Loading…
Add table
Reference in a new issue