74 lines
3.1 KiB
Python
74 lines
3.1 KiB
Python
import importlib.util
|
|
import sys
|
|
import unittest
|
|
from unittest import mock
|
|
from pathlib import Path
|
|
|
|
|
|
_SCRIPT = Path(__file__).parents[1] / "scripts" / "preflight-deployment.py"
|
|
_SPEC = importlib.util.spec_from_file_location("deployment_preflight", _SCRIPT)
|
|
assert _SPEC is not None and _SPEC.loader is not None
|
|
preflight = importlib.util.module_from_spec(_SPEC)
|
|
sys.modules[_SPEC.name] = preflight
|
|
_SPEC.loader.exec_module(preflight)
|
|
|
|
|
|
class DeploymentPreflightTests(unittest.TestCase):
|
|
def test_maps_nested_container_path_to_host_source(self):
|
|
container = {
|
|
"Mounts": [
|
|
{"Type": "bind", "Source": "/srv/data", "Destination": "/data"},
|
|
{"Type": "bind", "Source": "/srv/sync", "Destination": "/data/Sync"},
|
|
]
|
|
}
|
|
mapping = preflight.map_path(container, "/data/Sync/routes/route-1")
|
|
self.assertEqual(mapping.source, Path("/srv/sync/routes/route-1"))
|
|
self.assertEqual(str(mapping.destination), "/data/Sync")
|
|
|
|
def test_rejects_api_and_client_roots_from_different_host_paths(self):
|
|
with self.assertRaisesRegex(preflight.CheckFailure, "host paths differ"):
|
|
preflight.require_same_path(
|
|
"Syncthing api_root/local_root",
|
|
preflight.Mapping(Path("/srv/syncthing-config/routes"), Path("/var/syncthing")),
|
|
preflight.Mapping(Path("/srv/sync/routes"), Path("/data/storage")),
|
|
)
|
|
|
|
def test_rejects_unmounted_container_path(self):
|
|
with self.assertRaisesRegex(preflight.CheckFailure, "not backed"):
|
|
preflight.map_path({"Mounts": []}, "/missing")
|
|
|
|
def test_maps_future_route_through_syncthing_override(self):
|
|
config = {
|
|
"api_root": "/var/syncthing",
|
|
"local_root": "/data/sync",
|
|
"local_path_overrides": {
|
|
"/var/syncthing/routes": "/data/qb/.archive-control-routes",
|
|
},
|
|
}
|
|
self.assertEqual(
|
|
preflight.route_local_path(config),
|
|
"/data/qb/.archive-control-routes",
|
|
)
|
|
|
|
def test_runs_hardlink_probe_with_qb_and_route_roots(self):
|
|
completed = __import__("subprocess").CompletedProcess(
|
|
args=[], returncode=0, stdout="hard-link staging probe passed\n", stderr=""
|
|
)
|
|
with mock.patch.object(preflight.subprocess, "run", return_value=completed) as run:
|
|
preflight.run_hardlink_probe("client", "/data/qb", "/data/qb/routes")
|
|
args = run.call_args.args[0]
|
|
self.assertEqual(args[:5], ["docker", "exec", "client", "python", "-c"])
|
|
self.assertEqual(args[-2:], ["/data/qb", "/data/qb/routes"])
|
|
|
|
def test_hardlink_probe_failure_is_actionable(self):
|
|
completed = __import__("subprocess").CompletedProcess(
|
|
args=[], returncode=1, stdout="", stderr="[Errno 18] Invalid cross-device link"
|
|
)
|
|
with mock.patch.object(preflight.subprocess, "run", return_value=completed):
|
|
with self.assertRaisesRegex(preflight.CheckFailure, "hard-link staging probe failed"):
|
|
preflight.run_hardlink_probe("client", "/data/qb", "/data/qb/routes")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|