From a5ff281badd1c4e2219cd278bd319a15871f35bd Mon Sep 17 00:00:00 2001 From: Light1YT Date: Sat, 13 Jun 2026 13:12:30 +0500 Subject: [PATCH] =?UTF-8?q?fix(admin-jobs):=20UPSERT=20=D0=B2=20PUT=20/adm?= =?UTF-8?q?in/jobs/settings=20=E2=80=94=20=D0=BD=D0=B5=20=D1=82=D0=B5?= =?UTF-8?q?=D1=80=D1=8F=D1=82=D1=8C=20=D0=B8=D0=B7=D0=BC=D0=B5=D0=BD=D0=B5?= =?UTF-8?q?=D0=BD=D0=B8=D1=8F=20(#1223)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- backend/app/services/job_settings.py | 99 ++++++++-- .../tests/api/v1/test_admin_jobs_settings.py | 136 +++++++++++++ backend/tests/services/test_job_settings.py | 186 ++++++++++++++++++ 3 files changed, 409 insertions(+), 12 deletions(-) create mode 100644 backend/tests/api/v1/test_admin_jobs_settings.py create mode 100644 backend/tests/services/test_job_settings.py diff --git a/backend/app/services/job_settings.py b/backend/app/services/job_settings.py index 1be15f7a..4d12d32d 100644 --- a/backend/app/services/job_settings.py +++ b/backend/app/services/job_settings.py @@ -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) diff --git a/backend/tests/api/v1/test_admin_jobs_settings.py b/backend/tests/api/v1/test_admin_jobs_settings.py new file mode 100644 index 00000000..c409c322 --- /dev/null +++ b/backend/tests/api/v1/test_admin_jobs_settings.py @@ -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() diff --git a/backend/tests/services/test_job_settings.py b/backend/tests/services/test_job_settings.py new file mode 100644 index 00000000..b2e0a311 --- /dev/null +++ b/backend/tests/services/test_job_settings.py @@ -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