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()