fix: make CORS reconciliation recoverable

tests/test_contract.py

@@ -131,6 +131,8 @@ class ConfigContractTests(unittest.TestCase):
             "https://consumer.example:443",
             "https://consumer.example/",
             "https://consumer.example:invalid",
+            "https://[fe80::1%eth0]",
+            "https://[fe80::1%25eth0]",
         ):
             with self.subTest(origin=origin):
                 self.assert_invalid(
@@ -427,6 +429,51 @@ class PublishingTests(unittest.TestCase):
             )
         self.assertIn("delete-bucket-cors", request.call_args.args[0])
 
+    def test_cors_reconciliation_restores_prior_policies_on_failure(self):
+        artifacts = [
+            {
+                "bucket": "first", "cors_origins": ["https://new.example"],
+                "s3_endpoint": "http://garage-s3.storage.svc:3900", "credentials": {},
+            },
+            {
+                "bucket": "second", "cors_origins": [],
+                "s3_endpoint": "http://garage-s3.storage.svc:3900", "credentials": {},
+            },
+        ]
+        prior = [
+            {"CORSRules": [{"AllowedOrigins": ["https://old.example"]}]},
+            None,
+        ]
+        writes = []
+
+        def write(bucket, config, *_args):
+            writes.append((bucket, config))
+            if bucket == "second" and len(writes) == 2:
+                raise RuntimeError("write failed")
+
+        with patch.object(deploy, "publication_aws_env", return_value={}), patch.object(
+            deploy, "get_cors_configuration", side_effect=prior,
+        ) as read, patch.object(deploy, "set_cors_configuration", side_effect=write), \
+                self.assertRaisesRegex(RuntimeError, "write failed"):
+            deploy.reconcile_artifact_cors(artifacts)
+
+        self.assertEqual(2, read.call_count)
+        self.assertEqual(
+            [
+                ("first", {"CORSRules": [{
+                    "AllowedOrigins": ["https://new.example"],
+                    "AllowedMethods": ["GET", "HEAD"],
+                    "AllowedHeaders": ["*"],
+                    "ExposeHeaders": ["ETag"],
+                    "MaxAgeSeconds": 3600,
+                }]}),
+                ("second", None),
+                ("second", None),
+                ("first", prior[0]),
+            ],
+            writes,
+        )
+
     def test_apps_clone_never_places_token_in_argv_or_log(self):
         calls = []
         secret = "clone-secret-must-not-appear"
@@ -732,15 +779,15 @@ class PublishingTests(unittest.TestCase):
                         deploy, "publish_route_immutables",
                         side_effect=[None, RuntimeError("immutable failed")],
                     ) as immutable_publish, patch.object(
-                        deploy, "configure_artifact_cors"
-                    ) as cors_configure, patch.object(deploy, "s3_sync") as mutable_sync, \
+                        deploy, "reconcile_artifact_cors"
+                    ) as cors_reconcile, patch.object(deploy, "s3_sync") as mutable_sync, \
                     self.assertRaisesRegex(
                 RuntimeError, "immutable failed"
             ):
                 deploy.deploy_static("baseline", root, root, "token", cfg)
 
             self.assertEqual(2, immutable_publish.call_count)
-            cors_configure.assert_not_called()
+            cors_reconcile.assert_not_called()
             mutable_sync.assert_not_called()
 
     def test_all_cors_policies_complete_before_mutable_publication(self):
@@ -755,7 +802,7 @@ class PublishingTests(unittest.TestCase):
             ), patch.object(
                 deploy, "publish_route_immutables", side_effect=lambda *_args: events.append("immutable")
             ), patch.object(
-                deploy, "configure_artifact_cors", side_effect=lambda *_args: events.append("cors")
+                deploy, "reconcile_artifact_cors", side_effect=lambda *_args: events.append("cors")
             ), patch.object(
                 deploy, "s3_sync", side_effect=lambda *_args: events.append("mutable")
             ), patch.object(deploy, "clone_apps", return_value=apps), patch.object(
@@ -764,7 +811,7 @@ class PublishingTests(unittest.TestCase):
                 deploy.deploy_static("baseline", root, root, "token", cfg)
 
         self.assertEqual(
-            ["immutable", "immutable", "cors", "cors", "mutable", "mutable"], events
+            ["immutable", "immutable", "cors", "mutable", "mutable"], events
         )