Light1YT
6400ebde24
All checks were successful
CI / changes (push) Successful in 7s
CI / frontend-tests (push) Has been skipped
CI / changes (pull_request) Successful in 6s
CI / frontend-tests (pull_request) Has been skipped
Deploy / changes (push) Successful in 7s
Deploy / build-backend (push) Successful in 1m41s
Deploy / build-frontend (push) Has been skipped
Deploy / build-worker (push) Successful in 2m50s
Deploy / deploy (push) Successful in 1m22s
CI / backend-tests (push) Successful in 6m35s
CI / backend-tests (pull_request) Successful in 6m32s
Insight = аналитик пишет free-form intel про локацию/участок с пометкой «непублично» (is_confidential, §7.13/§8.9). CRUD под /api/v1/insights; created_by из X-Authenticated-User (не из тела — спуфинг автора невозможен). Зеркало custom_pois: raw-SQL service + Pydantic v2 schemas, sync def handlers (threadpool, off event loop). - data/sql/145_insight.sql: table insight (idempotent, geom GENERATED из lon/lat, индексы district/cad_num/is_confidential/created_by/GIST); аддитивно расширяет CHECK audit_log.action += insight_write (тот же constraint ck_audit_log_action, to_regclass-guarded, superset — не ломает #962-аудит; НЕ правит applied 144). - audit_middleware.classify_path: method-aware insight_write для POST/PUT/DELETE /api/v1/insights (GET-чтения не аудируются); обратно совместимо с parcels. - ACL: backend rbac_guard hard-блокирует только /api/v1/admin/*; pilot отсекается frontend RouteGuard + Caddy (как parcels/custom_pois). is_confidential — стораемая пометка без per-record backend-гейта (analyst видит, per #962-политике). - tests: 28 insight (CRUD+filters+required+#261 commit) + 5 audit. 86 зелёных, ruff. Part B (Location first-class, §8.2) — отдельный follow-up. Refs #948.
465 lines
18 KiB
Python
465 lines
18 KiB
Python
"""Тесты для insight CRUD (#948 Part A, ТЗ §7.13/§8.9).
|
||
|
||
Покрывает (зеркало test_custom_pois.py):
|
||
1. POST /api/v1/insights → 201, created_by из X-Authenticated-User
|
||
2. POST без title/body → 422 (Pydantic required-field validation)
|
||
3. POST без X-Authenticated-User → 401
|
||
4. GET /api/v1/insights → InsightList (total/limit/offset/rows)
|
||
5. GET /api/v1/insights?<filters> → фильтры (district/cad_num/category/is_confidential)
|
||
пробрасываются в сервис
|
||
6. GET /api/v1/insights/{id} → 200; 404 если нет
|
||
7. PUT /api/v1/insights/{id} → 200 updated; 404 если нет
|
||
8. DELETE /api/v1/insights/{id} → 204; 404 если нет
|
||
9. is_confidential хранится и фильтруется
|
||
10. Service-level: db.commit() в create/update/delete (#261 regression guard)
|
||
|
||
Стратегия mock: сервисные функции патчим через unittest.mock.patch,
|
||
DB — MagicMock (как test_custom_pois.py). settings.testing=True (conftest)
|
||
отключает rbac_guard, поэтому X-Authenticated-User в тесте проверяет роутерный
|
||
_require_user напрямую.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.main import app
|
||
from app.schemas.insight import InsightCreate, InsightList, InsightOut, InsightUpdate
|
||
|
||
# ── Константы ─────────────────────────────────────────────────────────────────
|
||
|
||
_CAD = "66:41:0204016:10"
|
||
_USER = "analysttest"
|
||
_TS = datetime(2026, 6, 7, 10, 0, 0, tzinfo=UTC)
|
||
_AUTH = {"X-Authenticated-User": _USER}
|
||
|
||
|
||
# ── Helpers ────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _make_out(
|
||
insight_id: int = 1,
|
||
created_by: str = _USER,
|
||
title: str = "Конкурент строит ЖК рядом",
|
||
body: str = "Замечен котлован, забор ГК «Х».",
|
||
category: str | None = "competition",
|
||
is_confidential: bool = False,
|
||
district: str | None = "Октябрьский",
|
||
cad_num: str | None = None,
|
||
lon: float | None = None,
|
||
lat: float | None = None,
|
||
) -> InsightOut:
|
||
return InsightOut(
|
||
id=insight_id,
|
||
created_by=created_by,
|
||
title=title,
|
||
body=body,
|
||
category=category,
|
||
is_confidential=is_confidential,
|
||
district=district,
|
||
cad_num=cad_num,
|
||
lon=lon,
|
||
lat=lat,
|
||
source=None,
|
||
created_at=_TS,
|
||
updated_at=_TS,
|
||
)
|
||
|
||
|
||
# ── Tests: CREATE ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_create_insight_returns_201_and_sets_created_by() -> None:
|
||
"""POST /insights → 201; created_by берётся из X-Authenticated-User, не из тела."""
|
||
expected = _make_out()
|
||
with patch("app.api.v1.insights.create_insight", return_value=expected) as mock_create:
|
||
client = TestClient(app)
|
||
resp = client.post(
|
||
"/api/v1/insights",
|
||
json={
|
||
"title": "Конкурент строит ЖК рядом",
|
||
"body": "Замечен котлован, забор ГК «Х».",
|
||
"category": "competition",
|
||
"district": "Октябрьский",
|
||
},
|
||
headers=_AUTH,
|
||
)
|
||
assert resp.status_code == 201, resp.text
|
||
body = resp.json()
|
||
assert body["title"] == "Конкурент строит ЖК рядом"
|
||
assert body["created_by"] == _USER
|
||
assert body["is_confidential"] is False
|
||
mock_create.assert_called_once()
|
||
# created_by пробрасывается из заголовка во 2-й позиционный аргумент сервиса.
|
||
call_args = mock_create.call_args
|
||
assert call_args[0][1] == _USER
|
||
|
||
|
||
def test_create_confidential_insight_stores_flag() -> None:
|
||
"""is_confidential=True в теле → пробрасывается в payload сервиса."""
|
||
expected = _make_out(is_confidential=True)
|
||
with patch("app.api.v1.insights.create_insight", return_value=expected) as mock_create:
|
||
client = TestClient(app)
|
||
resp = client.post(
|
||
"/api/v1/insights",
|
||
json={"title": "Инсайдер", "body": "Закрытая инфа", "is_confidential": True},
|
||
headers=_AUTH,
|
||
)
|
||
assert resp.status_code == 201, resp.text
|
||
assert resp.json()["is_confidential"] is True
|
||
payload: InsightCreate = mock_create.call_args[0][2]
|
||
assert payload.is_confidential is True
|
||
|
||
|
||
def test_create_insight_missing_title_returns_422() -> None:
|
||
"""title обязателен → 422."""
|
||
client = TestClient(app)
|
||
resp = client.post(
|
||
"/api/v1/insights",
|
||
json={"body": "только тело без заголовка"},
|
||
headers=_AUTH,
|
||
)
|
||
assert resp.status_code == 422, resp.text
|
||
|
||
|
||
def test_create_insight_missing_body_returns_422() -> None:
|
||
"""body обязателен → 422."""
|
||
client = TestClient(app)
|
||
resp = client.post(
|
||
"/api/v1/insights",
|
||
json={"title": "только заголовок"},
|
||
headers=_AUTH,
|
||
)
|
||
assert resp.status_code == 422, resp.text
|
||
|
||
|
||
def test_create_insight_empty_title_returns_422() -> None:
|
||
"""Пустой title (min_length=1) → 422."""
|
||
client = TestClient(app)
|
||
resp = client.post(
|
||
"/api/v1/insights",
|
||
json={"title": "", "body": "непустое тело"},
|
||
headers=_AUTH,
|
||
)
|
||
assert resp.status_code == 422, resp.text
|
||
|
||
|
||
def test_create_insight_bad_category_returns_422() -> None:
|
||
"""category вне enum → 422 (Literal)."""
|
||
client = TestClient(app)
|
||
resp = client.post(
|
||
"/api/v1/insights",
|
||
json={"title": "t", "body": "b", "category": "totally-unknown"},
|
||
headers=_AUTH,
|
||
)
|
||
assert resp.status_code == 422, resp.text
|
||
|
||
|
||
def test_create_insight_without_auth_header_returns_401() -> None:
|
||
"""Нет X-Authenticated-User → 401 (роутерный _require_user; rbac выключен в тестах)."""
|
||
client = TestClient(app)
|
||
resp = client.post(
|
||
"/api/v1/insights",
|
||
json={"title": "t", "body": "b"},
|
||
)
|
||
assert resp.status_code == 401, resp.text
|
||
|
||
|
||
# ── Tests: LIST ──────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_list_insights_returns_envelope() -> None:
|
||
"""GET /insights → InsightList {total, limit, offset, rows}."""
|
||
listing = InsightList(total=2, limit=50, offset=0, rows=[_make_out(1), _make_out(2)])
|
||
with patch("app.api.v1.insights.list_insights", return_value=listing):
|
||
client = TestClient(app)
|
||
resp = client.get("/api/v1/insights", headers=_AUTH)
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
assert body["total"] == 2
|
||
assert body["limit"] == 50
|
||
assert body["offset"] == 0
|
||
assert len(body["rows"]) == 2
|
||
assert body["rows"][0]["id"] == 1
|
||
|
||
|
||
def test_list_insights_passes_filters_to_service() -> None:
|
||
"""Фильтры district/cad_num/category/is_confidential → в сервис как kwargs."""
|
||
listing = InsightList(total=0, limit=50, offset=0, rows=[])
|
||
with patch("app.api.v1.insights.list_insights", return_value=listing) as mock_list:
|
||
client = TestClient(app)
|
||
resp = client.get(
|
||
"/api/v1/insights",
|
||
params={
|
||
"district": "Октябрьский",
|
||
"cad_num": _CAD,
|
||
"category": "risk",
|
||
"is_confidential": "true",
|
||
},
|
||
headers=_AUTH,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
kwargs = mock_list.call_args.kwargs
|
||
assert kwargs["district"] == "Октябрьский"
|
||
assert kwargs["cad_num"] == _CAD
|
||
assert kwargs["category"] == "risk"
|
||
assert kwargs["is_confidential"] is True
|
||
|
||
|
||
def test_list_insights_filter_confidential_false() -> None:
|
||
"""is_confidential=false тоже пробрасывается (не путается с «не задано»)."""
|
||
listing = InsightList(total=0, limit=50, offset=0, rows=[])
|
||
with patch("app.api.v1.insights.list_insights", return_value=listing) as mock_list:
|
||
client = TestClient(app)
|
||
resp = client.get(
|
||
"/api/v1/insights", params={"is_confidential": "false"}, headers=_AUTH
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
assert mock_list.call_args.kwargs["is_confidential"] is False
|
||
|
||
|
||
# ── Tests: GET one ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_get_insight_returns_200() -> None:
|
||
with patch("app.api.v1.insights.get_insight", return_value=_make_out(7)):
|
||
client = TestClient(app)
|
||
resp = client.get("/api/v1/insights/7", headers=_AUTH)
|
||
assert resp.status_code == 200, resp.text
|
||
assert resp.json()["id"] == 7
|
||
|
||
|
||
def test_get_insight_not_found_returns_404() -> None:
|
||
with patch("app.api.v1.insights.get_insight", return_value=None):
|
||
client = TestClient(app)
|
||
resp = client.get("/api/v1/insights/999", headers=_AUTH)
|
||
assert resp.status_code == 404, resp.text
|
||
|
||
|
||
# ── Tests: UPDATE ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_put_insight_returns_updated() -> None:
|
||
updated = _make_out(title="Обновлено", is_confidential=True)
|
||
with patch("app.api.v1.insights.update_insight", return_value=updated):
|
||
client = TestClient(app)
|
||
resp = client.put(
|
||
"/api/v1/insights/1",
|
||
json={"title": "Обновлено", "is_confidential": True},
|
||
headers=_AUTH,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
assert body["title"] == "Обновлено"
|
||
assert body["is_confidential"] is True
|
||
|
||
|
||
def test_put_insight_not_found_returns_404() -> None:
|
||
with patch("app.api.v1.insights.update_insight", return_value=None):
|
||
client = TestClient(app)
|
||
resp = client.put(
|
||
"/api/v1/insights/999", json={"title": "x"}, headers=_AUTH
|
||
)
|
||
assert resp.status_code == 404, resp.text
|
||
|
||
|
||
def test_put_insight_without_auth_returns_401() -> None:
|
||
client = TestClient(app)
|
||
resp = client.put("/api/v1/insights/1", json={"title": "x"})
|
||
assert resp.status_code == 401, resp.text
|
||
|
||
|
||
# ── Tests: DELETE ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_delete_insight_returns_204() -> None:
|
||
with patch("app.api.v1.insights.delete_insight", return_value=True):
|
||
client = TestClient(app)
|
||
resp = client.delete("/api/v1/insights/1", headers=_AUTH)
|
||
assert resp.status_code == 204, resp.text
|
||
|
||
|
||
def test_delete_insight_not_found_returns_404() -> None:
|
||
with patch("app.api.v1.insights.delete_insight", return_value=False):
|
||
client = TestClient(app)
|
||
resp = client.delete("/api/v1/insights/999", headers=_AUTH)
|
||
assert resp.status_code == 404, resp.text
|
||
|
||
|
||
def test_delete_insight_without_auth_returns_401() -> None:
|
||
client = TestClient(app)
|
||
resp = client.delete("/api/v1/insights/1")
|
||
assert resp.status_code == 401, resp.text
|
||
|
||
|
||
# ── Service-level: commit + mapping (#261 regression guard) ─────────────────────
|
||
|
||
|
||
def _make_row_data(
|
||
insight_id: int = 1,
|
||
created_by: str = _USER,
|
||
title: str = "T",
|
||
body: str = "B",
|
||
category: str | None = "competition",
|
||
is_confidential: bool = False,
|
||
district: str | None = "Октябрьский",
|
||
cad_num: str | None = None,
|
||
lon: float | None = None,
|
||
lat: float | None = None,
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"id": insight_id,
|
||
"created_by": created_by,
|
||
"title": title,
|
||
"body": body,
|
||
"category": category,
|
||
"is_confidential": is_confidential,
|
||
"district": district,
|
||
"cad_num": cad_num,
|
||
"lon": lon,
|
||
"lat": lat,
|
||
"source": None,
|
||
"created_at": _TS,
|
||
"updated_at": _TS,
|
||
}
|
||
|
||
|
||
def _make_db_with_row(row_data: dict[str, Any] | None) -> MagicMock:
|
||
"""MagicMock-сессия: execute().mappings().first() → row_data mapping."""
|
||
db = MagicMock()
|
||
row_mock: MagicMock | None = None
|
||
if row_data is not None:
|
||
row_mock = MagicMock()
|
||
row_mock.__getitem__ = lambda self, k: row_data[k]
|
||
row_mock.get = lambda k, default=None: row_data.get(k, default)
|
||
|
||
mapping_mock = MagicMock()
|
||
mapping_mock.first.return_value = row_mock
|
||
|
||
exec_mock = MagicMock()
|
||
exec_mock.mappings.return_value = mapping_mock
|
||
exec_mock.first.return_value = row_mock # DELETE RETURNING id (не mappings)
|
||
|
||
db.execute.return_value = exec_mock
|
||
return db
|
||
|
||
|
||
def test_service_create_insight_commits_and_sets_created_by() -> None:
|
||
"""create_insight вызывает db.commit() и мапит created_by — regression #261."""
|
||
from app.services.insights import create_insight
|
||
|
||
db = _make_db_with_row(_make_row_data(is_confidential=True))
|
||
payload = InsightCreate(title="T", body="B", is_confidential=True)
|
||
|
||
result = create_insight(db, _USER, payload)
|
||
|
||
db.commit.assert_called_once()
|
||
assert result.id == 1
|
||
assert result.created_by == _USER
|
||
assert result.is_confidential is True
|
||
|
||
|
||
def test_service_update_insight_commits_when_fields_given() -> None:
|
||
"""update_insight вызывает db.commit() при наличии изменяемых полей — regression #261."""
|
||
from app.services.insights import update_insight
|
||
|
||
db = _make_db_with_row(_make_row_data(title="new"))
|
||
payload = InsightUpdate(title="new")
|
||
|
||
result = update_insight(db, insight_id=1, payload=payload)
|
||
|
||
db.commit.assert_called_once()
|
||
assert result is not None
|
||
|
||
|
||
def test_service_update_insight_confidential_false_triggers_update() -> None:
|
||
"""is_confidential=False — это изменение (False is not None) → commit."""
|
||
from app.services.insights import update_insight
|
||
|
||
db = _make_db_with_row(_make_row_data(is_confidential=False))
|
||
payload = InsightUpdate(is_confidential=False)
|
||
|
||
update_insight(db, insight_id=1, payload=payload)
|
||
|
||
db.commit.assert_called_once()
|
||
|
||
|
||
def test_service_update_insight_no_commit_when_payload_empty() -> None:
|
||
"""Пустой payload → только updated_at в sets → UPDATE не выполняется, commit не зовётся."""
|
||
from app.services.insights import update_insight
|
||
|
||
db = _make_db_with_row(_make_row_data())
|
||
update_insight(db, insight_id=1, payload=InsightUpdate())
|
||
|
||
db.commit.assert_not_called()
|
||
|
||
|
||
def test_service_update_insight_no_commit_when_not_found() -> None:
|
||
"""update_insight НЕ коммитит если insight не найден."""
|
||
from app.services.insights import update_insight
|
||
|
||
db = _make_db_with_row(None)
|
||
result = update_insight(db, insight_id=999, payload=InsightUpdate(title="x"))
|
||
|
||
db.commit.assert_not_called()
|
||
assert result is None
|
||
|
||
|
||
def test_service_delete_insight_commits_when_found() -> None:
|
||
"""delete_insight коммитит когда строка найдена — regression #261."""
|
||
from app.services.insights import delete_insight
|
||
|
||
db = _make_db_with_row(_make_row_data())
|
||
db.execute.return_value.first.return_value = MagicMock() # RETURNING id truthy
|
||
|
||
assert delete_insight(db, insight_id=1) is True
|
||
db.commit.assert_called_once()
|
||
|
||
|
||
def test_service_delete_insight_no_commit_when_not_found() -> None:
|
||
"""delete_insight НЕ коммитит если строки нет."""
|
||
from app.services.insights import delete_insight
|
||
|
||
db = _make_db_with_row(None)
|
||
db.execute.return_value.first.return_value = None # RETURNING id пусто
|
||
|
||
assert delete_insight(db, insight_id=999) is False
|
||
db.commit.assert_not_called()
|
||
|
||
|
||
def test_service_list_insights_builds_filtered_query() -> None:
|
||
"""list_insights собирает WHERE по фильтрам + возвращает InsightList с total."""
|
||
from app.services.insights import list_insights
|
||
|
||
db = MagicMock()
|
||
row = _make_row_data(1, is_confidential=True)
|
||
row_mock = MagicMock()
|
||
row_mock.__getitem__ = lambda self, k: row[k]
|
||
|
||
rows_result = MagicMock()
|
||
rows_result.mappings.return_value.all.return_value = [row_mock]
|
||
count_result = MagicMock()
|
||
count_result.scalar_one.return_value = 1
|
||
db.execute.side_effect = [rows_result, count_result]
|
||
|
||
out = list_insights(db, district="Октябрьский", is_confidential=True, limit=10, offset=0)
|
||
|
||
assert isinstance(out, InsightList)
|
||
assert out.total == 1
|
||
assert out.limit == 10
|
||
assert len(out.rows) == 1
|
||
assert out.rows[0].is_confidential is True
|
||
|
||
# Первый execute — SELECT с WHERE по district + is_confidential, psycopg v3 CAST.
|
||
select_sql = str(db.execute.call_args_list[0].args[0])
|
||
assert "WHERE" in select_sql
|
||
assert "district = :district" in select_sql
|
||
assert "is_confidential = CAST(:is_confidential AS boolean)" in select_sql
|
||
assert "::" not in select_sql
|
||
select_params = db.execute.call_args_list[0].args[1]
|
||
assert select_params["district"] == "Октябрьский"
|
||
assert select_params["is_confidential"] is True
|