291 lines
9.1 KiB
Python
291 lines
9.1 KiB
Python
import importlib.util
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
class FakeNotFound(Exception):
|
|
pass
|
|
|
|
|
|
class FakeAPIError(Exception):
|
|
pass
|
|
|
|
|
|
class FakeContainer:
|
|
def __init__(self, status="running"):
|
|
self.stop_calls = []
|
|
self.start_calls = 0
|
|
self.reload_calls = 0
|
|
self.status = status
|
|
|
|
def reload(self):
|
|
self.reload_calls += 1
|
|
|
|
def stop(self, timeout=10):
|
|
self.stop_calls.append(timeout)
|
|
|
|
def start(self):
|
|
self.start_calls += 1
|
|
self.status = "running"
|
|
|
|
|
|
class FakeContainers:
|
|
def __init__(self, container=None):
|
|
self.container = container or FakeContainer()
|
|
self.requested_names = []
|
|
self.run_calls = []
|
|
self.raise_not_found = False
|
|
self.raise_api_error_on_run = None
|
|
|
|
def get(self, name):
|
|
self.requested_names.append(name)
|
|
if self.raise_not_found:
|
|
self.raise_not_found = False
|
|
raise FakeNotFound(name)
|
|
return self.container
|
|
|
|
def run(self, image, **kwargs):
|
|
self.run_calls.append((image, kwargs))
|
|
if self.raise_api_error_on_run is not None:
|
|
raise self.raise_api_error_on_run
|
|
return self.container
|
|
|
|
|
|
class FakeVolumes:
|
|
def __init__(self):
|
|
self.created = []
|
|
self.existing = set()
|
|
|
|
def get(self, name):
|
|
if name not in self.existing:
|
|
raise FakeNotFound(name)
|
|
return name
|
|
|
|
def create(self, name):
|
|
self.created.append(name)
|
|
self.existing.add(name)
|
|
return name
|
|
|
|
|
|
class FakeDockerClient:
|
|
def __init__(self):
|
|
self.containers = FakeContainers()
|
|
self.volumes = FakeVolumes()
|
|
|
|
|
|
def load_manager_module():
|
|
fake_client = FakeDockerClient()
|
|
fake_docker = types.SimpleNamespace(
|
|
from_env=lambda: fake_client,
|
|
errors=types.SimpleNamespace(NotFound=FakeNotFound, APIError=FakeAPIError),
|
|
)
|
|
sys.modules["docker"] = fake_docker
|
|
|
|
module_path = Path(__file__).with_name("manager.py")
|
|
spec = importlib.util.spec_from_file_location("manager_under_test", module_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
module._docker_client = fake_client
|
|
return module, fake_client
|
|
|
|
|
|
class ManagerConfigTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.manager, self.fake_client = load_manager_module()
|
|
|
|
def write_config(self, payload):
|
|
temp = tempfile.NamedTemporaryFile("w", delete=False, encoding="utf-8")
|
|
with temp:
|
|
json.dump(payload, temp)
|
|
return temp.name
|
|
|
|
def test_load_config_applies_defaults_and_placeholders(self):
|
|
path = self.write_config(
|
|
{
|
|
"defaults": {
|
|
"container_prefix": "fis-",
|
|
"volumes": {"v-conf-{id}": "/config"},
|
|
"environment": {"PUID": "1000", "PGID": "1000"},
|
|
"mem_limit": "3g",
|
|
"privileged": True,
|
|
},
|
|
"users": [
|
|
{
|
|
"id": "mario",
|
|
"port": 3101,
|
|
"environment": {"PASSWORD": "secret"},
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
config = self.manager.load_config(path)
|
|
user = config["users"][0]
|
|
|
|
self.assertEqual(user["container_name"], "fis-mario")
|
|
self.assertEqual(user["volumes"], {"v-conf-mario": "/config"})
|
|
self.assertEqual(user["environment"]["PUID"], "1000")
|
|
self.assertEqual(user["environment"]["PASSWORD"], "secret")
|
|
self.assertEqual(user["mem_limit"], "3g")
|
|
self.assertTrue(user["privileged"])
|
|
|
|
def test_load_config_rejects_duplicate_ports(self):
|
|
path = self.write_config(
|
|
{
|
|
"users": [
|
|
{"id": "user1", "port": 3001, "environment": {"PASSWORD": "a"}},
|
|
{"id": "user2", "port": 3001, "environment": {"PASSWORD": "b"}},
|
|
]
|
|
}
|
|
)
|
|
|
|
with self.assertRaisesRegex(ValueError, "Duplicate port"):
|
|
self.manager.load_config(path)
|
|
|
|
def test_load_config_requires_password(self):
|
|
path = self.write_config({"users": [{"id": "user1", "port": 3001}]})
|
|
|
|
with self.assertRaisesRegex(ValueError, "PASSWORD"):
|
|
self.manager.load_config(path)
|
|
|
|
def test_stop_user_container_uses_configured_container_name(self):
|
|
user = {"id": "user1", "container_name": "custom-name"}
|
|
|
|
self.manager.stop_user_container(user)
|
|
|
|
self.assertEqual(self.fake_client.containers.requested_names, ["custom-name"])
|
|
self.assertEqual(self.fake_client.containers.container.stop_calls, [10])
|
|
|
|
def test_start_user_container_recovers_from_concurrent_create_conflict(self):
|
|
container = FakeContainer(status="created")
|
|
self.fake_client.containers.container = container
|
|
self.fake_client.containers.raise_not_found = True
|
|
self.fake_client.containers.raise_api_error_on_run = FakeAPIError(
|
|
'Conflict. The container name "/fis-user1" is already in use'
|
|
)
|
|
self.manager.wait_until_ready = lambda name: True
|
|
|
|
user = {
|
|
"id": "user1",
|
|
"container_name": "fis-user1",
|
|
"volumes": {"v-conf-user1": "/config"},
|
|
"environment": {"PASSWORD": "secret"},
|
|
"privileged": False,
|
|
}
|
|
|
|
self.manager.start_user_container(user)
|
|
|
|
self.assertEqual(self.fake_client.containers.requested_names, ["fis-user1", "fis-user1"])
|
|
self.assertEqual(container.start_calls, 1)
|
|
self.assertEqual(self.fake_client.volumes.created, ["v-conf-user1"])
|
|
|
|
def test_extract_user_from_host_header(self):
|
|
request = (
|
|
b"GET / HTTP/1.1\r\n"
|
|
b"Host: user2.example.local\r\n"
|
|
b"Connection: keep-alive\r\n"
|
|
b"\r\n"
|
|
)
|
|
users = {"user2": {"id": "user2"}}
|
|
|
|
user, rewritten, redirect_target = self.manager.extract_user_from_http_request(request, users)
|
|
|
|
self.assertEqual(user["id"], "user2")
|
|
self.assertEqual(rewritten, request)
|
|
self.assertIsNone(redirect_target)
|
|
|
|
def test_extract_user_from_path_rewrites_request_for_direct_proxying(self):
|
|
request = (
|
|
b"GET /user1/websockify?token=abc HTTP/1.1\r\n"
|
|
b"Host: localhost:8080\r\n"
|
|
b"Upgrade: websocket\r\n"
|
|
b"\r\n"
|
|
)
|
|
users = {"user1": {"id": "user1"}}
|
|
|
|
user, rewritten, redirect_target = self.manager.extract_user_from_http_request(request, users)
|
|
|
|
self.assertEqual(user["id"], "user1")
|
|
self.assertEqual(
|
|
rewritten,
|
|
(
|
|
b"GET /websockify?token=abc HTTP/1.1\r\n"
|
|
b"Host: localhost:8080\r\n"
|
|
b"Upgrade: websocket\r\n"
|
|
b"\r\n"
|
|
),
|
|
)
|
|
self.assertIsNone(redirect_target)
|
|
|
|
def test_extract_user_from_bare_user_path_rewrites_to_root(self):
|
|
request = (
|
|
b"GET /user1 HTTP/1.1\r\n"
|
|
b"Host: 10.200.200.113:8080\r\n"
|
|
b"Connection: keep-alive\r\n"
|
|
b"\r\n"
|
|
)
|
|
users = {"user1": {"id": "user1"}}
|
|
|
|
user, rewritten, redirect_target = self.manager.extract_user_from_http_request(request, users)
|
|
|
|
self.assertEqual(user["id"], "user1")
|
|
self.assertEqual(
|
|
rewritten,
|
|
(
|
|
b"GET / HTTP/1.1\r\n"
|
|
b"Host: 10.200.200.113:8080\r\n"
|
|
b"Connection: keep-alive\r\n"
|
|
b"\r\n"
|
|
),
|
|
)
|
|
self.assertIsNone(redirect_target)
|
|
|
|
def test_extract_user_from_absolute_form_request_rewrites_to_root(self):
|
|
request = (
|
|
b"GET http://10.200.200.113:8080/user1 HTTP/1.1\r\n"
|
|
b"Host: 10.200.200.113:8080\r\n"
|
|
b"Connection: keep-alive\r\n"
|
|
b"\r\n"
|
|
)
|
|
users = {"user1": {"id": "user1"}}
|
|
|
|
user, rewritten, redirect_target = self.manager.extract_user_from_http_request(request, users)
|
|
|
|
self.assertEqual(user["id"], "user1")
|
|
self.assertEqual(
|
|
rewritten,
|
|
(
|
|
b"GET / HTTP/1.1\r\n"
|
|
b"Host: 10.200.200.113:8080\r\n"
|
|
b"Connection: keep-alive\r\n"
|
|
b"\r\n"
|
|
),
|
|
)
|
|
self.assertIsNone(redirect_target)
|
|
|
|
def test_wait_until_ready_applies_stabilization_delay(self):
|
|
sleep_calls = []
|
|
clock = {"now": 0.0}
|
|
|
|
self.manager.STARTUP_TIMEOUT_SECONDS = 10
|
|
self.manager.STARTUP_STABILIZATION_SECONDS = 5
|
|
self.manager.get_container_ip = lambda name: "10.0.0.2"
|
|
self.manager.is_port_open = lambda host, port: True
|
|
self.manager.time = types.SimpleNamespace(
|
|
time=lambda: clock["now"],
|
|
sleep=lambda seconds: (sleep_calls.append(seconds), clock.__setitem__("now", clock["now"] + seconds)),
|
|
)
|
|
|
|
ready = self.manager.wait_until_ready("fis-user1")
|
|
|
|
self.assertTrue(ready)
|
|
self.assertEqual(sleep_calls, [5])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|