didericis-claude
508c537deb
Closes #255. Without timeouts, a hung upstream during the access-hook or git http-backend CGI call (git_http_backend.py) and a stalled Gitea API during deploy-key provisioning (contrib/gitea/deploy_key_provisioner.py) could wedge a sidecar indefinitely. Adds GIT_HTTP_BACKEND_TIMEOUT_SECS (30s) to both subprocess.run calls in the HTTP backend, mirroring the existing GIT_GATE_DAEMON_TIMEOUT_SECS on the daemon path. Adds _API_TIMEOUT_SECS (30s) and _KEYGEN_TIMEOUT_SECS (10s) to the Gitea provisioner's urlopen and ssh-keygen calls. Tests verify the timeout values are forwarded in all four call sites.
221 lines
8.2 KiB
Python
221 lines
8.2 KiB
Python
"""Unit: GiteaDeployKeyProvisioner (PRD 0048, contrib/gitea)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
import urllib.error
|
|
from io import BytesIO
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from bot_bottle.contrib.gitea.deploy_key_provisioner import (
|
|
GiteaDeployKeyProvisioner,
|
|
_API_TIMEOUT_SECS,
|
|
_KEYGEN_TIMEOUT_SECS,
|
|
_split_owner_repo,
|
|
)
|
|
from bot_bottle.deploy_key_provisioner import DeployKeyCollisionError
|
|
|
|
|
|
def _provisioner() -> GiteaDeployKeyProvisioner:
|
|
return GiteaDeployKeyProvisioner(
|
|
token="test-token", api_url="https://gitea.example.com"
|
|
)
|
|
|
|
|
|
def _urlopen_response(body: dict, status: int = 200) -> MagicMock: # type: ignore
|
|
resp = MagicMock()
|
|
resp.read.return_value = json.dumps(body).encode()
|
|
resp.status = status
|
|
resp.__enter__ = lambda s: s # type: ignore
|
|
resp.__exit__ = MagicMock(return_value=False)
|
|
return resp
|
|
|
|
|
|
def _http_error(code: int, body: str = "") -> urllib.error.HTTPError:
|
|
return urllib.error.HTTPError(
|
|
url="http://x",
|
|
code=code,
|
|
msg="err",
|
|
hdrs=None, # type: ignore[arg-type]
|
|
fp=BytesIO(body.encode()),
|
|
)
|
|
|
|
|
|
class TestCreate(unittest.TestCase):
|
|
def test_create_calls_ssh_keygen_and_posts_to_api(self):
|
|
provisioner = _provisioner()
|
|
fake_key_id = 42
|
|
fake_private = b"PRIVATE_KEY"
|
|
fake_public = "ssh-ed25519 AAAA fake"
|
|
|
|
with patch(
|
|
"bot_bottle.contrib.gitea.deploy_key_provisioner.subprocess.run"
|
|
) as mock_run, patch(
|
|
"bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen"
|
|
) as mock_urlopen, patch(
|
|
"bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_bytes",
|
|
return_value=fake_private,
|
|
), patch(
|
|
"bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_text",
|
|
return_value=fake_public + "\n",
|
|
):
|
|
mock_urlopen.return_value = _urlopen_response({"id": fake_key_id})
|
|
key_id, private_bytes = provisioner.create(
|
|
"didericis/bot-bottle", "bot-bottle:slug:repo"
|
|
)
|
|
|
|
# ssh-keygen called with ed25519
|
|
mock_run.assert_called_once()
|
|
run_args = mock_run.call_args.args[0]
|
|
self.assertIn("ssh-keygen", run_args)
|
|
self.assertIn("-t", run_args)
|
|
self.assertIn("ed25519", run_args)
|
|
|
|
# POST body contains public key
|
|
post_call = mock_urlopen.call_args.args[0]
|
|
payload = json.loads(post_call.data)
|
|
self.assertEqual(fake_public, payload["key"])
|
|
self.assertFalse(payload["read_only"])
|
|
|
|
# Correct URL
|
|
self.assertIn(
|
|
"/api/v1/repos/didericis/bot-bottle/keys", post_call.full_url
|
|
)
|
|
self.assertEqual(str(fake_key_id), key_id)
|
|
self.assertEqual(fake_private, private_bytes)
|
|
|
|
def test_create_passes_timeout_to_ssh_keygen_and_urlopen(self):
|
|
provisioner = _provisioner()
|
|
with patch(
|
|
"bot_bottle.contrib.gitea.deploy_key_provisioner.subprocess.run"
|
|
) as mock_run, patch(
|
|
"bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen"
|
|
) as mock_urlopen, patch(
|
|
"bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_bytes",
|
|
return_value=b"PRIVATE",
|
|
), patch(
|
|
"bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_text",
|
|
return_value="ssh-ed25519 AAAA\n",
|
|
):
|
|
mock_urlopen.return_value = _urlopen_response({"id": 1})
|
|
provisioner.create("owner/repo", "title")
|
|
|
|
self.assertEqual(_KEYGEN_TIMEOUT_SECS, mock_run.call_args.kwargs.get("timeout"))
|
|
self.assertEqual(_API_TIMEOUT_SECS, mock_urlopen.call_args.kwargs.get("timeout"))
|
|
|
|
def test_create_raises_on_http_error(self):
|
|
provisioner = _provisioner()
|
|
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(403, "forbidden"),
|
|
), 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(RuntimeError) as ctx:
|
|
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):
|
|
provisioner = _provisioner()
|
|
with patch(
|
|
"bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen"
|
|
) as mock_urlopen:
|
|
mock_urlopen.return_value = _urlopen_response({})
|
|
provisioner.delete("didericis/bot-bottle", "99")
|
|
|
|
req = mock_urlopen.call_args.args[0]
|
|
self.assertIn("/api/v1/repos/didericis/bot-bottle/keys/99", req.full_url)
|
|
self.assertEqual("DELETE", req.get_method())
|
|
|
|
def test_delete_passes_timeout_to_urlopen(self):
|
|
provisioner = _provisioner()
|
|
with patch(
|
|
"bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen"
|
|
) as mock_urlopen:
|
|
mock_urlopen.return_value = _urlopen_response({})
|
|
provisioner.delete("owner/repo", "7")
|
|
|
|
self.assertEqual(_API_TIMEOUT_SECS, mock_urlopen.call_args.kwargs.get("timeout"))
|
|
|
|
def test_delete_tolerates_404(self):
|
|
provisioner = _provisioner()
|
|
with patch(
|
|
"bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen",
|
|
side_effect=_http_error(404),
|
|
):
|
|
provisioner.delete("owner/repo", "123") # must not raise
|
|
|
|
def test_delete_raises_on_non_404_http_error(self):
|
|
provisioner = _provisioner()
|
|
with patch(
|
|
"bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen",
|
|
side_effect=_http_error(500, "internal server error"),
|
|
):
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
provisioner.delete("owner/repo", "7")
|
|
self.assertIn("500", str(ctx.exception))
|
|
|
|
def test_delete_raises_on_url_error(self):
|
|
provisioner = _provisioner()
|
|
with patch(
|
|
"bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen",
|
|
side_effect=urllib.error.URLError("connection refused"),
|
|
):
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
provisioner.delete("owner/repo", "7")
|
|
self.assertIn("connection refused", str(ctx.exception))
|
|
|
|
|
|
class TestSplitOwnerRepo(unittest.TestCase):
|
|
def test_simple(self):
|
|
self.assertEqual(("owner", "repo"), _split_owner_repo("owner/repo"))
|
|
|
|
def test_raises_on_missing_slash(self):
|
|
with self.assertRaises(ValueError):
|
|
_split_owner_repo("noslash")
|
|
|
|
def test_raises_on_empty_owner(self):
|
|
with self.assertRaises(ValueError):
|
|
_split_owner_repo("/repo")
|
|
|
|
def test_raises_on_empty_repo(self):
|
|
with self.assertRaises(ValueError):
|
|
_split_owner_repo("owner/")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|