mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
maziggy
454457a0af
Everything the completion path needs to split a print's filament across the trays it fed from lived only in memory: the dispatched plate and slot-to-tray mapping, the spool-assignment snapshot, and the tray-change log. A print that outlived a restart lost all of it and fell back to what the printer reports at completion -- which, with AMS Filament Backup on, is the substitute tray. The whole print was charged to the spool that only finished it while the spool that ran dry was charged nothing. Persist that context in a new active_print_sessions row, append tray changes as they happen, and restore both the session and the printer's tray-change log at restart recovery. Seed the log from the current tray when there is nothing to restore, since last_loaded_tray advances even when no change is logged. Rank the queue item's stored ams_mapping above the printer's live mapping field, which is what backup rewrites. Recover plate_id from the archive or queue item, and give extract_layer_filament_usage_from_3mf a plate_id instead of taking the first .gcode member -- a Bambu Studio export stores plate 2 first, so per-layer figures were measured against the wrong plate for both inventory backends. Stop auto-unlinking a spool assignment when its slot reports empty during a running print. At a runout the spool is still in the AMS, and dropping the link leaves the completion path nothing to charge. Capture the print-start context for both inventory backends. Spoolman's own durable row (#1820) carries its plate-scoped figures and dispatched mapping but not the tray-change log, and its slot assignments -- the way. Registration in _active_sessions stays gated, since on_ams_change reads it to decide whether to skip the remain%-based weight sync (#880).
802 lines
29 KiB
Python
802 lines
29 KiB
Python
"""Shared test fixtures for BamBuddy backend tests."""
|
|
|
|
import asyncio
|
|
import atexit
|
|
import json
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from collections.abc import AsyncGenerator
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# IMPORTANT: Set environment variables BEFORE any app imports
|
|
# This must happen before settings/config are loaded
|
|
os.environ["LOG_TO_FILE"] = "false"
|
|
os.environ["DEBUG"] = "false"
|
|
|
|
# Point the app's own engine at a throwaway database before anything reads
|
|
# DATABASE_URL.
|
|
#
|
|
# The fixtures below build their own SQLite engine, but that is not the only
|
|
# engine in play: `core/config.py` snapshots ``DATABASE_URL`` at import time and
|
|
# `core/database.py` builds a module-level ``engine`` / ``async_session`` from
|
|
# it. Any app code that opens its own session rather than receiving the fixture
|
|
# one therefore talks to whatever database the developer's `.env` names. The
|
|
# clearest example is ``run_with_retry`` (used by the print-completion path),
|
|
# whose sessions come from ``backend.app.core.database`` — so the widespread
|
|
# ``patch("backend.app.main.async_session")`` does not intercept them.
|
|
#
|
|
# Left alone that is not a hypothetical: on a plain checkout it means the suite
|
|
# writes to the developer's real SQLite file, and with a PostgreSQL `.env` it
|
|
# means a live install. A completion test calling ``on_print_complete(1, ...)``
|
|
# closed a queue item belonging to an actual running print that way.
|
|
_TEST_APP_DB_DIR = Path(tempfile.mkdtemp(prefix="bambuddy_test_appdb_"))
|
|
APP_DATABASE_URL = f"sqlite+aiosqlite:///{_TEST_APP_DB_DIR / 'app.db'}"
|
|
os.environ["DATABASE_URL"] = APP_DATABASE_URL
|
|
|
|
|
|
def _cleanup_test_app_db_dir():
|
|
shutil.rmtree(_TEST_APP_DB_DIR, ignore_errors=True)
|
|
|
|
|
|
atexit.register(_cleanup_test_app_db_dir)
|
|
|
|
|
|
def _assert_disposable_database(url, source: str) -> None:
|
|
"""Abort the run unless *url* is the throwaway database created above.
|
|
|
|
A guard rather than a comment because the failure it prevents is silent and
|
|
destructive: the suite would appear to pass while having mutated real print
|
|
history. Anything that reintroduces a real ``DATABASE_URL`` — an `.env` read
|
|
later in the import order, a fixture rebuilding the engine — trips this
|
|
instead of reaching the database.
|
|
"""
|
|
database = str(getattr(url, "database", "") or "")
|
|
if not str(getattr(url, "drivername", "")).startswith("sqlite") or not database.startswith(str(_TEST_APP_DB_DIR)):
|
|
raise RuntimeError(
|
|
f"Refusing to run tests: {source} resolves to {url!r}, which is not the "
|
|
f"disposable SQLite database under {_TEST_APP_DB_DIR}. Tests must never "
|
|
f"open a session against a real Bambuddy database."
|
|
)
|
|
|
|
|
|
from httpx import ASGITransport, AsyncClient # noqa: E402
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine # noqa: E402
|
|
|
|
# Ensure settings use our env vars - import and override before database import
|
|
from backend.app.core.config import settings # noqa: E402
|
|
|
|
settings.log_to_file = False
|
|
if settings.database_url != APP_DATABASE_URL:
|
|
raise RuntimeError(
|
|
f"Refusing to run tests: settings.database_url is {settings.database_url!r} "
|
|
f"rather than the disposable test database. Something read DATABASE_URL "
|
|
f"before conftest could override it."
|
|
)
|
|
|
|
# Use a temp directory for plate calibration to avoid deleting real calibration files
|
|
_test_plate_cal_dir = Path(tempfile.mkdtemp(prefix="bambuddy_test_plate_cal_"))
|
|
settings.plate_calibration_dir = _test_plate_cal_dir
|
|
|
|
|
|
# Clean up temp directory when tests finish
|
|
def _cleanup_test_plate_cal_dir():
|
|
if _test_plate_cal_dir.exists():
|
|
shutil.rmtree(_test_plate_cal_dir, ignore_errors=True)
|
|
|
|
|
|
atexit.register(_cleanup_test_plate_cal_dir)
|
|
|
|
from backend.app.core.database import Base, engine as _app_engine # noqa: E402
|
|
|
|
# The engine is built at import time from the URL above, so this catches the
|
|
# case where that override did not take effect for whatever reason.
|
|
_assert_disposable_database(_app_engine.url, "backend.app.core.database.engine")
|
|
|
|
# Use in-memory SQLite for tests
|
|
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mfa_encryption_isolation(monkeypatch, tmp_path):
|
|
"""Per-test isolation for MFA encryption state.
|
|
|
|
- Sets ``DATA_DIR`` to an isolated tmp path so the auto-bootstrap can
|
|
never write ``.mfa_encryption_key`` into the repo or share state
|
|
across tests / xdist workers.
|
|
- Removes any inherited ``MFA_ENCRYPTION_KEY`` env var.
|
|
- With ``DATA_DIR`` pointing at a writable ``tmp_path``, the default
|
|
bootstrap path on first ``_get_fernet()`` call is **auto-generation**
|
|
(key_source='generated'), NOT plaintext fallback. Tests that need the
|
|
plaintext fallback path must monkeypatch ``_load_or_generate_key`` to
|
|
return ``(None, 'none')`` (or 'none_write_failed' / 'none_corrupted')
|
|
explicitly — see ``test_plaintext_passthrough_without_key`` for an
|
|
example.
|
|
- Resets the ``encryption`` module-level singletons before AND after the
|
|
test so reorder doesn't leak cached Fernet instances.
|
|
|
|
Tests that want to exercise an active key should call
|
|
``monkeypatch.setenv("MFA_ENCRYPTION_KEY", valid_key)`` and
|
|
``enc_mod._fernet_instance = None`` inside the test body — the autouse
|
|
fixture only sets defaults, it doesn't lock them in.
|
|
"""
|
|
from backend.app.core import encryption as enc_mod
|
|
|
|
monkeypatch.setenv("DATA_DIR", str(tmp_path))
|
|
monkeypatch.delenv("MFA_ENCRYPTION_KEY", raising=False)
|
|
enc_mod._fernet_instance = None
|
|
enc_mod._warn_shown = False
|
|
enc_mod._key_source = None
|
|
yield
|
|
enc_mod._fernet_instance = None
|
|
enc_mod._warn_shown = False
|
|
enc_mod._key_source = None
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_spoolman_location_sync_cache():
|
|
"""Drop the per-URL Spoolman location-sync TTL cache between tests.
|
|
|
|
Without this, a test that runs the sync against `http://localhost:7912`
|
|
will skip the sync in any later test that uses the same URL within 60
|
|
real seconds — test ordering would then leak assertions across runs."""
|
|
from backend.app.services.location_service import _spoolman_location_sync_cache_clear
|
|
|
|
_spoolman_location_sync_cache_clear()
|
|
yield
|
|
_spoolman_location_sync_cache_clear()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_auth_enabled_cache():
|
|
"""Drop the module-level auth-enabled cache between tests (issue #2572).
|
|
|
|
``is_auth_enabled`` caches an enabled=True result for a TTL. Without this
|
|
reset a test that enables auth would leave ``True`` cached, so a later test
|
|
running in auth-disabled mode (without going through ``set_auth_enabled``)
|
|
would wrongly see auth as enabled until the TTL expired — order-dependent
|
|
flakiness."""
|
|
from backend.app.core.auth import invalidate_auth_enabled_cache
|
|
|
|
invalidate_auth_enabled_cache()
|
|
yield
|
|
invalidate_auth_enabled_cache()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def event_loop():
|
|
"""Create an instance of the default event loop for each test session."""
|
|
loop = asyncio.get_event_loop_policy().new_event_loop()
|
|
yield loop
|
|
# Dispose the module-level engine so aiosqlite worker threads finish
|
|
# before the event loop closes, preventing "Event loop is closed" errors.
|
|
from backend.app.core.database import engine
|
|
|
|
loop.run_until_complete(engine.dispose())
|
|
loop.run_until_complete(asyncio.sleep(0.05))
|
|
loop.close()
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_engine():
|
|
"""Create a test database engine."""
|
|
engine = create_async_engine(TEST_DATABASE_URL, echo=False)
|
|
|
|
# Import all models to register them
|
|
from backend.app.models import (
|
|
active_print_session, # noqa: F401
|
|
ams_history,
|
|
ams_label,
|
|
api_key,
|
|
archive,
|
|
auth_ephemeral,
|
|
color_catalog,
|
|
external_link,
|
|
filament,
|
|
group,
|
|
kprofile_note,
|
|
maintenance,
|
|
notification,
|
|
notification_template,
|
|
oidc_provider,
|
|
print_log,
|
|
print_queue,
|
|
printer,
|
|
project,
|
|
project_bom,
|
|
settings,
|
|
slot_preset,
|
|
smart_plug,
|
|
smart_plug_energy_snapshot, # noqa: F401
|
|
sponsor_toast_state, # noqa: F401
|
|
spool,
|
|
spool_assignment,
|
|
spool_catalog,
|
|
spool_k_profile,
|
|
spool_usage_history,
|
|
spoolbuddy_device,
|
|
spoolman_k_profile,
|
|
spoolman_slot_assignment,
|
|
user,
|
|
user_email_pref,
|
|
user_otp_code,
|
|
user_totp,
|
|
virtual_printer,
|
|
)
|
|
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
yield engine
|
|
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.drop_all)
|
|
await engine.dispose()
|
|
# Allow aiosqlite's background thread to finish processing the close
|
|
# response before the per-function event loop shuts down, preventing
|
|
# "RuntimeError: Event loop is closed" in call_soon_threadsafe.
|
|
await asyncio.sleep(0.1)
|
|
|
|
|
|
@pytest.fixture
|
|
async def db_session(test_engine) -> AsyncGenerator[AsyncSession, None]:
|
|
"""Create a test database session."""
|
|
async_session_maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
|
|
async with async_session_maker() as session:
|
|
yield session
|
|
|
|
|
|
@pytest.fixture
|
|
async def async_client(test_engine, db_session) -> AsyncGenerator[AsyncClient, None]:
|
|
"""Create an async test client."""
|
|
from backend.app.core.database import async_session, get_db
|
|
from backend.app.main import app
|
|
|
|
# Create a new session maker for the test engine
|
|
test_async_session = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
|
async def override_get_db():
|
|
# Mirror production get_db (core/database.py): commit on success,
|
|
# rollback on error. Endpoints that rely on the request-scoped
|
|
# implicit commit (e.g. create_project, which only flushes) would
|
|
# otherwise silently lose their writes in tests (#1897).
|
|
async with test_async_session() as session:
|
|
try:
|
|
yield session
|
|
await session.commit()
|
|
except BaseException:
|
|
await session.rollback()
|
|
raise
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
|
|
# Mock init_printer_connections to prevent MQTT connection attempts during tests
|
|
async def mock_init_printer_connections(db):
|
|
pass # No-op - don't connect to real printers
|
|
|
|
# Also patch the module-level async_session used by services, auth, and middleware
|
|
with (
|
|
patch("backend.app.core.database.async_session", test_async_session),
|
|
patch("backend.app.core.auth.async_session", test_async_session),
|
|
patch("backend.app.main.async_session", test_async_session),
|
|
# Obico endpoints load settings through the service's module-level binding;
|
|
# without this patch they'd read whatever DB the cwd resolves to (#1546).
|
|
patch("backend.app.services.obico_detection.async_session", test_async_session),
|
|
patch("backend.app.main.init_printer_connections", mock_init_printer_connections),
|
|
):
|
|
# Seed default groups for tests that need them
|
|
from backend.app.core.database import seed_default_groups
|
|
|
|
await seed_default_groups()
|
|
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
|
yield client
|
|
|
|
# The app lifespan called init_db() which used the module-level engine
|
|
# (not the test engine), creating aiosqlite connections. Dispose those
|
|
# connections so their background threads finish before the event loop closes.
|
|
from backend.app.core.database import engine as real_engine
|
|
|
|
await real_engine.dispose()
|
|
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
# ============================================================================
|
|
# Mock External Services
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_tasmota_service():
|
|
"""Mock the Tasmota service for smart plug tests."""
|
|
# Patch both the module where it's defined and where it's imported
|
|
with (
|
|
patch("backend.app.services.tasmota.tasmota_service") as mock,
|
|
patch("backend.app.api.routes.smart_plugs.tasmota_service") as mock2,
|
|
):
|
|
mock.turn_on = AsyncMock(return_value=True)
|
|
mock.turn_off = AsyncMock(return_value=True)
|
|
mock.toggle = AsyncMock(return_value=True)
|
|
mock.get_status = AsyncMock(return_value={"state": "ON", "reachable": True, "device_name": "Test Plug"})
|
|
mock.get_energy = AsyncMock(
|
|
return_value={
|
|
"power": 150.5,
|
|
"voltage": 120.0,
|
|
"current": 1.25,
|
|
"today": 2.5,
|
|
"total": 100.0,
|
|
"factor": 0.95,
|
|
}
|
|
)
|
|
mock.test_connection = AsyncMock(return_value={"success": True, "state": "ON", "device_name": "Test Plug"})
|
|
# Copy mocks to second patch target
|
|
mock2.turn_on = mock.turn_on
|
|
mock2.turn_off = mock.turn_off
|
|
mock2.toggle = mock.toggle
|
|
mock2.get_status = mock.get_status
|
|
mock2.get_energy = mock.get_energy
|
|
mock2.test_connection = mock.test_connection
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_homeassistant_service():
|
|
"""Mock the Home Assistant service for smart plug tests."""
|
|
# Patch both the module where it's defined and where it's imported
|
|
with (
|
|
patch("backend.app.services.homeassistant.homeassistant_service") as mock,
|
|
patch("backend.app.api.routes.smart_plugs.homeassistant_service") as mock2,
|
|
):
|
|
mock.turn_on = AsyncMock(return_value=True)
|
|
mock.turn_off = AsyncMock(return_value=True)
|
|
mock.toggle = AsyncMock(return_value=True)
|
|
mock.get_status = AsyncMock(return_value={"state": "ON", "reachable": True, "device_name": "Test HA Entity"})
|
|
mock.get_energy = AsyncMock(return_value=None) # Most HA entities don't have power monitoring
|
|
mock.test_connection = AsyncMock(return_value={"success": True, "message": "API running", "error": None})
|
|
mock.list_entities = AsyncMock(
|
|
return_value=[
|
|
{
|
|
"entity_id": "switch.printer_plug",
|
|
"friendly_name": "Printer Plug",
|
|
"state": "on",
|
|
"domain": "switch",
|
|
},
|
|
{"entity_id": "switch.test", "friendly_name": "Test Switch", "state": "off", "domain": "switch"},
|
|
]
|
|
)
|
|
mock.configure = MagicMock()
|
|
# Copy mocks to second patch target
|
|
mock2.turn_on = mock.turn_on
|
|
mock2.turn_off = mock.turn_off
|
|
mock2.toggle = mock.toggle
|
|
mock2.get_status = mock.get_status
|
|
mock2.get_energy = mock.get_energy
|
|
mock2.test_connection = mock.test_connection
|
|
mock2.list_entities = mock.list_entities
|
|
mock2.configure = mock.configure
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_mqtt_client():
|
|
"""Mock the MQTT client for printer communication tests."""
|
|
with patch("backend.app.services.bambu_mqtt.BambuMQTTClient") as mock:
|
|
instance = MagicMock()
|
|
instance.state = MagicMock(connected=True, state="IDLE", progress=0, temperatures={"nozzle": 25, "bed": 25})
|
|
instance.connect = MagicMock()
|
|
instance.disconnect = MagicMock()
|
|
mock.return_value = instance
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_mqtt_smart_plug_service():
|
|
"""Mock the MQTT smart plug service for MQTT plug tests."""
|
|
with patch("backend.app.api.routes.smart_plugs.mqtt_relay") as mock:
|
|
# Create a mock smart_plug_service
|
|
mock_service = MagicMock()
|
|
mock_service.is_configured = MagicMock(return_value=True)
|
|
mock_service.has_broker_settings = MagicMock(return_value=True)
|
|
mock_service.configure = AsyncMock(return_value=True)
|
|
mock_service.subscribe = MagicMock()
|
|
mock_service.unsubscribe = MagicMock()
|
|
mock_service.get_plug_data = MagicMock(return_value=None)
|
|
mock_service.is_reachable = MagicMock(return_value=False)
|
|
|
|
mock.smart_plug_service = mock_service
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_ftp_client():
|
|
"""Mock the FTP client for file transfer tests."""
|
|
with (
|
|
patch("backend.app.services.bambu_ftp.download_file_async") as download_mock,
|
|
patch("backend.app.services.bambu_ftp.list_files_async") as list_mock,
|
|
):
|
|
download_mock.return_value = True
|
|
list_mock.return_value = []
|
|
yield {"download": download_mock, "list": list_mock}
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_httpx_client():
|
|
"""Mock httpx for webhook/notification HTTP calls."""
|
|
with patch("httpx.AsyncClient") as mock_class:
|
|
mock_instance = AsyncMock()
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.text = "OK"
|
|
mock_response.json.return_value = {}
|
|
|
|
mock_instance.get = AsyncMock(return_value=mock_response)
|
|
mock_instance.post = AsyncMock(return_value=mock_response)
|
|
mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
|
|
mock_instance.__aexit__ = AsyncMock()
|
|
|
|
mock_class.return_value = mock_instance
|
|
yield mock_instance
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_printer_manager():
|
|
"""Mock the printer manager for status checks."""
|
|
with patch("backend.app.services.printer_manager.printer_manager") as mock:
|
|
mock.get_status = MagicMock(
|
|
return_value=MagicMock(
|
|
connected=True,
|
|
state="IDLE",
|
|
progress=0,
|
|
temperatures={"nozzle": 25, "bed": 25, "chamber": 25},
|
|
raw_data={},
|
|
)
|
|
)
|
|
mock.mark_printer_offline = MagicMock()
|
|
yield mock
|
|
|
|
|
|
# ============================================================================
|
|
# Factory Fixtures for Test Data
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def smart_plug_factory(db_session):
|
|
"""Factory to create test smart plugs."""
|
|
|
|
async def _create_plug(**kwargs):
|
|
from backend.app.models.smart_plug import SmartPlug
|
|
|
|
# Determine defaults based on plug_type
|
|
plug_type = kwargs.get("plug_type", "tasmota")
|
|
|
|
defaults = {
|
|
"name": "Test Plug",
|
|
"plug_type": plug_type,
|
|
"enabled": True,
|
|
"auto_on": True,
|
|
"auto_off": True,
|
|
"off_delay_mode": "time",
|
|
"off_delay_minutes": 5,
|
|
"off_temp_threshold": 70,
|
|
"schedule_enabled": False,
|
|
"power_alert_enabled": False,
|
|
}
|
|
|
|
# Set required fields based on plug_type
|
|
if plug_type == "homeassistant":
|
|
defaults["ha_entity_id"] = "switch.test"
|
|
defaults["ip_address"] = None
|
|
elif plug_type == "mqtt":
|
|
# Legacy fields (for backward compatibility tests)
|
|
defaults["mqtt_topic"] = kwargs.get("mqtt_topic", "test/topic")
|
|
defaults["mqtt_multiplier"] = kwargs.get("mqtt_multiplier", 1.0)
|
|
# New separate topic/path/multiplier fields
|
|
defaults["mqtt_power_topic"] = kwargs.get("mqtt_power_topic")
|
|
defaults["mqtt_power_path"] = kwargs.get("mqtt_power_path", "power")
|
|
defaults["mqtt_power_multiplier"] = kwargs.get("mqtt_power_multiplier", 1.0)
|
|
defaults["mqtt_energy_topic"] = kwargs.get("mqtt_energy_topic")
|
|
defaults["mqtt_energy_path"] = kwargs.get("mqtt_energy_path")
|
|
defaults["mqtt_energy_multiplier"] = kwargs.get("mqtt_energy_multiplier", 1.0)
|
|
defaults["mqtt_state_topic"] = kwargs.get("mqtt_state_topic")
|
|
defaults["mqtt_state_path"] = kwargs.get("mqtt_state_path")
|
|
defaults["mqtt_state_on_value"] = kwargs.get("mqtt_state_on_value")
|
|
defaults["ip_address"] = None
|
|
defaults["ha_entity_id"] = None
|
|
elif plug_type == "rest":
|
|
defaults["rest_on_url"] = kwargs.get("rest_on_url", "http://192.168.1.100/api/plug/on")
|
|
defaults["rest_off_url"] = kwargs.get("rest_off_url", "http://192.168.1.100/api/plug/off")
|
|
defaults["rest_method"] = kwargs.get("rest_method", "POST")
|
|
defaults["ip_address"] = None
|
|
defaults["ha_entity_id"] = None
|
|
else:
|
|
defaults["ip_address"] = "192.168.1.100"
|
|
defaults["ha_entity_id"] = None
|
|
|
|
defaults.update(kwargs)
|
|
|
|
plug = SmartPlug(**defaults)
|
|
db_session.add(plug)
|
|
await db_session.commit()
|
|
await db_session.refresh(plug)
|
|
return plug
|
|
|
|
return _create_plug
|
|
|
|
|
|
@pytest.fixture
|
|
def printer_factory(db_session):
|
|
"""Factory to create test printers."""
|
|
_counter = [0] # Use list to allow mutation in nested function
|
|
|
|
async def _create_printer(**kwargs):
|
|
from backend.app.models.printer import Printer
|
|
|
|
_counter[0] += 1
|
|
counter = _counter[0]
|
|
|
|
defaults = {
|
|
"name": "Test Printer",
|
|
"serial_number": f"00M09A{counter:09d}", # Unique serial per printer
|
|
"ip_address": f"192.168.1.{100 + counter}", # Unique IP per printer
|
|
"access_code": "12345678",
|
|
"is_active": True,
|
|
"auto_archive": True,
|
|
"model": "X1C",
|
|
}
|
|
defaults.update(kwargs)
|
|
|
|
printer = Printer(**defaults)
|
|
db_session.add(printer)
|
|
await db_session.commit()
|
|
await db_session.refresh(printer)
|
|
return printer
|
|
|
|
return _create_printer
|
|
|
|
|
|
@pytest.fixture
|
|
def notification_provider_factory(db_session):
|
|
"""Factory to create test notification providers."""
|
|
|
|
async def _create_provider(**kwargs):
|
|
from backend.app.models.notification import NotificationProvider
|
|
|
|
config = kwargs.pop("config", {"server": "https://ntfy.sh", "topic": "test-topic"})
|
|
if isinstance(config, dict):
|
|
config = json.dumps(config)
|
|
|
|
defaults = {
|
|
"name": "Test Provider",
|
|
"provider_type": "ntfy",
|
|
"enabled": True,
|
|
"config": config,
|
|
"on_print_start": True,
|
|
"on_print_complete": True,
|
|
"on_print_failed": True,
|
|
"on_print_stopped": True,
|
|
"on_print_progress": False,
|
|
"on_print_missing_spool_assignment": False,
|
|
"on_billing_charge_failed": True,
|
|
"on_printer_offline": False,
|
|
"on_printer_error": False,
|
|
"on_filament_low": False,
|
|
"on_maintenance_due": False,
|
|
"on_ams_humidity_high": False,
|
|
"on_ams_temperature_high": False,
|
|
"on_bed_cooled": False,
|
|
"quiet_hours_enabled": False,
|
|
"daily_digest_enabled": False,
|
|
}
|
|
defaults.update(kwargs)
|
|
|
|
provider = NotificationProvider(**defaults)
|
|
db_session.add(provider)
|
|
await db_session.commit()
|
|
await db_session.refresh(provider)
|
|
return provider
|
|
|
|
return _create_provider
|
|
|
|
|
|
@pytest.fixture
|
|
def archive_factory(db_session):
|
|
"""Factory to create test archives.
|
|
|
|
Also synthesizes one PrintLogEntry per archive (matching the production
|
|
flow where statistics are aggregated from PrintLogEntry, not PrintArchive,
|
|
per #1378). Pass ``with_run=False`` to skip — useful for testing the
|
|
"archived but never printed" state. Pass ``run_status=...`` to override
|
|
the run's status independently of the archive's status field.
|
|
"""
|
|
|
|
async def _create_archive(printer_id: int, **kwargs):
|
|
from backend.app.models.archive import PrintArchive
|
|
from backend.app.models.print_log import PrintLogEntry
|
|
|
|
with_run = kwargs.pop("with_run", True)
|
|
run_status = kwargs.pop("run_status", None)
|
|
|
|
defaults = {
|
|
"printer_id": printer_id,
|
|
"filename": "test_print.gcode.3mf",
|
|
"print_name": "Test Print",
|
|
"file_path": "archives/test/test_print.gcode.3mf",
|
|
"file_size": 1024000,
|
|
"status": "completed",
|
|
"filament_type": "PLA",
|
|
"filament_used_grams": 50.0,
|
|
"print_time_seconds": 3600,
|
|
}
|
|
defaults.update(kwargs)
|
|
|
|
archive = PrintArchive(**defaults)
|
|
db_session.add(archive)
|
|
await db_session.commit()
|
|
await db_session.refresh(archive)
|
|
|
|
if with_run:
|
|
duration = None
|
|
if archive.started_at and archive.completed_at:
|
|
duration = int((archive.completed_at - archive.started_at).total_seconds()) or None
|
|
run = PrintLogEntry(
|
|
archive_id=archive.id,
|
|
printer_id=archive.printer_id,
|
|
status=run_status or archive.status,
|
|
started_at=archive.started_at,
|
|
completed_at=archive.completed_at,
|
|
duration_seconds=duration,
|
|
filament_type=archive.filament_type,
|
|
filament_color=archive.filament_color,
|
|
filament_used_grams=archive.filament_used_grams,
|
|
cost=archive.cost,
|
|
energy_kwh=archive.energy_kwh,
|
|
energy_cost=archive.energy_cost,
|
|
failure_reason=archive.failure_reason,
|
|
print_name=archive.print_name,
|
|
created_by_id=archive.created_by_id,
|
|
# Sync the event's created_at with the archive's so date-range
|
|
# filtered tests that backdate an archive still find its event.
|
|
created_at=archive.created_at,
|
|
)
|
|
db_session.add(run)
|
|
await db_session.commit()
|
|
|
|
return archive
|
|
|
|
return _create_archive
|
|
|
|
|
|
# ============================================================================
|
|
# Sample Data Fixtures
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_mqtt_print_start():
|
|
"""Sample MQTT message for print start."""
|
|
return {
|
|
"print": {
|
|
"command": "project_file",
|
|
"param": "/sdcard/test.gcode.3mf",
|
|
"subtask_name": "test_print",
|
|
"gcode_state": "RUNNING",
|
|
"mc_percent": 0,
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_mqtt_print_complete():
|
|
"""Sample MQTT message for print complete."""
|
|
return {
|
|
"print": {
|
|
"gcode_state": "FINISH",
|
|
"mc_percent": 100,
|
|
"subtask_name": "test_print",
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_printer_status():
|
|
"""Sample printer status data."""
|
|
return {
|
|
"connected": True,
|
|
"state": "IDLE",
|
|
"progress": 0,
|
|
"layer_num": 0,
|
|
"total_layers": 0,
|
|
"temperatures": {
|
|
"nozzle": 25.0,
|
|
"bed": 25.0,
|
|
"chamber": 25.0,
|
|
},
|
|
"remaining_time": 0,
|
|
"filename": None,
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# Log Capture Fixtures for Error Detection
|
|
# ============================================================================
|
|
|
|
|
|
class LogCapture(logging.Handler):
|
|
"""Handler that captures log records for testing."""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.records: list[logging.LogRecord] = []
|
|
|
|
def emit(self, record: logging.LogRecord):
|
|
self.records.append(record)
|
|
|
|
def clear(self):
|
|
self.records.clear()
|
|
|
|
def get_errors(self) -> list[logging.LogRecord]:
|
|
"""Get all ERROR and CRITICAL level records."""
|
|
return [r for r in self.records if r.levelno >= logging.ERROR]
|
|
|
|
def get_warnings(self) -> list[logging.LogRecord]:
|
|
"""Get all WARNING level records."""
|
|
return [r for r in self.records if r.levelno == logging.WARNING]
|
|
|
|
def has_errors(self) -> bool:
|
|
"""Check if any errors were logged."""
|
|
return len(self.get_errors()) > 0
|
|
|
|
def format_errors(self) -> str:
|
|
"""Format all errors as a string for assertion messages."""
|
|
errors = self.get_errors()
|
|
if not errors:
|
|
return "No errors"
|
|
formatter = logging.Formatter("%(name)s - %(levelname)s - %(message)s")
|
|
return "\n".join(formatter.format(r) for r in errors)
|
|
|
|
|
|
@pytest.fixture
|
|
def capture_logs():
|
|
"""Fixture that captures log output during a test.
|
|
|
|
Usage:
|
|
def test_something(capture_logs):
|
|
# Do something that might log errors
|
|
some_function()
|
|
|
|
# Check no errors were logged
|
|
assert not capture_logs.has_errors(), capture_logs.format_errors()
|
|
"""
|
|
handler = LogCapture()
|
|
handler.setLevel(logging.DEBUG)
|
|
|
|
# Attach to root logger to capture all logs
|
|
root_logger = logging.getLogger()
|
|
root_logger.addHandler(handler)
|
|
|
|
yield handler
|
|
|
|
root_logger.removeHandler(handler)
|
|
|
|
|
|
@pytest.fixture
|
|
def assert_no_log_errors(capture_logs):
|
|
"""Fixture that automatically asserts no errors were logged.
|
|
|
|
Usage:
|
|
def test_something(assert_no_log_errors):
|
|
# If any ERROR logs occur during this test, it will fail
|
|
some_function()
|
|
"""
|
|
yield capture_logs
|
|
|
|
errors = capture_logs.get_errors()
|
|
if errors:
|
|
pytest.fail(f"Unexpected log errors:\n{capture_logs.format_errors()}")
|