didericis
5dfb9b0d75
git-gate provisioning into the running gateway was hard-wired to docker exec/cp. Extract a backend-neutral `GatewayTransport` (exec + cp_into) so the same provisioning logic serves both the docker gateway container and the firecracker gateway VM (over SSH, added with the launch swap). - `provision_git_gate` / `deprovision_git_gate` now take a transport instead of a gateway name; `DockerGatewayTransport` wraps the existing docker exec/cp behavior. deprovision is best-effort (catches the transport error) — matching the prior idempotent teardown. - both consolidated_launch callers pass `DockerGatewayTransport(name)` — no behavior change; the firecracker swap flips only its own to SSH. Pure refactor: docker path unchanged, gateway_provision tests updated to construct the transport, all green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
115 lines
4.5 KiB
Python
115 lines
4.5 KiB
Python
"""Unit: git-gate provisioning into the running shared gateway (PRD 0070)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from bot_bottle.backend.docker.gateway_provision import (
|
|
GatewayProvisionError,
|
|
DockerGatewayTransport,
|
|
deprovision_git_gate,
|
|
provision_git_gate,
|
|
)
|
|
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
|
|
|
|
_RUN = "bot_bottle.backend.docker.gateway_provision.run_docker"
|
|
|
|
|
|
def _proc(returncode: int = 0, stderr: str = "") -> Mock:
|
|
return Mock(returncode=returncode, stderr=stderr)
|
|
|
|
|
|
def _recorder(calls: list[list[str]]):
|
|
"""A run_docker side_effect that records argv and returns success."""
|
|
def fake(argv: list[str]) -> Mock:
|
|
calls.append(argv)
|
|
return _proc()
|
|
return fake
|
|
|
|
|
|
def _plan(*upstreams: GitGateUpstream) -> GitGatePlan:
|
|
return GitGatePlan(
|
|
slug="demo",
|
|
entrypoint_script=Path("/stage/entrypoint.sh"),
|
|
hook_script=Path("/stage/pre-receive"),
|
|
access_hook_script=Path("/stage/access-hook"),
|
|
upstreams=tuple(upstreams),
|
|
)
|
|
|
|
|
|
def _up(name: str, *, key: str = "/host/keys/id", known_hosts: str = "") -> GitGateUpstream:
|
|
return GitGateUpstream(
|
|
name=name,
|
|
upstream_url=f"ssh://git@github.com/x/{name}.git",
|
|
upstream_host="github.com",
|
|
upstream_port="22",
|
|
identity_file=key,
|
|
known_host_key="",
|
|
known_hosts_file=Path(known_hosts) if known_hosts else Path(),
|
|
)
|
|
|
|
|
|
class TestProvisionGitGate(unittest.TestCase):
|
|
def test_copies_creds_and_runs_namespaced_init(self) -> None:
|
|
calls: list[list[str]] = []
|
|
with patch(_RUN, side_effect=_recorder(calls)):
|
|
provision_git_gate(DockerGatewayTransport("gw"), "bottle1", _plan(_up("foo", known_hosts="/host/kh")))
|
|
|
|
cps = [c for c in calls if c[:2] == ["docker", "cp"]]
|
|
self.assertIn(["docker", "cp", "/host/keys/id", "gw:/git-gate/creds/bottle1/foo-key"], cps)
|
|
self.assertIn(
|
|
["docker", "cp", "/host/kh", "gw:/git-gate/creds/bottle1/foo-known_hosts"], cps,
|
|
)
|
|
# The shared (bottle-agnostic) hooks are installed into the gateway.
|
|
self.assertIn(["docker", "cp", "/stage/pre-receive", "gw:/etc/git-gate/pre-receive"], cps)
|
|
# The init script runs in the gateway, namespaced under the bottle id.
|
|
exec_scripts = [c for c in calls if c[:3] == ["docker", "exec", "gw"] and c[3] == "sh"]
|
|
self.assertEqual(1, len(exec_scripts))
|
|
self.assertIn("repo=/git/bottle1/${name}.git", exec_scripts[0][-1])
|
|
|
|
def test_omits_known_hosts_copy_when_absent(self) -> None:
|
|
calls: list[list[str]] = []
|
|
with patch(_RUN, side_effect=_recorder(calls)):
|
|
provision_git_gate(DockerGatewayTransport("gw"), "b1", _plan(_up("foo"))) # no known_hosts
|
|
creds_cps = [c for c in calls if c[:2] == ["docker", "cp"] and "/git-gate/creds/" in c[3]]
|
|
self.assertEqual(1, len(creds_cps)) # only the key, not known_hosts
|
|
self.assertTrue(creds_cps[0][3].endswith("/foo-key"))
|
|
|
|
def test_no_upstreams_is_noop(self) -> None:
|
|
with patch(_RUN) as m:
|
|
provision_git_gate(DockerGatewayTransport("gw"), "b1", _plan())
|
|
m.assert_not_called()
|
|
|
|
def test_raises_on_docker_failure(self) -> None:
|
|
with patch(_RUN, return_value=_proc(returncode=1, stderr="boom")):
|
|
with self.assertRaises(GatewayProvisionError):
|
|
provision_git_gate(DockerGatewayTransport("gw"), "b1", _plan(_up("foo")))
|
|
|
|
def test_rejects_unsafe_bottle_id_before_any_docker(self) -> None:
|
|
with patch(_RUN) as m:
|
|
with self.assertRaises(GatewayProvisionError):
|
|
provision_git_gate(DockerGatewayTransport("gw"), "../etc", _plan(_up("foo")))
|
|
m.assert_not_called() # rejected before a single docker call
|
|
|
|
|
|
class TestDeprovision(unittest.TestCase):
|
|
def test_removes_repo_and_creds(self) -> None:
|
|
with patch(_RUN, return_value=_proc()) as m:
|
|
deprovision_git_gate(DockerGatewayTransport("gw"), "b1")
|
|
argv = m.call_args.args[0]
|
|
self.assertEqual(["docker", "exec", "gw", "rm", "-rf"], argv[:5])
|
|
self.assertIn("/git/b1", argv)
|
|
self.assertIn("/git-gate/creds/b1", argv)
|
|
|
|
def test_rejects_unsafe_bottle_id(self) -> None:
|
|
with patch(_RUN) as m:
|
|
with self.assertRaises(GatewayProvisionError):
|
|
deprovision_git_gate(DockerGatewayTransport("gw"), "a/b")
|
|
m.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|