"""Тесты завершения поиска (кнопка «Завершить» → архив + модель + completed).""" 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' @pytest.fixture() def complete_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', elapsed_hours=3.0, tnp_lat=53.9, tnp_lon=27.56) db.add(case); db.flush(); db.commit() op = SearchOperation(title='Операция завершения', case_id=case.id, status='active', unit_id=rcu.id, created_by=root.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 # closed-cases читает через legacy-глобальный db + модульный SessionLocal — # направляю оба в тестовую БД import backend.database as database_mod import backend.routers.closed_cases as closed_mod from backend.database import SQLCaseRepository test_session = sessionmaker(bind=engine) monkeypatch.setattr(database_mod, 'SessionLocal', test_session) monkeypatch.setattr(closed_mod, 'db', SQLCaseRepository()) client = TestClient(app) tok = client.post('/api/v1/auth/login', data={'username': 'root', 'password': 'rootpass123'}).json()['access_token'] yield {'db': db, 'client': client, 'token': tok, 'op': op, 'case': case} app.dependency_overrides.pop(auth_router.get_db, None) class TestCompleteOperation: def test_complete_flow(self, complete_env): env = complete_env r = env['client'].post( f"/api/v1/operations/{env['op'].id}/complete", headers={'Authorization': f"Bearer {env['token']}"}, json={'found_alive': True, 'found_distance_km': 4.2, 'found_direction': 'NE', 'search_duration_hours': 5.5, 'who_found': 'лесники'}) assert r.status_code == 200, r.json() body = r.json() assert body['status'] == 'completed' assert body['closed_at'] is not None # Кейс закрыт + фактические данные from backend.models import Case case = env['db'].query(Case).filter(Case.id == env['case'].id).first() assert case.status == 'closed' assert case.found_alive is True assert case.found_distance_km == 4.2 assert case.who_found == 'лесники' assert case.analysis_log.get('closed_case') is True assert case.analysis_log.get('model_prediction') # прогон модели записан # Операция завершена from backend.models import SearchOperation op = env['db'].query(SearchOperation).filter(SearchOperation.id == env['op'].id).first() assert op.status == 'completed' assert op.closed_at is not None def test_double_complete_400(self, complete_env): env = complete_env h = {'Authorization': f"Bearer {env['token']}"} payload = {'found_alive': False} r1 = env['client'].post(f"/api/v1/operations/{env['op'].id}/complete", headers=h, json=payload) assert r1.status_code == 200 r2 = env['client'].post(f"/api/v1/operations/{env['op'].id}/complete", headers=h, json=payload) assert r2.status_code == 400 assert 'уже завершена' in r2.json()['detail'] def test_complete_audited(self, complete_env): env = complete_env env['client'].post( f"/api/v1/operations/{env['op'].id}/complete", headers={'Authorization': f"Bearer {env['token']}"}, json={'found_alive': True}) from backend.models import AuditEvent types = {e.event_type for e in env['db'].query(AuditEvent).all()} assert 'operation_complete' in types assert 'case_closed' in types def test_completed_case_in_closed_list(self, complete_env): # Закрытый через операцию кейс попадает в архив завершённых (/admin) env = complete_env env['client'].post( f"/api/v1/operations/{env['op'].id}/complete", headers={'Authorization': f"Bearer {env['token']}"}, json={'found_alive': True, 'found_distance_km': 2.0}) r = env['client'].get('/api/v1/closed-cases', headers={'Authorization': f"Bearer {env['token']}"}) assert r.status_code == 200 items = r.json()['items'] assert any(it.get('age_years') == 10 for it in items)