mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
@@ -102,7 +102,7 @@ jobs:
|
||||
timeout-minutes: 10
|
||||
run: |
|
||||
cd backend
|
||||
python -m pytest tests/ -v --tb=short --timeout=60 --timeout-method=thread
|
||||
python -m pytest tests/ -v --tb=short --timeout=60 --timeout-method=thread -n auto
|
||||
|
||||
# ============================================================================
|
||||
# Frontend Checks
|
||||
|
||||
@@ -2,8 +2,13 @@
|
||||
|
||||
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
|
||||
|
||||
@@ -13,6 +18,8 @@ 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):
|
||||
@@ -30,15 +37,16 @@ def _find_free_port() -> int:
|
||||
return s.getsockname()[1]
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def ftp_root(tmp_path):
|
||||
@pytest.fixture(scope="class")
|
||||
def ftp_root(tmp_path_factory):
|
||||
"""Create temp directory with standard Bambu printer directory structure."""
|
||||
for d in ("cache", "timelapse", "model", "data", "data/Metadata"):
|
||||
(tmp_path / d).mkdir(parents=True, exist_ok=True)
|
||||
return tmp_path
|
||||
root = tmp_path_factory.mktemp("ftp_root")
|
||||
for d in BAMBU_DIRS:
|
||||
(root / d).mkdir(parents=True, exist_ok=True)
|
||||
return root
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
@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
|
||||
@@ -56,6 +64,35 @@ def ftp_server(ftp_certs, ftp_root):
|
||||
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."""
|
||||
|
||||
@@ -85,32 +85,6 @@ class TestConnection:
|
||||
client.disconnect() # Should not raise
|
||||
assert client._ftp is None
|
||||
|
||||
def test_disconnect_after_server_gone(self, ftp_certs, ftp_root):
|
||||
"""Disconnect after server has stopped raises EOFError.
|
||||
|
||||
Note: The current disconnect() catches (OSError, ftplib.Error) but
|
||||
EOFError is neither. This documents actual behavior — a future fix
|
||||
could add EOFError to the except clause.
|
||||
"""
|
||||
from backend.tests.unit.services.mock_ftp_server import (
|
||||
MockBambuFTPServer,
|
||||
)
|
||||
|
||||
from .conftest import _find_free_port
|
||||
|
||||
cert_path, key_path = ftp_certs
|
||||
port = _find_free_port()
|
||||
server = MockBambuFTPServer("127.0.0.1", port, str(ftp_root), cert_path, key_path)
|
||||
server.start()
|
||||
|
||||
client = BambuFTPClient("127.0.0.1", "12345678", timeout=5.0)
|
||||
client.FTP_PORT = port
|
||||
client.connect()
|
||||
|
||||
server.stop()
|
||||
with pytest.raises(EOFError):
|
||||
client.disconnect()
|
||||
|
||||
def test_x1c_uses_prot_p(self, ftp_client_factory):
|
||||
"""X1C model connects with prot_p (protected data channel)."""
|
||||
client = ftp_client_factory(printer_model="X1C")
|
||||
@@ -141,6 +115,40 @@ class TestConnection:
|
||||
client.disconnect()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestDisconnectServerGone — isolated class because server.stop() calls
|
||||
# close_all() which nukes all asyncore sockets globally.
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestDisconnectServerGone:
|
||||
"""Test disconnect behavior when the server has stopped."""
|
||||
|
||||
def test_disconnect_after_server_gone(self, ftp_certs, tmp_path):
|
||||
"""Disconnect after server has stopped raises EOFError.
|
||||
|
||||
Note: The current disconnect() catches (OSError, ftplib.Error) but
|
||||
EOFError is neither. This documents actual behavior — a future fix
|
||||
could add EOFError to the except clause.
|
||||
"""
|
||||
from backend.tests.unit.services.mock_ftp_server import (
|
||||
MockBambuFTPServer,
|
||||
)
|
||||
|
||||
from .conftest import _find_free_port
|
||||
|
||||
cert_path, key_path = ftp_certs
|
||||
port = _find_free_port()
|
||||
server = MockBambuFTPServer("127.0.0.1", port, str(tmp_path), cert_path, key_path)
|
||||
server.start()
|
||||
|
||||
client = BambuFTPClient("127.0.0.1", "12345678", timeout=5.0)
|
||||
client.FTP_PORT = port
|
||||
client.connect()
|
||||
|
||||
server.stop()
|
||||
with pytest.raises(EOFError):
|
||||
client.disconnect()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestListFiles
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user