from __future__ import annotations from datetime import date, time from app.schemas.itinerary import ( DailyItinerary, FlightSelection, HotelSelection, ItineraryActivity, ItineraryPlan, ) from app.schemas.place import ( MapResearchResult, PlaceCandidate, ) from app.schemas.route import ( HotelRouteEvaluation, RouteEvaluationResult, ) from app.schemas.selection import ( CandidateSelectionResult, RankedHotelCandidate, RoundTripOption, ) from app.schemas.travel_request import ( TravelRequest, ) from app.services.plan_validation_service import ( PlanValidationService, ) def build_context( *, total_budget: float = 8000, ): request = TravelRequest( origin_city="上海", destination_city="成都", departure_date=date(2026, 8, 10), return_date=date(2026, 8, 13), trip_days=4, nights=3, adults=2, total_budget=total_budget, pace="relaxed", ) outbound = { "option_id": "out-001", "flight_numbers": ["MU001"], "departure_time": ( "2026-08-10 08:00" ), "arrival_time": ( "2026-08-10 11:00" ), "price": 2000, "currency": "CNY", } return_flight = { "option_id": "ret-001", "flight_numbers": ["MU002"], "departure_time": ( "2026-08-13 12:00" ), "arrival_time": ( "2026-08-13 15:00" ), "price": 2000, "currency": "CNY", } hotel = { "hotel_id": "hotel-001", "name": "成都测试酒店", "price_per_night": 500, "currency": "CNY", } selection = CandidateSelectionResult( flight_type="round_trip", round_trip_options=[ RoundTripOption( combination_id="round-001", score=90, quoted_price=2000, currency="CNY", outbound=outbound, return_flight=return_flight, reasons=[], warnings=[], ) ], hotels=[ RankedHotelCandidate( hotel_id="hotel-001", score=90, reasons=[], warnings=[], hotel=hotel, ) ], ) map_result = MapResearchResult( city="成都", attractions=[ PlaceCandidate( poi_id="poi-001", name="测试景点", location="104.01,30.01", ) ], restaurants=[ PlaceCandidate( poi_id="food-001", name="测试餐厅", location="104.02,30.02", ) ], ) route_result = RouteEvaluationResult( evaluated_hotels=[ HotelRouteEvaluation( hotel_id="hotel-001", hotel_name="成都测试酒店", final_score=90, base_hotel_score=85, centrality_score=95, subway_score=90, hotel=hotel, ) ], selected_hotel_id="hotel-001", ) plan = ItineraryPlan( title="成都四日旅行", overview="轻松游览成都。", selected_flight=FlightSelection( flight_type="round_trip", combination_id="round-001", outbound_option_id="out-001", return_option_id="ret-001", quoted_price=2000, currency="CNY", selection_reason="综合价格和时间。", ), selected_hotel=HotelSelection( hotel_id="hotel-001", name="成都测试酒店", check_in_date=date(2026, 8, 10), check_out_date=date(2026, 8, 13), nights=3, price_per_night=500, estimated_total_price=1500, currency="CNY", selection_reason="位置较方便。", ), days=[ DailyItinerary( day_index=1, date=date(2026, 8, 10), theme="抵达成都", activities=[ ItineraryActivity( sequence=1, activity_type="flight", name="上海飞往成都", start_time=time(8, 0), end_time=time(11, 0), ), ItineraryActivity( sequence=2, activity_type="transport", name="机场前往酒店", start_time=time(11, 30), end_time=time(13, 0), ), ItineraryActivity( sequence=3, activity_type=( "hotel_check_in" ), name="办理入住", start_time=time(14, 0), end_time=time(14, 30), ), ], ), DailyItinerary( day_index=2, date=date(2026, 8, 11), theme="景点游览", activities=[ ItineraryActivity( sequence=1, activity_type="attraction", reference_id="poi-001", name="测试景点", start_time=time(9, 0), end_time=time(12, 0), ), ItineraryActivity( sequence=2, activity_type="restaurant", reference_id="food-001", name="测试餐厅", start_time=time(12, 30), end_time=time(14, 0), ), ], ), DailyItinerary( day_index=3, date=date(2026, 8, 12), theme="自由活动", activities=[ ItineraryActivity( sequence=1, activity_type="free_time", name="自由活动", start_time=time(10, 0), end_time=time(16, 0), ) ], ), DailyItinerary( day_index=4, date=date(2026, 8, 13), theme="返程", activities=[ ItineraryActivity( sequence=1, activity_type=( "hotel_check_out" ), name="办理退房", start_time=time(8, 0), end_time=time(9, 0), ), ItineraryActivity( sequence=2, activity_type="transport", name="前往机场", start_time=time(9, 0), end_time=time(10, 0), ), ItineraryActivity( sequence=3, activity_type="flight", name="成都飞往上海", start_time=time(12, 0), end_time=time(15, 0), ), ], ), ], ) return ( request, plan, selection, map_result, route_result, ) def test_valid_plan_passes() -> None: service = PlanValidationService() ( request, plan, selection, map_result, route_result, ) = build_context() result = service.validate( request=request, plan=plan, selection=selection, map_result=map_result, route_result=route_result, ) assert result.is_valid is True assert result.error_count == 0 assert result.budget.known_total == 3500 assert ( result.budget.status == "within_known_budget" ) def test_unknown_flight_combination_fails() -> None: service = PlanValidationService() ( request, plan, selection, map_result, route_result, ) = build_context() plan = plan.model_copy( update={ "selected_flight": ( plan.selected_flight.model_copy( update={ "combination_id": ( "round-not-found" ) } ) ) } ) result = service.validate( request=request, plan=plan, selection=selection, map_result=map_result, route_result=route_result, ) assert result.is_valid is False codes = { issue.code for issue in result.issues } assert ( "FLIGHT_COMBINATION_NOT_FOUND" in codes ) def test_unknown_attraction_fails() -> None: service = PlanValidationService() ( request, plan, selection, map_result, route_result, ) = build_context() invalid_activity = ( plan.days[1] .activities[0] .model_copy( update={ "reference_id": "unknown-poi", "name": "不存在的景点", } ) ) invalid_day = plan.days[1].model_copy( update={ "activities": [ invalid_activity, plan.days[1].activities[1], ] } ) invalid_plan = plan.model_copy( update={ "days": [ plan.days[0], invalid_day, plan.days[2], plan.days[3], ] } ) result = service.validate( request=request, plan=invalid_plan, selection=selection, map_result=map_result, route_result=route_result, ) assert result.is_valid is False assert any( issue.code.startswith("UNKNOWN_") for issue in result.issues ) def test_known_cost_over_budget_fails() -> None: service = PlanValidationService() ( request, plan, selection, map_result, route_result, ) = build_context( total_budget=3000 ) result = service.validate( request=request, plan=plan, selection=selection, map_result=map_result, route_result=route_result, ) assert result.is_valid is False assert result.budget.status == "over_budget" assert any( issue.code == "KNOWN_BUDGET_EXCEEDED" for issue in result.issues )