"""Retain useful Apple refusal evidence without inventing a cap or exposing response data.""" import contextlib import io import json import os import sys import unittest from pathlib import Path from unittest import mock import requests sys.path.insert(0, str(Path(__file__).resolve().parent)) import cert_factory # noqa: E402 class CertificateFailureTests(unittest.TestCase): def request(self, body): response = requests.Response() response.status_code = 409 response._content = body.encode() output = io.StringIO() with mock.patch.dict(os.environ, {"CERTIFICATE_CAP_POLICY": "fail"}, clear=True): with mock.patch.object(cert_factory, "request", return_value=response) as request: with contextlib.redirect_stdout(output), self.assertRaises(SystemExit) as failed: cert_factory.create_distribution_cert("private-token", "private-csr") request.assert_called_once() self.assertEqual(request.call_args.args[:2], ("POST", "/certificates")) text = output.getvalue() + str(failed.exception) self.assertNotIn("private-", text) return json.loads(output.getvalue().split("::", 2)[2]), str(failed.exception) def test_limit_requires_the_provider_limit_detail(self): for detail in ["Maximum number of certificates generated.", "You already have a current Distribution certificate or a pending certificate request."]: with self.subTest(detail=detail): result, message = self.request(json.dumps({"errors": [{ "code": "ENTITY_ERROR.ATTRIBUTE.INVALID", "detail": detail, "source": {"pointer": "private-source"}, "links": {"about": "private-url"}}]})) self.assertEqual(result["reason"], "certificate_limit") self.assertEqual(result["error_codes"], ["ENTITY_ERROR.ATTRIBUTE.INVALID"]) self.assertIn("refuses automatic revocation", message) def test_other_conflicts_are_not_reported_as_quota(self): cases = [{"errors": [{"code": "ENTITY_ERROR.ATTRIBUTE.INVALID", "detail": "Invalid private-csr"}]}, {"errors": [{"code": "private-secret", "detail": "private-response"}]}, {"errors": [{"code": ["private-value"]}]}, {"errors": "private-response"}, {"errors": ["private-response"]}, [], None] for case in cases: with self.subTest(case=case): result, message = self.request(json.dumps(case)) self.assertEqual(result["reason"], "unclassified_conflict") self.assertNotIn("cap reached", message) self.assertEqual(result["http_status"], 409) self.assertEqual(result["operation"], "certificates_createInstance") def test_non_json_response_is_an_unclassified_conflict(self): result, _ = self.request("private-response") self.assertEqual(result["reason"], "unclassified_conflict") self.assertEqual(result["error_codes"], []) if __name__ == "__main__": unittest.main()