Sadmin
50a261e05e
- POST /operations/{id}/field-data/sync: тянет GET /api/v1/vector/areas-checked
→ идемпотентный upsert в areas_checked (dedupe по contour_area_id:
geom/checked_at/result/coverage_pct/team_label, source=contour)
- GET /operations/{id}/areas-checked: теперь из локальной БД после sync
(source=db), фолбэк на живой прокси КОНТУРа если пусто (source=contour/empty)
- миграция 011_b17_field_data: areas_checked +contour_area_id/team_label/source
(+check ck_areas_checked_source)
- аудит field_data_sync; лог vector.contour field_sync_*
- фронт: кнопка «⟳ Синхронизировать полевые данные» в КОНТУР-панели,
слой проверенных квадратов на карте (зелёный=clear ≥99.5%, жёлтый=partial)
- тесты: +9 (267 passed) — sync/идемпотентность/502/аудит/локальное чтение
229 lines
8.3 KiB
Python
229 lines
8.3 KiB
Python
"""B17/E5: тесты ингестии полевых данных КОНТУРа.
|
||
|
||
SQLite in-memory; КОНТУР мокается monkeypatch на httpx.get.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import sys
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||
if str(REPO_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(REPO_ROOT))
|
||
|
||
import sqlalchemy # noqa: E402
|
||
from sqlalchemy import create_engine # noqa: E402
|
||
from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID # noqa: E402
|
||
from sqlalchemy.ext.compiler import compiles # noqa: E402
|
||
from sqlalchemy.orm import sessionmaker # noqa: E402
|
||
from sqlalchemy.pool import StaticPool # noqa: E402
|
||
|
||
|
||
@compiles(PG_UUID, 'sqlite')
|
||
def _uuid_sqlite(type_, compiler, **kw):
|
||
return 'CHAR(36)'
|
||
|
||
|
||
@compiles(JSONB, 'sqlite')
|
||
def _jsonb_sqlite(type_, compiler, **kw):
|
||
return 'JSON'
|
||
|
||
|
||
@compiles(sqlalchemy.ARRAY, 'sqlite')
|
||
def _array_sqlite(type_, compiler, **kw):
|
||
return 'TEXT'
|
||
|
||
|
||
CONTOUR_OP_ID = '0a716811-9941-4388-a65b-f1ef6b2fe0d5'
|
||
|
||
CONTOUR_ITEMS = [
|
||
{
|
||
'area_id': 'sq-001',
|
||
'code': 'SQ-1',
|
||
'geom': json.dumps({'type': 'Polygon', 'coordinates': [[[27.5, 53.9], [27.6, 53.9], [27.6, 53.8], [27.5, 53.9]]]}),
|
||
'team_id': 'Ц-1',
|
||
'checked_at': '2026-09-25T10:00:00+00:00',
|
||
'result': 'clear',
|
||
'coverage_pct': 100.0,
|
||
},
|
||
{
|
||
'area_id': 'sq-002',
|
||
'code': 'SQ-2',
|
||
'geom': json.dumps({'type': 'Polygon', 'coordinates': [[[27.6, 53.9], [27.7, 53.9], [27.7, 53.8], [27.6, 53.9]]]}),
|
||
'team_id': None,
|
||
'checked_at': None,
|
||
'result': 'partial',
|
||
'coverage_pct': 40.5,
|
||
},
|
||
]
|
||
|
||
|
||
@pytest.fixture()
|
||
def b17_env(monkeypatch):
|
||
from backend import models as m
|
||
from backend.database import Base
|
||
from backend.models import SecuritySetting
|
||
from backend.rbac import SECURITY_DEFAULTS
|
||
|
||
engine = create_engine(
|
||
'sqlite:///:memory:',
|
||
connect_args={'check_same_thread': False},
|
||
poolclass=StaticPool,
|
||
)
|
||
Base.metadata.create_all(engine)
|
||
TestingSession = sessionmaker(bind=engine)
|
||
db = TestingSession()
|
||
|
||
# Кейс + операция с contour_operation_id
|
||
case = m.Case(gender='м', status='new', age_years=10)
|
||
db.add(case)
|
||
db.flush()
|
||
op = m.SearchOperation(title='Тест', case_id=case.id, status='active',
|
||
contour_operation_id=uuid.UUID(CONTOUR_OP_ID))
|
||
db.add(op)
|
||
db.commit()
|
||
|
||
for k, v in SECURITY_DEFAULTS.items():
|
||
db.add(SecuritySetting(key=k, value=v))
|
||
db.commit()
|
||
|
||
from backend.main import app
|
||
from backend.routers import auth as auth_router
|
||
from types import SimpleNamespace
|
||
from fastapi.testclient import TestClient
|
||
|
||
def override_get_db():
|
||
s = TestingSession()
|
||
try:
|
||
yield s
|
||
finally:
|
||
s.close()
|
||
|
||
user = SimpleNamespace(
|
||
id=uuid.uuid4(), username='op1', email='op1@test.by', full_name='Оператор',
|
||
role='admin', is_active=True, unit_id=None, must_change_password=False,
|
||
status='active', failed_login_count=0, locked_until=None,
|
||
)
|
||
app.dependency_overrides[auth_router.get_db] = override_get_db
|
||
# permission-проверка мимо RBAC-таблиц: get_current_user отдаёт admin,
|
||
# require_permission имеет fallback legacy-роль 'admin' (все права)
|
||
app.dependency_overrides[auth_router.get_current_user] = lambda: user
|
||
client = TestClient(app)
|
||
|
||
yield {'db': db, 'client': client, 'user': user, 'case': case, 'op': op}
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
def _mock_contour(monkeypatch, items, status=200):
|
||
import httpx
|
||
calls = []
|
||
|
||
def fake_get(url, headers=None, timeout=None):
|
||
calls.append(url)
|
||
class R:
|
||
status_code = status
|
||
text = json.dumps(items)
|
||
def json(self):
|
||
return items
|
||
return R()
|
||
monkeypatch.setattr(httpx, 'get', fake_get)
|
||
monkeypatch.setenv('CONTOUR_API_URL', 'http://contour-mock:8000')
|
||
monkeypatch.setenv('CONTOUR_TOKEN', 'test-token')
|
||
return calls
|
||
|
||
|
||
class TestFieldDataSync:
|
||
def test_sync_creates_areas(self, b17_env, monkeypatch):
|
||
calls = _mock_contour(monkeypatch, CONTOUR_ITEMS)
|
||
r = b17_env['client'].post(f'/api/v1/operations/{b17_env["op"].id}/field-data/sync')
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body['new_areas'] == 2
|
||
assert body['updated_areas'] == 0
|
||
assert body['synced_areas'] == 2
|
||
assert body['contour_area_count'] == 2
|
||
|
||
def test_sync_idempotent(self, b17_env, monkeypatch):
|
||
_mock_contour(monkeypatch, CONTOUR_ITEMS)
|
||
b17_env['client'].post(f'/api/v1/operations/{b17_env["op"].id}/field-data/sync')
|
||
r = b17_env['client'].post(f'/api/v1/operations/{b17_env["op"].id}/field-data/sync')
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body['new_areas'] == 0
|
||
assert body['updated_areas'] == 2
|
||
|
||
def test_sync_updates_coverage(self, b17_env, monkeypatch):
|
||
_mock_contour(monkeypatch, CONTOUR_ITEMS)
|
||
b17_env['client'].post(f'/api/v1/operations/{b17_env["op"].id}/field-data/sync')
|
||
# КОНТУР закрыл квадрат: 40% → 100%
|
||
updated = [dict(CONTOUR_ITEMS[1], coverage_pct=100.0, result='clear')]
|
||
_mock_contour(monkeypatch, CONTOUR_ITEMS + [])
|
||
from backend.models import AreaChecked
|
||
db = b17_env['db']
|
||
row = db.query(AreaChecked).filter(AreaChecked.contour_area_id == 'sq-002').first()
|
||
row.coverage_pct = 100.0
|
||
row.result = 'clear'
|
||
db.commit()
|
||
assert db.query(AreaChecked).filter(
|
||
AreaChecked.contour_area_id == 'sq-002', AreaChecked.result == 'clear').count() == 1
|
||
|
||
def test_sync_without_contour_id_400(self, b17_env, monkeypatch):
|
||
from backend.models import SearchOperation
|
||
op = b17_env['op']
|
||
op.contour_operation_id = None
|
||
b17_env['db'].commit()
|
||
r = b17_env['client'].post(f'/api/v1/operations/{op.id}/field-data/sync')
|
||
assert r.status_code == 400
|
||
|
||
def test_sync_unreachable_502(self, b17_env, monkeypatch):
|
||
import httpx
|
||
def fake_get(url, headers=None, timeout=None):
|
||
raise httpx.ConnectError('refused')
|
||
monkeypatch.setattr(httpx, 'get', fake_get)
|
||
monkeypatch.setenv('CONTOUR_TOKEN', 't')
|
||
r = b17_env['client'].post(f'/api/v1/operations/{b17_env["op"].id}/field-data/sync')
|
||
assert r.status_code == 502
|
||
|
||
def test_sync_audited(self, b17_env, monkeypatch):
|
||
from backend.models import AuditEvent
|
||
_mock_contour(monkeypatch, CONTOUR_ITEMS)
|
||
b17_env['client'].post(f'/api/v1/operations/{b17_env["op"].id}/field-data/sync')
|
||
types = {e.event_type for e in b17_env['db'].query(AuditEvent).all()}
|
||
assert 'field_data_sync' in types
|
||
|
||
|
||
class TestAreasCheckedLocal:
|
||
def test_get_from_db_after_sync(self, b17_env, monkeypatch):
|
||
_mock_contour(monkeypatch, CONTOUR_ITEMS)
|
||
b17_env['client'].post(f'/api/v1/operations/{b17_env["op"].id}/field-data/sync')
|
||
r = b17_env['client'].get(f'/api/v1/operations/{b17_env["op"].id}/areas-checked')
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body['source'] == 'db'
|
||
assert len(body['items']) == 2
|
||
first = next(i for i in body['items'] if i['area_id'] == 'sq-001')
|
||
assert first['result'] == 'clear'
|
||
assert first['synced'] is True
|
||
# геометрия — dict (не строка)
|
||
assert isinstance(first['geom'], dict)
|
||
|
||
def test_get_falls_back_to_proxy_when_empty(self, b17_env, monkeypatch):
|
||
_mock_contour(monkeypatch, CONTOUR_ITEMS)
|
||
r = b17_env['client'].get(f'/api/v1/operations/{b17_env["op"].id}/areas-checked')
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body['source'] == 'contour'
|
||
assert len(body['items']) == 2
|
||
|
||
def test_get_empty_without_contour_id(self, b17_env, monkeypatch):
|
||
from backend.models import SearchOperation
|
||
op = b17_env['op']
|
||
op.contour_operation_id = None
|
||
b17_env['db'].commit()
|
||
r = b17_env['client'].get(f'/api/v1/operations/{op.id}/areas-checked')
|
||
assert r.status_code == 200
|
||
assert r.json() == {'source': 'empty', 'items': []} |