File size: 13,736 Bytes
fcf8749 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 | """
Unit tests for Phase 5 Agent Workflow Visualization APIs.
Tests agent timeline and driver allocation story endpoints.
"""
import pytest
from datetime import date, datetime, timedelta
from uuid import uuid4, UUID
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import (
Driver, Route, Assignment, AllocationRun, DecisionLog,
AllocationRunStatus, VehicleType,
)
from app.services.admin_service import (
get_agent_timeline,
get_driver_allocation_story,
_generate_short_message,
_extract_details,
)
from app.schemas.admin import AgentTimelineResponse, DriverAllocationStoryResponse
class TestAgentTimelineShortMessages:
"""Tests for short message generation."""
def test_ml_effort_message(self):
"""ML_EFFORT should include driver/route counts."""
log = type('Log', (), {
'agent_name': 'ML_EFFORT',
'step_type': 'MATRIX_GENERATION',
'input_snapshot': {'num_drivers': 50, 'num_routes': 50},
'output_snapshot': {},
})()
message = _generate_short_message(log)
assert "50 drivers" in message
assert "50 routes" in message
def test_route_planner_proposal(self):
"""ROUTE_PLANNER proposals should describe appropriately."""
log = type('Log', (), {
'agent_name': 'ROUTE_PLANNER',
'step_type': 'PROPOSAL_1',
'input_snapshot': {},
'output_snapshot': {},
})()
message = _generate_short_message(log)
assert "initial" in message.lower()
def test_route_planner_resolution(self):
"""FINAL_RESOLUTION should include swap count."""
log = type('Log', (), {
'agent_name': 'ROUTE_PLANNER',
'step_type': 'FINAL_RESOLUTION',
'input_snapshot': {},
'output_snapshot': {'swaps_applied': 4},
})()
message = _generate_short_message(log)
assert "4 swaps" in message
def test_fairness_manager_reoptimize(self):
"""FAIRNESS_MANAGER with REOPTIMIZE status."""
log = type('Log', (), {
'agent_name': 'FAIRNESS_MANAGER',
'step_type': 'FAIRNESS_CHECK_PROPOSAL_1',
'input_snapshot': {},
'output_snapshot': {'status': 'REOPTIMIZE'},
})()
message = _generate_short_message(log)
assert "re-optimization" in message.lower()
def test_driver_liaison_counts(self):
"""DRIVER_LIAISON should include decision counts."""
log = type('Log', (), {
'agent_name': 'DRIVER_LIAISON',
'step_type': 'NEGOTIATION_DECISIONS',
'input_snapshot': {},
'output_snapshot': {'num_accept': 32, 'num_counter': 10, 'num_force_accept': 8},
})()
message = _generate_short_message(log)
assert "32 ACCEPT" in message
assert "10 COUNTER" in message
assert "8 FORCE_ACCEPT" in message
def test_explainability_categories(self):
"""EXPLAINABILITY should include explanation count."""
log = type('Log', (), {
'agent_name': 'EXPLAINABILITY',
'step_type': 'EXPLANATIONS_GENERATED',
'input_snapshot': {},
'output_snapshot': {'total_explanations': 50, 'category_counts': {'NEAR_AVG': 20, 'HEAVY': 10}},
})()
message = _generate_short_message(log)
assert "50" in message
assert "2 categories" in message
class TestExtractDetails:
"""Tests for details extraction from logs."""
def test_extracts_relevant_keys(self):
"""Should extract whitelisted keys from snapshots."""
log = type('Log', (), {
'input_snapshot': {'num_drivers': 50, 'irrelevant_key': 'ignored'},
'output_snapshot': {'gini_index': 0.15, 'std_dev': 12.0, 'also_irrelevant': []},
})()
details = _extract_details(log)
assert details.get('num_drivers') == 50
assert details.get('gini_index') == 0.15
assert details.get('std_dev') == 12.0
assert 'irrelevant_key' not in details
assert 'also_irrelevant' not in details
def test_prefers_output_over_input(self):
"""Output snapshot should take precedence."""
log = type('Log', (), {
'input_snapshot': {'gini_index': 0.20},
'output_snapshot': {'gini_index': 0.15},
})()
details = _extract_details(log)
assert details['gini_index'] == 0.15
@pytest.fixture
async def test_data(db_session: AsyncSession):
"""Create test data for visualization APIs."""
# Create driver
driver = Driver(
external_id="VIZ-001",
name="Visualization Test Driver",
vehicle_type=VehicleType.ICE,
vehicle_capacity_kg=100.0,
)
db_session.add(driver)
# Create route
route = Route(
date=date.today(),
cluster_id=1,
num_packages=25,
total_weight_kg=60.0,
num_stops=15,
route_difficulty_score=2.5,
estimated_time_minutes=180,
)
db_session.add(route)
await db_session.flush()
# Create allocation run
allocation_run = AllocationRun(
date=date.today(),
num_drivers=10,
num_routes=10,
num_packages=100,
global_gini_index=0.15,
global_std_dev=12.0,
global_max_gap=25.0,
status=AllocationRunStatus.SUCCESS,
started_at=datetime.utcnow() - timedelta(minutes=5),
finished_at=datetime.utcnow(),
)
db_session.add(allocation_run)
await db_session.flush()
# Create assignment
assignment = Assignment(
date=date.today(),
driver_id=driver.id,
route_id=route.id,
workload_score=65.0,
fairness_score=0.85,
explanation="Test explanation",
allocation_run_id=allocation_run.id,
)
db_session.add(assignment)
# Create decision logs
logs = [
DecisionLog(
allocation_run_id=allocation_run.id,
agent_name="ML_EFFORT",
step_type="MATRIX_GENERATION",
input_snapshot={"num_drivers": 10, "num_routes": 10},
output_snapshot={"avg_effort": 60.0},
),
DecisionLog(
allocation_run_id=allocation_run.id,
agent_name="ROUTE_PLANNER",
step_type="PROPOSAL_1",
input_snapshot={},
output_snapshot={"total_effort": 600.0},
),
DecisionLog(
allocation_run_id=allocation_run.id,
agent_name="FAIRNESS_MANAGER",
step_type="FAIRNESS_CHECK_PROPOSAL_1",
input_snapshot={},
output_snapshot={"status": "ACCEPT", "gini_index": 0.15},
),
DecisionLog(
allocation_run_id=allocation_run.id,
agent_name="EXPLAINABILITY",
step_type="EXPLANATIONS_GENERATED",
input_snapshot={},
output_snapshot={"total_explanations": 10, "category_counts": {"NEAR_AVG": 8}},
),
]
for log in logs:
db_session.add(log)
await db_session.commit()
return {
"driver_id": driver.id,
"route_id": route.id,
"allocation_run_id": allocation_run.id,
"assignment_id": assignment.id,
"date": date.today(),
}
class TestAgentTimelineEndpoint:
"""Tests for GET /admin/agent_timeline."""
@pytest.mark.asyncio
async def test_timeline_returns_allocation_run_info(self, db_session: AsyncSession, test_data):
"""Timeline should include allocation run info."""
result = await get_agent_timeline(db_session, test_data["allocation_run_id"])
assert result.allocation_run.id == test_data["allocation_run_id"]
assert result.allocation_run.num_drivers == 10
assert result.allocation_run.num_routes == 10
assert result.allocation_run.status == "SUCCESS"
assert "gini_index" in result.allocation_run.global_metrics
@pytest.mark.asyncio
async def test_timeline_contains_all_logs(self, db_session: AsyncSession, test_data):
"""Timeline should contain all decision logs."""
result = await get_agent_timeline(db_session, test_data["allocation_run_id"])
assert len(result.timeline) == 4
agents = [e.agent_name for e in result.timeline]
assert "ML_EFFORT" in agents
assert "ROUTE_PLANNER" in agents
assert "FAIRNESS_MANAGER" in agents
assert "EXPLAINABILITY" in agents
@pytest.mark.asyncio
async def test_timeline_events_have_short_messages(self, db_session: AsyncSession, test_data):
"""Each timeline event should have a short_message."""
result = await get_agent_timeline(db_session, test_data["allocation_run_id"])
for event in result.timeline:
assert event.short_message
assert len(event.short_message) > 0
@pytest.mark.asyncio
async def test_timeline_events_sorted_by_time(self, db_session: AsyncSession, test_data):
"""Events should be sorted by timestamp."""
result = await get_agent_timeline(db_session, test_data["allocation_run_id"])
timestamps = [e.timestamp for e in result.timeline]
assert timestamps == sorted(timestamps)
@pytest.mark.asyncio
async def test_nonexistent_run_returns_empty(self, db_session: AsyncSession):
"""Non-existent allocation run should return empty timeline."""
fake_id = uuid4()
result = await get_agent_timeline(db_session, fake_id)
assert result.allocation_run.status == "NOT_FOUND"
assert len(result.timeline) == 0
class TestDriverAllocationStoryEndpoint:
"""Tests for GET /admin/driver_allocation_story."""
@pytest.mark.asyncio
async def test_story_returns_driver_info(self, db_session: AsyncSession, test_data):
"""Story should include driver info."""
result = await get_driver_allocation_story(
db_session, test_data["driver_id"], test_data["date"]
)
assert result is not None
assert result.driver.id == test_data["driver_id"]
assert result.driver.name == "Visualization Test Driver"
@pytest.mark.asyncio
async def test_story_returns_today_info(self, db_session: AsyncSession, test_data):
"""Story should include today's assignment info."""
result = await get_driver_allocation_story(
db_session, test_data["driver_id"], test_data["date"]
)
assert result.today.assignment_id == test_data["assignment_id"]
assert result.today.route.id == test_data["route_id"]
assert result.today.effort.value == 65.0
assert result.today.fairness_score == 0.85
@pytest.mark.asyncio
async def test_story_returns_global_metrics(self, db_session: AsyncSession, test_data):
"""Story should include allocation run metrics."""
result = await get_driver_allocation_story(
db_session, test_data["driver_id"], test_data["date"]
)
assert result.allocation_run.id == test_data["allocation_run_id"]
assert result.allocation_run.global_metrics.gini_index == 0.15
assert result.allocation_run.global_metrics.std_dev == 12.0
@pytest.mark.asyncio
async def test_story_includes_timeline_slice(self, db_session: AsyncSession, test_data):
"""Story should include agent timeline slice."""
result = await get_driver_allocation_story(
db_session, test_data["driver_id"], test_data["date"]
)
assert len(result.agent_timeline_slice) >= 1
for event in result.agent_timeline_slice:
assert event.agent_name
assert event.description
@pytest.mark.asyncio
async def test_story_returns_none_for_no_assignment(self, db_session: AsyncSession, test_data):
"""Should return None when no assignment exists."""
fake_driver_id = uuid4()
result = await get_driver_allocation_story(
db_session, fake_driver_id, test_data["date"]
)
assert result is None
@pytest.mark.asyncio
async def test_story_returns_none_for_wrong_date(self, db_session: AsyncSession, test_data):
"""Should return None when no assignment for date."""
wrong_date = date.today() - timedelta(days=30)
result = await get_driver_allocation_story(
db_session, test_data["driver_id"], wrong_date
)
assert result is None
@pytest.mark.asyncio
async def test_story_includes_recovery_info(self, db_session: AsyncSession, test_data):
"""Story should include recovery information."""
result = await get_driver_allocation_story(
db_session, test_data["driver_id"], test_data["date"]
)
assert hasattr(result.recovery, 'is_recovery_day')
assert hasattr(result.recovery, 'recent_hard_days')
@pytest.mark.asyncio
async def test_story_includes_negotiation_info(self, db_session: AsyncSession, test_data):
"""Story should include negotiation information."""
result = await get_driver_allocation_story(
db_session, test_data["driver_id"], test_data["date"]
)
assert hasattr(result.negotiation, 'swap_applied')
assert hasattr(result.negotiation, 'manual_override')
assert hasattr(result.negotiation.manual_override, 'affected')
|