beam: add canonical JSON serializer with tests

wizard-work/beam/json_canonical.py

@@ -0,0 +1,39 @@
+"""Canonical JSON serialization (RFC 8785-ish): sorted keys, compact separators,
+no NaN/Infinity, deterministic float repr, UTF-8, ensure_ascii=False."""
+from __future__ import annotations
+import json
+import math
+from typing import Any
+
+
+class CanonicalError(ValueError):
+    pass
+
+
+def _normalize(obj):
+    if obj is None or isinstance(obj, bool) or isinstance(obj, int) or isinstance(obj, str):
+        return obj
+    if isinstance(obj, float):
+        if not math.isfinite(obj):
+            raise CanonicalError("non-finite float: %r" % obj)
+        return obj
+    if isinstance(obj, (list, tuple)):
+        return [_normalize(x) for x in obj]
+    if isinstance(obj, dict):
+        out = {}
+        for k, v in obj.items():
+            if not isinstance(k, str):
+                raise CanonicalError("non-string key: %r" % (k,))
+            out[k] = _normalize(v)
+        return out
+    raise CanonicalError("unsupported type: %r" % (type(obj),))
+
+
+def canonical(obj) -> str:
+    n = _normalize(obj)
+    return json.dumps(n, sort_keys=True, separators=(",", ":"), ensure_ascii=False,
+                      allow_nan=False)
+
+
+def canonical_bytes(obj) -> bytes:
+    return canonical(obj).encode("utf-8")

wizard-work/beam/test_json_canonical.py

@@ -0,0 +1,39 @@
+import sys, os, json
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from json_canonical import canonical, canonical_bytes, CanonicalError
+
+def test_basic():
+    assert canonical({"b":2,"a":1}) == '{"a":1,"b":2}'
+    assert canonical([3,1,2]) == '[3,1,2]'
+    assert canonical(True) == 'true'
+    assert canonical(None) == 'null'
+    assert canonical("x") == '"x"'
+
+def test_nested():
+    assert canonical({"z":{"y":1,"x":2},"a":[1,2]}) == '{"a":[1,2],"z":{"x":2,"y":1}}'
+
+def test_unicode():
+    assert canonical({"é":1}) == '{"é":1}'
+
+def test_reject_nan():
+    try:
+        canonical(float('nan'))
+        raise AssertionError("expected error")
+    except CanonicalError:
+        pass
+
+def test_reject_nonstr_key():
+    try:
+        canonical({1:2})
+        raise AssertionError("expected error")
+    except CanonicalError:
+        pass
+
+def test_bytes():
+    assert canonical_bytes({"a":1}) == b'{"a":1}'
+
+def test_roundtrip():
+    o = {"a":[1,2,3],"b":{"c":True}}
+    assert json.loads(canonical(o)) == o
+
+print("ALL PASS")