Serialize OCSP Redis snapshots with CBOR

server/tests/test_routes_extra.py

@@ -1216,6 +1216,7 @@ class OcspSnapshotTests(unittest.TestCase):
 
     @async_test
     async def test_publisher_marks_dirty_and_uploads(self) -> None:
+        import cbor2
         from dataclasses import replace
         from unittest.mock import patch
 
@@ -1252,8 +1253,9 @@ class OcspSnapshotTests(unittest.TestCase):
                 assert len(fake.calls) == 1
                 call = fake.calls[0]
                 assert call["name"] == f"ocsp:{ca['id']}"
-                assert b"kca-ocsp-status/v2" in call["value"]
-                assert b"generated_at" not in call["value"]
+                published = cbor2.loads(call["value"])
+                assert published["schema"] == "kca-ocsp-status/v2"
+                assert "generated_at" not in published
                 assert call["ex"] == 60
                 row = await OcspSnapshot.get(ca_id=ca["id"])
                 assert row.dirty is False
@@ -1334,7 +1336,7 @@ class OcspSnapshotTests(unittest.TestCase):
 
     @async_test
     async def test_imported_ca_cannot_overwrite_other_vault_snapshot(self) -> None:
-        import json
+        import cbor2
         from dataclasses import replace
         from unittest.mock import patch
 
@@ -1414,9 +1416,9 @@ class OcspSnapshotTests(unittest.TestCase):
         assert imported_key != target_key
         assert set(fake.objects) == {target_key, imported_key}
         assert fake.expiries == {target_key: 60, imported_key: 60}
-        target_snapshot = json.loads(fake.objects[target_key])
+        target_snapshot = cbor2.loads(fake.objects[target_key])
         target_entries = {entry["serial_hex"]: entry for entry in target_snapshot["entries"]}
         assert target_entries[leaf["serial_hex"].lower()]["status"] == "revoked"
-        imported_snapshot = json.loads(fake.objects[imported_key])
+        imported_snapshot = cbor2.loads(fake.objects[imported_key])
         imported_serials = {entry["serial_hex"] for entry in imported_snapshot["entries"]}
         assert leaf["serial_hex"].lower() not in imported_serials