199 lines
12 KiB
Python
199 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import importlib.util
|
|
import hashlib
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "tools/provider-policy-compiler"))
|
|
from provider_policy import DELIVERY_FILES, build_policy, create_delivery_zip, export_package, fingerprint, validate_package, validate_policy
|
|
|
|
spec = importlib.util.spec_from_file_location("pipeline", ROOT / "tools/provider-pipeline/run_provider_pipeline.py")
|
|
pipeline = importlib.util.module_from_spec(spec)
|
|
assert spec and spec.loader
|
|
spec.loader.exec_module(pipeline)
|
|
|
|
|
|
class ProviderPipelineContractTests(unittest.TestCase):
|
|
def test_v2_full_catalog_evidence_and_fail_closed(self):
|
|
style = {"layers": [{"id": "layer-a", "source-layer": "points"}, {"id": "layer-b", "source-layer": "points"}]}
|
|
components = [
|
|
{"style_layer_id": "layer-a", "style_index": 0, "role": "icon", "resource_key": "ICON", "resource_visual_hash": "a" * 64},
|
|
{"style_layer_id": "layer-b", "style_index": 1, "role": "text", "resource_key": None, "resource_visual_hash": None},
|
|
]
|
|
catalog = {"visual_objects": [{"visual_object_id": "VO-a", "provenance": {"source_layer": "points"}, "geometry_role": "Point", "occurrence_count": 123, "components": components}]}
|
|
rows = [{"visualObjectId": "VO-a", "status": "LAYER_ONLY", "standardLayer": "navigation_marks"}, {"visualObjectId": "VO-missing", "status": "UNKNOWN", "standardLayer": None}]
|
|
self.assertEqual(pipeline.attach_render_evidence(rows, catalog), ["VO-missing"])
|
|
self.assertIsNone(rows[1]["renderEvidence"])
|
|
self.assertEqual([x["role"] for x in rows[0]["renderEvidence"]["components"]], ["icon", "text"])
|
|
self.assertIsNone(rows[0]["renderEvidence"]["components"][1]["resourceKey"])
|
|
self.assertEqual(rows[0]["renderEvidence"]["occurrenceCount"], 123)
|
|
from provider_policy import validate_mapping
|
|
mapping = {"schemaVersion": "navsea-standard-layer-mapping/v2", "renderEvidenceSource": {"schema": "actual-visual-object-catalog/v1"}, "assignments": rows}
|
|
self.assertEqual(validate_mapping(mapping, style), [])
|
|
for key, value, expected in (("styleLayerId", "missing", "styleLayerId:missing"), ("styleIndex", 9, "styleIndex:mismatch")):
|
|
rows[0]["renderEvidence"]["components"][0][key] = value
|
|
self.assertTrue(any(expected in x for x in validate_mapping(mapping, style)))
|
|
rows[0]["renderEvidence"]["components"][0][key] = components[0][{"styleLayerId": "style_layer_id", "styleIndex": "style_index"}[key]]
|
|
rows[0]["renderEvidence"]["sourceLayer"] = "wrong"
|
|
self.assertTrue(any("sourceLayer:mismatch" in x for x in validate_mapping(mapping, style)))
|
|
rows[0]["renderEvidence"]["sourceLayer"] = "points"
|
|
rows[0]["renderEvidence"]["geometryRole"] = "eval(1)"
|
|
self.assertTrue(any("script_forbidden" in x for x in validate_mapping(mapping, style)))
|
|
rows[0]["renderEvidence"]["geometryRole"] = "Point"
|
|
duplicate = json.loads(json.dumps(rows[0]))
|
|
duplicate["renderEvidence"]["occurrenceCount"] = 124
|
|
mapping["assignments"].append(duplicate)
|
|
self.assertTrue(any("nondeterministic_duplicate" in x for x in validate_mapping(mapping, style)))
|
|
|
|
def test_execution_catalog_style_hash_and_schema(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
base = Path(tmp)
|
|
style = base / "style.json"
|
|
style.write_bytes(b"{}")
|
|
catalog = base / "catalog.json"
|
|
catalog.write_text(json.dumps({"schema": "actual-visual-object-catalog/v1", "inputs": {"style": str(style)}, "summary": {}, "visual_objects": []}))
|
|
manifest = base / "manifest.json"
|
|
expected = hashlib.sha256(b"{}").hexdigest()
|
|
self.assertEqual(pipeline.load_execution_catalog(manifest, "catalog.json", expected)[1]["styleSha256"], expected)
|
|
with self.assertRaisesRegex(ValueError, "SHA-256"):
|
|
pipeline.load_execution_catalog(manifest, "catalog.json", "0" * 64)
|
|
|
|
def test_cm95_existing_semantics_and_specific_visual_objects(self):
|
|
base = ROOT / "dist/cm95/s52-full/1.0.0/mapping-layer-assignment.json"
|
|
catalog_file = ROOT / "tools/visual-object-instantiation/output/cm95-current/actual-anonymous-catalog.json"
|
|
if not base.exists() or not catalog_file.exists():
|
|
self.skipTest("本机未提供 CM95 样本与全量 Catalog")
|
|
previous = json.loads(base.read_text())["assignments"]
|
|
rows = json.loads(json.dumps(previous))
|
|
catalog = json.loads(catalog_file.read_text())
|
|
self.assertEqual(len(rows), 102)
|
|
self.assertEqual(pipeline.attach_render_evidence(rows, catalog), [])
|
|
self.assertEqual([(x["visualObjectId"], x["status"], x["standardLayer"]) for x in rows], [(x["visualObjectId"], x["status"], x["standardLayer"]) for x in previous])
|
|
boy = next(x for x in rows if x["visualObjectId"] == "VO-2329594761200ffd")
|
|
self.assertEqual(boy["standardLayer"], "navigation_marks")
|
|
self.assertEqual(boy["renderEvidence"]["sourceLayer"], "s52_points")
|
|
self.assertTrue(any(c["styleLayerId"] == "s52-p08-sy-s52-points" and c["role"] == "icon" and c["resourceKey"] == "BOYCON61" for c in boy["renderEvidence"]["components"]))
|
|
self.assertTrue(any(x["standardLayer"] == "navigation_marks" and len(x["renderEvidence"]["components"]) > 1 and any((c["resourceKey"] or "").startswith("LIGHTS") or c["role"] == "text" for c in x["renderEvidence"]["components"]) for x in rows))
|
|
|
|
def test_generated_mapping_policy_package_and_zip_agree(self):
|
|
base = ROOT / "dist/cm95/s52-full/1.1.0"
|
|
if not base.exists():
|
|
self.skipTest("正式 CM95 1.1.0 Package 尚未生成")
|
|
mapping = json.loads((base / "mapping-layer-assignment.json").read_text())
|
|
policy = json.loads((base / "provider-policy.json").read_text())
|
|
package = json.loads((base / "provider-package.json").read_text())
|
|
self.assertEqual(mapping["schemaVersion"], "navsea-standard-layer-mapping/v2")
|
|
self.assertEqual({k: v for k, v in mapping.items() if k != "source"}, policy["mapping"])
|
|
self.assertEqual(policy["mapping"], package["standardLayerMapping"])
|
|
z = base.parent / "cm95-s52-full-1.1.0.navsea-provider.zip"
|
|
with zipfile.ZipFile(z) as archive:
|
|
self.assertEqual(len(archive.namelist()), 9)
|
|
self.assertIsNone(archive.testzip())
|
|
first = fingerprint(mapping)
|
|
second = fingerprint(json.loads((base / "mapping-layer-assignment.json").read_text()))
|
|
self.assertEqual(first, second)
|
|
|
|
def data(self, provider_id: str) -> dict:
|
|
return {
|
|
"provider": {"providerId": provider_id, "name": provider_id, "version": "1"},
|
|
"dataset": {"datasetId": "dataset-1"},
|
|
"package": {"packageId": f"{provider_id}-package", "version": "1.0.0"},
|
|
"standardCatalog": {"id": "navsea-frozen-standard", "fingerprint": "a" * 64},
|
|
"resources": {"style": {"uri": "https://example.invalid/style.json"}, "pbfSamples": [{"tile": "1/0/0", "sha256": "b" * 64}]},
|
|
"mapping": [{"visualObjectId": "VO-a", "status": "LAYER_ONLY", "standardLayer": "navigation_marks", "candidateStandardLayers": [], "basis": "review"}],
|
|
"display": {}, "pickup": {}, "planning": {}, "danger": {}, "capabilities": {},
|
|
"validationRequirements": {"requireStyle": True, "requirePbfSample": True},
|
|
"unresolved": [], "limitations": [], "provenance": {},
|
|
"fingerprints": {"source": "c" * 64, "style": "d" * 64, "mapping": "e" * 64},
|
|
}
|
|
|
|
def test_fingerprint_is_key_order_independent(self):
|
|
self.assertEqual(fingerprint({"a": 1, "b": 2}), fingerprint({"b": 2, "a": 1}))
|
|
self.assertEqual(fingerprint({"snapshot": "/one", "generatedAt": "now", "sha256": "abc"}), fingerprint({"snapshot": "/two", "generatedAt": "later", "sha256": "abc"}))
|
|
|
|
def test_policy_schema_same_for_navsea_and_cm95(self):
|
|
navsea = build_policy(self.data("navsea"))
|
|
cm95 = build_policy(self.data("cm95"))
|
|
self.assertEqual(navsea["schemaVersion"], cm95["schemaVersion"])
|
|
self.assertEqual(set(navsea), set(cm95))
|
|
self.assertEqual(validate_policy(navsea)["result"], "PASS")
|
|
self.assertEqual(validate_policy(cm95)["result"], "PASS")
|
|
|
|
def test_ambiguous_requires_multiple_candidates(self):
|
|
policy = build_policy(self.data("provider-a"))
|
|
policy["mapping"]["assignments"] = [{"status": "AMBIGUOUS", "standardLayer": None, "candidateStandardLayers": ["one"]}]
|
|
self.assertEqual(validate_policy(policy)["result"], "FAIL")
|
|
|
|
def test_fail_prevents_export(self):
|
|
data = self.data("provider-a")
|
|
policy = build_policy(data)
|
|
package = {"providerPolicy": policy}
|
|
invalid = validate_package(package)
|
|
self.assertEqual(invalid["result"], "FAIL")
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
with self.assertRaises(ValueError):
|
|
export_package(Path(tmp) / "forbidden", package, policy, invalid)
|
|
self.assertFalse((Path(tmp) / "forbidden").exists())
|
|
|
|
def test_export_accepts_only_two_non_fail_results(self):
|
|
data = self.data("provider-a")
|
|
policy = build_policy(data)
|
|
package = {"provider": data["provider"], "dataset": data["dataset"], "package": data["package"], "fingerprints": {}, "limitations": []}
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
with self.assertRaises(ValueError):
|
|
export_package(Path(tmp) / "unknown", package, policy, {"result": "NOT_FAIL"})
|
|
|
|
def test_both_provider_manifests_share_policy_shape(self):
|
|
manifests = [json.loads((ROOT / f"providers/{key}/manifest.json").read_text()) for key in ("navsea", "cm95")]
|
|
self.assertEqual({x["mapping"]["mode"] for x in manifests}, {"frozen-standard-visual", "manual-assignment"})
|
|
for manifest in manifests:
|
|
self.assertTrue(manifest["provider"]["providerId"])
|
|
self.assertIn("style", manifest["resources"])
|
|
self.assertIn("sprite", manifest["resources"])
|
|
|
|
def test_delivery_zip_contains_only_required_navsea_files(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
export_dir = Path(tmp) / "1.0.0"
|
|
export_dir.mkdir()
|
|
for name in DELIVERY_FILES:
|
|
path = export_dir / name
|
|
if name == "validation.json":
|
|
path.write_text(json.dumps({"result": "PASS_WITH_WARNINGS"}), encoding="utf-8")
|
|
elif name.endswith(".png"):
|
|
path.write_bytes(b"png")
|
|
else:
|
|
path.write_text(name, encoding="utf-8")
|
|
(export_dir / "not-for-navsea.tmp").write_text("exclude me", encoding="utf-8")
|
|
info = create_delivery_zip(export_dir, "provider-a", "dataset-1", "1.0.0")
|
|
self.assertEqual(info["validationResult"], "PASS_WITH_WARNINGS")
|
|
self.assertEqual(len(info["sha256"]), 64)
|
|
with zipfile.ZipFile(info["path"], "r") as archive:
|
|
expected = [f"provider-a-dataset-1-1.0.0/{name}" for name in DELIVERY_FILES]
|
|
self.assertEqual(archive.namelist(), expected)
|
|
self.assertNotIn("provider-a-dataset-1-1.0.0/not-for-navsea.tmp", archive.namelist())
|
|
|
|
def test_delivery_zip_fails_when_required_file_is_missing(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
export_dir = Path(tmp) / "1.0.0"
|
|
export_dir.mkdir()
|
|
for name in DELIVERY_FILES:
|
|
if name == "resources-sprite.png":
|
|
continue
|
|
path = export_dir / name
|
|
if name == "validation.json":
|
|
path.write_text(json.dumps({"result": "PASS"}), encoding="utf-8")
|
|
else:
|
|
path.write_text(name, encoding="utf-8")
|
|
with self.assertRaises(ValueError):
|
|
create_delivery_zip(export_dir, "provider-a", "dataset-1", "1.0.0")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|