#!/usr/bin/env python3 """Local-only map regression: real graphs + a fresh JSON-imported scratch graph. Existing graphs are read-only. Only the UUID graph created by this invocation is written/deleted; PostgreSQL POI access is deliberately forbidden. """ from __future__ import annotations import argparse import asyncio import json from pathlib import Path import sys import time import uuid from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from fastapi import FastAPI from fastapi.testclient import TestClient from app.api import plaza from app.auth import get_current_user from app.config import settings from app.project_context import ProjectContext from app.project_lifecycle import _falkor_client, import_falkor_graph, normalize_provision_payload def map_result(graph_name): with patch("app.api.plaza.get_conn", side_effect=AssertionError("Map reads must not require PostgreSQL POIs")): return asyncio.run(plaza.map_pois(context=ProjectContext("local_map_test", graph_name, graph_name))) def verify(map_json: Path | None = None): if settings.falkordb_host.lower() not in {"localhost", "127.0.0.1", "::1"}: raise RuntimeError("This test is restricted to the local FalkorDB service") report = {"test": "graph-backed-map", "read_only_existing_graphs": True} source_nodes = None if map_json is not None: with map_json.open(encoding="utf-8") as stream: source_nodes = json.load(stream)["nodes"] assert len(source_nodes) > 2000, "Provide a map JSON with more than 2,000 POIs" for graph_name in ("yunyou_libo", "guiyang_new2"): start = time.monotonic() payload = map_result(graph_name) ids = {item["id"] for item in payload["items"]} assert len(ids) == payload["total"], "Duplicate map POI ids" assert payload["source"] == "falkordb" city_count = sum(str(item["adcode"]).startswith("5201") for item in payload["items"]) report[graph_name] = { "graph": payload["graph_name"], "points": payload["total"], "guiyang_city_points": city_count, "categories": payload["categories"], "elapsed_seconds": round(time.monotonic() - start, 3), } if graph_name == "yunyou_libo": assert payload["total"] == 2799, "Libo point count changed" else: assert city_count > 0, "Guiyang map has no points in its selected city" if source_nodes is not None: source_ids = {str(node["properties"].get("element_id") or node["id"]) for node in source_nodes} assert not source_ids - ids, "Exported POIs are missing from the local source graph" report["export_snapshot"] = { "points": len(source_ids), "missing_from_local_graph": 0, "extra_points_in_local_graph": len(ids - source_ids), } client = _falkor_client() scratch_name = "codex_map_verify_" + uuid.uuid4().hex created = False try: existing = {name.decode() if isinstance(name, bytes) else str(name) for name in client.list_graphs()} assert scratch_name not in existing source_graph = client.select_graph("guiyang_spatial_v1") if source_nodes is None: rows = source_graph.ro_query( "MATCH (n) WHERE n.element_id IS NOT NULL " "AND (n.lng IS NOT NULL OR n.longitude IS NOT NULL) " "AND (n.lat IS NOT NULL OR n.latitude IS NOT NULL) " "AND n.adcode STARTS WITH '5201' RETURN labels(n),properties(n) LIMIT 2600", timeout=30_000, ).result_set else: rows = [(node.get("labels") or [node["type"]], node["properties"]) for node in source_nodes[:2600]] nodes, seen, entity_types = [], set(), {} for labels, properties in rows: external_id = str(properties["element_id"]) if external_id in seen: continue seen.add(external_id) for label in labels: entity_types[label] = {"fields": {"name": {"type": "string", "required": True}}} nodes.append({"id": external_id, "type": labels[0], "labels": labels, "properties": properties}) assert len(nodes) > 2000, "Test must cross the map query page boundary" body = { "project_id": scratch_name, "display_name": "城市知识图谱-贵州*贵阳", "spatial_map": {"enabled": True, "scope": "guizhou", "region_name": "贵阳市", "region_adcode": "520100", "region_level": "city"}, "schema": {"namespace": scratch_name, "version": "1.0.0", "entity_types": entity_types, "relation_types": {}}, "graph_data": {"nodes": nodes, "relations": []}, } normalized = normalize_provision_payload(body) expected = {item["id"]: item for item in plaza._graph_map_poi_items( [[node["id"], node["labels"], node["properties"]] for node in nodes])} created = True counts = import_falkor_graph(scratch_name, normalized["graph_data"]) assert counts == {"nodes": len(nodes), "relations": 0} actual = map_result(scratch_name) indexed = {item["id"]: item for item in actual["items"]} assert indexed == expected, "Imported map fields do not round-trip intact" # Exercise real HTTP routing/serialization as well as direct readers. app = FastAPI() app.include_router(plaza.router, prefix="/v1/admin") app.dependency_overrides[get_current_user] = lambda: {"username": "local_map_test", "roles": ["admin"]} headers = {"X-KG-Tenant-Id": scratch_name, "X-KG-Project-Id": scratch_name, "X-KG-Graph-Name": scratch_name} with TestClient(app) as http, patch( "app.api.plaza.get_conn", side_effect=AssertionError("No PostgreSQL POI rows in a new import") ): response = http.get("/v1/admin/plaza/map-pois", headers=headers) assert response.status_code == 200, response.text[:300] assert response.json()["total"] == len(expected) sample = next(iter(expected.values())) detail = http.get("/v1/admin/plaza/map-pois/" + sample["id"], headers=headers) assert detail.status_code == 200, detail.text[:300] data = detail.json() assert (data["id"], data["name"], data["lng"], data["lat"]) == ( sample["id"], sample["name"], sample["lng"], sample["lat"]) report["fresh_json_import"] = { "display_name": body["display_name"], "region": body["spatial_map"], "nodes": counts["nodes"], "map_pois": len(expected), "http_map_status": 200, "http_detail_status": 200, "map_payload_matches_input": True, "postgres_pois_required": False, } report["status"] = "passed" finally: try: if created: remaining = {name.decode() if isinstance(name, bytes) else str(name) for name in client.list_graphs()} if scratch_name in remaining: client.select_graph(scratch_name).delete() report["scratch_graph_removed"] = True finally: client.close() return report if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--map-json", type=Path, help="Optional exported map POI JSON used for the import sample") args = parser.parse_args() print(json.dumps(verify(args.map_json), ensure_ascii=False, indent=2))