didericis
b96a8b44e0
test / integration-docker (pull_request) Successful in 22s
lint / lint (push) Successful in 1m7s
test / unit (pull_request) Successful in 2m10s
test / integration-firecracker (pull_request) Successful in 3m35s
test / coverage (pull_request) Successful in 15s
test / publish-infra (pull_request) Has been skipped
tracker-policy-pr / check-pr (pull_request) Failing after 13m2s
OrchestratorService wasn't the orchestrator — it's the host-side lifecycle of the docker infra *container* (the one that runs the Orchestrator). It read as "the orchestrator as a service" and lived in orchestrator/lifecycle.py, while its siblings (macOS MacosInfraService, Firecracker infra_vm) live under their backend package. Rename it DockerInfraService and move it to backend/docker/infra.py alongside the docker backend, with its docker-only constants (INFRA_*/ORCHESTRATOR_* image + container names, daemon list, mount paths). orchestrator/lifecycle.py keeps only the backend-neutral pieces the other infra services share — DEFAULT_PORT, DEFAULT_STARTUP_TIMEOUT_SECONDS, OrchestratorStartError, source_hash — which macOS / firecracker / client still import from there. backend/docker/infra.py imports those (backend -> orchestrator is an allowed direction). Renamed the unit test to test_docker_infra.py. Full unit suite green (2243). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
60 lines
2.1 KiB
Python
60 lines
2.1 KiB
Python
"""Unit: the `rotate_ca` one-shot CLI (issue #450). Docker mocked."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from bot_bottle.orchestrator import rotate_ca
|
|
from bot_bottle.gateway import GATEWAY_NAME
|
|
from bot_bottle.backend.docker.infra import INFRA_NAME
|
|
from bot_bottle.paths import host_gateway_ca_dir
|
|
from tests.unit import use_bottle_root
|
|
|
|
_RUN = "bot_bottle.orchestrator.rotate_ca.run_docker"
|
|
|
|
|
|
def _proc(returncode: int = 0, stdout: str = "", stderr: str = "") -> Mock:
|
|
return Mock(returncode=returncode, stdout=stdout, stderr=stderr)
|
|
|
|
|
|
class TestRotateCaCli(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self._tmp.cleanup)
|
|
self.addCleanup(use_bottle_root(Path(self._tmp.name)))
|
|
|
|
def test_clears_ca_and_drops_gateway_containers(self) -> None:
|
|
ca_dir = host_gateway_ca_dir()
|
|
(ca_dir / "mitmproxy-ca.pem").write_text("x")
|
|
(ca_dir / "mitmproxy-ca-cert.pem").write_text("x")
|
|
|
|
calls: list[list[str]] = []
|
|
|
|
def fake(argv: list[str], **_kw: object) -> Mock:
|
|
calls.append(argv)
|
|
# Report a removed container name so the CLI logs it.
|
|
return _proc(stdout=argv[-1])
|
|
|
|
with patch(_RUN, side_effect=fake):
|
|
self.assertEqual(0, rotate_ca.main([]))
|
|
|
|
# Persisted CA is gone → next start remints it.
|
|
self.assertEqual([], list(ca_dir.glob("mitmproxy-ca*")))
|
|
# Both the infra container and the standalone gateway are force-removed
|
|
# so no mitmproxy keeps serving the old CA from memory.
|
|
removed = {c[-1] for c in calls if c[:3] == ["docker", "rm", "--force"]}
|
|
self.assertEqual({INFRA_NAME, GATEWAY_NAME}, removed)
|
|
|
|
def test_succeeds_with_no_persisted_ca(self) -> None:
|
|
with patch(_RUN, return_value=_proc()) as m:
|
|
self.assertEqual(0, rotate_ca.main([]))
|
|
# Still tears down any running gateway even when there was no CA on disk.
|
|
self.assertTrue(m.called)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|