Batch certificate summary lookups

2026-10-27 14:09:00 +00:00
parent a2457ab1f4
commit 0301cbff6f
3 changed files with 129 additions and 7 deletions
@@ -604,6 +604,77 @@ class ExtendedRouteTests(unittest.TestCase):
                assert response.json()["reason"] == "superseded"


+class CertificateListTests(unittest.TestCase):
+    @async_test
+    async def test_summaries_batch_revocation_and_private_key_queries(self) -> None:
+        from unittest.mock import patch
+
+        from kca import service
+        from kca.models import CertificateRevocation, Key
+
+        async with api_client() as client:
+            with oidc_user("alice"):
+                vault_id = await _vault(client)
+                url = f"/api/vaults/{vault_id}/certificates"
+
+                with (
+                    patch.object(
+                        CertificateRevocation,
+                        "filter",
+                        wraps=CertificateRevocation.filter,
+                    ) as revocation_filter,
+                    patch.object(Key, "filter", wraps=Key.filter) as key_filter,
+                ):
+                    response = await client.get(url)
+                assert response.status_code == 200, response.text
+                assert response.json()["results"] == []
+                revocation_filter.assert_not_called()
+                key_filter.assert_not_called()
+
+                ca_key = await _key(client, vault_id, "list ca")
+                ca = await _cert(client, vault_id, ca_key["id"], "List CA", ca=True)
+                leaf_key = await _key(client, vault_id, "list leaf")
+                leaf = await _cert(
+                    client,
+                    vault_id,
+                    leaf_key["id"],
+                    "List Leaf",
+                    signer_id=ca["id"],
+                )
+                response = await client.post(
+                    f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke",
+                    json={"reason": "superseded"},
+                )
+                assert response.status_code == 200, response.text
+
+                with (
+                    patch.object(
+                        CertificateRevocation,
+                        "filter",
+                        wraps=CertificateRevocation.filter,
+                    ) as revocation_filter,
+                    patch.object(Key, "filter", wraps=Key.filter) as key_filter,
+                    patch.object(
+                        service,
+                        "revoked_for",
+                        side_effect=AssertionError("per-certificate revocation query"),
+                    ) as single_revocation,
+                ):
+                    response = await client.get(url)
+
+                assert response.status_code == 200, response.text
+                revocation_filter.assert_called_once()
+                key_filter.assert_called_once()
+                single_revocation.assert_not_called()
+                summaries = {row["id"]: row for row in response.json()["results"]}
+                assert summaries[ca["id"]]["revoked"] is False
+                assert summaries[ca["id"]]["revoked_at"] is None
+                assert summaries[ca["id"]]["has_private_key"] is True
+                assert summaries[leaf["id"]]["revoked"] is True
+                assert summaries[leaf["id"]]["revoked_at"] is not None
+                assert summaries[leaf["id"]]["has_private_key"] is True
+
+
 class SerialRenewalTests(unittest.TestCase):
    @async_test
    async def test_explicit_serial_must_be_unique_and_not_revoked(self) -> None: