Sadmin
2aa45e8932
Решение Виктора: СП(лес) 0.5→1.0. Прежняя таблица учитывала лес дважды (в НормС и в СП) — подросток 14 лет за 3 часа получал 2.55 км вместо реалистичных 5.1. Согласуется с априором ПСО (94% находок в 3 км): радиусы стали верхне-перцентилями, не медианой. - terrain_map пересчитан относительно среднего леса: лес/простой лес/поле/ луг/город = 1.0; сложный/густой лес 0.5; болото 0.4; горы/овраг 0.6; дорога/тропа 0.8; дефолт 1.0. - Контрольные точки тестов пересчитаны (6 шт: 1.92→3.84, 0.33→0.665, 2.43→2.7, 5.4→10.8, 13.5→15.0, stats 0.5→1.0) + операционный тест подростка 5.1 км. - settlement_score пока НЕ тронут — на обдумывание (Виктор). 193 passed.
202 lines
8.3 KiB
Python
202 lines
8.3 KiB
Python
"""B14: юнит-тесты чистого движка services/search_engine.py.
|
||
|
||
Анализатор мокается (движок тестируем изолированно, без geo/httpx).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from services.rules_analysis import AnalysisResult, PrimaryZone
|
||
from services.search_engine import (
|
||
SearchInput,
|
||
build_search_model,
|
||
derive_profiles,
|
||
derive_time_of_day,
|
||
unmodeled_profiles,
|
||
)
|
||
|
||
|
||
def _fake_analyzer(urgency: str = 'высокая', radius: float = 1.5) -> Any:
|
||
async def analyzer(case_data: dict[str, Any]) -> AnalysisResult:
|
||
analyzer.captured = case_data
|
||
return AnalysisResult(
|
||
urgency=urgency,
|
||
primary_zones=[
|
||
PrimaryZone(priority=1, name='Тест', direction='N', distance=0.5, reason='тест')
|
||
],
|
||
search_radius_km=radius,
|
||
key_locations=['водоёмы'],
|
||
behavioral_prediction='тест',
|
||
immediate_actions=['действие'],
|
||
summary='тест',
|
||
fallback_used=True,
|
||
)
|
||
|
||
analyzer.captured = None
|
||
return analyzer
|
||
|
||
|
||
def _base_input(**overrides: Any) -> SearchInput:
|
||
data = dict(
|
||
age=8,
|
||
gender='м',
|
||
terrain=['лес'],
|
||
elapsed_hours=2.0,
|
||
has_transport='bike',
|
||
time_of_day='день',
|
||
)
|
||
data.update(overrides)
|
||
return SearchInput(**data)
|
||
|
||
|
||
async def test_build_search_model_bike_8yo_matches_old_contract():
|
||
"""bike 8yo 2h лес день → max_distance.
|
||
B1-ревью 2026-09-09 (Виктор): СП(лес) 0.5→1.0 (НормС — уже скорость
|
||
смещения по среднему лесу; 0.5 учитывал лес дважды).
|
||
Было 5.4 (2 × 1.2 × 0.5 × 0.9 × 5.0), стало 10.8 (2 × 1.2 × 1.0 × 0.9 × 5.0)."""
|
||
model = await build_search_model(_base_input(), analyzer=_fake_analyzer())
|
||
assert model.max_distance_km == pytest.approx(10.8, abs=1e-6)
|
||
assert model.coefficients['base_speed'] == 1.2
|
||
assert model.coefficients['terrain'] == 1.0
|
||
assert model.coefficients['urgency'] == 5.0 # bike
|
||
|
||
|
||
async def test_teen_3h_forest_realistic_distance():
|
||
"""Операционная проверка Виктора: здоровый подросток за 3 часа проходит
|
||
заметно больше 2.5 км. При СП(лес)=1.0: 3 × 2.0 × 1.0 × 0.85 = 5.1 км."""
|
||
model = await build_search_model(
|
||
_base_input(age=14, has_transport=None, elapsed_hours=3.0),
|
||
analyzer=_fake_analyzer(),
|
||
)
|
||
assert model.max_distance_km == pytest.approx(5.1, abs=1e-6)
|
||
|
||
|
||
async def test_profiles_derived_from_diagnosis_and_transport():
|
||
analyzer = _fake_analyzer()
|
||
model = await build_search_model(
|
||
_base_input(diagnosis_type=['РАС'], cant_swim=True),
|
||
analyzer=analyzer,
|
||
)
|
||
assert 'РАС' in [p['name'] for p in model.active_profiles]
|
||
assert model.critical_warnings == [] or isinstance(model.critical_warnings, list)
|
||
# Профили дошли и до анализатора (case_data), и в скорер
|
||
assert 'РАС' in analyzer.captured['profiles']
|
||
assert 'велосипед' in analyzer.captured['profiles']
|
||
assert 'не_умеет_плавать' in analyzer.captured['profiles']
|
||
|
||
|
||
async def test_ras_bike_profiles_case_sensitive_keys():
|
||
"""Ключи профилей — русские (§8): РАС + велосипед, не bike."""
|
||
assert derive_profiles(['рас'], 'bike', False) == ['РАС', 'велосипед']
|
||
assert derive_profiles([], 'scooter', True) == ['самокат', 'не_умеет_плавать']
|
||
|
||
|
||
async def test_explicit_profiles_win_over_derivation():
|
||
assert derive_profiles(['РАС'], 'bike', True, explicit=['СДВГ']) == ['СДВГ']
|
||
|
||
|
||
async def test_unmodeled_diagnoses_flagged():
|
||
flagged = unmodeled_profiles(['ДЦП', 'РАС', 'слабый_слух'])
|
||
assert [f['profile'] for f in flagged] == ['ДЦП', 'слабый_слух']
|
||
assert all(f['note'] for f in flagged)
|
||
|
||
|
||
async def test_time_of_day_from_loss_time():
|
||
assert derive_time_of_day('2026-09-09T14:00:00') == 'день'
|
||
assert derive_time_of_day('2026-09-09T20:00:00') == 'сумерки'
|
||
assert derive_time_of_day('2026-09-09T02:00:00') == 'ночь'
|
||
assert derive_time_of_day('мусор') == 'день'
|
||
assert derive_time_of_day(None) == 'день'
|
||
assert derive_time_of_day(datetime(2026, 9, 9, 23, 0)) == 'ночь'
|
||
|
||
|
||
async def test_time_of_day_auto_applied_in_model():
|
||
model = await build_search_model(
|
||
_base_input(time_of_day=None, loss_time='2026-09-09T21:00:00'),
|
||
analyzer=_fake_analyzer(),
|
||
)
|
||
assert model.time_of_day == 'сумерки'
|
||
assert model.coefficients['time_of_day'] == 0.5 # суметки замедляют (видимость)
|
||
|
||
|
||
async def test_psychotype_detected_and_applied():
|
||
answers = {
|
||
'unfamiliar_behavior': 'explore',
|
||
'stress_reaction': 'angry',
|
||
'leadership': 'always_leader',
|
||
'risk_taking': 'very',
|
||
}
|
||
analyzer = _fake_analyzer()
|
||
model = await build_search_model(
|
||
_base_input(psychotype_answers=answers),
|
||
analyzer=analyzer,
|
||
)
|
||
assert model.psychotype == 'dominant'
|
||
assert model.psychotype_modifiers is not None
|
||
assert model.psychotype_recommendations is not None
|
||
assert analyzer.captured['psychotype_modifiers'] == model.psychotype_modifiers
|
||
|
||
|
||
async def test_no_psychotype_when_no_answers():
|
||
model = await build_search_model(_base_input(), analyzer=_fake_analyzer())
|
||
assert model.psychotype is None
|
||
assert model.psychotype_modifiers is None
|
||
|
||
|
||
async def test_engine_does_not_touch_db_or_http():
|
||
"""Чистота границы: движок не импортирует DB/auth/HTTP-клиентов."""
|
||
import services.search_engine as se
|
||
source = open(se.__file__, encoding='utf-8').read()
|
||
assert 'backend.database' not in source
|
||
assert 'backend.routers' not in source
|
||
assert 'SessionLocal' not in source
|
||
assert 'Depends' not in source
|
||
# И никакого ATAK/Meshtastic/CoT (критерий B20/B14)
|
||
assert 'ATAK' not in source and 'Meshtastic' not in source and 'CoT' not in source
|
||
|
||
|
||
async def test_case_id_merge_loses_payload_empty_containers():
|
||
"""Пустые контейнеры выбрасываются из case_data (старое поведение
|
||
_as_case_data): пустой diagnosis_type не доходит до сервисов."""
|
||
si = SearchInput(diagnosis_type=[])
|
||
assert 'diagnosis_type' not in si.to_case_data()
|
||
|
||
|
||
async def test_none_fields_dropped_from_case_data():
|
||
"""Регресс 2026-09-09: None-поля не должны попадать в case_data.
|
||
|
||
Сервисы читают case_data.get(key, DEFAULT); ключ с None обманывает
|
||
дефолт → elapsed_hours=None ломал calculate_max_distance
|
||
(TypeError: float * NoneType) — «не удаётся запустить анализ».
|
||
"""
|
||
si = SearchInput(age=None, elapsed_hours=None, weather=None, gender=None)
|
||
d = si.to_case_data()
|
||
assert 'elapsed_hours' not in d
|
||
assert 'age' not in d
|
||
assert 'weather' not in d
|
||
assert 'diagnosis_type' not in d # пустой список тоже выбрасывается
|
||
|
||
# И полный прогон не падает:
|
||
model = await build_search_model(si, analyzer=_fake_analyzer())
|
||
assert model.max_distance_km > 0 # elapsed None → дефолт 1.0
|
||
|
||
|
||
async def test_case_card_none_does_not_override_payload():
|
||
"""Слияние карточки (case.to_detail) с payload: None из карточки
|
||
не должен ломать расчёт при пустом payload."""
|
||
si = SearchInput(age=8, elapsed_hours=2.0, terrain=['лес'],
|
||
has_transport='bike', time_of_day='день')
|
||
d = si.to_case_data()
|
||
assert d['elapsed_hours'] == 2.0
|
||
|
||
|
||
async def test_fallback_fields_flow_through():
|
||
analyzer = _fake_analyzer(urgency='критическая', radius=2.5)
|
||
model = await build_search_model(_base_input(), analyzer=analyzer)
|
||
assert model.urgency == 'критическая'
|
||
assert model.search_radius_km == 2.5
|
||
assert model.fallback_used is True
|
||
assert model.primary_zones[0]['direction'] == 'N' |