File size: 5,407 Bytes
e7cf451 e510416 e7cf451 e510416 e7cf451 e510416 e7cf451 e510416 e7cf451 e510416 e7cf451 e510416 e7cf451 e510416 e7cf451 |
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 |
"""
Integration tests for the employee scheduling solver.
Tests that the solver can find feasible solutions for demo data
and that the REST API works correctly.
"""
from employee_scheduling.rest_api import app
from employee_scheduling.domain import EmployeeScheduleModel
from employee_scheduling.converters import model_to_schedule
from fastapi.testclient import TestClient
from time import sleep
from pytest import fail
import pytest
client = TestClient(app)
@pytest.mark.timeout(120)
def test_feasible():
"""Test that the solver can find a feasible solution for SMALL demo data."""
demo_data_response = client.get("/demo-data/SMALL")
assert demo_data_response.status_code == 200
job_id_response = client.post("/schedules", json=demo_data_response.json())
assert job_id_response.status_code == 200
job_id = job_id_response.text[1:-1]
ATTEMPTS = 1_000
best_score = None
for _ in range(ATTEMPTS):
sleep(0.1)
schedule_response = client.get(f"/schedules/{job_id}")
schedule_json = schedule_response.json()
schedule_model = EmployeeScheduleModel.model_validate(schedule_json)
schedule = model_to_schedule(schedule_model)
if schedule.score is not None:
best_score = schedule.score
if schedule.score.is_feasible:
stop_solving_response = client.delete(f"/schedules/{job_id}")
assert stop_solving_response.status_code == 200
return
client.delete(f"/schedules/{job_id}")
fail(f"Solution is not feasible after 100 seconds. Best score: {best_score}")
def test_demo_data_list():
"""Test that demo data list endpoint returns available datasets."""
response = client.get("/demo-data")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) > 0
assert "SMALL" in data
def test_demo_data_small_structure():
"""Test that SMALL demo data has expected structure."""
response = client.get("/demo-data/SMALL")
assert response.status_code == 200
data = response.json()
# Check required fields
assert "employees" in data
assert "shifts" in data
# Validate employees
assert len(data["employees"]) > 0
for employee in data["employees"]:
assert "name" in employee
# Validate shifts
assert len(data["shifts"]) > 0
for shift in data["shifts"]:
assert "id" in shift
assert "start" in shift
assert "end" in shift
assert "location" in shift
assert "requiredSkill" in shift
def test_solver_start_and_stop():
"""Test that solver can be started and stopped."""
demo_data_response = client.get("/demo-data/SMALL")
assert demo_data_response.status_code == 200
# Start solving
start_response = client.post("/schedules", json=demo_data_response.json())
assert start_response.status_code == 200
job_id = start_response.text[1:-1]
# Wait a bit
sleep(0.5)
# Check status
status_response = client.get(f"/schedules/{job_id}")
assert status_response.status_code == 200
schedule = status_response.json()
assert "solverStatus" in schedule
# Stop solving
stop_response = client.delete(f"/schedules/{job_id}")
assert stop_response.status_code == 200
def test_solver_assigns_employees():
"""Test that solver actually assigns employees to shifts."""
demo_data_response = client.get("/demo-data/SMALL")
assert demo_data_response.status_code == 200
job_id_response = client.post("/schedules", json=demo_data_response.json())
assert job_id_response.status_code == 200
job_id = job_id_response.text[1:-1]
# Wait for some solving
sleep(2)
schedule_response = client.get(f"/schedules/{job_id}")
schedule_json = schedule_response.json()
# Check that some shifts have employees assigned
assigned_shifts = [s for s in schedule_json["shifts"] if s.get("employee") is not None]
assert len(assigned_shifts) > 0, "Solver should assign some employees to shifts"
client.delete(f"/schedules/{job_id}")
def test_score_analysis():
"""Test that score analysis endpoint returns constraint analysis."""
demo_data_response = client.get("/demo-data/SMALL")
assert demo_data_response.status_code == 200
# Start solving and get a scored solution
job_id_response = client.post("/schedules", json=demo_data_response.json())
assert job_id_response.status_code == 200
job_id = job_id_response.text[1:-1]
# Wait for solver to produce a score
sleep(2)
schedule_response = client.get(f"/schedules/{job_id}")
schedule_json = schedule_response.json()
# Stop solving
client.delete(f"/schedules/{job_id}")
# Call analyze endpoint
analyze_response = client.put("/schedules/analyze", json=schedule_json)
assert analyze_response.status_code == 200
analysis = analyze_response.json()
# Verify structure
assert "constraints" in analysis
assert isinstance(analysis["constraints"], list)
assert len(analysis["constraints"]) > 0
# Check constraint structure
for constraint in analysis["constraints"]:
assert "name" in constraint
assert "weight" in constraint
assert "score" in constraint
assert "matches" in constraint
assert isinstance(constraint["matches"], list)
|