bambuddy/backend/tests/unit/services/test_ftp_session_close_logging_3009.py

"""Every FTP session the client opens says how it closed (#3009).

The reporter of #3009 read a print-completion trace that showed two FTP
connects, one DELE and then nothing, and concluded the connections were never
closed -- the SD-card corruption they were chasing being the consequence.

They were closed. ``disconnect()`` and ``_abandon_connection()`` simply logged
nothing at any level, so a clean close and a genuinely leaked socket produced
the same log: silence. These tests pin the close line down, because a
diagnostic that only exists until someone tidies it away is worth nothing to
the next person reading a support bundle.
"""

import logging

import pytest

from backend.app.services.bambu_ftp import BambuFTPClient
from backend.tests.unit.services.mock_ftp_server import MockBambuFTPServer

from .conftest import _find_free_port


def _close_lines(caplog) -> list[str]:
    return [r.getMessage() for r in caplog.records if "FTP session to" in r.getMessage()]


class TestACleanSessionSaysSo:
    """The ordinary path: connect, work, QUIT."""

    def test_a_clean_close_is_logged_once(self, ftp_client_factory, caplog):
        client = ftp_client_factory()
        assert client.connect() is True
        with caplog.at_level(logging.DEBUG, logger="backend.app.services.bambu_ftp"):
            client.disconnect()

        lines = _close_lines(caplog)
        assert len(lines) == 1, lines
        assert "closed after QUIT" in lines[0]
        assert "127.0.0.1" in lines[0]

    def test_the_line_carries_how_long_the_session_was_held(self, ftp_client_factory, caplog):
        """Without a duration the line cannot distinguish a short delete from a
        session that sat open for the length of a print -- which is the exact
        question #3009 asked."""
        client = ftp_client_factory()
        client.connect()
        with caplog.at_level(logging.DEBUG, logger="backend.app.services.bambu_ftp"):
            client.disconnect()

        assert "held 0." in _close_lines(caplog)[0]

    def test_a_delete_through_the_async_wrapper_closes_and_says_so(self, ftp_server, ftp_root, caplog):
        """The path #3009 actually traced: the post-print SD-card cleanup in
        ``on_print_complete`` calls ``delete_file_async`` once per candidate."""
        import asyncio

        from backend.app.services.bambu_ftp import DeleteResult, delete_file_async

        (ftp_root / "cube.gcode").write_bytes(b"G28\n")
        original_port = BambuFTPClient.FTP_PORT
        BambuFTPClient.FTP_PORT = ftp_server.port
        try:
            with caplog.at_level(logging.DEBUG, logger="backend.app.services.bambu_ftp"):
                result = asyncio.run(delete_file_async("127.0.0.1", "12345678", "/cube.gcode", printer_model="X1C"))
        finally:
            BambuFTPClient.FTP_PORT = original_port

        assert result == DeleteResult.DELETED
        assert len(_close_lines(caplog)) == 1

    def test_the_550_path_closes_too(self, ftp_server, caplog):
        """The line #3009's log ends on. A candidate the printer does not have
        answers 550, and that session has to close like any other."""
        import asyncio

        from backend.app.services.bambu_ftp import DeleteResult, delete_file_async

        original_port = BambuFTPClient.FTP_PORT
        BambuFTPClient.FTP_PORT = ftp_server.port
        try:
            with caplog.at_level(logging.DEBUG, logger="backend.app.services.bambu_ftp"):
                result = asyncio.run(delete_file_async("127.0.0.1", "12345678", "/not_here.3mf", printer_model="X1C"))
        finally:
            BambuFTPClient.FTP_PORT = original_port

        assert result == DeleteResult.NOT_FOUND
        lines = _close_lines(caplog)
        assert len(lines) == 1, lines
        assert "closed after QUIT" in lines[0]

    def test_disconnect_without_a_session_says_nothing(self, ftp_client_factory, caplog):
        """No socket was opened, so there is no session to account for. A line
        here would be worse than none: it would pair with no connect."""
        client = ftp_client_factory()
        with caplog.at_level(logging.DEBUG, logger="backend.app.services.bambu_ftp"):
            client.disconnect()

        assert _close_lines(caplog) == []


class TestAFailedConnectIsAccountedForToo:
    """A connect that opens a socket and then fails still closed something."""

    def test_a_rejected_login_reports_the_close(self, ftp_client_factory, caplog):
        with caplog.at_level(logging.DEBUG, logger="backend.app.services.bambu_ftp"):
            assert ftp_client_factory(access_code="wrongcode").connect() is False

        lines = _close_lines(caplog)
        assert len(lines) == 1, lines
        assert "closed without QUIT" in lines[0]
        assert "login rejected" in lines[0]

    def test_an_unreachable_printer_reports_the_close(self, ftp_server, caplog):
        client = BambuFTPClient("192.0.2.1", "12345678", timeout=1.0, printer_model="X1C")
        client.FTP_PORT = ftp_server.port
        with caplog.at_level(logging.DEBUG, logger="backend.app.services.bambu_ftp"):
            assert client.connect() is False

        lines = _close_lines(caplog)
        assert len(lines) == 1, lines
        assert "closed without QUIT" in lines[0]
        # No socket was ever established, so there is no duration to claim.
        assert "held unknown" in lines[0]


class TestTheSessionIsNotDoubleCounted:
    """Isolated class: ``server.stop()`` calls ``close_all()``, which nukes every
    asyncore socket in the process."""

    def test_a_failing_quit_reports_one_close_not_two(self, ftp_certs, tmp_path, caplog):
        """``disconnect()`` falls through to ``_abandon_connection()`` when QUIT
        cannot be sent. Both log, so the fallback must not produce a second line
        for one session."""
        cert_path, key_path = ftp_certs
        server = MockBambuFTPServer("127.0.0.1", _find_free_port(), str(tmp_path), cert_path, key_path)
        server.start()

        client = BambuFTPClient("127.0.0.1", "12345678", timeout=5.0)
        client.FTP_PORT = server.port
        assert client.connect() is True

        server.stop()
        with caplog.at_level(logging.DEBUG, logger="backend.app.services.bambu_ftp"):
            client.disconnect()

        lines = _close_lines(caplog)
        assert len(lines) == 1, lines
        assert "closed without QUIT" in lines[0]
        assert "QUIT failed" in lines[0]
        assert client._ftp is None


class TestTheCoolOffSkipStaysSilent:
    """No connect was attempted, so there is nothing to close."""

    def test_a_skipped_connect_logs_no_close(self, ftp_client_factory, caplog):
        import time

        BambuFTPClient._handshake_blocked_until["127.0.0.1"] = time.monotonic() + 300
        client = ftp_client_factory()
        with caplog.at_level(logging.DEBUG, logger="backend.app.services.bambu_ftp"):
            assert client.connect() is False

        assert _close_lines(caplog) == []