From d38120d4cee5e4f3f2ed15a2542fc3ff10a0f264 Mon Sep 17 00:00:00 2001 From: didericis Date: Thu, 25 Jun 2026 05:45:16 -0400 Subject: [PATCH] test: add coverage for git gate and supervise server --- tests/unit/test_git_gate.py | 65 +++++++++++++++++++++++++++++ tests/unit/test_supervise_server.py | 49 ++++++++++++++++++++++ 2 files changed, 114 insertions(+) diff --git a/tests/unit/test_git_gate.py b/tests/unit/test_git_gate.py index fa6bf3b..3eaf3a0 100644 --- a/tests/unit/test_git_gate.py +++ b/tests/unit/test_git_gate.py @@ -4,6 +4,7 @@ import os import tempfile import unittest from pathlib import Path +from unittest.mock import patch from bot_bottle.git_gate import ( GitGate, @@ -13,6 +14,8 @@ from bot_bottle.git_gate import ( git_gate_render_access_hook, git_gate_render_entrypoint, git_gate_render_hook, + revoke_git_gate_provisioned_keys, + _resolve_identity_file, git_gate_upstreams_for_bottle, ) from bot_bottle.manifest import ManifestIndex @@ -328,6 +331,68 @@ class TestPrepare(unittest.TestCase): self.assertIn("exec git daemon", content) +class TestDynamicKeyProvisioning(unittest.TestCase): + def setUp(self): + self.stage = Path(tempfile.mkdtemp()) + + def tearDown(self): + import shutil + + shutil.rmtree(self.stage, ignore_errors=True) + + def _gitea_manifest(self): + return ManifestIndex.from_json_obj({ + "bottles": { + "dev": { + "git-gate": { + "repos": { + "repo": { + "url": "ssh://git@gitea.example.com/org/repo.git", + "key": { + "provider": "gitea", + "forge_token_env": "GITEA_TOKEN", + }, + "host_key": "ssh-ed25519 AAAA...", + }, + }, + } + } + }, + "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, + }) + + def test_resolve_identity_file_static_uses_entry_path(self): + entry = fixture_with_git().bottles["dev"].git[0] + self.assertEqual(entry.IdentityFile, _resolve_identity_file(entry, "demo", self.stage)) + + def test_resolve_identity_file_gitea_provisions_key(self): + entry = self._gitea_manifest().bottles["dev"].git[0] + with patch("bot_bottle.git_gate._provision_dynamic_key", return_value="/tmp/provisioned-key") as mock_provision: + self.assertEqual("/tmp/provisioned-key", _resolve_identity_file(entry, "demo", self.stage)) + mock_provision.assert_called_once() + + def test_revoke_skips_non_gitea_and_missing_id_file(self): + revoke_git_gate_provisioned_keys(fixture_with_git().bottles["dev"], self.stage) + + def test_revoke_calls_delete_for_gitea_entry(self): + bottle = self._gitea_manifest().bottles["dev"] + (self.stage / "repo-deploy-key-id").write_text("123\n") + with patch.dict("os.environ", {"GITEA_TOKEN": "token"}), patch( + "bot_bottle.deploy_key_provisioner.get_provisioner" + ) as mock_get_provisioner: + provisioner = mock_get_provisioner.return_value + revoke_git_gate_provisioned_keys(bottle, self.stage) + mock_get_provisioner.assert_called_once() + provisioner.delete.assert_called_once_with("org/repo", "123") + + def test_revoke_missing_token_raises(self): + bottle = self._gitea_manifest().bottles["dev"] + (self.stage / "repo-deploy-key-id").write_text("123\n") + with patch.dict("os.environ", {}, clear=True), self.assertRaises(RuntimeError) as cm: + revoke_git_gate_provisioned_keys(bottle, self.stage) + self.assertIn("env var is not set", str(cm.exception)) + + class TestShellEscaping(unittest.TestCase): """Regression tests: all three render functions must produce syntactically valid sh code even when names and upstream URLs contain shell-special diff --git a/tests/unit/test_supervise_server.py b/tests/unit/test_supervise_server.py index 0b4efa8..cb121bb 100644 --- a/tests/unit/test_supervise_server.py +++ b/tests/unit/test_supervise_server.py @@ -364,6 +364,23 @@ class TestHandleToolsCall(unittest.TestCase): self.config, ) + def test_missing_name_raises(self): + with self.assertRaises(_RpcError) as cm: + handle_tools_call({"arguments": {}}, self.config) + self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code) + + def test_arguments_must_be_object(self): + with self.assertRaises(_RpcError) as cm: + handle_tools_call( + { + "name": _sv.TOOL_EGRESS_ALLOW, + "arguments": [], + }, + self.config, + ) + self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code) + self.assertIn("must be an object", cm.exception.message) + def test_capability_block_call_raises_unknown_tool(self): with self.assertRaises(_RpcError) as cm: handle_tools_call( @@ -426,6 +443,31 @@ class TestHandleToolsCall(unittest.TestCase): class TestHandleListEgressRoutes(unittest.TestCase): + def test_success_returns_body_text(self): + class _Resp: + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def read(self): + return b"[{\"host\": \"example.com\"}]" + + class _Opener: + def open(self, *args, **kwargs): # noqa: ANN001, ANN002, ANN003 # type: ignore + return _Resp() + + with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()): + result = handle_list_egress_routes( + {}, + ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")), + ) + + self.assertFalse(result["isError"]) # type: ignore[index] + text = result["content"][0]["text"] # type: ignore[index] + self.assertIn("example.com", text) + def test_url_error_returns_tool_error(self): class _Opener: def open(self, *args, **kwargs): # noqa: ANN001, ANN002, ANN003 # type: ignore @@ -485,6 +527,13 @@ class TestFormatResponseText(unittest.TestCase): self.assertIn("the operator modified", text.lower()) +class TestFormatPendingResponseText(unittest.TestCase): + def test_formats_timeout_message(self): + text = supervise_server.format_pending_response_text(12.5) + self.assertIn("status: pending", text) + self.assertIn("12.5s", text) + + # --- End-to-end HTTP sanity ------------------------------------------------