wehub-resource-sync
753 lines
28 KiB
Python
753 lines
28 KiB
Python
"""
|
||
Map task correctness tests.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import inspect
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from bench_env.task.base import BaseTask
|
||
from bench_env.task.common_tasks import AnswerTask, CriteriaTask
|
||
from bench_env.task.judge import JudgeInput
|
||
from bench_env.task.map import tasks as _tasks_module
|
||
from bench_env.task.map.app import Map
|
||
from bench_env.tests.conftest import make_judge_input
|
||
|
||
ALL_TASK_CLASSES: list[type[BaseTask]] = [
|
||
obj
|
||
for _, obj in inspect.getmembers(_tasks_module, inspect.isclass)
|
||
if issubclass(obj, BaseTask) and obj is not BaseTask and obj.__module__ == _tasks_module.__name__
|
||
]
|
||
ALL_TASK_IDS = [cls.__name__ for cls in ALL_TASK_CLASSES]
|
||
ANSWER_TASK_CLASSES = [cls for cls in ALL_TASK_CLASSES if issubclass(cls, AnswerTask)]
|
||
|
||
TEST_OS_STATE = {"time": {"timestamp": 1742025600000}}
|
||
DEFAULT_ROUTE = {"app": "map", "path": "/"}
|
||
|
||
|
||
def _load_defaults() -> dict[str, Any]:
|
||
path = Path(__file__).resolve().parents[3] / "apps" / "Map" / "data" / "defaults.json"
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def _base_state() -> dict[str, Any]:
|
||
state = copy.deepcopy(_load_defaults())
|
||
state["currentLocation"] = {"latitude": 39.9042, "longitude": 116.4074}
|
||
state["currentView"] = {
|
||
"searchResults": [],
|
||
"poi": None,
|
||
"route": None,
|
||
"routeModes": {},
|
||
"autocomplete": None,
|
||
}
|
||
return state
|
||
|
||
|
||
DEFAULTS = _load_defaults()
|
||
BASE_STATE = _base_state()
|
||
|
||
|
||
def _make_task_input(
|
||
init_state: dict[str, Any],
|
||
curr_state: dict[str, Any],
|
||
*,
|
||
route: dict[str, Any] | None = None,
|
||
init_route: dict[str, Any] | None = None,
|
||
answer: str | None = None,
|
||
) -> JudgeInput:
|
||
return make_judge_input(
|
||
{"apps": {"map": init_state}, "os": TEST_OS_STATE},
|
||
{"apps": {"map": curr_state}, "os": TEST_OS_STATE},
|
||
route=route or DEFAULT_ROUTE,
|
||
init_route=init_route,
|
||
answer=answer,
|
||
)
|
||
|
||
|
||
def _deep_update(target: dict[str, Any], patch: dict[str, Any]) -> None:
|
||
for key, value in patch.items():
|
||
if isinstance(value, dict) and isinstance(target.get(key), dict):
|
||
_deep_update(target[key], value)
|
||
else:
|
||
target[key] = value
|
||
|
||
|
||
def _state(
|
||
*,
|
||
search_results: list[dict[str, Any]] | None = None,
|
||
active_poi: dict[str, Any] | None = None,
|
||
route: dict[str, Any] | None = None,
|
||
route_modes: dict[str, Any] | None = None,
|
||
autocomplete: dict[str, Any] | None = None,
|
||
settings_patch: dict[str, Any] | None = None,
|
||
user_patch: dict[str, Any] | None = None,
|
||
) -> dict[str, Any]:
|
||
state = copy.deepcopy(BASE_STATE)
|
||
if search_results is not None:
|
||
state["currentView"]["searchResults"] = copy.deepcopy(search_results)
|
||
if active_poi is not None:
|
||
state["currentView"]["poi"] = copy.deepcopy(active_poi)
|
||
if route is not None:
|
||
state["currentView"]["route"] = copy.deepcopy(route)
|
||
if route_modes is not None:
|
||
state["currentView"]["routeModes"] = copy.deepcopy(route_modes)
|
||
if autocomplete is not None:
|
||
state["currentView"]["autocomplete"] = copy.deepcopy(autocomplete)
|
||
if settings_patch is not None:
|
||
_deep_update(state["settings"], copy.deepcopy(settings_patch))
|
||
if user_patch is not None:
|
||
_deep_update(state["user"], copy.deepcopy(user_patch))
|
||
return state
|
||
|
||
|
||
def _place(
|
||
name: str,
|
||
*,
|
||
address: str,
|
||
rating: float | None = None,
|
||
review_count: int | None = None,
|
||
phone: str | None = None,
|
||
distance: str | None = None,
|
||
distance_meters: int | None = None,
|
||
) -> dict[str, Any]:
|
||
result: dict[str, Any] = {
|
||
"name": name,
|
||
"address": address,
|
||
"formatted_address": address,
|
||
}
|
||
if rating is not None:
|
||
result["rating"] = rating
|
||
if review_count is not None:
|
||
result["user_ratings_total"] = review_count
|
||
if phone is not None:
|
||
result["formatted_phone_number"] = phone
|
||
if distance is not None:
|
||
result["distance"] = distance
|
||
if distance_meters is not None:
|
||
result["distance_meters"] = distance_meters
|
||
return result
|
||
|
||
|
||
def _route(
|
||
*,
|
||
mode: str,
|
||
destination: str,
|
||
origin: str = "当前位置",
|
||
distance: str = "2.4公里",
|
||
steps: list[dict[str, Any]] | None = None,
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"mode": mode,
|
||
"origin": {"name": origin},
|
||
"destination": {"name": destination},
|
||
"distance": distance,
|
||
"steps": copy.deepcopy(steps or []),
|
||
}
|
||
|
||
|
||
def _route_modes() -> dict[str, Any]:
|
||
return {
|
||
"modes": {
|
||
"WALKING": {
|
||
"distance": "3.2公里",
|
||
"distance_meters": 3200,
|
||
"duration": "45分钟",
|
||
"duration_seconds": 2700,
|
||
},
|
||
"DRIVING": {
|
||
"distance": "4.8公里",
|
||
"distance_meters": 4800,
|
||
"duration": "18分钟",
|
||
"duration_seconds": 1080,
|
||
},
|
||
}
|
||
}
|
||
|
||
|
||
NATIONAL_MUSEUM = _place(
|
||
"中国国家博物馆",
|
||
address="北京市东城区东长安街16号",
|
||
rating=4.8,
|
||
review_count=1280,
|
||
phone="010-65116400",
|
||
distance="1.2公里",
|
||
distance_meters=1200,
|
||
)
|
||
FORBIDDEN_CITY = _place(
|
||
"故宫",
|
||
address="北京市东城区景山前街4号",
|
||
rating=4.8,
|
||
review_count=5621,
|
||
phone="400-950-1925",
|
||
distance="2.6公里",
|
||
distance_meters=2600,
|
||
)
|
||
PIZZA_HUT = _place(
|
||
"必胜客(西单店)",
|
||
address="北京市西城区西单北大街131号",
|
||
rating=4.5,
|
||
review_count=256,
|
||
phone="010-66012345",
|
||
distance="500米",
|
||
distance_meters=500,
|
||
)
|
||
HAIDILAO = _place(
|
||
"海底捞(西单店)",
|
||
address="北京市西城区西单北大街133号",
|
||
rating=4.9,
|
||
review_count=1600,
|
||
phone="010-66056789",
|
||
distance="900米",
|
||
distance_meters=900,
|
||
)
|
||
BURGER_KING = _place(
|
||
"汉堡王(西单店)",
|
||
address="北京市西城区西单北大街129号",
|
||
rating=4.2,
|
||
review_count=380,
|
||
phone="010-66011111",
|
||
distance="300米",
|
||
distance_meters=300,
|
||
)
|
||
MANNER = _place(
|
||
"Manner Coffee(王府井店)",
|
||
address="北京市东城区王府井大街138号",
|
||
rating=4.9,
|
||
review_count=420,
|
||
phone="010-67010001",
|
||
distance="260米",
|
||
distance_meters=260,
|
||
)
|
||
LUCKIN = _place(
|
||
"瑞幸咖啡(东华门店)",
|
||
address="北京市东城区东华门大街20号",
|
||
rating=4.7,
|
||
review_count=320,
|
||
phone="010-67010002",
|
||
distance="180米",
|
||
distance_meters=180,
|
||
)
|
||
STARBUCKS = _place(
|
||
"星巴克(王府井店)",
|
||
address="北京市东城区王府井大街88号",
|
||
rating=4.6,
|
||
review_count=510,
|
||
phone="010-67010003",
|
||
distance="320米",
|
||
distance_meters=320,
|
||
)
|
||
APM = _place(
|
||
"apm购物中心",
|
||
address="北京市东城区王府井大街138号",
|
||
rating=4.6,
|
||
review_count=980,
|
||
phone="010-85186688",
|
||
distance="300米",
|
||
distance_meters=300,
|
||
)
|
||
JOY_CITY = _place(
|
||
"大悦城",
|
||
address="北京市西城区西单北大街131号",
|
||
rating=4.7,
|
||
review_count=2100,
|
||
phone="010-66018888",
|
||
distance="500米",
|
||
distance_meters=500,
|
||
)
|
||
SKP = _place(
|
||
"SKP",
|
||
address="北京市朝阳区建国路87号",
|
||
rating=4.8,
|
||
review_count=3200,
|
||
phone="010-65307788",
|
||
distance="1.8公里",
|
||
distance_meters=1800,
|
||
)
|
||
|
||
def _with_new_search(state: dict[str, Any], keyword: str = "测试搜索") -> dict[str, Any]:
|
||
"""Add a new searchHistory entry so check_searched passes."""
|
||
state = copy.deepcopy(state)
|
||
history = state.setdefault("searchHistory", [])
|
||
history.append({"id": "test_new_search", "kind": "query", "text": keyword})
|
||
return state
|
||
|
||
|
||
RESTAURANT_RESULTS = [BURGER_KING, PIZZA_HUT, HAIDILAO]
|
||
CAFE_RESULTS = [LUCKIN, MANNER, STARBUCKS]
|
||
SHOPPING_RESULTS = [APM, JOY_CITY, SKP]
|
||
MUSEUM_RESULTS = [NATIONAL_MUSEUM]
|
||
FORBIDDEN_CITY_RESULTS = [FORBIDDEN_CITY]
|
||
AUTOCOMPLETE_RESULTS = {
|
||
"query": "银行",
|
||
"suggestions": [
|
||
{"main_text": "中国银行(王府井支行)", "description": "中国银行(王府井支行)"},
|
||
{"main_text": "工商银行(东长安街支行)", "description": "工商银行(东长安街支行)"},
|
||
{"main_text": "建设银行(天安门支行)", "description": "建设银行(天安门支行)"},
|
||
],
|
||
}
|
||
ROUTE_STEPS = [
|
||
{"instruction": "向东出发", "distance": "120米"},
|
||
{"instruction": "右转进入东华门大街", "distance": "300米"},
|
||
{"instruction": "继续直行到达目的地", "distance": "500米"},
|
||
]
|
||
|
||
|
||
def _parse_path(path: str) -> list[str]:
|
||
return [part for part in path.split(".") if part]
|
||
|
||
|
||
def _set_by_path(state: dict[str, Any], path: str, value: Any) -> None:
|
||
current: Any = state
|
||
parts = _parse_path(path)
|
||
for part in parts[:-1]:
|
||
current = current[part]
|
||
current[parts[-1]] = value
|
||
|
||
|
||
def _resolve_criteria_value(value: Any, params: dict[str, Any]) -> Any:
|
||
if isinstance(value, str):
|
||
match = __import__("re").fullmatch(r"\{(\w+)\}", value)
|
||
if match:
|
||
return params[match.group(1)]
|
||
return value
|
||
|
||
|
||
SLOT_LABELS = {
|
||
"name": "名称",
|
||
"rating": "评分",
|
||
"address": "地址",
|
||
"phone": "电话",
|
||
}
|
||
|
||
|
||
def _natural_answer(expected: Any) -> str:
|
||
if isinstance(expected, dict):
|
||
parts = []
|
||
for key, value in expected.items():
|
||
label = SLOT_LABELS.get(key, key)
|
||
parts.append(f"{label}是{value}")
|
||
return "，".join(parts)
|
||
if isinstance(expected, float) and expected.is_integer():
|
||
expected = int(expected)
|
||
return f"我看到的是{expected}"
|
||
|
||
|
||
def _wrong_answer(expected: Any) -> str:
|
||
if isinstance(expected, dict):
|
||
return "名称是别的地方，评分是1，地址是错误地址，电话是0000"
|
||
if isinstance(expected, int):
|
||
return f"我看到的是{expected + 1}"
|
||
if isinstance(expected, float):
|
||
return f"我看到的是{expected + 0.5:g}"
|
||
return "我看到的是另一个结果"
|
||
|
||
|
||
def _search_place_address_expected(place: str) -> str:
|
||
return Map.extract_address(Map.resolve_places(place)[0])
|
||
|
||
|
||
def _place_detail_full_expected(place: str) -> dict[str, Any]:
|
||
resolved = Map.resolve_places(place)[0]
|
||
return {
|
||
"rating": Map.extract_rating(resolved),
|
||
"address": Map.extract_address(resolved),
|
||
"phone": Map.extract_phone(resolved),
|
||
}
|
||
|
||
|
||
def _positive_answer_case(
|
||
task: BaseTask,
|
||
curr_state: dict[str, Any],
|
||
*,
|
||
route: dict[str, Any] | None = None,
|
||
) -> tuple[BaseTask, JudgeInput]:
|
||
curr = copy.deepcopy(curr_state)
|
||
inp = _make_task_input(copy.deepcopy(BASE_STATE), curr, route=route)
|
||
expected = task.get_answer(inp) # type: ignore[attr-defined]
|
||
return task, _make_task_input(copy.deepcopy(BASE_STATE), curr, route=route, answer=_natural_answer(expected))
|
||
|
||
|
||
def _negative_answer_case(
|
||
task: BaseTask,
|
||
curr_state: dict[str, Any],
|
||
*,
|
||
route: dict[str, Any] | None = None,
|
||
) -> tuple[BaseTask, JudgeInput]:
|
||
curr = copy.deepcopy(curr_state)
|
||
inp = _make_task_input(copy.deepcopy(BASE_STATE), curr, route=route)
|
||
expected = task.get_answer(inp) # type: ignore[attr-defined]
|
||
return task, _make_task_input(copy.deepcopy(BASE_STATE), curr, route=route, answer=_wrong_answer(expected))
|
||
|
||
|
||
def _positive_criteria_case(task: CriteriaTask) -> tuple[BaseTask, JudgeInput]:
|
||
curr = copy.deepcopy(BASE_STATE)
|
||
route = DEFAULT_ROUTE
|
||
for path, raw_value in task.criteria.items():
|
||
value = _resolve_criteria_value(raw_value, task.params)
|
||
if path == "route":
|
||
route = {"app": "map", "path": str(value)}
|
||
continue
|
||
_set_by_path(curr, path, value)
|
||
return task, _make_task_input(copy.deepcopy(BASE_STATE), curr, route=route)
|
||
|
||
|
||
def _negative_criteria_case(task: CriteriaTask) -> tuple[BaseTask, JudgeInput]:
|
||
return task, _make_task_input(copy.deepcopy(BASE_STATE), copy.deepcopy(BASE_STATE))
|
||
|
||
|
||
class TestTaskDefinitions:
|
||
@pytest.mark.parametrize("cls", ALL_TASK_CLASSES, ids=ALL_TASK_IDS)
|
||
def test_instantiation(self, cls):
|
||
task = cls()
|
||
assert task.name == cls.__name__
|
||
assert task.templates
|
||
assert "map" in task.apps
|
||
|
||
@pytest.mark.parametrize("cls", ALL_TASK_CLASSES, ids=ALL_TASK_IDS)
|
||
def test_description_renders(self, cls):
|
||
task = cls()
|
||
task._env_state = {"os": TEST_OS_STATE}
|
||
desc = task.description
|
||
assert desc
|
||
assert "{" not in desc
|
||
|
||
@pytest.mark.parametrize("cls", ALL_TASK_CLASSES, ids=ALL_TASK_IDS)
|
||
def test_required_class_attrs(self, cls):
|
||
assert cls.scope in ("S1", "S2", "S3")
|
||
assert cls.objective in ("operate", "query", "hybrid")
|
||
assert cls.composition in ("atomic", "sequential", "transfer", "deep_dive")
|
||
assert cls.difficulty in ("L1", "L2", "L3", "L4")
|
||
|
||
@pytest.mark.parametrize("cls", ALL_TASK_CLASSES, ids=ALL_TASK_IDS)
|
||
def test_parameter_defaults_present(self, cls):
|
||
for key, schema in cls.parameters.items():
|
||
if key.startswith("_"):
|
||
continue
|
||
assert "default" in schema
|
||
|
||
@pytest.mark.parametrize("cls", ANSWER_TASK_CLASSES, ids=[c.__name__ for c in ANSWER_TASK_CLASSES])
|
||
def test_answer_task_has_answer_or_get_answer(self, cls):
|
||
has_answer_attr = cls.answer is not None
|
||
has_get_answer_override = cls.get_answer is not AnswerTask.get_answer
|
||
has_check_goals_override = cls.check_goals is not AnswerTask.check_goals
|
||
assert has_answer_attr or has_get_answer_override or has_check_goals_override
|
||
|
||
|
||
class TestMapAccessor:
|
||
@pytest.fixture
|
||
def map_app(self) -> Map:
|
||
return Map(
|
||
_state(
|
||
search_results=RESTAURANT_RESULTS,
|
||
active_poi=NATIONAL_MUSEUM,
|
||
route=_route(mode="DRIVING", origin="故宫", destination="中国国家博物馆", distance="1.2公里", steps=ROUTE_STEPS),
|
||
route_modes=_route_modes(),
|
||
autocomplete=AUTOCOMPLETE_RESULTS,
|
||
)
|
||
)
|
||
|
||
def test_user_properties(self, map_app: Map):
|
||
assert map_app.user_name == "pure"
|
||
assert map_app.favorite_place_count == 0
|
||
|
||
def test_search_and_settings_access(self, map_app: Map):
|
||
assert len(map_app.search_history) == 2
|
||
assert map_app.get_setting("navigation.keepMapNorthUp") is False
|
||
|
||
def test_runtime_view_access(self, map_app: Map):
|
||
assert map_app.active_poi["name"] == "中国国家博物馆"
|
||
assert map_app.active_route["mode"] == "DRIVING"
|
||
assert len(map_app.search_results) == 3
|
||
assert "WALKING" in map_app.route_modes["modes"]
|
||
|
||
def test_place_queries(self, map_app: Map):
|
||
poi = map_app.active_poi
|
||
assert isinstance(poi, dict)
|
||
assert Map.extract_address(poi) == "北京市东城区东长安街16号"
|
||
assert Map.extract_phone(poi) == "010-65116400"
|
||
assert Map.extract_rating(poi) == 4.8
|
||
|
||
def test_normalize_place_prefers_details_national_phone(self):
|
||
place = Map._normalize_place(
|
||
{
|
||
"placeId": "test-place",
|
||
"name": "测试地点",
|
||
"formattedAddress": "北京市海淀区测试路1号",
|
||
"nationalPhoneNumber": "010-11112222",
|
||
"internationalPhoneNumber": "+86 10 1111 2222",
|
||
"details": {
|
||
"displayName": "测试地点",
|
||
"formattedAddress": "北京市海淀区测试路1号",
|
||
"nationalPhoneNumber": "010-33334444",
|
||
"internationalPhoneNumber": "+86 10 3333 4444",
|
||
},
|
||
}
|
||
)
|
||
assert Map.extract_phone(place) == "010-33334444"
|
||
|
||
def test_search_result_helpers(self, map_app: Map):
|
||
pizza_candidates = [
|
||
r for r in map_app.search_results
|
||
if isinstance(r, dict) and "必胜客" in str(r.get("name") or "")
|
||
]
|
||
assert pizza_candidates and pizza_candidates[0]["name"] == "必胜客(西单店)"
|
||
assert Map.best_rated_from_results(RESTAURANT_RESULTS)["name"] == "海底捞(西单店)"
|
||
assert Map.nearest_from_results(RESTAURANT_RESULTS)["name"] == "汉堡王(西单店)"
|
||
assert Map.rating_rank_from_results(RESTAURANT_RESULTS, "必胜客") == 2
|
||
|
||
def test_radius_filter_uses_frontend_display_distance(self):
|
||
results = [
|
||
{"name": "1995m", "distance_meters": 1995},
|
||
{"name": "2010m", "distance_meters": 2010},
|
||
{"name": "2048m", "distance_meters": 2048},
|
||
{"name": "2059m", "distance_meters": 2059},
|
||
]
|
||
|
||
filtered = Map.filter_results(results, max_distance_meters=2000)
|
||
|
||
assert [item["name"] for item in filtered] == ["1995m", "2010m", "2048m"]
|
||
|
||
def test_alias_resolution_helpers(self):
|
||
forbidden_city_names = [place["name"] for place in Map.resolve_places("故宫")]
|
||
assert forbidden_city_names[:2] == ["故宫博物院", "故宫"]
|
||
assert "故宫角楼" not in forbidden_city_names
|
||
|
||
library_names = [place["name"] for place in Map.resolve_places("国家图书馆")]
|
||
assert "中国国家图书馆" in library_names
|
||
assert "国家图书馆" in library_names
|
||
assert library_names.count("国家图书馆") >= 2
|
||
assert "国家图书馆古籍馆" not in library_names
|
||
|
||
assert Map.geo_resolve("圆明园")[3] == "圆明园遗址公园"
|
||
|
||
def test_route_helpers(self, map_app: Map):
|
||
assert map_app.route_distance_meters("中国国家博物馆") == 1200
|
||
|
||
def test_check_helpers(self, map_app: Map):
|
||
assert map_app.check_route(
|
||
mode="DRIVING",
|
||
origin_hint="故宫",
|
||
destination_hint="中国国家博物馆",
|
||
)["passed"]
|
||
assert Map.check_place_rating_answer(
|
||
"评分是4.8分。",
|
||
4.8,
|
||
)["passed"]
|
||
assert Map.check_place_address_answer(
|
||
"地址是北京市东城区东长安街16号。",
|
||
"北京市东城区东长安街16号",
|
||
)["passed"]
|
||
|
||
def test_accessor_raise_on_missing_data(self, map_app: Map):
|
||
with pytest.raises(ValueError):
|
||
Map.extract_address({})
|
||
with pytest.raises(ValueError):
|
||
Map.extract_rating({})
|
||
assert Map.extract_phone({}) is None
|
||
|
||
def test_geo_search_filters_registered_subbrand_pois_by_default(self):
|
||
expected_filtered_keywords = {
|
||
"肯德基": ("甜品站", "宅急送", "Select"),
|
||
"麦当劳": ("甜品站",),
|
||
"必胜客": ("宅急送",),
|
||
}
|
||
for query, keywords in expected_filtered_keywords.items():
|
||
raw_names = [str(place["name"]) for place in Map.geo_search(query, limit=0, excludes=None)]
|
||
filtered_names = [str(place["name"]) for place in Map.geo_search(query, limit=0)]
|
||
assert any(any(keyword in name for keyword in keywords) for name in raw_names), query
|
||
assert all(not any(keyword in name for keyword in keywords) for name in filtered_names), query
|
||
assert len(filtered_names) < len(raw_names), query
|
||
|
||
|
||
def _check_drive_route_positive_case():
|
||
task = _tasks_module.CheckDriveRoute()
|
||
curr = _state(route=_route(mode="DRIVING", destination="故宫", distance="2.6公里"))
|
||
return task, _make_task_input(copy.deepcopy(BASE_STATE), curr)
|
||
|
||
|
||
def _check_drive_route_negative_case():
|
||
task = _tasks_module.CheckDriveRoute()
|
||
curr = _state(route=_route(mode="WALKING", destination="故宫", distance="2.6公里"))
|
||
return task, _make_task_input(copy.deepcopy(BASE_STATE), curr)
|
||
|
||
|
||
def _query_driving_distance_alias_positive_case():
|
||
task = _tasks_module.QueryDrivingDistance(place="国家图书馆")
|
||
place = next(
|
||
place for place in Map.resolve_places("国家图书馆")
|
||
if place["name"] == "国家图书馆"
|
||
)
|
||
route = Map.geo_route_from_current(str(place["place_id"]), "DRIVING")
|
||
return task, _make_task_input(
|
||
copy.deepcopy(BASE_STATE),
|
||
copy.deepcopy(BASE_STATE),
|
||
answer=f"我看到的是{route['distance']}",
|
||
)
|
||
|
||
|
||
def _query_driving_distance_same_name_duplicate_positive_case():
|
||
task = _tasks_module.QueryDrivingDistance(place="国家图书馆")
|
||
places = [
|
||
place for place in Map.resolve_places("国家图书馆")
|
||
if place["name"] == "国家图书馆"
|
||
]
|
||
place = places[-1]
|
||
route = Map.geo_route_from_current(str(place["place_id"]), "DRIVING")
|
||
return task, _make_task_input(
|
||
copy.deepcopy(BASE_STATE),
|
||
copy.deepcopy(BASE_STATE),
|
||
answer=f"我看到的是{route['distance']}",
|
||
)
|
||
def _best_rated_route_case_state(
|
||
task: BaseTask,
|
||
*,
|
||
mode: str,
|
||
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
|
||
results = Map.geo_search(task.p.category, limit=0)
|
||
best = Map.best_rated_from_results(results, max_distance_meters=float(task.p.radius))
|
||
route = Map.geo_route_from_current(str(best["place_id"]), mode)
|
||
curr_state = _with_new_search(
|
||
_state(
|
||
route=_route(
|
||
mode=mode,
|
||
destination=str(best["name"]),
|
||
distance=str(route["distance"]),
|
||
)
|
||
),
|
||
str(task.p.category),
|
||
)
|
||
return best, route, curr_state
|
||
|
||
|
||
def _filtered_subbrand_with_route(
|
||
query: str,
|
||
*,
|
||
mode: str,
|
||
max_distance_meters: float | None = None,
|
||
) -> tuple[dict[str, Any], dict[str, Any]]:
|
||
filtered_ids = {
|
||
str(place.get("place_id"))
|
||
for place in Map.geo_search(query, limit=0)
|
||
if place.get("place_id") is not None
|
||
}
|
||
for place in Map.geo_search(query, limit=0, excludes=None):
|
||
place_id = place.get("place_id")
|
||
if place_id is None or str(place_id) in filtered_ids:
|
||
continue
|
||
if max_distance_meters is not None and float(place.get("distance_meters") or 0) > max_distance_meters:
|
||
continue
|
||
try:
|
||
route = Map.geo_route_from_current(str(place_id), mode)
|
||
except ValueError:
|
||
continue
|
||
return place, route
|
||
raise AssertionError(f"no filtered sub-brand result with {mode} route for {query!r}")
|
||
|
||
|
||
def _find_best_rated_and_route_mcdonalds_positive_case():
|
||
task = _tasks_module.FindBestRatedAndRoute(category="麦当劳", radius=3000)
|
||
best, route, curr_state = _best_rated_route_case_state(task, mode="DRIVING")
|
||
answer = f"我看评分最高且最近的是{best['name']}，开车大概{route['distance']}"
|
||
return task, _make_task_input(copy.deepcopy(BASE_STATE), curr_state, answer=answer)
|
||
|
||
|
||
def _find_best_rated_and_route_filtered_subbrand_negative_case():
|
||
task = _tasks_module.FindBestRatedAndRoute(category="麦当劳", radius=3000)
|
||
_, _, curr_state = _best_rated_route_case_state(task, mode="DRIVING")
|
||
removed_place, removed_route = _filtered_subbrand_with_route(
|
||
task.p.category,
|
||
mode="DRIVING",
|
||
max_distance_meters=float(task.p.radius),
|
||
)
|
||
answer = f"我看评分最高且最近的是{removed_place['name']}，开车大概{removed_route['distance']}"
|
||
return task, _make_task_input(copy.deepcopy(BASE_STATE), curr_state, answer=answer)
|
||
|
||
|
||
def _best_rated_with_walk_route_kfc_positive_case():
|
||
task = _tasks_module.BestRatedWithWalkRoute(category="肯德基", radius=3000)
|
||
best, route, curr_state = _best_rated_route_case_state(task, mode="WALKING")
|
||
answer = f"我看到的是{best['name']}，走过去大概{route['distance']}"
|
||
return task, _make_task_input(copy.deepcopy(BASE_STATE), curr_state, answer=answer)
|
||
|
||
|
||
def _best_rated_with_walk_route_filtered_subbrand_negative_case():
|
||
task = _tasks_module.BestRatedWithWalkRoute(category="肯德基", radius=3000)
|
||
_, _, curr_state = _best_rated_route_case_state(task, mode="WALKING")
|
||
removed_place, removed_route = _filtered_subbrand_with_route(
|
||
task.p.category,
|
||
mode="WALKING",
|
||
max_distance_meters=float(task.p.radius),
|
||
)
|
||
answer = f"我看到的是{removed_place['name']}，走过去大概{removed_route['distance']}"
|
||
return task, _make_task_input(copy.deepcopy(BASE_STATE), curr_state, answer=answer)
|
||
|
||
|
||
PRIMARY_POSITIVE_CASES = [
|
||
("CheckDriveRoute", _check_drive_route_positive_case),
|
||
("CheckHighestRatedPlace", lambda: _positive_answer_case(_tasks_module.CheckHighestRatedPlace(category="咖啡馆"), _with_new_search(_state(search_results=CAFE_RESULTS), "咖啡馆"))),
|
||
("CheckNearestPlaceAddress", lambda: _positive_answer_case(_tasks_module.CheckNearestPlaceAddress(category="咖啡馆"), _with_new_search(_state(search_results=CAFE_RESULTS), "咖啡馆"))),
|
||
("SetMapNorthUp", lambda: _positive_criteria_case(_tasks_module.SetMapNorthUp())),
|
||
("ModifyMultiSettings", lambda: _positive_criteria_case(_tasks_module.ModifyMultiSettings())),
|
||
("DarkModeSettings", lambda: _positive_criteria_case(_tasks_module.DarkModeSettings())),
|
||
("FindNearestWithRating", lambda: _positive_answer_case(_tasks_module.FindNearestWithRating(category="咖啡馆"), _with_new_search(_state(search_results=CAFE_RESULTS), "咖啡馆"))),
|
||
]
|
||
|
||
PRIMARY_NEGATIVE_CASES = [
|
||
("CheckDriveRoute", _check_drive_route_negative_case),
|
||
("CheckHighestRatedPlace", lambda: _negative_answer_case(_tasks_module.CheckHighestRatedPlace(category="咖啡馆"), _with_new_search(_state(search_results=CAFE_RESULTS), "咖啡馆"))),
|
||
("CheckNearestPlaceAddress", lambda: _negative_answer_case(_tasks_module.CheckNearestPlaceAddress(category="咖啡馆"), _with_new_search(_state(search_results=CAFE_RESULTS), "咖啡馆"))),
|
||
("SetMapNorthUp", lambda: _negative_criteria_case(_tasks_module.SetMapNorthUp())),
|
||
("ModifyMultiSettings", lambda: _negative_criteria_case(_tasks_module.ModifyMultiSettings())),
|
||
("DarkModeSettings", lambda: _negative_criteria_case(_tasks_module.DarkModeSettings())),
|
||
("FindNearestWithRating", lambda: _negative_answer_case(_tasks_module.FindNearestWithRating(category="咖啡馆"), _with_new_search(_state(search_results=CAFE_RESULTS), "咖啡馆"))),
|
||
]
|
||
|
||
EXTRA_POSITIVE_CASES: list[tuple[str, Any]] = [
|
||
("QueryDrivingDistance_alias_name", _query_driving_distance_alias_positive_case),
|
||
("QueryDrivingDistance_same_name_duplicate", _query_driving_distance_same_name_duplicate_positive_case),
|
||
("FindBestRatedAndRoute_mcdonalds_filtered_brand", _find_best_rated_and_route_mcdonalds_positive_case),
|
||
("BestRatedWithWalkRoute_kfc_filtered_brand", _best_rated_with_walk_route_kfc_positive_case),
|
||
]
|
||
|
||
EXTRA_NEGATIVE_CASES: list[tuple[str, Any]] = [
|
||
("FindBestRatedAndRoute_filtered_subbrand_answer", _find_best_rated_and_route_filtered_subbrand_negative_case),
|
||
("BestRatedWithWalkRoute_filtered_subbrand_answer", _best_rated_with_walk_route_filtered_subbrand_negative_case),
|
||
]
|
||
|
||
PRIMARY_TASK_NAMES = {name for name, _ in PRIMARY_POSITIVE_CASES}
|
||
|
||
|
||
class TestTaskJudgeMatrixOffline:
|
||
def test_offline_judge_matrix_complete(self):
|
||
positive = {name for name, _ in PRIMARY_POSITIVE_CASES}
|
||
negative = {name for name, _ in PRIMARY_NEGATIVE_CASES}
|
||
assert positive == negative, "positive/negative case sets must match"
|
||
all_names = {cls.__name__ for cls in ALL_TASK_CLASSES}
|
||
assert positive <= all_names, f"test references unknown tasks: {positive - all_names}"
|
||
|
||
@pytest.mark.parametrize(
|
||
"task_name,builder",
|
||
PRIMARY_POSITIVE_CASES + EXTRA_POSITIVE_CASES,
|
||
ids=[name for name, _ in PRIMARY_POSITIVE_CASES + EXTRA_POSITIVE_CASES],
|
||
)
|
||
def test_positive_case(self, task_name, builder):
|
||
task, inp = builder()
|
||
result = task.evaluate(inp)
|
||
assert result.success, f"{task_name} positive failed: issues={result.issues}, warnings={result.warnings}"
|
||
|
||
@pytest.mark.parametrize(
|
||
"task_name,builder",
|
||
PRIMARY_NEGATIVE_CASES + EXTRA_NEGATIVE_CASES,
|
||
ids=[name for name, _ in PRIMARY_NEGATIVE_CASES + EXTRA_NEGATIVE_CASES],
|
||
)
|
||
def test_negative_case(self, task_name, builder):
|
||
task, inp = builder()
|
||
result = task.evaluate(inp)
|
||
assert not result.success, f"{task_name} negative unexpectedly passed"
|