diff --git a/bot_bottle/contrib/gitea/deploy_key_provisioner.py b/bot_bottle/contrib/gitea/deploy_key_provisioner.py index 0ceadd7..9a4a9dc 100644 --- a/bot_bottle/contrib/gitea/deploy_key_provisioner.py +++ b/bot_bottle/contrib/gitea/deploy_key_provisioner.py @@ -19,7 +19,7 @@ import urllib.error import urllib.request from pathlib import Path -from ...deploy_key_provisioner import DeployKeyProvisioner +from ...deploy_key_provisioner import DeployKeyCollisionError, DeployKeyProvisioner class GiteaDeployKeyProvisioner(DeployKeyProvisioner): @@ -71,6 +71,11 @@ class GiteaDeployKeyProvisioner(DeployKeyProvisioner): body = json.loads(resp.read()) except urllib.error.HTTPError as exc: _body = _read_error_body(exc) + if exc.code == 422: + raise DeployKeyCollisionError( + f"deploy key collision for {owner_repo!r} " + f"(title={title!r}): key title or content already registered — {_body}" + ) from exc raise RuntimeError( f"failed to create deploy key for {owner_repo}: " f"HTTP {exc.code} — {_body}" diff --git a/bot_bottle/deploy_key_provisioner.py b/bot_bottle/deploy_key_provisioner.py index b7cac3c..e733bc1 100644 --- a/bot_bottle/deploy_key_provisioner.py +++ b/bot_bottle/deploy_key_provisioner.py @@ -11,6 +11,10 @@ from __future__ import annotations from abc import ABC, abstractmethod +class DeployKeyCollisionError(RuntimeError): + """Raised when a deploy key title or public key already exists on the repo.""" + + class DeployKeyProvisioner(ABC): """Manages a single deploy-key lifecycle on a remote forge.""" diff --git a/tests/unit/test_contrib_gitea_deploy_key.py b/tests/unit/test_contrib_gitea_deploy_key.py index b5a0402..469f757 100644 --- a/tests/unit/test_contrib_gitea_deploy_key.py +++ b/tests/unit/test_contrib_gitea_deploy_key.py @@ -12,6 +12,7 @@ from bot_bottle.contrib.gitea.deploy_key_provisioner import ( GiteaDeployKeyProvisioner, _split_owner_repo, ) +from bot_bottle.deploy_key_provisioner import DeployKeyCollisionError def _provisioner() -> GiteaDeployKeyProvisioner: @@ -100,6 +101,30 @@ class TestCreate(unittest.TestCase): provisioner.create("owner/repo", "title") self.assertIn("403", str(ctx.exception)) + def test_create_raises_collision_error_on_422(self): + provisioner = _provisioner() + collision_body = json.dumps({ + "errors": ["Key content already exists on this repository"], + "message": "422 Unprocessable Entity", + }) + with patch( + "bot_bottle.contrib.gitea.deploy_key_provisioner.subprocess.run" + ), patch( + "bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen", + side_effect=_http_error(422, collision_body), + ), patch( + "bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_bytes", + return_value=b"pk", + ), patch( + "bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_text", + return_value="ssh-ed25519 AAAA\n", + ): + with self.assertRaises(DeployKeyCollisionError) as ctx: + provisioner.create("owner/repo", "my-title") + msg = str(ctx.exception) + self.assertIn("owner/repo", msg) + self.assertIn("my-title", msg) + class TestDelete(unittest.TestCase): def test_delete_calls_correct_endpoint(self):