# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""
A fake FTP server for testing.

Do not use this in production!  It has only the features required for the
tests that currently use it, and no security.
"""

from pathlib import Path
from threading import Event, Thread
from typing import override

from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.handlers import FTPHandler
from pyftpdlib.ioloop import IOLoop
from pyftpdlib.servers import ThreadedFTPServer


# mypy complains that IOLoop is untyped, which is true, but we can't fix
# that here.
class InterruptibleIOLoop(IOLoop):  # type: ignore[misc]
    """A poller that can be interrupted by an event."""

    stop_event: Event | None = None

    def poll(self, timeout: float) -> None:
        """Poll once."""
        if (
            self.stop_event is not None and self.stop_event.is_set()
        ):  # pragma: no cover
            return
        super().poll(timeout)


class FakeFTPServerThread(Thread):
    """Thread that runs a fake FTP server for testing."""

    def __init__(self, username: str, home_directory: Path) -> None:
        """Initialize the server and start listening."""
        super().__init__()
        self.name = "FakeFTPServer"
        authorizer = DummyAuthorizer()
        authorizer.add_user(username, "", str(home_directory), perm="ew")
        handler = FTPHandler
        handler.authorizer = authorizer
        ioloop_class = InterruptibleIOLoop
        ioloop_class.stop_event = self.stop_event = Event()
        self.server = ThreadedFTPServer(("", 0), handler, ioloop=ioloop_class())

    @property
    def address(self) -> tuple[str, int]:
        """Return the (host, port) to be used by clients."""
        host, port = self.server.socket.getsockname()
        assert isinstance(host, str)
        assert isinstance(port, int)
        return host, port

    @override
    def run(self) -> None:
        """Run the server."""
        self.server.serve_forever()

    def stop(self) -> None:
        """Stop the server."""
        self.stop_event.set()
        self.server.close_all()
        self.join()
