113 lines
5.8 KiB
Python
113 lines
5.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
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
|
|
|
|
|
|
class ProviderPipelineContractTests(unittest.TestCase):
|
|
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()
|