mcbebop/tests/test_ftp.py

"""FTP: the area-to-port map, path containment, and the size cap.

ftplib is replaced with a fake; nothing opens a socket.
"""

import ftplib
from typing import ClassVar

import pytest

from mcbebop.files import ftp

# --- the area map ------------------------------------------------------------


def test_the_three_read_only_areas_map_to_the_ports_found_on_the_aircraft():
    assert {a.name: a.port for a in ftp.areas()} == {"media": 21, "flightplans": 61, "logs": 21}


def test_the_firmware_write_port_is_not_reachable_through_the_api():
    # Port 51 serves /update as root. It is how firmware is pushed, it has no
    # read use case, and the drone's Wi-Fi is open. Exposing it would hand the
    # aircraft's weakest point to anything that can call a tool.
    assert ftp.FIRMWARE_WRITE_PORT == 51
    assert 51 not in {a.port for a in ftp.areas()}
    for name in ("update", "firmware", "51"):
        with pytest.raises(ftp.FtpError):
            ftp.resolve_area(name)


def test_an_unknown_area_says_what_is_available():
    with pytest.raises(ftp.FtpError, match="flightplans"):
        ftp.resolve_area("nope")


def test_logs_is_the_media_ftpd_narrowed_to_the_blackbox_tree():
    logs = ftp.resolve_area("logs")
    media = ftp.resolve_area("media")
    assert logs.port == media.port == 21
    assert logs.prefix == "internal_000/Debug/current"
    assert media.prefix == ""


def test_every_area_describes_itself():
    assert all(a.describe for a in ftp.areas())


# --- path containment --------------------------------------------------------


def test_a_logs_path_stays_inside_the_debug_tree():
    area = ftp.resolve_area("logs")
    assert ftp._safe_path(area, "") == "internal_000/Debug/current"
    assert ftp._safe_path(area, "boot.log") == "internal_000/Debug/current/boot.log"


def test_dot_dot_cannot_climb_out_of_an_area():
    area = ftp.resolve_area("logs")
    for path in ("../../..", "a/../../b", ".."):
        with pytest.raises(ftp.FtpError, match=r"\.\."):
            ftp._safe_path(area, path)


def test_an_absolute_path_is_refused_rather_than_silently_rebased():
    with pytest.raises(ftp.FtpError, match="relative"):
        ftp._safe_path(ftp.resolve_area("media"), "/data/ftp/internal_000")


def test_an_empty_media_path_lists_the_area_root():
    assert ftp._safe_path(ftp.resolve_area("media"), "") == "."


def test_redundant_separators_and_dots_collapse():
    area = ftp.resolve_area("media")
    assert ftp._safe_path(area, "internal_000//./media/") == "internal_000/media"


# --- listing -----------------------------------------------------------------


UNIX_LINE = "-rw-r--r--    1 root     root       1048576 Jan  1 00:00 flightPlan.mavlink"
DIR_LINE = "drwxr-xr-x    2 root     root          4096 Jan  1 00:00 internal_000"


def test_a_unix_list_line_parses():
    entry = ftp._parse_line(UNIX_LINE)
    assert entry.name == "flightPlan.mavlink"
    assert entry.size == 1048576
    assert entry.is_dir is False


def test_a_directory_line_is_marked_as_one():
    assert ftp._parse_line(DIR_LINE).is_dir is True


def test_an_unrecognised_line_still_yields_a_name():
    entry = ftp._parse_line("weird-output.txt")
    assert entry.name == "weird-output.txt"
    assert entry.size is None
    assert entry.raw == "weird-output.txt"


# --- a fake ftpd -------------------------------------------------------------


class FakeFTP:
    """Just enough ftplib surface, recording what was asked of it.

    `FakeFTP.config` is what a test uses to describe the ftpd it wants, since
    ftplib.FTP is constructed inside the module under test.
    """

    instances: ClassVar[list["FakeFTP"]] = []
    config: ClassVar[dict] = {}

    def __init__(self, timeout=None):
        self.timeout = timeout
        self.connected_to = None
        self.logged_in = False
        self.commands: list[str] = []
        self.quit_called = False
        self.listing = FakeFTP.config.get("listing", [DIR_LINE, UNIX_LINE])
        self.content = FakeFTP.config.get("content", b"x" * 64)
        self.declared_size = FakeFTP.config.get("declared_size")
        self.size_raises = FakeFTP.config.get("size_raises", False)
        FakeFTP.instances.append(self)

    def connect(self, host, port):
        self.connected_to = (host, port)

    def login(self, *a):
        self.logged_in = True

    def retrlines(self, cmd, callback):
        self.commands.append(cmd)
        for line in self.listing:
            callback(line)

    def retrbinary(self, cmd, callback):
        self.commands.append(cmd)
        for i in range(0, len(self.content), 16):
            callback(self.content[i : i + 16])

    def voidcmd(self, cmd):
        self.commands.append(cmd)

    def size(self, path):
        if self.size_raises:
            raise ftplib.error_perm("550 SIZE not understood")
        return self.declared_size

    def quit(self):
        self.quit_called = True

    def close(self):
        pass


@pytest.fixture
def fake_ftp(monkeypatch):
    FakeFTP.instances = []
    FakeFTP.config = {}
    monkeypatch.setattr(ftp.ftplib, "FTP", FakeFTP)
    return FakeFTP


def test_list_dir_connects_to_the_area_port_and_logs_in_anonymously(fake_ftp):
    entries = ftp.list_dir("flightplans", host="192.168.42.1")
    conn = fake_ftp.instances[0]
    assert conn.connected_to == ("192.168.42.1", 61)
    assert conn.logged_in
    assert conn.timeout == ftp.DEFAULT_TIMEOUT
    assert [e.name for e in entries] == ["internal_000", "flightPlan.mavlink"]
    assert conn.quit_called


def test_list_dir_scopes_the_logs_area_in_the_command_it_sends(fake_ftp):
    ftp.list_dir("logs", host="h")
    assert fake_ftp.instances[0].commands[0] == "LIST internal_000/Debug/current"


def test_a_refused_connection_explains_the_access_point(fake_ftp, monkeypatch):
    def refuse(self, host, port):
        raise OSError("connection refused")

    monkeypatch.setattr(FakeFTP, "connect", refuse)
    with pytest.raises(ftp.FtpError, match="access point"):
        ftp.list_dir("media", host="h")


# --- fetching ----------------------------------------------------------------


def test_fetch_streams_to_capture_dir_and_returns_the_path(fake_ftp, tmp_path):
    dest = ftp.fetch("flightplans", "flightPlan.mavlink", host="h", capture_dir=tmp_path)
    assert dest == tmp_path / "flightplans" / "flightPlan.mavlink"
    assert dest.read_bytes() == b"x" * 64
    assert "RETR flightPlan.mavlink" in fake_ftp.instances[0].commands


def test_fetch_asks_for_the_size_in_binary_mode_first(fake_ftp, tmp_path):
    # busybox will not answer SIZE in ASCII mode, so TYPE I has to come first.
    ftp.fetch("media", "internal_000/a.jpg", host="h", capture_dir=tmp_path)
    cmds = fake_ftp.instances[0].commands
    assert cmds[0] == "TYPE I"


def test_a_file_the_drone_declares_as_oversized_is_never_transferred(fake_ftp, tmp_path):
    fake_ftp.config["declared_size"] = 50_000_000
    with pytest.raises(ftp.TooLarge, match="cap"):
        ftp.fetch("media", "internal_000/big.mp4", host="h", capture_dir=tmp_path)
    assert not any(c.startswith("RETR") for c in fake_ftp.instances[0].commands)
    assert not (tmp_path / "media" / "big.mp4").exists()


def test_a_cap_passed_mid_transfer_aborts_and_removes_the_partial_file(fake_ftp, tmp_path):
    # The second line of defence: an ftpd that will not answer SIZE still
    # cannot stream a video into memory, because the callback counts bytes.
    fake_ftp.config.update(content=b"y" * 4096, size_raises=True)
    with pytest.raises(ftp.TooLarge, match="mid-transfer"):
        ftp.fetch("media", "internal_000/big.mp4", host="h", capture_dir=tmp_path, max_bytes=100)
    assert not (tmp_path / "media" / "big.mp4").exists()


def test_fetch_refuses_a_directory(fake_ftp, tmp_path):
    with pytest.raises(ftp.FtpError, match="file path"):
        ftp.fetch("media", "", host="h", capture_dir=tmp_path)


def test_fetch_cannot_be_talked_out_of_its_area(fake_ftp, tmp_path):
    with pytest.raises(ftp.FtpError):
        ftp.fetch("logs", "../../internal_000/video.mp4", host="h", capture_dir=tmp_path)


def test_a_server_error_during_fetch_leaves_no_stub_file(fake_ftp, tmp_path, monkeypatch):
    def fail(self, cmd, callback):
        raise ftplib.error_perm("550 No such file")

    monkeypatch.setattr(FakeFTP, "retrbinary", fail)
    with pytest.raises(ftp.FtpError, match="Could not fetch"):
        ftp.fetch("media", "internal_000/ghost.jpg", host="h", capture_dir=tmp_path)
    assert not (tmp_path / "media" / "ghost.jpg").exists()