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

"""Test fixtures for FTP service tests.

Provides a real implicit FTPS server (via mock_ftp_server) and client factory
for integration-style testing of BambuFTPClient against a live server.

The server fixture is class-scoped to avoid the overhead of starting a new
TLS server for every test (~67 TLS handshakes → ~9 per class).
"""

import os
import shutil
import socket
from unittest.mock import patch

import pytest

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

BAMBU_DIRS = ("cache", "timelapse", "model", "data", "data/Metadata")


@pytest.fixture(scope="session")
def ftp_certs(tmp_path_factory):
    """Generate self-signed TLS certificates once per test session."""
    cert_dir = tmp_path_factory.mktemp("ftp_certs")
    svc = CertificateService(cert_dir, serial="TEST_FTP_SERVER")
    cert_path, key_path = svc.generate_certificates()
    return str(cert_path), str(key_path)


def _find_free_port() -> int:
    """Find a free TCP port on localhost."""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


@pytest.fixture(scope="class")
def ftp_root(tmp_path_factory):
    """Create temp directory with standard Bambu printer directory structure."""
    root = tmp_path_factory.mktemp("ftp_root")
    for d in BAMBU_DIRS:
        (root / d).mkdir(parents=True, exist_ok=True)
    return root


@pytest.fixture(scope="class")
def ftp_server(ftp_certs, ftp_root):
    """Start a mock implicit FTPS server, yield it, stop on cleanup."""
    cert_path, key_path = ftp_certs
    port = _find_free_port()
    server = MockBambuFTPServer(
        host="127.0.0.1",
        port=port,
        root_dir=str(ftp_root),
        cert_path=cert_path,
        key_path=key_path,
        access_code="12345678",
    )
    server.start()
    yield server
    server.stop()


@pytest.fixture(autouse=True)
def _ftp_test_cleanup(request):
    """Reset server state between tests within a class.

    Clears injected failures and restores the Bambu directory structure
    so each test starts with a clean filesystem.  Skips cleanup for test
    classes that don't use the class-scoped ftp_server (e.g.
    TestDisconnectServerGone).
    """
    yield
    # Only clean up if this test class uses the class-scoped fixtures
    ftp_root = request.node.funcargs.get("ftp_root")
    if ftp_root is None:
        return
    server = request.node.funcargs.get("ftp_server")
    if server is not None:
        server.clear_failures()
    # Restore clean directory structure
    root = str(ftp_root)
    for entry in os.listdir(root):
        path = os.path.join(root, entry)
        if os.path.isdir(path):
            shutil.rmtree(path)
        else:
            os.remove(path)
    for d in BAMBU_DIRS:
        os.makedirs(os.path.join(root, d), exist_ok=True)


@pytest.fixture()
def ftp_client_factory(ftp_server):
    """Factory that creates BambuFTPClient instances pointed at the mock server."""

    def _make_client(
        printer_model: str = "X1C",
        force_prot_c: bool = False,
        access_code: str = "12345678",
        timeout: float = 10.0,
    ) -> BambuFTPClient:
        client = BambuFTPClient(
            ip_address="127.0.0.1",
            access_code=access_code,
            timeout=timeout,
            printer_model=printer_model,
            force_prot_c=force_prot_c,
        )
        # Override port to point at mock server
        client.FTP_PORT = ftp_server.port
        return client

    return _make_client


@pytest.fixture(autouse=True)
def clear_ftp_mode_cache():
    """Clear BambuFTPClient's per-printer caches before and after each test.

    Both are class-level dicts keyed by IP, and every test here talks to
    127.0.0.1 — a handshake cool-off left behind by one test would make the
    next one's ``connect()`` return False without touching the server (#2780).
    """
    BambuFTPClient._mode_cache.clear()
    BambuFTPClient._handshake_blocked_until.clear()
    yield
    BambuFTPClient._mode_cache.clear()
    BambuFTPClient._handshake_blocked_until.clear()


@pytest.fixture()
def patch_ftp_port(ftp_server):
    """Patch FTP_PORT at class level for async wrapper tests.

    Async wrappers create their own BambuFTPClient instances internally,
    so we need to patch the class-level default port.
    """
    with patch.object(BambuFTPClient, "FTP_PORT", ftp_server.port):
        yield ftp_server