blackopsrepl/meeting-scheduling-python

1from meeting_scheduling.rest_api import app2from meeting_scheduling.converters import MeetingScheduleModel, model_to_schedule3 4from fastapi.testclient import TestClient5from time import sleep6from pytest import fail7import json8 9client = TestClient(app)10 11 12def json_to_meeting_schedule(schedule_json):13    """Convert JSON response to MeetingSchedule domain object with proper score."""14    # Parse JSON to Pydantic model first15    schedule_model = MeetingScheduleModel.model_validate(schedule_json)16 17    # Convert to domain model18    schedule = model_to_schedule(schedule_model)19 20    return schedule21 22 23def test_feasible():24    demo_data_response = client.get("/demo-data")25    assert demo_data_response.status_code == 20026 27    job_id_response = client.post("/schedules", json=demo_data_response.json())28    assert job_id_response.status_code == 20029    job_id = job_id_response.text[1:-1]30 31    ATTEMPTS = 1_00032    for _ in range(ATTEMPTS):33        sleep(0.1)34        schedule_response = client.get(f"/schedules/{job_id}")35        schedule_json = schedule_response.json()36        schedule = json_to_meeting_schedule(schedule_json)37 38        if schedule.score is not None and schedule.score.is_feasible:39            # Additional validation like Java version40            assert all(41                assignment.starting_time_grain is not None42                and assignment.room is not None43                for assignment in schedule.meeting_assignments44            )45 46            stop_solving_response = client.delete(f"/schedules/{job_id}")47            assert stop_solving_response.status_code == 20048            return49 50    client.delete(f"/schedules/{job_id}")51    fail("solution is not feasible")52 53 54def test_analyze():55    demo_data_response = client.get("/demo-data")56    assert demo_data_response.status_code == 20057 58    job_id_response = client.post("/schedules", json=demo_data_response.json())59    assert job_id_response.status_code == 20060    job_id = job_id_response.text[1:-1]61 62    ATTEMPTS = 1_00063    for _ in range(ATTEMPTS):64        sleep(0.1)65        schedule_response = client.get(f"/schedules/{job_id}")66        schedule_json = schedule_response.json()67        schedule = json_to_meeting_schedule(schedule_json)68 69        if schedule.score is not None and schedule.score.is_feasible:70            # Test the analyze endpoint71            analysis_response = client.put("/schedules/analyze", json=schedule_json)72            assert analysis_response.status_code == 20073            analysis = analysis_response.text74            assert analysis is not None75 76            # Test with fetchPolicy parameter77            analysis_response_2 = client.put(78                "/schedules/analyze?fetchPolicy=FETCH_SHALLOW", json=schedule_json79            )80            assert analysis_response_2.status_code == 20081            analysis_2 = analysis_response_2.text82            assert analysis_2 is not None83 84            client.delete(f"/schedules/{job_id}")85            return86 87    client.delete(f"/schedules/{job_id}")88    fail("solution is not feasible for analyze test")89 90 91def test_analyze_constraint_scores():92    """Test that the analyze endpoint returns proper constraint scores instead of all zeros."""93    demo_data_response = client.get("/demo-data")94    assert demo_data_response.status_code == 20095 96    job_id_response = client.post("/schedules", json=demo_data_response.json())97    assert job_id_response.status_code == 20098    job_id = job_id_response.text[1:-1]99 100    ATTEMPTS = 1_000101    for _ in range(ATTEMPTS):102        sleep(0.1)103        schedule_response = client.get(f"/schedules/{job_id}")104        schedule_json = schedule_response.json()105        schedule = json_to_meeting_schedule(schedule_json)106 107        if schedule.score is not None and schedule.score.is_feasible:108            # Test the analyze endpoint and verify constraint scores109            analysis_response = client.put("/schedules/analyze", json=schedule_json)110            assert analysis_response.status_code == 200111 112            # Parse the analysis response113            analysis_data = json.loads(analysis_response.text)114            constraints = analysis_data.get("constraints", [])115 116            # Verify we have constraints117            assert len(constraints) > 0, "Should have at least one constraint"118 119            # Check that at least some constraints have non-zero scores120            # (since we have a feasible solution, some soft constraints should be violated)121            non_zero_scores = 0122            for constraint in constraints:123                score_str = constraint.get("score", "")124                if score_str and score_str != "0hard/0medium/0soft":125                    non_zero_scores += 1126                    print(127                        f"Found non-zero constraint score: {constraint.get('name')} = {score_str}"128                    )129 130            # We should have at least some non-zero scores for soft constraints131            assert non_zero_scores > 0, (132                f"Expected some non-zero constraint scores, but all were zero. Total constraints: {len(constraints)}"133            )134 135            print(136                f"✅ Analysis test passed: Found {non_zero_scores} constraints with non-zero scores out of {len(constraints)} total constraints"137            )138 139            client.delete(f"/schedules/{job_id}")140            return141 142    client.delete(f"/schedules/{job_id}")143    fail("solution is not feasible for analyze constraint scores test")144