mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 19:21:33 +02:00
maziggy
b886590339
Function-scoped ftp_server meant 67 TLS server start/stop cycles. Class-scoped reduces to 10. Adds per-test cleanup fixture to reset failure injections and filesystem between tests. Isolates test_disconnect_after_server_gone into its own class to prevent close_all() from nuking other servers' asyncore sockets.
137 lines
4.2 KiB
Python
137 lines
4.2 KiB
Python
"""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 mode cache before and after each test."""
|
|
BambuFTPClient._mode_cache.clear()
|
|
yield
|
|
BambuFTPClient._mode_cache.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
|