File size: 5,461 Bytes
666f6cf |
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 |
from meeting_scheduling.rest_api import app
from meeting_scheduling.converters import MeetingScheduleModel, model_to_schedule
from fastapi.testclient import TestClient
from time import sleep
from pytest import fail
import json
client = TestClient(app)
def json_to_meeting_schedule(schedule_json):
"""Convert JSON response to MeetingSchedule domain object with proper score."""
# Parse JSON to Pydantic model first
schedule_model = MeetingScheduleModel.model_validate(schedule_json)
# Convert to domain model
schedule = model_to_schedule(schedule_model)
return schedule
def test_feasible():
demo_data_response = client.get("/demo-data")
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
for _ in range(ATTEMPTS):
sleep(0.1)
schedule_response = client.get(f"/schedules/{job_id}")
schedule_json = schedule_response.json()
schedule = json_to_meeting_schedule(schedule_json)
if schedule.score is not None and schedule.score.is_feasible:
# Additional validation like Java version
assert all(
assignment.starting_time_grain is not None
and assignment.room is not None
for assignment in schedule.meeting_assignments
)
stop_solving_response = client.delete(f"/schedules/{job_id}")
assert stop_solving_response.status_code == 200
return
client.delete(f"/schedules/{job_id}")
fail("solution is not feasible")
def test_analyze():
demo_data_response = client.get("/demo-data")
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
for _ in range(ATTEMPTS):
sleep(0.1)
schedule_response = client.get(f"/schedules/{job_id}")
schedule_json = schedule_response.json()
schedule = json_to_meeting_schedule(schedule_json)
if schedule.score is not None and schedule.score.is_feasible:
# Test the analyze endpoint
analysis_response = client.put("/schedules/analyze", json=schedule_json)
assert analysis_response.status_code == 200
analysis = analysis_response.text
assert analysis is not None
# Test with fetchPolicy parameter
analysis_response_2 = client.put(
"/schedules/analyze?fetchPolicy=FETCH_SHALLOW", json=schedule_json
)
assert analysis_response_2.status_code == 200
analysis_2 = analysis_response_2.text
assert analysis_2 is not None
client.delete(f"/schedules/{job_id}")
return
client.delete(f"/schedules/{job_id}")
fail("solution is not feasible for analyze test")
def test_analyze_constraint_scores():
"""Test that the analyze endpoint returns proper constraint scores instead of all zeros."""
demo_data_response = client.get("/demo-data")
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
for _ in range(ATTEMPTS):
sleep(0.1)
schedule_response = client.get(f"/schedules/{job_id}")
schedule_json = schedule_response.json()
schedule = json_to_meeting_schedule(schedule_json)
if schedule.score is not None and schedule.score.is_feasible:
# Test the analyze endpoint and verify constraint scores
analysis_response = client.put("/schedules/analyze", json=schedule_json)
assert analysis_response.status_code == 200
# Parse the analysis response
analysis_data = json.loads(analysis_response.text)
constraints = analysis_data.get("constraints", [])
# Verify we have constraints
assert len(constraints) > 0, "Should have at least one constraint"
# Check that at least some constraints have non-zero scores
# (since we have a feasible solution, some soft constraints should be violated)
non_zero_scores = 0
for constraint in constraints:
score_str = constraint.get("score", "")
if score_str and score_str != "0hard/0medium/0soft":
non_zero_scores += 1
print(
f"Found non-zero constraint score: {constraint.get('name')} = {score_str}"
)
# We should have at least some non-zero scores for soft constraints
assert non_zero_scores > 0, (
f"Expected some non-zero constraint scores, but all were zero. Total constraints: {len(constraints)}"
)
print(
f"✅ Analysis test passed: Found {non_zero_scores} constraints with non-zero scores out of {len(constraints)} total constraints"
)
client.delete(f"/schedules/{job_id}")
return
client.delete(f"/schedules/{job_id}")
fail("solution is not feasible for analyze constraint scores test")
|