86 lines
2.8 KiB
Python
86 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
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,
|
|
"127.0.0.1:49152\n",
|
|
"",
|
|
)
|
|
with patch.object(INGRESS, "_run", return_value=completed) as run:
|
|
port = INGRESS._published_port("ingress", 8443)
|
|
|
|
self.assertEqual(49152, port)
|
|
self.assertEqual(
|
|
["docker", "port", "ingress", "8443/tcp"],
|
|
run.call_args.args[0],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|