128 lines
4.3 KiB
Python
128 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _load_module():
|
|
path = ROOT / "tools/checks/managed-ingress-drill.py"
|
|
spec = importlib.util.spec_from_file_location("managed_ingress_drill", path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
INGRESS = _load_module()
|
|
|
|
|
|
class ManagedIngressDrillTests(unittest.TestCase):
|
|
def test_config_is_streamed_into_a_daemon_visible_volume(self) -> None:
|
|
completed = subprocess.CompletedProcess([], 0, "", "")
|
|
with patch.object(INGRESS, "_run", return_value=completed) as run:
|
|
INGRESS._write_volume_file(
|
|
image="registry.example/caddy@sha256:" + "1" * 64,
|
|
volume="config-volume",
|
|
filename="Caddyfile",
|
|
content=":8080 { respond /health 200 }\n",
|
|
)
|
|
|
|
argv = run.call_args.args[0]
|
|
self.assertIn("type=volume,src=config-volume,dst=/govoplan-config", argv)
|
|
self.assertIn("0:0", argv)
|
|
self.assertNotIn("type=bind", " ".join(argv))
|
|
self.assertEqual(
|
|
":8080 { respond /health 200 }\n",
|
|
run.call_args.kwargs["input_text"],
|
|
)
|
|
|
|
def test_config_filename_cannot_escape_the_volume(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "invalid config filename"):
|
|
INGRESS._write_volume_file(
|
|
image="registry.example/caddy@sha256:" + "1" * 64,
|
|
volume="config-volume",
|
|
filename="../Caddyfile",
|
|
content="",
|
|
)
|
|
|
|
def test_drill_has_no_runner_local_bind_mounts(self) -> None:
|
|
source = (ROOT / "tools/checks/managed-ingress-drill.py").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
|
|
self.assertNotIn("type=bind", source)
|
|
self.assertIn('"--network-alias",\n "load-balancer"', source)
|
|
self.assertNotIn('"127.0.0.1::8080"', source)
|
|
self.assertIn("requested_http_port", source)
|
|
self.assertIn("requested_https_port", source)
|
|
|
|
def test_published_port_reads_the_docker_mapping(self) -> None:
|
|
completed = subprocess.CompletedProcess(
|
|
[],
|
|
0,
|
|
json.dumps(
|
|
{
|
|
"8443/tcp": [
|
|
{"HostIp": "127.0.0.1", "HostPort": "49152"}
|
|
]
|
|
}
|
|
),
|
|
"",
|
|
)
|
|
with patch.object(INGRESS, "_run", return_value=completed) as run:
|
|
port = INGRESS._published_port("ingress", 8443)
|
|
|
|
self.assertEqual(49152, port)
|
|
self.assertEqual(
|
|
[
|
|
"docker",
|
|
"inspect",
|
|
"--format",
|
|
"{{json .HostConfig.PortBindings}}",
|
|
"ingress",
|
|
],
|
|
run.call_args.args[0],
|
|
)
|
|
|
|
def test_published_port_rejects_non_loopback_binding(self) -> None:
|
|
completed = subprocess.CompletedProcess(
|
|
[],
|
|
0,
|
|
'{"8443/tcp":[{"HostIp":"0.0.0.0","HostPort":"49152"}]}',
|
|
"",
|
|
)
|
|
with patch.object(INGRESS, "_run", return_value=completed):
|
|
with self.assertRaisesRegex(RuntimeError, "loopback binding"):
|
|
INGRESS._published_port("ingress", 8443)
|
|
|
|
def test_probe_runs_as_a_network_sibling_from_a_digest_image(self) -> None:
|
|
completed = subprocess.CompletedProcess([], 0, "", "")
|
|
image = "registry.example/runtime-api@sha256:" + "1" * 64
|
|
with patch.object(INGRESS, "_run", return_value=completed) as run:
|
|
INGRESS._probe_ingress(
|
|
image=image,
|
|
network="deployment-network",
|
|
container="ingress",
|
|
)
|
|
|
|
argv = run.call_args.args[0]
|
|
self.assertEqual("docker", argv[0])
|
|
self.assertIn("deployment-network", argv)
|
|
self.assertIn(image, argv)
|
|
self.assertIn('(\"ingress\", port)', argv[-1])
|
|
self.assertIn("server_hostname=\"localhost\"", argv[-1])
|
|
self.assertNotIn("localhost:49152", argv[-1])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|