156 lines
7.6 KiB
Python
156 lines
7.6 KiB
Python
#!/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))
|