Fix admin import chain and tests
This commit is contained in:
root
@@ -0,0 +1,162 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
from uuid import uuid4
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import types
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(REPO_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(REPO_ROOT))
|
||||
|
||||
from backend import database as backend_database
|
||||
|
||||
|
||||
def utcnow() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
@dataclass
|
||||
class CaseRecord:
|
||||
id: str
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
age: int | None = None
|
||||
gender: str | None = None
|
||||
health_flags: list[str] = field(default_factory=list)
|
||||
elapsed_hours: int | None = None
|
||||
last_known_place: str | None = None
|
||||
direction: str | None = None
|
||||
terrain_primary: str | None = None
|
||||
weather: str | None = None
|
||||
resources: list[str] = field(default_factory=list)
|
||||
gps_lat: float | None = None
|
||||
gps_lon: float | None = None
|
||||
notes: str | None = None
|
||||
status: str = 'new'
|
||||
found_alive: bool | None = None
|
||||
found_distance_km: float | None = None
|
||||
found_direction: str | None = None
|
||||
found_location_type: str | None = None
|
||||
found_lat: float | None = None
|
||||
found_lon: float | None = None
|
||||
who_found: str | None = None
|
||||
raw_text: str | None = None
|
||||
analysis_log: str | None = None
|
||||
result: dict[str, Any] | None = None
|
||||
|
||||
def to_summary(self) -> dict[str, Any]:
|
||||
return {
|
||||
'id': self.id,
|
||||
'created_at': self.created_at.isoformat(),
|
||||
'updated_at': self.updated_at.isoformat(),
|
||||
'age': self.age,
|
||||
'gender': self.gender,
|
||||
'status': self.status,
|
||||
'found_alive': self.found_alive,
|
||||
'found_distance_km': self.found_distance_km,
|
||||
'found_direction': self.found_direction,
|
||||
'found_location_type': self.found_location_type,
|
||||
'who_found': self.who_found,
|
||||
'last_known_place': self.last_known_place,
|
||||
'direction': self.direction,
|
||||
'terrain_primary': self.terrain_primary,
|
||||
'weather': self.weather,
|
||||
}
|
||||
|
||||
def to_detail(self) -> dict[str, Any]:
|
||||
data = self.to_summary()
|
||||
data.update({
|
||||
'health_flags': list(self.health_flags),
|
||||
'elapsed_hours': self.elapsed_hours,
|
||||
'resources': list(self.resources),
|
||||
'gps_lat': self.gps_lat,
|
||||
'gps_lon': self.gps_lon,
|
||||
'notes': self.notes,
|
||||
'found_lat': self.found_lat,
|
||||
'found_lon': self.found_lon,
|
||||
'raw_text': self.raw_text,
|
||||
'analysis_log': self.analysis_log,
|
||||
'result': self.result,
|
||||
})
|
||||
return data
|
||||
|
||||
|
||||
class TestDB:
|
||||
def __init__(self) -> None:
|
||||
self._cases: dict[str, CaseRecord] = {}
|
||||
|
||||
def create_case(self, **data: Any) -> CaseRecord:
|
||||
now = utcnow()
|
||||
data.pop('mobile', None)
|
||||
data.setdefault('status', 'new')
|
||||
case = CaseRecord(id=str(uuid4()), created_at=now, updated_at=now, **data)
|
||||
self._cases[case.id] = case
|
||||
return case
|
||||
|
||||
def list_cases(self) -> list[CaseRecord]:
|
||||
return sorted(self._cases.values(), key=lambda c: c.created_at, reverse=True)
|
||||
|
||||
def get_case(self, case_id: str) -> CaseRecord | None:
|
||||
return self._cases.get(case_id)
|
||||
|
||||
def update_case(self, case_id: str, **patch: Any) -> CaseRecord | None:
|
||||
case = self._cases.get(case_id)
|
||||
if not case:
|
||||
return None
|
||||
for key, value in patch.items():
|
||||
if hasattr(case, key) and value is not None:
|
||||
setattr(case, key, value)
|
||||
case.updated_at = utcnow()
|
||||
return case
|
||||
|
||||
def delete_case(self, case_id: str) -> bool:
|
||||
return self._cases.pop(case_id, None) is not None
|
||||
|
||||
def stats(self) -> dict[str, Any]:
|
||||
cases = list(self._cases.values())
|
||||
total = len(cases)
|
||||
found_alive = sum(1 for c in cases if c.found_alive is True)
|
||||
found_deceased = sum(1 for c in cases if c.found_alive is False)
|
||||
unknown = total - found_alive - found_deceased
|
||||
distances = [c.found_distance_km for c in cases if c.found_distance_km is not None]
|
||||
median_distance = None
|
||||
if distances:
|
||||
distances = sorted(distances)
|
||||
mid = len(distances) // 2
|
||||
if len(distances) % 2:
|
||||
median_distance = distances[mid]
|
||||
else:
|
||||
median_distance = (distances[mid - 1] + distances[mid]) / 2
|
||||
return {
|
||||
'total_cases': total,
|
||||
'found_alive_count': found_alive,
|
||||
'found_deceased_count': found_deceased,
|
||||
'unknown_outcome_count': unknown,
|
||||
'median_found_distance_km': median_distance,
|
||||
'recent_activity_count': sum(1 for c in cases if (utcnow() - c.updated_at).days < 7),
|
||||
}
|
||||
|
||||
def heatmap(self) -> list[dict[str, Any]]:
|
||||
heat = []
|
||||
for case in self._cases.values():
|
||||
if case.gps_lat is not None and case.gps_lon is not None:
|
||||
heat.append({'lat': case.gps_lat, 'lon': case.gps_lon, 'weight': 1})
|
||||
return heat
|
||||
|
||||
|
||||
db = TestDB()
|
||||
backend_database.db = db
|
||||
|
||||
services_pkg = types.ModuleType('backend.services')
|
||||
services_pkg.__path__ = []
|
||||
stats_mod = types.ModuleType('backend.services.stats_service')
|
||||
stats_mod.summary = db.stats
|
||||
stats_mod.heatmap = db.heatmap
|
||||
services_pkg.stats_service = stats_mod
|
||||
sys.modules['backend.services'] = services_pkg
|
||||
sys.modules['backend.services.stats_service'] = stats_mod
|
||||
Reference in New Issue
Block a user