E4 (B21): отправка зон в КОНТУР по операции (контракт B20)
- contour_operation_id в PATCH /operations/{id} (mapping на операцию КОНТУРа).
- backend/routers/contour.py: POST /{id}/send-to-contour — последний анализ
(search_models → legacy analysis_log) → SearchZones[] GeoJSON-сектора
направлений (±22.5°, probability по приоритету) → POST {CONTOUR_API_URL}/
api/v1/vector/zones?operation_id=... (Bearer CONTOUR_TOKEN). Все шаги
аудируются (contour_send_ok/failed). GET /{id}/areas-checked — прокси
проверенных квадратов (inbound к B17).
- env: CONTOUR_API_URL (дефолт CT130:8000), CONTOUR_TOKEN (в secrets).
- case_id-фильтр в GET /operations.
- UI: панель «КОНТУР» на странице анализа: ввод/сохранение UUID операции
КОНТУРа, кнопка «Отправить зоны в КОНТУР» со статусом результата.
- Тесты E4 (4): build B20-payload (полигон замкнут, probability 0..1,
operation_id в query, Bearer), недостающий contour_id 400, отказ КОНТУРа
→ 502 + аудит, геометрия сектора.
226 passed, 5 skipped.
This commit is contained in:
@@ -0,0 +1,203 @@
|
||||
"""E4 (B21): тесты отправки зон в КОНТУР — build payload, мок httpx, аудиты."""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import bcrypt
|
||||
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'
|
||||
|
||||
|
||||
ANALYSIS = {
|
||||
'lat': 53.9, 'lon': 27.56, 'max_distance_km': 5.4,
|
||||
'analyzed_at': '2026-09-09T12:00:00+00:00',
|
||||
'primary_zones': [
|
||||
{'priority': 1, 'direction': 'NE', 'distance': 2.0, 'reason': 'лес+вода'},
|
||||
{'priority': 2, 'direction': 'E', 'distance': 2.0, 'reason': 'лес'},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def contour_env(monkeypatch):
|
||||
from backend import models as m
|
||||
from backend.database import Base
|
||||
from backend.models import (MchsUnit, Permission, Role, RolePermission,
|
||||
SecuritySetting, SearchOperation, UserRole)
|
||||
|
||||
engine = create_engine('sqlite:///:memory:', connect_args={'check_same_thread': False},
|
||||
poolclass=StaticPool)
|
||||
Base.metadata.create_all(engine)
|
||||
db = sessionmaker(bind=engine)()
|
||||
|
||||
rcu = MchsUnit(name='РЦУ РЧС', kind='rcu')
|
||||
db.add(rcu); db.flush()
|
||||
|
||||
def _hash(pw): return bcrypt.hashpw(pw.encode(), bcrypt.gensalt()).decode()
|
||||
|
||||
root = m.User(username='root', email='r@t.by', hashed_password=_hash('rootpass123'),
|
||||
role='admin', is_active=True, unit_id=rcu.id)
|
||||
db.add(root); db.commit()
|
||||
|
||||
case = m.Case(age_years=10, gender='м', status='analyzed')
|
||||
db.add(case); db.flush()
|
||||
case.analysis_log = dict(ANALYSIS)
|
||||
db.commit()
|
||||
|
||||
contour_op_id = '22222222-2222-4222-8222-000000000001' # как в КОНТУРе e2e
|
||||
|
||||
op = SearchOperation(title='Тестовая операция', case_id=case.id, status='active',
|
||||
unit_id=rcu.id, created_by=root.id,
|
||||
contour_operation_id=__import__('uuid').UUID(contour_op_id))
|
||||
db.add(op); db.commit()
|
||||
|
||||
roles = {}
|
||||
for name in ('admin', 'coordinator', 'operator', 'observer'):
|
||||
r = Role(name=name, is_system=True); db.add(r); roles[name] = r
|
||||
perms = {}
|
||||
for code in ('view', 'create', 'update', 'delete', 'export', 'manage_users',
|
||||
'manage_roles', 'view_audit', 'manage_security'):
|
||||
p = Permission(code=code); db.add(p); perms[code] = p
|
||||
db.commit()
|
||||
for rn, codes in {'admin': tuple(perms), 'observer': ('view',)}.items():
|
||||
for c in codes:
|
||||
db.add(RolePermission(role_id=roles[rn].id, permission_id=perms[c].id))
|
||||
db.add(UserRole(user_id=root.id, role_id=roles['admin'].id))
|
||||
db.commit()
|
||||
|
||||
from backend.rbac import SECURITY_DEFAULTS
|
||||
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 fastapi.testclient import TestClient
|
||||
|
||||
def override():
|
||||
yield db
|
||||
|
||||
app.dependency_overrides[auth_router.get_db] = override
|
||||
client = TestClient(app)
|
||||
tok = client.post('/api/v1/auth/login', data={'username': 'root', 'password': 'rootpass123'}).json()['access_token']
|
||||
|
||||
# env для contour
|
||||
monkeypatch.setenv('CONTOUR_API_URL', 'http://contour-mock:8000')
|
||||
monkeypatch.setenv('CONTOUR_TOKEN', 'test-token')
|
||||
|
||||
yield {'db': db, 'client': client, 'token': tok, 'op': op,
|
||||
'case': case, 'contour_op_id': contour_op_id}
|
||||
|
||||
app.dependency_overrides.pop(auth_router.get_db, None)
|
||||
|
||||
|
||||
class TestSendToContour:
|
||||
def test_missing_contour_id_400(self, contour_env):
|
||||
from backend.models import SearchOperation
|
||||
op = contour_env['op']
|
||||
op.contour_operation_id = None
|
||||
contour_env['db'].commit()
|
||||
r = contour_env['client'].post(
|
||||
f'/api/v1/operations/{op.id}/send-to-contour',
|
||||
headers={'Authorization': f"Bearer {contour_env['token']}"})
|
||||
assert r.status_code == 400
|
||||
assert 'contour_operation_id' in r.json()['detail']
|
||||
|
||||
def test_send_builds_valid_b20_payload(self, contour_env, monkeypatch):
|
||||
captured = {}
|
||||
|
||||
def fake_post(url, json=None, headers=None, timeout=None):
|
||||
captured['url'] = url
|
||||
captured['json'] = json
|
||||
captured['headers'] = headers
|
||||
|
||||
class R:
|
||||
status_code = 200
|
||||
text = ''
|
||||
def json(self):
|
||||
return {'status': 'applied', 'zones_total': 2, 'accepted': 2}
|
||||
return R()
|
||||
|
||||
import httpx
|
||||
monkeypatch.setattr(httpx, 'post', fake_post)
|
||||
|
||||
r = contour_env['client'].post(
|
||||
f"/api/v1/operations/{contour_env['op'].id}/send-to-contour",
|
||||
headers={'Authorization': f"Bearer {contour_env['token']}"})
|
||||
assert r.status_code == 200, r.json()
|
||||
body = r.json()
|
||||
assert body['sent_zones'] == 2
|
||||
assert body['contour_operation_id'] == contour_env['contour_op_id']
|
||||
|
||||
payload = captured['json']
|
||||
assert payload['schema_version'] == '1.0'
|
||||
assert payload['max_distance_km'] == 5.4
|
||||
assert len(payload['zones']) == 2
|
||||
z1 = payload['zones'][0]
|
||||
assert z1['zone_id'] == 'vec:NE:2000m'
|
||||
assert z1['geom']['type'] == 'Polygon'
|
||||
ring = z1['geom']['coordinates'][0]
|
||||
assert ring[0] == ring[-1] # замкнут
|
||||
assert 0 < z1['probability'] <= 1
|
||||
# operation_id в query
|
||||
assert contour_env['contour_op_id'] in captured['url']
|
||||
# auth на КОНТУР
|
||||
assert captured['headers']['Authorization'] == 'Bearer test-token'
|
||||
|
||||
def test_send_failure_audited(self, contour_env, monkeypatch):
|
||||
import httpx
|
||||
|
||||
def fake_post(url, json=None, headers=None, timeout=None):
|
||||
class R:
|
||||
status_code = 400
|
||||
text = 'bad zones'
|
||||
def json(self):
|
||||
return {'detail': 'bad zones'}
|
||||
return R()
|
||||
|
||||
monkeypatch.setattr(httpx, 'post', fake_post)
|
||||
r = contour_env['client'].post(
|
||||
f"/api/v1/operations/{contour_env['op'].id}/send-to-contour",
|
||||
headers={'Authorization': f"Bearer {contour_env['token']}"})
|
||||
assert r.status_code == 502
|
||||
from backend.models import AuditEvent
|
||||
types = {e.event_type for e in contour_env['db'].query(AuditEvent).all()}
|
||||
assert 'contour_send_failed' in types
|
||||
|
||||
def test_sector_polygon_geometry(self, contour_env):
|
||||
from backend.routers.contour import _sector_polygon
|
||||
poly = _sector_polygon(53.9, 27.56, 2.0, 'N')
|
||||
ring = poly['coordinates'][0]
|
||||
assert poly['type'] == 'Polygon'
|
||||
assert ring[0] == ring[-1]
|
||||
# Север: lat растёт
|
||||
lats = [p[1] for p in ring[1:-1]]
|
||||
assert max(lats) > 53.9
|
||||
# Дуга ~2 км
|
||||
assert max(lats) < 53.9 + (2.0 / 111.0) * 1.05
|
||||
Reference in New Issue
Block a user