373 lines
15 KiB
Python
373 lines
15 KiB
Python
import unittest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from app.api.plaza import (
|
|
_build_libo_bus_route_payload,
|
|
_graph_map_poi_items,
|
|
_graph_poi_detail_payload,
|
|
_graph_zone_payload,
|
|
_map_falkor_client,
|
|
_read_graph_poi_detail,
|
|
_read_graph_map_pois,
|
|
_resolve_spatial_graph_name,
|
|
map_pois,
|
|
map_poi_detail,
|
|
)
|
|
from app.config import settings
|
|
from app.project_context import ProjectContext
|
|
|
|
|
|
class PlazaGraphFallbackTests(unittest.TestCase):
|
|
def test_map_connection_uses_configured_credentials_and_finite_timeouts(self) -> None:
|
|
for password in ("", "local-unit-test-password"):
|
|
with (
|
|
self.subTest(password_configured=bool(password)),
|
|
patch.object(settings, "falkordb_password", password),
|
|
patch("app.api.plaza.FalkorDB") as factory,
|
|
):
|
|
_map_falkor_client()
|
|
kwargs = factory.call_args.kwargs
|
|
self.assertEqual(kwargs["host"], settings.falkordb_host)
|
|
self.assertEqual(kwargs["port"], settings.falkordb_port)
|
|
self.assertEqual(kwargs["socket_connect_timeout"], 5)
|
|
self.assertGreater(kwargs["socket_timeout"], 30)
|
|
if password:
|
|
self.assertEqual(kwargs["password"], password)
|
|
else:
|
|
self.assertNotIn("password", kwargs)
|
|
|
|
def test_imported_poi_keeps_json_id_when_no_amap_id_is_present(self) -> None:
|
|
items = _graph_map_poi_items([[
|
|
0, ["Hotel"],
|
|
{"__kg_node_id": "guiyang-hotel-001", "name": "贵阳测试酒店",
|
|
"longitude": 106.63, "latitude": 26.65},
|
|
]])
|
|
self.assertEqual(items[0]["id"], "guiyang-hotel-001")
|
|
graph = MagicMock()
|
|
graph.query.return_value.result_set = [[["Hotel"], {
|
|
"__kg_node_id": "guiyang-hotel-001", "name": "贵阳测试酒店",
|
|
"lng": None, "lat": None, "longitude": 106.63, "latitude": 26.65,
|
|
}]]
|
|
with patch("app.api.plaza._map_falkor_client") as factory:
|
|
factory.return_value.select_graph.return_value = graph
|
|
detail = _read_graph_poi_detail("guiyang_import", items[0]["id"])
|
|
self.assertEqual(detail["id"], items[0]["id"])
|
|
self.assertEqual((detail["lng"], detail["lat"]), (106.63, 26.65))
|
|
self.assertEqual(graph.query.call_args.args[1]["place_id"], "guiyang-hotel-001")
|
|
factory.return_value.close.assert_called_once()
|
|
|
|
def test_map_read_pages_before_filtering_and_merges_duplicate_poi_facets(self) -> None:
|
|
graph = MagicMock()
|
|
helper_rows = [[index, ["GeoCell"], {"lng": 106.6, "lat": 26.6}] for index in range(2000)]
|
|
business_rows = [
|
|
[2000, ["Hotel"], {"element_id": "amap:H1", "name": "同一地点", "lng": 106.6, "lat": 26.6}],
|
|
[2001, ["FoodPlace"], {"element_id": "amap:H1", "name": "同一地点", "lng": 106.6, "lat": 26.6}],
|
|
]
|
|
graph.query.side_effect = [
|
|
MagicMock(result_set=helper_rows), MagicMock(result_set=business_rows),
|
|
]
|
|
with patch("app.api.plaza._map_falkor_client") as factory:
|
|
factory.return_value.select_graph.return_value = graph
|
|
items = _read_graph_map_pois("fresh_guiyang")
|
|
self.assertEqual(len(items), 1)
|
|
self.assertEqual(items[0]["categories"], ["酒店", "美食"])
|
|
self.assertIn("SKIP 2000 LIMIT 2000", graph.query.call_args.args[0])
|
|
|
|
def test_graph_pois_keep_business_places_out_of_the_bus_route_layer(self) -> None:
|
|
rows = [
|
|
[
|
|
"poi:hotel-1",
|
|
["Hotel"],
|
|
{
|
|
"element_id": "amap:H1",
|
|
"gaode_poi_id": "H1",
|
|
"name": "荔波测试酒店",
|
|
"type_label": "酒店",
|
|
"place_type": "hotel",
|
|
"lng": 107.88,
|
|
"lat": 25.40,
|
|
"towncode": "522722100000",
|
|
"zone_id": "A01",
|
|
},
|
|
],
|
|
[
|
|
"poi:stop-1",
|
|
["BusStop"],
|
|
{
|
|
"element_id": "amap:S1",
|
|
"gaode_poi_id": "S1",
|
|
"name": "汽车站",
|
|
"type_label": "公交站",
|
|
"place_type": "transit_stop",
|
|
"lng": 107.89,
|
|
"lat": 25.41,
|
|
},
|
|
],
|
|
]
|
|
|
|
items = _graph_map_poi_items(rows)
|
|
|
|
self.assertEqual(len(items), 1)
|
|
self.assertEqual(items[0]["categories"], ["酒店"])
|
|
self.assertEqual(items[0]["zone_id"], "A01")
|
|
|
|
def test_city_place_nodes_use_place_type_as_dynamic_business_category(self) -> None:
|
|
items = _graph_map_poi_items(
|
|
[[
|
|
None,
|
|
["Place"],
|
|
{
|
|
"element_id": "city:mall-1",
|
|
"name": "测试商场",
|
|
"place_type": "mall",
|
|
"longitude": 106.63,
|
|
"latitude": 26.65,
|
|
"adcode": "520102",
|
|
},
|
|
]]
|
|
)
|
|
|
|
self.assertEqual(items[0]["category"], "商场")
|
|
self.assertEqual(items[0]["categories"], ["商场"])
|
|
self.assertEqual(items[0]["lng"], 106.63)
|
|
self.assertEqual(items[0]["adcode"], "520102")
|
|
|
|
def test_coordinate_helper_nodes_are_not_rendered_as_business_pois(self) -> None:
|
|
items = _graph_map_poi_items(
|
|
[[
|
|
"cell-1",
|
|
["GeoCell"],
|
|
{"name": "网格", "lng": 106.63, "lat": 26.65},
|
|
]]
|
|
)
|
|
|
|
self.assertEqual(items, [])
|
|
|
|
def test_explicit_semantic_only_node_is_excluded_but_legacy_poi_is_kept(self) -> None:
|
|
properties = {
|
|
"element_id": "amap:semantic-place", "name": "语义地点",
|
|
"lng": 106.63, "lat": 26.65, "place_type": "sight",
|
|
}
|
|
items = _graph_map_poi_items([
|
|
[1, ["ScenicSpot"], {**properties, "map_poi": False}],
|
|
[2, ["ScenicSpot"], {**properties, "element_id": "amap:legacy-place"}],
|
|
])
|
|
self.assertEqual([item["id"] for item in items], ["amap:legacy-place"])
|
|
|
|
def test_merged_semantic_labels_do_not_change_verified_map_categories(self) -> None:
|
|
items = _graph_map_poi_items([[
|
|
1, ["Place", "ScenicSpot", "FoodPlace"],
|
|
{
|
|
"element_id": "amap:merged", "name": "多业态地点",
|
|
"lng": 106.6, "lat": 26.6, "type_label": "景点",
|
|
"map_poi": True, "map_source_labels": ["Place", "ScenicSpot"],
|
|
},
|
|
]])
|
|
self.assertEqual(items[0]["categories"], ["景点"])
|
|
|
|
def test_bulk_graph_map_query_only_returns_lightweight_projection(self) -> None:
|
|
projected_row = [
|
|
7,
|
|
["Hotel"],
|
|
"amap:H7",
|
|
"H7",
|
|
"荔波轻量酒店",
|
|
"酒店",
|
|
"hotel",
|
|
None,
|
|
"107.88",
|
|
None,
|
|
"25.40",
|
|
"测试地址",
|
|
"荔波县",
|
|
"黔南布依族苗族自治州",
|
|
"贵州省",
|
|
"522722",
|
|
"522722100000",
|
|
"玉屏街道",
|
|
"A01",
|
|
"古城片区",
|
|
True,
|
|
["Place", "Hotel"],
|
|
]
|
|
graph = MagicMock()
|
|
graph.query.return_value = MagicMock(result_set=[projected_row])
|
|
client = MagicMock()
|
|
client.select_graph.return_value = graph
|
|
|
|
with patch("app.api.plaza.FalkorDB", return_value=client):
|
|
items = _read_graph_map_pois("server_import_graph")
|
|
|
|
self.assertEqual(len(items), 1)
|
|
self.assertEqual(items[0]["name"], "荔波轻量酒店")
|
|
self.assertEqual(items[0]["lng"], 107.88)
|
|
self.assertEqual(items[0]["lat"], 25.40)
|
|
query = graph.query.call_args.args[0]
|
|
self.assertNotIn("properties(n)", query)
|
|
self.assertIn("n.element_id", query)
|
|
self.assertIn("n.map_poi IS NULL OR n.map_poi = true", query)
|
|
self.assertEqual(graph.query.call_args.kwargs["timeout"], 30_000)
|
|
client.close.assert_called_once_with()
|
|
|
|
def test_invalid_graph_coordinates_are_not_returned_to_the_map(self) -> None:
|
|
items = _graph_map_poi_items(
|
|
[[
|
|
"poi:bad-coordinate",
|
|
["Hotel"],
|
|
{"name": "错误坐标", "lng": "not-a-number", "lat": 25.4},
|
|
]]
|
|
)
|
|
|
|
self.assertEqual(items, [])
|
|
|
|
def test_legacy_city_project_uses_its_explicit_spatial_companion(self) -> None:
|
|
self.assertEqual(_resolve_spatial_graph_name("guiyang_new2"), "guiyang_spatial_v1")
|
|
self.assertEqual(_resolve_spatial_graph_name("city_import_v3"), "city_import_v3")
|
|
|
|
def test_graph_zones_decode_serialized_geometry(self) -> None:
|
|
payload = _graph_zone_payload(
|
|
[[{"towncode": "522722100000", "name": "玉屏街道"}, 12, 107.8, 25.3, 107.9, 25.5]],
|
|
[[{"zone_id": "A01", "zone_name": "古城片区", "geometry": "[[[107.8,25.3],[107.9,25.4]]]"}, 6]],
|
|
"yunyou_libo_full_v2",
|
|
)
|
|
|
|
self.assertEqual(payload["towns"][0]["poi_count"], 12)
|
|
self.assertEqual(payload["towns"][0]["bbox"], [107.8, 25.3, 107.9, 25.5])
|
|
self.assertIsInstance(payload["zones"][0]["geometry"], list)
|
|
self.assertEqual(payload["zones"][0]["poi_count"], 6)
|
|
|
|
def test_graph_detail_uses_embedded_json_images(self) -> None:
|
|
payload = _graph_poi_detail_payload(
|
|
"yunyou_libo_full_v2",
|
|
["ScenicSpot"],
|
|
{
|
|
"element_id": "amap:P1",
|
|
"gaode_poi_id": "P1",
|
|
"name": "小七孔",
|
|
"type_label": "景点",
|
|
"lng": 107.72,
|
|
"lat": 25.25,
|
|
"images": '[{"image_url":"https://example.com/a.jpg"}]',
|
|
"spatial_raw_data": '{"visitor_value":"世界遗产"}',
|
|
},
|
|
)
|
|
|
|
self.assertEqual(payload["visitor_value"], "世界遗产")
|
|
self.assertEqual(payload["photo_urls"], ["https://example.com/a.jpg"])
|
|
self.assertEqual(payload["graph_labels"], ["ScenicSpot"])
|
|
|
|
def test_normalized_bus_rows_build_ordered_route_payload(self) -> None:
|
|
rows = [
|
|
["r1", "荔波1路", "上行", "甲", "乙", 2, 2, "07:00", "18:00", 5, "station_sequence", "s1", "甲", 107.8, 25.3, 1],
|
|
["r1", "荔波1路", "上行", "甲", "乙", 2, 2, "07:00", "18:00", 5, "station_sequence", "s2", "乙", 107.9, 25.4, 2],
|
|
]
|
|
|
|
payload = _build_libo_bus_route_payload(rows, "yunyou_libo_full_v2")
|
|
|
|
self.assertEqual(payload["line_count"], 1)
|
|
self.assertEqual(payload["direction_count"], 1)
|
|
self.assertEqual(payload["stop_count"], 2)
|
|
self.assertEqual(payload["route_stop_count"], 2)
|
|
|
|
def test_graph_detail_preserves_enrichment_photos(self) -> None:
|
|
payload = _graph_poi_detail_payload("imported", ["Hotel"], {
|
|
"element_id": "hotel-1", "lng": 106.6, "lat": 26.6,
|
|
"photo_urls": ["https://example.com/original.jpg"],
|
|
"dianping_shop_image": "https://example.com/food.jpg",
|
|
"hotel_image_samples": '["https://example.com/hotel.jpg","https://example.com/original.jpg"]',
|
|
})
|
|
self.assertEqual(payload["photo_urls"], [
|
|
"https://example.com/original.jpg", "https://example.com/food.jpg",
|
|
"https://example.com/hotel.jpg",
|
|
])
|
|
|
|
|
|
class PlazaMapEndpointSourceTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_detail_connection_failure_is_not_reported_as_missing_poi(self) -> None:
|
|
context = ProjectContext("city", "city", "fresh_guiyang")
|
|
with patch("app.api.plaza._read_graph_poi_detail", side_effect=RuntimeError("unavailable")):
|
|
with self.assertRaises(HTTPException) as raised:
|
|
await map_poi_detail("hotel-1", context=context)
|
|
self.assertEqual(raised.exception.status_code, 503)
|
|
|
|
async def test_poi_detail_does_not_require_a_postgresql_poi_table(self) -> None:
|
|
context = ProjectContext("city", "city_guizhou_guiyang", "fresh_guiyang")
|
|
with (
|
|
patch("app.api.plaza.get_conn", side_effect=AssertionError("No PostgreSQL POI table")),
|
|
patch("app.api.plaza._read_graph_poi_detail", return_value={"id": "hotel-1", "name": "贵阳酒店"}) as read,
|
|
):
|
|
detail = await map_poi_detail("hotel-1", context=context)
|
|
read.assert_called_once_with("fresh_guiyang", "hotel-1")
|
|
self.assertEqual(detail["name"], "贵阳酒店")
|
|
|
|
async def test_missing_poi_detail_is_not_taken_from_another_project(self) -> None:
|
|
context = ProjectContext("city", "city", "fresh_guiyang")
|
|
with patch("app.api.plaza._read_graph_poi_detail", return_value=None):
|
|
with self.assertRaises(HTTPException) as raised:
|
|
await map_poi_detail("missing", context=context)
|
|
self.assertEqual(raised.exception.status_code, 404)
|
|
|
|
async def test_map_endpoint_uses_current_falkordb_as_single_source(
|
|
self,
|
|
) -> None:
|
|
graph_items = [
|
|
{
|
|
"id": "amap:H1",
|
|
"gaode_poi_id": "H1",
|
|
"name": "荔波测试酒店",
|
|
"category": "酒店",
|
|
"categories": ["酒店"],
|
|
"place_type": "hotel",
|
|
"lng": 107.88,
|
|
"lat": 25.40,
|
|
"address": "",
|
|
"district": "荔波县",
|
|
"city": "黔南布依族苗族自治州",
|
|
"province": "贵州省",
|
|
"adcode": "522722",
|
|
"towncode": "",
|
|
"town_name": "",
|
|
"zone_id": "",
|
|
"zone_name": "",
|
|
}
|
|
]
|
|
context = ProjectContext("tenant", "project", "server_import_graph")
|
|
graph_read = AsyncMock(return_value=graph_items)
|
|
|
|
with (
|
|
patch(
|
|
"app.api.plaza.get_conn",
|
|
side_effect=AssertionError("地图运行时不应读取 PostgreSQL POI"),
|
|
) as relational_read,
|
|
patch("app.api.plaza.asyncio.to_thread", graph_read),
|
|
):
|
|
payload = await map_pois(context=context)
|
|
|
|
self.assertEqual(payload["source"], "falkordb")
|
|
self.assertEqual(payload["graph_name"], "server_import_graph")
|
|
self.assertEqual(payload["total"], 1)
|
|
self.assertEqual(payload["categories"], [{"category": "酒店", "count": 1}])
|
|
relational_read.assert_not_called()
|
|
graph_read.assert_awaited_once_with(_read_graph_map_pois, "server_import_graph")
|
|
|
|
async def test_map_endpoint_reports_graph_read_failure_instead_of_empty_data(
|
|
self,
|
|
) -> None:
|
|
context = ProjectContext("tenant", "project", "server_import_graph")
|
|
|
|
with patch(
|
|
"app.api.plaza.asyncio.to_thread",
|
|
AsyncMock(side_effect=RuntimeError("FalkorDB timeout")),
|
|
):
|
|
with self.assertRaises(HTTPException) as raised:
|
|
await map_pois(context=context)
|
|
|
|
self.assertEqual(raised.exception.status_code, 503)
|
|
self.assertIn("FalkorDB", raised.exception.detail)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|