Light1YT
e9585bcd3b
Suite was uncollectable for a while → code evolved, mocks didn't (42 failed / 5 errors → 1687 passed / 0 failed / 0 errors). TEST-ONLY, no app code changed, no real bugs found (code-reviewed ✅ — no test weakened to fake-green). - best_layouts: sum_deals→deals_window rename; velocity-scaling direction for new SF-01 divisors (last_month=1.0/last_year=12.0) - quarter_dump_lookup: 7→10-tuple mock rows (risks/opportunity/red_lines cols); fix early-exit call-counts for the cad_zouit fallback (#232) - cadastre_bulk: add xmin/ymin/xmax/ymax to harvest-quarter db mock - nspd_client: add required QuarterDump.opportunity - TopLayoutRow: add required is_oversold - analyze_{market_price,recent_permits,inline_weights}: SQL-signature dispatch instead of fragile positional db.execute indices (SF-B5 query reorder; also fixes a double-POST geom-starvation false-negative) - custom_pois: move centroid >15km so center_bonus=0 isolates custom-POI delta - layout/report PDF: runtime WeasyPrint probe (skip macOS dev, RUN on CI) - mv_layout SQL: normalize +psycopg DSN + connectivity-probe skipif - admin token tests: skip (X-Admin-Token gate removed #437/#426; protection is Caddy basic_auth + RBAC, covered by test_rbac) Refs #944.
315 lines
11 KiB
Python
315 lines
11 KiB
Python
"""Тесты для admin_cadastre.py API endpoints.
|
||
|
||
Использует FastAPI TestClient с mock DB и mock Celery tasks.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
# scrape_cadastre зависит от app.scrapers.nspd_bulk_client (PR 2/5).
|
||
# Используем importlib.util.find_spec вместо прямого import — иначе
|
||
# `import app.workers...` пересоздаёт `app` как Python package и
|
||
# перебивает FastAPI instance, привязанный в строке выше → AttributeError
|
||
# на app.dependency_overrides.
|
||
import importlib.util
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.main import app
|
||
|
||
_SCRAPE_CADASTRE_AVAILABLE = (
|
||
importlib.util.find_spec("app.workers.tasks.scrape_cadastre") is not None
|
||
)
|
||
|
||
requires_scrape_cadastre = pytest.mark.skipif(
|
||
not _SCRAPE_CADASTRE_AVAILABLE,
|
||
reason=(
|
||
"app.scrapers.nspd_bulk_client отсутствует — нужно смержить main (PR #170). "
|
||
"Команда: git rebase origin/main"
|
||
),
|
||
)
|
||
|
||
# Токен для тестов (не реальный)
|
||
ADMIN_TOKEN = "test-admin-token"
|
||
ADMIN_HEADERS = {"X-Admin-Token": ADMIN_TOKEN}
|
||
|
||
|
||
def _make_mock_db(quarters: list[str] | None = None, job_row: dict[str, Any] | None = None):
|
||
"""Создать mock db session с преднастроенными ответами."""
|
||
db = MagicMock()
|
||
|
||
# Результат для _resolve_quarters (SELECT cad_number FROM cad_quarters_geom)
|
||
quarter_rows = [(q,) for q in (quarters or ["66:41:0303161", "66:41:0303162"])]
|
||
|
||
# Mock execute chain для разных запросов
|
||
def mock_execute(stmt, params=None):
|
||
stmt_str = str(stmt)
|
||
result = MagicMock()
|
||
|
||
if "cad_quarters_geom" in stmt_str and "cad_number" in stmt_str:
|
||
# _resolve_quarters query
|
||
result.all.return_value = quarter_rows
|
||
elif "cadastre_jobs" in stmt_str and "RETURNING job_id" in stmt_str:
|
||
# _create_job INSERT
|
||
result.scalar_one.return_value = 42
|
||
elif "cadastre_jobs" in stmt_str and "SELECT" in stmt_str:
|
||
# get/list jobs
|
||
result.mappings.return_value.all.return_value = [job_row] if job_row else []
|
||
result.mappings.return_value.first.return_value = job_row
|
||
elif "RETURNING job_id" in stmt_str:
|
||
# cancel/resume
|
||
result.first.return_value = (42,)
|
||
else:
|
||
result.all.return_value = []
|
||
result.mappings.return_value.all.return_value = []
|
||
result.mappings.return_value.first.return_value = None
|
||
result.scalar_one.return_value = 1
|
||
|
||
return result
|
||
|
||
db.execute = mock_execute
|
||
db.commit = MagicMock()
|
||
return db
|
||
|
||
|
||
def _mock_get_db(db):
|
||
"""FastAPI dependency override factory."""
|
||
|
||
def _get_db_override():
|
||
yield db
|
||
|
||
return _get_db_override
|
||
|
||
|
||
def _make_sample_job_row() -> dict[str, Any]:
|
||
import datetime
|
||
|
||
return {
|
||
"job_id": 42,
|
||
"name": "cadastre_pilot",
|
||
"job_kind": "pilot",
|
||
"scope": "pilot",
|
||
"status": "queued",
|
||
"triggered_by": "manual",
|
||
"rate_ms": 333,
|
||
"created_at": datetime.datetime(2026, 5, 15, 10, 0, 0),
|
||
"started_at": None,
|
||
"finished_at": None,
|
||
"heartbeat_at": None,
|
||
"targets_total": 50,
|
||
"targets_done": 0,
|
||
"targets_failed": 0,
|
||
"targets_skipped": 0,
|
||
"requests_count": 0,
|
||
"waf_blocked_count": 0,
|
||
"error": None,
|
||
"phase_state": None,
|
||
"raw_targets": {"quarters": ["66:41:0303161"]},
|
||
}
|
||
|
||
|
||
# ── Tests ────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@requires_scrape_cadastre
|
||
@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
|
||
def test_create_job_pilot_returns_job_id() -> None:
|
||
"""POST /cadastre/jobs scope=pilot → job_id + targets_total."""
|
||
quarters_50 = [f"66:41:{i:07d}" for i in range(50)]
|
||
db = _make_mock_db(quarters=quarters_50)
|
||
|
||
from app.core.db import get_db
|
||
|
||
app.dependency_overrides[get_db] = _mock_get_db(db)
|
||
|
||
try:
|
||
with patch("app.workers.tasks.scrape_cadastre.enqueue_cadastre_harvest") as mock_task:
|
||
mock_task.apply_async = MagicMock()
|
||
client = TestClient(app)
|
||
response = client.post(
|
||
"/api/v1/admin/cadastre/jobs",
|
||
json={"scope": "pilot"},
|
||
headers=ADMIN_HEADERS,
|
||
)
|
||
|
||
assert response.status_code == 200, response.text
|
||
body = response.json()
|
||
assert "job_id" in body
|
||
assert body["scope"] == "pilot"
|
||
assert body["targets_total"] == 50
|
||
assert "estimate_minutes" in body
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
@requires_scrape_cadastre
|
||
@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
|
||
def test_create_job_manual_list() -> None:
|
||
"""POST /cadastre/jobs scope=manual_list с явным списком."""
|
||
db = _make_mock_db()
|
||
|
||
from app.core.db import get_db
|
||
|
||
app.dependency_overrides[get_db] = _mock_get_db(db)
|
||
|
||
try:
|
||
with patch("app.workers.tasks.scrape_cadastre.enqueue_cadastre_harvest") as mock_task:
|
||
mock_task.apply_async = MagicMock()
|
||
client = TestClient(app)
|
||
response = client.post(
|
||
"/api/v1/admin/cadastre/jobs",
|
||
json={
|
||
"scope": "manual_list",
|
||
"quarters": ["66:41:0303161", "66:41:0303162"],
|
||
},
|
||
headers=ADMIN_HEADERS,
|
||
)
|
||
|
||
assert response.status_code == 200, response.text
|
||
body = response.json()
|
||
assert body["targets_total"] == 2
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
@requires_scrape_cadastre
|
||
@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
|
||
def test_create_job_manual_list_empty_quarters_returns_400() -> None:
|
||
"""scope=manual_list без quarters → 400."""
|
||
db = _make_mock_db()
|
||
|
||
from app.core.db import get_db
|
||
|
||
app.dependency_overrides[get_db] = _mock_get_db(db)
|
||
|
||
try:
|
||
client = TestClient(app)
|
||
response = client.post(
|
||
"/api/v1/admin/cadastre/jobs",
|
||
json={"scope": "manual_list"},
|
||
headers=ADMIN_HEADERS,
|
||
)
|
||
assert response.status_code == 400
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
|
||
def test_list_jobs_returns_list() -> None:
|
||
"""GET /cadastre/jobs → список jobs."""
|
||
db = _make_mock_db(job_row=_make_sample_job_row())
|
||
|
||
# Переопределяем execute чтобы list query вернул список
|
||
job_row = _make_sample_job_row()
|
||
mock_result = MagicMock()
|
||
mock_result.mappings.return_value.all.return_value = [job_row]
|
||
db.execute = MagicMock(return_value=mock_result)
|
||
|
||
from app.core.db import get_db
|
||
|
||
app.dependency_overrides[get_db] = _mock_get_db(db)
|
||
|
||
try:
|
||
client = TestClient(app)
|
||
response = client.get("/api/v1/admin/cadastre/jobs", headers=ADMIN_HEADERS)
|
||
assert response.status_code == 200, response.text
|
||
body = response.json()
|
||
assert isinstance(body, list)
|
||
assert len(body) == 1
|
||
assert body[0]["job_id"] == 42
|
||
assert body[0]["scope"] == "pilot"
|
||
assert "progress_pct" in body[0]
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
|
||
def test_get_job_not_found_returns_404() -> None:
|
||
"""GET /cadastre/jobs/9999 → 404."""
|
||
db = MagicMock()
|
||
mock_result = MagicMock()
|
||
mock_result.mappings.return_value.first.return_value = None
|
||
db.execute = MagicMock(return_value=mock_result)
|
||
|
||
from app.core.db import get_db
|
||
|
||
app.dependency_overrides[get_db] = _mock_get_db(db)
|
||
|
||
try:
|
||
client = TestClient(app)
|
||
response = client.get("/api/v1/admin/cadastre/jobs/9999", headers=ADMIN_HEADERS)
|
||
assert response.status_code == 404
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
|
||
def test_cancel_job_success() -> None:
|
||
"""POST /cadastre/jobs/42/cancel → {cancelled: true}."""
|
||
db = MagicMock()
|
||
mock_result = MagicMock()
|
||
mock_result.first.return_value = (42,)
|
||
db.execute = MagicMock(return_value=mock_result)
|
||
db.commit = MagicMock()
|
||
|
||
from app.core.db import get_db
|
||
|
||
app.dependency_overrides[get_db] = _mock_get_db(db)
|
||
|
||
try:
|
||
client = TestClient(app)
|
||
response = client.post(
|
||
"/api/v1/admin/cadastre/jobs/42/cancel",
|
||
headers=ADMIN_HEADERS,
|
||
)
|
||
assert response.status_code == 200, response.text
|
||
body = response.json()
|
||
assert body["cancelled"] is True
|
||
assert body["job_id"] == 42
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
|
||
def test_cancel_job_not_found_returns_404() -> None:
|
||
"""POST /cadastre/jobs/9999/cancel когда job не найден → 404."""
|
||
db = MagicMock()
|
||
mock_result = MagicMock()
|
||
mock_result.first.return_value = None # job не найден / уже cancelled
|
||
db.execute = MagicMock(return_value=mock_result)
|
||
db.commit = MagicMock()
|
||
|
||
from app.core.db import get_db
|
||
|
||
app.dependency_overrides[get_db] = _mock_get_db(db)
|
||
|
||
try:
|
||
client = TestClient(app)
|
||
response = client.post(
|
||
"/api/v1/admin/cadastre/jobs/9999/cancel",
|
||
headers=ADMIN_HEADERS,
|
||
)
|
||
assert response.status_code == 404
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
@pytest.mark.skip(
|
||
reason=(
|
||
"X-Admin-Token gate удалён в #437 (refactor(security): убрать X-Admin-Token — "
|
||
"Caddy basic_auth + RBAC middleware достаточны). Admin endpoint'ы больше не "
|
||
"несут verify_admin_token dependency; защита — на уровне Caddy/RBAC. В test-mode "
|
||
"RBAC bypass'ится (CI-rehab 1/3), поэтому 401/503 здесь больше недостижим без "
|
||
"реверта security-решения #437. Тест проверял удалённое поведение."
|
||
)
|
||
)
|
||
def test_create_job_no_token_returns_401() -> None:
|
||
"""Без X-Admin-Token → 401/503 (устарело: токен-гейт удалён в #437)."""
|
||
client = TestClient(app)
|
||
response = client.post(
|
||
"/api/v1/admin/cadastre/jobs",
|
||
json={"scope": "pilot"},
|
||
)
|
||
assert response.status_code in (401, 503)
|