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_remembered_gateway_route_returns_user_within_ttl(self):
        self.manager.users_by_id = {"user1": {"id": "user1"}}
        self.manager.gateway_client_routes.clear()
        self.manager.time = types.SimpleNamespace(time=lambda: 100.0)
        self.manager.GATEWAY_ROUTE_TTL_SECONDS = 60

        self.manager.remember_gateway_route("10.200.200.136", "user1")

        user = self.manager.get_remembered_gateway_user("10.200.200.136")

        self.assertEqual(user["id"], "user1")

    def test_remembered_gateway_route_expires_after_ttl(self):
        now = {"value": 100.0}
        self.manager.users_by_id = {"user1": {"id": "user1"}}
        self.manager.gateway_client_routes.clear()
        self.manager.time = types.SimpleNamespace(time=lambda: now["value"])
        self.manager.GATEWAY_ROUTE_TTL_SECONDS = 60

        self.manager.remember_gateway_route("10.200.200.136", "user1")
        now["value"] = 161.0

        user = self.manager.get_remembered_gateway_user("10.200.200.136")

        self.assertIsNone(user)
        self.assertEqual(self.manager.gateway_client_routes, {})

    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()