Add comprehensive automated testing infrastructure

Backend:
  - pytest configuration with async support and coverage
  - Unit tests for notification service (23 tests)
  - Unit tests for smart plug manager (12 tests)
  - Unit tests for archive service (16 tests)
  - Integration tests for API endpoints
  - Fix: notifications now send immediately (digest is summary only)

  Frontend:
  - Vitest configuration with jsdom and coverage
  - MSW for API mocking
  - Component tests for Toggle, Button, Card, ConfirmModal (77 tests)
  - Test utilities with custom render wrapper

  CI/CD:
  - GitHub Actions workflow for automated testing
  - Backend lint, unit tests, integration tests
  - Frontend lint, type-check, unit tests, build
This commit is contained in:
maziggy
2025-12-11 10:03:40 +01:00
parent c3026632da
commit ff53e62ef8
126 changed files with 101955 additions and 10 deletions
+1
View File
@@ -0,0 +1 @@
"""BamBuddy backend tests."""
+380
View File
@@ -0,0 +1,380 @@
"""Shared test fixtures for BamBuddy backend tests."""
import asyncio
import json
import os
import sys
import pytest
from typing import AsyncGenerator
from datetime import datetime
from unittest.mock import AsyncMock, MagicMock, patch
# 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"
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from httpx import AsyncClient, ASGITransport
# Ensure settings use our env vars - import and override before database import
from backend.app.core.config import settings
settings.log_to_file = False
from backend.app.core.database import Base
# Use in-memory SQLite for tests
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
@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
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 (
printer, archive, filament, settings, smart_plug,
print_queue, notification, maintenance, kprofile_note,
notification_template, external_link, project, api_key,
ams_history
)
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()
@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.main import app
from backend.app.core.database import get_db, async_session
# 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():
async with test_async_session() as session:
yield session
app.dependency_overrides[get_db] = override_get_db
# Also patch the module-level async_session used by services
with patch('backend.app.core.database.async_session', test_async_session):
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as client:
yield client
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_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_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
defaults = {
"name": "Test Plug",
"ip_address": "192.168.1.100",
"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,
}
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_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,
"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."""
async def _create_archive(printer_id: int, **kwargs):
from backend.app.models.archive import PrintArchive
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)
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,
}
+1
View File
@@ -0,0 +1 @@
"""Integration tests for BamBuddy API endpoints."""
@@ -0,0 +1,189 @@
"""Integration tests for AMS History API endpoints."""
import pytest
from datetime import datetime, timedelta
from httpx import AsyncClient
class TestAMSHistoryAPI:
"""Integration tests for /api/v1/ams-history endpoints."""
@pytest.fixture
async def ams_history_factory(self, db_session, printer_factory):
"""Factory to create test AMS history records."""
async def _create_history(printer_id=None, ams_id=0, **kwargs):
from backend.app.models.ams_history import AMSSensorHistory
if printer_id is None:
printer = await printer_factory()
printer_id = printer.id
defaults = {
"printer_id": printer_id,
"ams_id": ams_id,
"humidity": 45.0,
"humidity_raw": 4500,
"temperature": 25.0,
"recorded_at": datetime.now(),
}
defaults.update(kwargs)
history = AMSSensorHistory(**defaults)
db_session.add(history)
await db_session.commit()
await db_session.refresh(history)
return history
return _create_history
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_ams_history_empty(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify empty history returns empty data array."""
printer = await printer_factory()
response = await async_client.get(f"/api/v1/ams-history/{printer.id}/0")
assert response.status_code == 200
data = response.json()
assert data["printer_id"] == printer.id
assert data["ams_id"] == 0
assert data["data"] == []
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_ams_history_with_data(
self, async_client: AsyncClient, ams_history_factory, db_session
):
"""Verify history returns recorded data."""
# Create history records
history = await ams_history_factory()
printer_id = history.printer_id
response = await async_client.get(f"/api/v1/ams-history/{printer_id}/0")
assert response.status_code == 200
data = response.json()
assert len(data["data"]) >= 1
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_ams_history_with_stats(
self, async_client: AsyncClient, ams_history_factory, printer_factory, db_session
):
"""Verify history includes statistics."""
printer = await printer_factory()
# Create multiple records with different values
await ams_history_factory(printer_id=printer.id, humidity=40.0, temperature=24.0)
await ams_history_factory(printer_id=printer.id, humidity=50.0, temperature=26.0)
await ams_history_factory(printer_id=printer.id, humidity=45.0, temperature=25.0)
response = await async_client.get(f"/api/v1/ams-history/{printer.id}/0")
assert response.status_code == 200
data = response.json()
# Check statistics
assert data["min_humidity"] == 40.0
assert data["max_humidity"] == 50.0
assert data["min_temperature"] == 24.0
assert data["max_temperature"] == 26.0
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_ams_history_with_hours_filter(
self, async_client: AsyncClient, ams_history_factory, printer_factory, db_session
):
"""Verify hours parameter filters data."""
printer = await printer_factory()
# Create a recent record
await ams_history_factory(
printer_id=printer.id,
recorded_at=datetime.now()
)
# Create an old record (outside default 24h)
await ams_history_factory(
printer_id=printer.id,
recorded_at=datetime.now() - timedelta(hours=48)
)
# Request only last 24 hours (default)
response = await async_client.get(f"/api/v1/ams-history/{printer.id}/0")
assert response.status_code == 200
data = response.json()
# Should only get the recent record
assert len(data["data"]) == 1
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_ams_history_custom_hours(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify custom hours parameter works."""
printer = await printer_factory()
response = await async_client.get(
f"/api/v1/ams-history/{printer.id}/0",
params={"hours": 48}
)
assert response.status_code == 200
data = response.json()
assert data["printer_id"] == printer.id
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_ams_history_different_ams_units(
self, async_client: AsyncClient, ams_history_factory, printer_factory, db_session
):
"""Verify filtering by AMS unit ID."""
printer = await printer_factory()
await ams_history_factory(printer_id=printer.id, ams_id=0, humidity=40.0)
await ams_history_factory(printer_id=printer.id, ams_id=1, humidity=50.0)
# Get AMS unit 0
response = await async_client.get(f"/api/v1/ams-history/{printer.id}/0")
assert response.status_code == 200
data0 = response.json()
assert len(data0["data"]) == 1
assert data0["data"][0]["humidity"] == 40.0
# Get AMS unit 1
response = await async_client.get(f"/api/v1/ams-history/{printer.id}/1")
assert response.status_code == 200
data1 = response.json()
assert len(data1["data"]) == 1
assert data1["data"][0]["humidity"] == 50.0
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_old_history(
self, async_client: AsyncClient, ams_history_factory, printer_factory, db_session
):
"""Verify old history can be deleted."""
printer = await printer_factory()
# Create an old record
await ams_history_factory(
printer_id=printer.id,
recorded_at=datetime.now() - timedelta(days=60)
)
# Delete records older than 30 days
response = await async_client.delete(
f"/api/v1/ams-history/{printer.id}",
params={"days": 30}
)
assert response.status_code == 200
data = response.json()
assert data["deleted"] >= 1
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_old_history_no_records(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify delete with no old records returns 0."""
printer = await printer_factory()
response = await async_client.delete(
f"/api/v1/ams-history/{printer.id}",
params={"days": 30}
)
assert response.status_code == 200
data = response.json()
assert data["deleted"] == 0
@@ -0,0 +1,294 @@
"""Integration tests for Archives API endpoints.
Tests the full request/response cycle for /api/v1/archives/ endpoints.
"""
import pytest
from httpx import AsyncClient
class TestArchivesAPI:
"""Integration tests for /api/v1/archives/ endpoints."""
# ========================================================================
# List endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_archives_empty(self, async_client: AsyncClient):
"""Verify empty list is returned when no archives exist."""
response = await async_client.get("/api/v1/archives/")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) == 0
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_archives_with_data(
self, async_client: AsyncClient, archive_factory, printer_factory, db_session
):
"""Verify list returns existing archives."""
printer = await printer_factory()
await archive_factory(printer.id, print_name="Test Archive")
response = await async_client.get("/api/v1/archives/")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) >= 1
assert any(a["print_name"] == "Test Archive" for a in data)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_archives_pagination(
self, async_client: AsyncClient, archive_factory, printer_factory, db_session
):
"""Verify pagination works correctly."""
printer = await printer_factory()
# Create 5 archives
for i in range(5):
await archive_factory(printer.id, print_name=f"Archive {i}")
# Get first page with limit 2
response = await async_client.get("/api/v1/archives/?limit=2&offset=0")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) == 2
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_archives_filter_by_printer(
self, async_client: AsyncClient, archive_factory, printer_factory, db_session
):
"""Verify filtering by printer_id works."""
printer1 = await printer_factory(name="Printer 1", serial_number="00M09A000000001")
printer2 = await printer_factory(name="Printer 2", serial_number="00M09A000000002")
await archive_factory(printer1.id, print_name="Printer 1 Archive")
await archive_factory(printer2.id, print_name="Printer 2 Archive")
response = await async_client.get(
f"/api/v1/archives/?printer_id={printer1.id}"
)
assert response.status_code == 200
data = response.json()
assert all(a["printer_id"] == printer1.id for a in data)
# ========================================================================
# Get single endpoint
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_archive(
self, async_client: AsyncClient, archive_factory, printer_factory, db_session
):
"""Verify single archive can be retrieved."""
printer = await printer_factory()
archive = await archive_factory(printer.id, print_name="Get Test Archive")
response = await async_client.get(f"/api/v1/archives/{archive.id}")
assert response.status_code == 200
result = response.json()
assert result["id"] == archive.id
assert result["print_name"] == "Get Test Archive"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_archive_not_found(self, async_client: AsyncClient):
"""Verify 404 for non-existent archive."""
response = await async_client.get("/api/v1/archives/9999")
assert response.status_code == 404
# ========================================================================
# Update endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_archive_name(
self, async_client: AsyncClient, archive_factory, printer_factory, db_session
):
"""Verify archive name can be updated."""
printer = await printer_factory()
archive = await archive_factory(printer.id, print_name="Original Name")
response = await async_client.patch(
f"/api/v1/archives/{archive.id}",
json={"print_name": "Updated Name"}
)
assert response.status_code == 200
assert response.json()["print_name"] == "Updated Name"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_archive_notes(
self, async_client: AsyncClient, archive_factory, printer_factory, db_session
):
"""Verify archive notes can be updated."""
printer = await printer_factory()
archive = await archive_factory(printer.id)
response = await async_client.patch(
f"/api/v1/archives/{archive.id}",
json={"notes": "Great print!"}
)
assert response.status_code == 200
assert response.json()["notes"] == "Great print!"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_archive_favorite(
self, async_client: AsyncClient, archive_factory, printer_factory, db_session
):
"""Verify archive favorite status can be updated."""
printer = await printer_factory()
archive = await archive_factory(printer.id)
response = await async_client.patch(
f"/api/v1/archives/{archive.id}",
json={"is_favorite": True}
)
assert response.status_code == 200
assert response.json()["is_favorite"] is True
# ========================================================================
# Delete endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_archive(
self, async_client: AsyncClient, archive_factory, printer_factory, db_session
):
"""Verify archive can be deleted."""
printer = await printer_factory()
archive = await archive_factory(printer.id)
archive_id = archive.id
response = await async_client.delete(f"/api/v1/archives/{archive_id}")
assert response.status_code == 200
# Verify deleted
response = await async_client.get(f"/api/v1/archives/{archive_id}")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_nonexistent_archive(self, async_client: AsyncClient):
"""Verify deleting non-existent archive returns 404."""
response = await async_client.delete("/api/v1/archives/9999")
assert response.status_code == 404
# ========================================================================
# Statistics endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_archive_stats(
self, async_client: AsyncClient, archive_factory, printer_factory, db_session
):
"""Verify archive statistics can be retrieved."""
printer = await printer_factory()
await archive_factory(
printer.id,
status="completed",
print_time_seconds=3600,
filament_used_grams=50.0,
)
await archive_factory(
printer.id,
status="completed",
print_time_seconds=7200,
filament_used_grams=100.0,
)
response = await async_client.get("/api/v1/archives/stats")
assert response.status_code == 200
result = response.json()
# Check for actual stats fields
assert "total_prints" in result
assert "successful_prints" in result
class TestArchiveDataIntegrity:
"""Tests for archive data integrity."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_archive_linked_to_printer(
self, async_client: AsyncClient, archive_factory, printer_factory, db_session
):
"""Verify archive is properly linked to printer."""
printer = await printer_factory(name="My Printer")
archive = await archive_factory(printer.id)
response = await async_client.get(f"/api/v1/archives/{archive.id}")
assert response.status_code == 200
result = response.json()
assert result["printer_id"] == printer.id
@pytest.mark.asyncio
@pytest.mark.integration
async def test_archive_stores_print_data(
self, async_client: AsyncClient, archive_factory, printer_factory, db_session
):
"""Verify archive stores all print data correctly."""
printer = await printer_factory()
archive = await archive_factory(
printer.id,
print_name="Test Print",
filename="test.3mf",
status="completed",
filament_type="PLA",
filament_used_grams=75.5,
print_time_seconds=5400,
)
response = await async_client.get(f"/api/v1/archives/{archive.id}")
assert response.status_code == 200
result = response.json()
assert result["print_name"] == "Test Print"
assert result["filename"] == "test.3mf"
assert result["status"] == "completed"
assert result["filament_type"] == "PLA"
assert result["filament_used_grams"] == 75.5
assert result["print_time_seconds"] == 5400
@pytest.mark.asyncio
@pytest.mark.integration
async def test_archive_update_persists(
self, async_client: AsyncClient, archive_factory, printer_factory, db_session
):
"""CRITICAL: Verify archive updates persist."""
printer = await printer_factory()
archive = await archive_factory(printer.id, notes="Original notes")
# Update
await async_client.patch(
f"/api/v1/archives/{archive.id}",
json={"notes": "Updated notes", "is_favorite": True}
)
# Verify persistence
response = await async_client.get(f"/api/v1/archives/{archive.id}")
result = response.json()
assert result["notes"] == "Updated notes"
assert result["is_favorite"] is True
@@ -0,0 +1,184 @@
"""Integration tests for External Links API endpoints."""
import pytest
from httpx import AsyncClient
class TestExternalLinksAPI:
"""Integration tests for /api/v1/external-links endpoints."""
@pytest.fixture
async def link_factory(self, db_session):
"""Factory to create test external links."""
_counter = [0]
async def _create_link(**kwargs):
from backend.app.models.external_link import ExternalLink
_counter[0] += 1
counter = _counter[0]
defaults = {
"name": f"Test Link {counter}",
"url": f"https://example.com/{counter}",
"icon": "Link",
"sort_order": counter,
}
defaults.update(kwargs)
link = ExternalLink(**defaults)
db_session.add(link)
await db_session.commit()
await db_session.refresh(link)
return link
return _create_link
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_external_links_empty(self, async_client: AsyncClient):
"""Verify empty list when no links exist."""
response = await async_client.get("/api/v1/external-links/")
assert response.status_code == 200
assert isinstance(response.json(), list)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_external_links_with_data(
self, async_client: AsyncClient, link_factory, db_session
):
"""Verify list returns existing links."""
await link_factory(name="My Link")
response = await async_client.get("/api/v1/external-links/")
assert response.status_code == 200
data = response.json()
assert any(link["name"] == "My Link" for link in data)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_external_link(self, async_client: AsyncClient):
"""Verify external link can be created."""
data = {
"name": "New Link",
"url": "https://new-link.example.com",
"icon": "ExternalLink",
}
response = await async_client.post("/api/v1/external-links/", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "New Link"
assert result["url"] == "https://new-link.example.com"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_external_link(
self, async_client: AsyncClient, link_factory, db_session
):
"""Verify single link can be retrieved."""
link = await link_factory(name="Get Test Link")
response = await async_client.get(f"/api/v1/external-links/{link.id}")
assert response.status_code == 200
assert response.json()["name"] == "Get Test Link"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_external_link_not_found(self, async_client: AsyncClient):
"""Verify 404 for non-existent link."""
response = await async_client.get("/api/v1/external-links/9999")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_external_link(
self, async_client: AsyncClient, link_factory, db_session
):
"""Verify link can be updated."""
link = await link_factory(name="Original")
response = await async_client.patch(
f"/api/v1/external-links/{link.id}",
json={"name": "Updated", "url": "https://updated.example.com"}
)
assert response.status_code == 200
result = response.json()
assert result["name"] == "Updated"
assert result["url"] == "https://updated.example.com"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_external_link(
self, async_client: AsyncClient, link_factory, db_session
):
"""Verify link can be deleted."""
link = await link_factory()
response = await async_client.delete(f"/api/v1/external-links/{link.id}")
assert response.status_code == 200
# Verify deleted
response = await async_client.get(f"/api/v1/external-links/{link.id}")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_reorder_external_links(
self, async_client: AsyncClient, link_factory, db_session
):
"""Verify links can be reordered."""
link1 = await link_factory(name="Link 1")
link2 = await link_factory(name="Link 2")
link3 = await link_factory(name="Link 3")
# Reorder: 3, 1, 2
response = await async_client.put(
"/api/v1/external-links/reorder",
json={"ids": [link3.id, link1.id, link2.id]}
)
assert response.status_code == 200
data = response.json()
# First link should be link3
assert data[0]["id"] == link3.id
assert data[0]["sort_order"] == 0
class TestExternalLinksIconAPI:
"""Tests for external link icon upload/delete."""
@pytest.fixture
async def link_factory(self, db_session):
"""Factory to create test external links."""
async def _create_link(**kwargs):
from backend.app.models.external_link import ExternalLink
defaults = {
"name": "Icon Test Link",
"url": "https://example.com",
"icon": "Link",
"sort_order": 0,
}
defaults.update(kwargs)
link = ExternalLink(**defaults)
db_session.add(link)
await db_session.commit()
await db_session.refresh(link)
return link
return _create_link
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_icon_not_set(
self, async_client: AsyncClient, link_factory, db_session
):
"""Verify 404 when no custom icon is set."""
link = await link_factory()
response = await async_client.get(f"/api/v1/external-links/{link.id}/icon")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_icon_when_none(
self, async_client: AsyncClient, link_factory, db_session
):
"""Verify deleting non-existent icon succeeds silently."""
link = await link_factory()
response = await async_client.delete(f"/api/v1/external-links/{link.id}/icon")
assert response.status_code == 200
@@ -0,0 +1,117 @@
"""Integration tests for Filaments API endpoints."""
import pytest
from httpx import AsyncClient
class TestFilamentsAPI:
"""Integration tests for /api/v1/filaments/ endpoints."""
@pytest.fixture
async def filament_factory(self, db_session):
"""Factory to create test filaments."""
async def _create_filament(**kwargs):
from backend.app.models.filament import Filament
defaults = {
"name": "Test PLA",
"type": "PLA",
"color": "Red",
"color_hex": "#FF0000",
"brand": "Generic",
"cost_per_kg": 25.0,
}
defaults.update(kwargs)
filament = Filament(**defaults)
db_session.add(filament)
await db_session.commit()
await db_session.refresh(filament)
return filament
return _create_filament
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_filaments_empty(self, async_client: AsyncClient):
"""Verify empty list when no filaments exist."""
response = await async_client.get("/api/v1/filaments/")
assert response.status_code == 200
assert isinstance(response.json(), list)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_filaments_with_data(
self, async_client: AsyncClient, filament_factory, db_session
):
"""Verify list returns existing filaments."""
await filament_factory(name="Test Filament")
response = await async_client.get("/api/v1/filaments/")
assert response.status_code == 200
data = response.json()
assert any(f["name"] == "Test Filament" for f in data)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_filament(self, async_client: AsyncClient):
"""Verify filament can be created."""
data = {
"name": "New PETG",
"type": "PETG",
"color": "Blue",
"color_hex": "#0000FF",
"brand": "Bambu",
"cost_per_kg": 30.0,
}
response = await async_client.post("/api/v1/filaments/", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "New PETG"
assert result["type"] == "PETG"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_filament(
self, async_client: AsyncClient, filament_factory, db_session
):
"""Verify single filament can be retrieved."""
filament = await filament_factory(name="Get Test")
response = await async_client.get(f"/api/v1/filaments/{filament.id}")
assert response.status_code == 200
assert response.json()["name"] == "Get Test"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_filament_not_found(self, async_client: AsyncClient):
"""Verify 404 for non-existent filament."""
response = await async_client.get("/api/v1/filaments/9999")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_filament(
self, async_client: AsyncClient, filament_factory, db_session
):
"""Verify filament can be updated."""
filament = await filament_factory(name="Original")
response = await async_client.patch(
f"/api/v1/filaments/{filament.id}",
json={"name": "Updated", "cost_per_kg": 35.0}
)
assert response.status_code == 200
result = response.json()
assert result["name"] == "Updated"
assert result["cost_per_kg"] == 35.0
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_filament(
self, async_client: AsyncClient, filament_factory, db_session
):
"""Verify filament can be deleted."""
filament = await filament_factory()
response = await async_client.delete(f"/api/v1/filaments/{filament.id}")
assert response.status_code == 200
# Verify deleted
response = await async_client.get(f"/api/v1/filaments/{filament.id}")
assert response.status_code == 404
@@ -0,0 +1,268 @@
"""Integration tests for Maintenance API endpoints."""
import pytest
from httpx import AsyncClient
class TestMaintenanceTypesAPI:
"""Integration tests for /api/v1/maintenance/types endpoints."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_maintenance_types(self, async_client: AsyncClient):
"""Verify maintenance types list returns data with defaults."""
response = await async_client.get("/api/v1/maintenance/types")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
# Should have default system types
assert len(data) >= 1
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_includes_system_types(self, async_client: AsyncClient):
"""Verify default system types are created."""
response = await async_client.get("/api/v1/maintenance/types")
assert response.status_code == 200
data = response.json()
names = [t["name"] for t in data]
# Check for some default types
assert "Lubricate Linear Rails" in names or len(data) > 0
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_custom_maintenance_type(self, async_client: AsyncClient):
"""Verify custom maintenance type can be created."""
data = {
"name": "Custom Test Task",
"description": "Test description",
"default_interval_hours": 200.0,
"interval_type": "hours",
"icon": "Wrench",
}
response = await async_client.post("/api/v1/maintenance/types", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "Custom Test Task"
assert result["is_system"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_maintenance_type(self, async_client: AsyncClient):
"""Verify maintenance type can be updated."""
# First create a custom type
create_data = {
"name": "Update Test",
"description": "Original",
"default_interval_hours": 100.0,
}
create_response = await async_client.post(
"/api/v1/maintenance/types", json=create_data
)
assert create_response.status_code == 200
type_id = create_response.json()["id"]
# Update it
update_data = {"description": "Updated description"}
response = await async_client.patch(
f"/api/v1/maintenance/types/{type_id}", json=update_data
)
assert response.status_code == 200
assert response.json()["description"] == "Updated description"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_custom_maintenance_type(self, async_client: AsyncClient):
"""Verify custom maintenance type can be deleted."""
# Create a custom type
create_data = {
"name": "Delete Test",
"description": "To be deleted",
"default_interval_hours": 50.0,
}
create_response = await async_client.post(
"/api/v1/maintenance/types", json=create_data
)
type_id = create_response.json()["id"]
# Delete it
response = await async_client.delete(f"/api/v1/maintenance/types/{type_id}")
assert response.status_code == 200
class TestPrinterMaintenanceAPI:
"""Integration tests for /api/v1/maintenance/printers endpoints."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_printer_maintenance_not_found(self, async_client: AsyncClient):
"""Verify 404 for non-existent printer."""
response = await async_client.get("/api/v1/maintenance/printers/9999")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_printer_maintenance(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify maintenance overview for a printer."""
printer = await printer_factory(name="Maintenance Test Printer")
response = await async_client.get(f"/api/v1/maintenance/printers/{printer.id}")
assert response.status_code == 200
data = response.json()
assert data["printer_id"] == printer.id
assert data["printer_name"] == "Maintenance Test Printer"
assert "maintenance_items" in data
assert "total_print_hours" in data
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_all_maintenance_overview(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify overview endpoint returns all printers."""
await printer_factory(name="Overview Printer 1")
await printer_factory(name="Overview Printer 2")
response = await async_client.get("/api/v1/maintenance/overview")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_maintenance_summary(self, async_client: AsyncClient):
"""Verify summary endpoint returns counts."""
response = await async_client.get("/api/v1/maintenance/summary")
assert response.status_code == 200
data = response.json()
assert "total_due" in data
assert "total_warning" in data
assert "printers_with_issues" in data
class TestMaintenanceItemsAPI:
"""Integration tests for /api/v1/maintenance/items endpoints."""
@pytest.fixture
async def maintenance_item(self, async_client: AsyncClient, printer_factory, db_session):
"""Create a maintenance item for testing."""
printer = await printer_factory(name="Item Test Printer")
# Get the printer's maintenance overview to create items
response = await async_client.get(f"/api/v1/maintenance/printers/{printer.id}")
assert response.status_code == 200
data = response.json()
# Return the first maintenance item
if data["maintenance_items"]:
return data["maintenance_items"][0]
return None
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_maintenance_item(
self, async_client: AsyncClient, maintenance_item
):
"""Verify maintenance item can be updated."""
if not maintenance_item:
pytest.skip("No maintenance items available")
item_id = maintenance_item["id"]
response = await async_client.patch(
f"/api/v1/maintenance/items/{item_id}",
json={"custom_interval_hours": 150.0}
)
assert response.status_code == 200
@pytest.mark.asyncio
@pytest.mark.integration
async def test_disable_maintenance_item(
self, async_client: AsyncClient, maintenance_item
):
"""Verify maintenance item can be disabled."""
if not maintenance_item:
pytest.skip("No maintenance items available")
item_id = maintenance_item["id"]
response = await async_client.patch(
f"/api/v1/maintenance/items/{item_id}",
json={"enabled": False}
)
assert response.status_code == 200
assert response.json()["enabled"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_perform_maintenance(
self, async_client: AsyncClient, maintenance_item
):
"""Verify maintenance can be marked as performed."""
if not maintenance_item:
pytest.skip("No maintenance items available")
item_id = maintenance_item["id"]
response = await async_client.post(
f"/api/v1/maintenance/items/{item_id}/perform",
json={"notes": "Test maintenance performed"}
)
assert response.status_code == 200
data = response.json()
assert data["last_performed_at"] is not None
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_maintenance_history(
self, async_client: AsyncClient, maintenance_item
):
"""Verify maintenance history can be retrieved."""
if not maintenance_item:
pytest.skip("No maintenance items available")
item_id = maintenance_item["id"]
# First perform maintenance to create history
await async_client.post(
f"/api/v1/maintenance/items/{item_id}/perform",
json={"notes": "History test"}
)
response = await async_client.get(f"/api/v1/maintenance/items/{item_id}/history")
assert response.status_code == 200
history = response.json()
assert isinstance(history, list)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_maintenance_item_not_found(self, async_client: AsyncClient):
"""Verify 404 for non-existent maintenance item."""
response = await async_client.patch(
"/api/v1/maintenance/items/9999",
json={"enabled": False}
)
assert response.status_code == 404
class TestPrinterHoursAPI:
"""Integration tests for /api/v1/maintenance/printers/{id}/hours endpoint."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_set_printer_hours(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify printer hours can be set."""
printer = await printer_factory(name="Hours Test Printer")
response = await async_client.patch(
f"/api/v1/maintenance/printers/{printer.id}/hours",
params={"total_hours": 500.0}
)
assert response.status_code == 200
data = response.json()
assert data["total_hours"] == 500.0
@pytest.mark.asyncio
@pytest.mark.integration
async def test_set_printer_hours_not_found(self, async_client: AsyncClient):
"""Verify 404 for non-existent printer."""
response = await async_client.patch(
"/api/v1/maintenance/printers/9999/hours",
params={"total_hours": 100.0}
)
assert response.status_code == 404
@@ -0,0 +1,445 @@
"""Integration tests for Notifications API endpoints.
Tests the full request/response cycle for /api/v1/notifications/ endpoints.
"""
import pytest
from httpx import AsyncClient
class TestNotificationsAPI:
"""Integration tests for /api/v1/notifications/ endpoints."""
# ========================================================================
# List endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_notification_providers_empty(
self, async_client: AsyncClient
):
"""Verify empty list is returned when no providers exist."""
response = await async_client.get("/api/v1/notifications/")
assert response.status_code == 200
assert response.json() == []
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_notification_providers_with_data(
self, async_client: AsyncClient, notification_provider_factory, db_session
):
"""Verify list returns existing providers."""
provider = await notification_provider_factory(name="Test Provider")
response = await async_client.get("/api/v1/notifications/")
assert response.status_code == 200
data = response.json()
assert len(data) >= 1
assert any(p["name"] == "Test Provider" for p in data)
# ========================================================================
# Create endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_callmebot_provider(self, async_client: AsyncClient):
"""Verify callmebot notification provider can be created."""
data = {
"name": "Test CallMeBot",
"provider_type": "callmebot",
"enabled": True,
"config": {"phone_number": "+1234567890", "api_key": "test-api-key"},
"on_print_start": True,
"on_print_complete": True,
"on_print_failed": True,
"on_print_stopped": False,
}
response = await async_client.post("/api/v1/notifications/", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "Test CallMeBot"
assert result["provider_type"] == "callmebot"
assert result["on_print_start"] is True
assert result["on_print_stopped"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_ntfy_provider(self, async_client: AsyncClient):
"""Verify ntfy notification provider can be created."""
data = {
"name": "Test Ntfy",
"provider_type": "ntfy",
"enabled": True,
"config": {
"server": "https://ntfy.sh",
"topic": "test-topic",
},
"on_print_complete": True,
}
response = await async_client.post("/api/v1/notifications/", json=data)
assert response.status_code == 200
result = response.json()
assert result["provider_type"] == "ntfy"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_provider_with_printer(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify provider can be linked to specific printer."""
printer = await printer_factory(name="Test Printer")
data = {
"name": "Printer Ntfy",
"provider_type": "ntfy",
"config": {"server": "https://ntfy.sh", "topic": "test-topic"},
"printer_id": printer.id,
}
response = await async_client.post("/api/v1/notifications/", json=data)
assert response.status_code == 200
result = response.json()
assert result["printer_id"] == printer.id
# ========================================================================
# Get single endpoint
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_notification_provider(
self, async_client: AsyncClient, notification_provider_factory, db_session
):
"""Verify single provider can be retrieved."""
provider = await notification_provider_factory(name="Get Test Provider")
response = await async_client.get(f"/api/v1/notifications/{provider.id}")
assert response.status_code == 200
result = response.json()
assert result["id"] == provider.id
assert result["name"] == "Get Test Provider"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_provider_not_found(self, async_client: AsyncClient):
"""Verify 404 for non-existent provider."""
response = await async_client.get("/api/v1/notifications/9999")
assert response.status_code == 404
# ========================================================================
# Update endpoints (CRITICAL - toggle persistence)
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_event_toggles(
self, async_client: AsyncClient, notification_provider_factory, db_session
):
"""CRITICAL: Verify notification event toggles persist correctly."""
provider = await notification_provider_factory(
on_print_start=True,
on_print_complete=True,
on_print_stopped=False,
)
# Toggle on_print_stopped to True
response = await async_client.patch(
f"/api/v1/notifications/{provider.id}",
json={"on_print_stopped": True}
)
assert response.status_code == 200
assert response.json()["on_print_stopped"] is True
# Verify change persisted
response = await async_client.get(f"/api/v1/notifications/{provider.id}")
assert response.json()["on_print_stopped"] is True
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_ams_alarm_toggles(
self, async_client: AsyncClient, notification_provider_factory, db_session
):
"""CRITICAL: Verify AMS alarm toggles persist correctly."""
provider = await notification_provider_factory(
on_ams_humidity_high=False,
on_ams_temperature_high=False,
)
# Enable AMS alarms
response = await async_client.patch(
f"/api/v1/notifications/{provider.id}",
json={
"on_ams_humidity_high": True,
"on_ams_temperature_high": True,
}
)
assert response.status_code == 200
result = response.json()
assert result["on_ams_humidity_high"] is True
assert result["on_ams_temperature_high"] is True
# Verify persistence
response = await async_client.get(f"/api/v1/notifications/{provider.id}")
result = response.json()
assert result["on_ams_humidity_high"] is True
assert result["on_ams_temperature_high"] is True
@pytest.mark.asyncio
@pytest.mark.integration
async def test_enable_disable_provider(
self, async_client: AsyncClient, notification_provider_factory, db_session
):
"""Verify provider can be enabled/disabled."""
provider = await notification_provider_factory(enabled=True)
# Disable
response = await async_client.patch(
f"/api/v1/notifications/{provider.id}",
json={"enabled": False}
)
assert response.status_code == 200
assert response.json()["enabled"] is False
# Enable
response = await async_client.patch(
f"/api/v1/notifications/{provider.id}",
json={"enabled": True}
)
assert response.status_code == 200
assert response.json()["enabled"] is True
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_quiet_hours(
self, async_client: AsyncClient, notification_provider_factory, db_session
):
"""Verify quiet hours can be configured."""
provider = await notification_provider_factory(quiet_hours_enabled=False)
response = await async_client.patch(
f"/api/v1/notifications/{provider.id}",
json={
"quiet_hours_enabled": True,
"quiet_hours_start": "22:00",
"quiet_hours_end": "07:00",
}
)
assert response.status_code == 200
result = response.json()
assert result["quiet_hours_enabled"] is True
assert result["quiet_hours_start"] == "22:00"
assert result["quiet_hours_end"] == "07:00"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_daily_digest(
self, async_client: AsyncClient, notification_provider_factory, db_session
):
"""Verify daily digest can be configured."""
provider = await notification_provider_factory(daily_digest_enabled=False)
response = await async_client.patch(
f"/api/v1/notifications/{provider.id}",
json={
"daily_digest_enabled": True,
"daily_digest_time": "09:00",
}
)
assert response.status_code == 200
result = response.json()
assert result["daily_digest_enabled"] is True
assert result["daily_digest_time"] == "09:00"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_multiple_event_toggles(
self, async_client: AsyncClient, notification_provider_factory, db_session
):
"""Verify multiple event toggles can be updated at once."""
provider = await notification_provider_factory(
on_print_start=True,
on_print_complete=True,
on_print_failed=True,
on_print_stopped=False,
on_printer_offline=False,
)
response = await async_client.patch(
f"/api/v1/notifications/{provider.id}",
json={
"on_print_start": False,
"on_print_stopped": True,
"on_printer_offline": True,
}
)
assert response.status_code == 200
result = response.json()
assert result["on_print_start"] is False
assert result["on_print_stopped"] is True
assert result["on_printer_offline"] is True
# Unchanged fields should remain
assert result["on_print_complete"] is True
assert result["on_print_failed"] is True
# ========================================================================
# Test notification endpoint
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_test_notification(
self, async_client: AsyncClient, notification_provider_factory,
mock_httpx_client, db_session
):
"""Verify test notification can be sent."""
provider = await notification_provider_factory()
response = await async_client.post(
f"/api/v1/notifications/{provider.id}/test"
)
assert response.status_code == 200
result = response.json()
assert result["success"] is True
@pytest.mark.asyncio
@pytest.mark.integration
async def test_test_notification_disabled_provider(
self, async_client: AsyncClient, notification_provider_factory, db_session
):
"""Verify test notification works even for disabled provider."""
provider = await notification_provider_factory(enabled=False)
response = await async_client.post(
f"/api/v1/notifications/{provider.id}/test"
)
# Test should still work for disabled providers
assert response.status_code == 200
# ========================================================================
# Delete endpoint
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_notification_provider(
self, async_client: AsyncClient, notification_provider_factory, db_session
):
"""Verify notification provider can be deleted."""
provider = await notification_provider_factory()
provider_id = provider.id
response = await async_client.delete(f"/api/v1/notifications/{provider_id}")
assert response.status_code == 200
# Verify deleted
response = await async_client.get(f"/api/v1/notifications/{provider_id}")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_nonexistent_provider(self, async_client: AsyncClient):
"""Verify deleting non-existent provider returns 404."""
response = await async_client.delete("/api/v1/notifications/9999")
assert response.status_code == 404
class TestNotificationTemplatesAPI:
"""Integration tests for /api/v1/notification-templates/ endpoints."""
@pytest.fixture
async def seeded_templates(self, db_session):
"""Seed notification templates for tests."""
from backend.app.models.notification_template import NotificationTemplate, DEFAULT_TEMPLATES
templates = []
for template_data in DEFAULT_TEMPLATES:
template = NotificationTemplate(**template_data)
db_session.add(template)
templates.append(template)
await db_session.commit()
for template in templates:
await db_session.refresh(template)
return templates
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_templates(self, async_client: AsyncClient, seeded_templates):
"""Verify default templates are seeded and can be listed."""
response = await async_client.get("/api/v1/notification-templates/")
assert response.status_code == 200
templates = response.json()
# Should have default templates seeded
assert len(templates) >= 1
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_template_by_id(self, async_client: AsyncClient, seeded_templates):
"""Verify template can be retrieved by ID."""
# Get first template ID from seeded data
template_id = seeded_templates[0].id
response = await async_client.get(
f"/api/v1/notification-templates/{template_id}"
)
assert response.status_code == 200
template = response.json()
assert template["id"] == template_id
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_template(self, async_client: AsyncClient, seeded_templates):
"""Verify template can be updated."""
# Get first template
template_id = seeded_templates[0].id
# Update it (route uses PUT, not PATCH)
response = await async_client.put(
f"/api/v1/notification-templates/{template_id}",
json={
"title_template": "Custom Title: {printer}",
"body_template": "Custom body for {filename}",
}
)
assert response.status_code == 200
result = response.json()
assert result["title_template"] == "Custom Title: {printer}"
assert result["body_template"] == "Custom body for {filename}"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_reset_template_to_default(self, async_client: AsyncClient, seeded_templates):
"""Verify template can be reset to default."""
template_id = seeded_templates[0].id
response = await async_client.post(
f"/api/v1/notification-templates/{template_id}/reset"
)
assert response.status_code == 200
result = response.json()
assert result["is_default"] is True
@@ -0,0 +1,284 @@
"""Integration tests for Printers API endpoints.
Tests the full request/response cycle for /api/v1/printers/ endpoints.
"""
import pytest
from httpx import AsyncClient
from unittest.mock import patch, MagicMock, AsyncMock
class TestPrintersAPI:
"""Integration tests for /api/v1/printers/ endpoints."""
# ========================================================================
# List endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_printers_empty(self, async_client: AsyncClient):
"""Verify empty list is returned when no printers exist."""
response = await async_client.get("/api/v1/printers/")
assert response.status_code == 200
assert response.json() == []
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_printers_with_data(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify list returns existing printers."""
printer = await printer_factory(name="Test Printer")
response = await async_client.get("/api/v1/printers/")
assert response.status_code == 200
data = response.json()
assert len(data) >= 1
assert any(p["name"] == "Test Printer" for p in data)
# ========================================================================
# Create endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_printer(self, async_client: AsyncClient):
"""Verify printer can be created."""
data = {
"name": "New Printer",
"serial_number": "00M09A111111111",
"ip_address": "192.168.1.100",
"access_code": "12345678",
"is_active": True,
"model": "X1C",
}
response = await async_client.post("/api/v1/printers/", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "New Printer"
assert result["serial_number"] == "00M09A111111111"
assert result["model"] == "X1C"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_printer_duplicate_serial(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify duplicate serial number is rejected."""
await printer_factory(serial_number="00M09A222222222")
data = {
"name": "Duplicate Printer",
"serial_number": "00M09A222222222",
"ip_address": "192.168.1.101",
"access_code": "12345678",
}
response = await async_client.post("/api/v1/printers/", json=data)
# Should fail due to duplicate serial
assert response.status_code in [400, 409, 422, 500]
# ========================================================================
# Get single endpoint
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_printer(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify single printer can be retrieved."""
printer = await printer_factory(name="Get Test Printer")
response = await async_client.get(f"/api/v1/printers/{printer.id}")
assert response.status_code == 200
result = response.json()
assert result["id"] == printer.id
assert result["name"] == "Get Test Printer"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_printer_not_found(self, async_client: AsyncClient):
"""Verify 404 for non-existent printer."""
response = await async_client.get("/api/v1/printers/9999")
assert response.status_code == 404
# ========================================================================
# Update endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_printer_name(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify printer name can be updated."""
printer = await printer_factory(name="Original Name")
response = await async_client.patch(
f"/api/v1/printers/{printer.id}",
json={"name": "Updated Name"}
)
assert response.status_code == 200
assert response.json()["name"] == "Updated Name"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_printer_active_status(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify printer active status can be updated."""
printer = await printer_factory(is_active=True)
response = await async_client.patch(
f"/api/v1/printers/{printer.id}",
json={"is_active": False}
)
assert response.status_code == 200
assert response.json()["is_active"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_printer_auto_archive(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify auto_archive setting can be updated."""
printer = await printer_factory(auto_archive=True)
response = await async_client.patch(
f"/api/v1/printers/{printer.id}",
json={"auto_archive": False}
)
assert response.status_code == 200
assert response.json()["auto_archive"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_nonexistent_printer(self, async_client: AsyncClient):
"""Verify updating non-existent printer returns 404."""
response = await async_client.patch(
"/api/v1/printers/9999",
json={"name": "New Name"}
)
assert response.status_code == 404
# ========================================================================
# Delete endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_printer(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify printer can be deleted."""
printer = await printer_factory()
printer_id = printer.id
response = await async_client.delete(f"/api/v1/printers/{printer_id}")
assert response.status_code == 200
# Verify deleted
response = await async_client.get(f"/api/v1/printers/{printer_id}")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_nonexistent_printer(self, async_client: AsyncClient):
"""Verify deleting non-existent printer returns 404."""
response = await async_client.delete("/api/v1/printers/9999")
assert response.status_code == 404
# ========================================================================
# Status endpoint
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_printer_status(
self, async_client: AsyncClient, printer_factory, mock_printer_manager, db_session
):
"""Verify printer status can be retrieved."""
printer = await printer_factory()
response = await async_client.get(f"/api/v1/printers/{printer.id}/status")
assert response.status_code == 200
result = response.json()
assert "connected" in result
assert "state" in result
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_printer_status_not_found(self, async_client: AsyncClient):
"""Verify 404 for status of non-existent printer."""
response = await async_client.get("/api/v1/printers/9999/status")
assert response.status_code == 404
# ========================================================================
# Test connection endpoint
# ========================================================================
class TestPrinterDataIntegrity:
"""Tests for printer data integrity."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_printer_stores_all_fields(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify printer stores all fields correctly."""
printer = await printer_factory(
name="Full Test Printer",
serial_number="00M09A444444444",
ip_address="192.168.1.150",
model="P1S",
is_active=True,
auto_archive=False,
)
response = await async_client.get(f"/api/v1/printers/{printer.id}")
assert response.status_code == 200
result = response.json()
assert result["name"] == "Full Test Printer"
assert result["serial_number"] == "00M09A444444444"
assert result["ip_address"] == "192.168.1.150"
assert result["model"] == "P1S"
assert result["is_active"] is True
assert result["auto_archive"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_printer_update_persists(
self, async_client: AsyncClient, printer_factory, db_session
):
"""CRITICAL: Verify printer updates persist."""
printer = await printer_factory(name="Original", is_active=True)
# Update
await async_client.patch(
f"/api/v1/printers/{printer.id}",
json={"name": "Updated", "is_active": False}
)
# Verify persistence
response = await async_client.get(f"/api/v1/printers/{printer.id}")
result = response.json()
assert result["name"] == "Updated"
assert result["is_active"] is False
@@ -0,0 +1,160 @@
"""Integration tests for Projects API endpoints."""
import pytest
from httpx import AsyncClient
class TestProjectsAPI:
"""Integration tests for /api/v1/projects endpoints."""
@pytest.fixture
async def project_factory(self, db_session):
"""Factory to create test projects."""
_counter = [0]
async def _create_project(**kwargs):
from backend.app.models.project import Project
_counter[0] += 1
counter = _counter[0]
defaults = {
"name": f"Test Project {counter}",
"description": "Test project description",
"color": "#FF0000",
}
defaults.update(kwargs)
project = Project(**defaults)
db_session.add(project)
await db_session.commit()
await db_session.refresh(project)
return project
return _create_project
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_projects_empty(self, async_client: AsyncClient):
"""Verify empty list when no projects exist."""
response = await async_client.get("/api/v1/projects/")
assert response.status_code == 200
assert isinstance(response.json(), list)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_projects_with_data(
self, async_client: AsyncClient, project_factory, db_session
):
"""Verify list returns existing projects."""
await project_factory(name="My Project")
response = await async_client.get("/api/v1/projects/")
assert response.status_code == 200
data = response.json()
assert any(p["name"] == "My Project" for p in data)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_project(self, async_client: AsyncClient):
"""Verify project can be created."""
data = {
"name": "New Project",
"description": "A new project",
"color": "#00FF00",
}
response = await async_client.post("/api/v1/projects/", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "New Project"
assert result["color"] == "#00FF00"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_project(
self, async_client: AsyncClient, project_factory, db_session
):
"""Verify single project can be retrieved."""
project = await project_factory(name="Get Test Project")
response = await async_client.get(f"/api/v1/projects/{project.id}")
assert response.status_code == 200
assert response.json()["name"] == "Get Test Project"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_project_not_found(self, async_client: AsyncClient):
"""Verify 404 for non-existent project."""
response = await async_client.get("/api/v1/projects/9999")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_project(
self, async_client: AsyncClient, project_factory, db_session
):
"""Verify project can be updated."""
project = await project_factory(name="Original")
response = await async_client.patch(
f"/api/v1/projects/{project.id}",
json={"name": "Updated", "description": "Updated description"}
)
assert response.status_code == 200
result = response.json()
assert result["name"] == "Updated"
assert result["description"] == "Updated description"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_project(
self, async_client: AsyncClient, project_factory, db_session
):
"""Verify project can be deleted."""
project = await project_factory()
response = await async_client.delete(f"/api/v1/projects/{project.id}")
assert response.status_code == 200
data = response.json()
assert data["message"] == "Project deleted"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_project_not_found(self, async_client: AsyncClient):
"""Verify 404 for deleting non-existent project."""
response = await async_client.delete("/api/v1/projects/9999")
assert response.status_code == 404
class TestProjectArchivesAPI:
"""Tests for project-archive relationships."""
@pytest.fixture
async def project_factory(self, db_session):
"""Factory to create test projects."""
async def _create_project(**kwargs):
from backend.app.models.project import Project
defaults = {
"name": "Archive Test Project",
"description": "Test project",
"color": "#0000FF",
}
defaults.update(kwargs)
project = Project(**defaults)
db_session.add(project)
await db_session.commit()
await db_session.refresh(project)
return project
return _create_project
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_project_with_archives(
self, async_client: AsyncClient, project_factory, db_session
):
"""Verify project can be retrieved with archive count."""
project = await project_factory()
response = await async_client.get(f"/api/v1/projects/{project.id}")
assert response.status_code == 200
# Project should have an archive count (may be 0)
data = response.json()
assert "name" in data
@@ -0,0 +1,215 @@
"""Integration tests for Settings API endpoints.
Tests the full request/response cycle for /api/v1/settings/ endpoints.
"""
import pytest
from httpx import AsyncClient
class TestSettingsAPI:
"""Integration tests for /api/v1/settings/ endpoints."""
# ========================================================================
# Get settings
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_settings(self, async_client: AsyncClient):
"""Verify settings can be retrieved."""
response = await async_client.get("/api/v1/settings/")
assert response.status_code == 200
result = response.json()
# Check for actual settings fields
assert "auto_archive" in result
assert "currency" in result
assert "date_format" in result
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_settings_has_defaults(self, async_client: AsyncClient):
"""Verify default settings values are returned."""
response = await async_client.get("/api/v1/settings/")
assert response.status_code == 200
result = response.json()
# Verify some default values
assert isinstance(result["auto_archive"], bool)
assert isinstance(result["currency"], str)
# ========================================================================
# Update settings
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_auto_archive(self, async_client: AsyncClient):
"""Verify auto_archive can be updated."""
# First get current value
response = await async_client.get("/api/v1/settings/")
original = response.json()["auto_archive"]
# Update to opposite value
new_value = not original
response = await async_client.put(
"/api/v1/settings/",
json={"auto_archive": new_value}
)
assert response.status_code == 200
assert response.json()["auto_archive"] == new_value
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_currency(self, async_client: AsyncClient):
"""Verify currency can be updated."""
response = await async_client.put(
"/api/v1/settings/",
json={"currency": "EUR"}
)
assert response.status_code == 200
assert response.json()["currency"] == "EUR"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_date_format(self, async_client: AsyncClient):
"""Verify date format can be updated."""
response = await async_client.put(
"/api/v1/settings/",
json={"date_format": "eu"}
)
assert response.status_code == 200
assert response.json()["date_format"] == "eu"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_time_format(self, async_client: AsyncClient):
"""Verify time format can be updated."""
response = await async_client.put(
"/api/v1/settings/",
json={"time_format": "24h"}
)
assert response.status_code == 200
assert response.json()["time_format"] == "24h"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_filament_cost(self, async_client: AsyncClient):
"""Verify default filament cost can be updated."""
response = await async_client.put(
"/api/v1/settings/",
json={"default_filament_cost": 30.0}
)
assert response.status_code == 200
assert response.json()["default_filament_cost"] == 30.0
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_energy_cost(self, async_client: AsyncClient):
"""Verify energy cost can be updated."""
response = await async_client.put(
"/api/v1/settings/",
json={"energy_cost_per_kwh": 0.20}
)
assert response.status_code == 200
assert response.json()["energy_cost_per_kwh"] == 0.20
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_multiple_settings(self, async_client: AsyncClient):
"""Verify multiple settings can be updated at once."""
response = await async_client.put(
"/api/v1/settings/",
json={
"currency": "GBP",
"date_format": "iso",
"time_format": "12h",
"save_thumbnails": False,
}
)
assert response.status_code == 200
result = response.json()
assert result["currency"] == "GBP"
assert result["date_format"] == "iso"
assert result["time_format"] == "12h"
assert result["save_thumbnails"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_spoolman_settings(self, async_client: AsyncClient):
"""Verify Spoolman settings can be updated."""
response = await async_client.put(
"/api/v1/settings/",
json={
"spoolman_enabled": True,
"spoolman_url": "http://localhost:7912",
"spoolman_sync_mode": "manual",
}
)
assert response.status_code == 200
result = response.json()
assert result["spoolman_enabled"] is True
assert result["spoolman_url"] == "http://localhost:7912"
assert result["spoolman_sync_mode"] == "manual"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_ams_thresholds(self, async_client: AsyncClient):
"""Verify AMS threshold settings can be updated."""
response = await async_client.put(
"/api/v1/settings/",
json={
"ams_humidity_good": 35,
"ams_humidity_fair": 55,
"ams_temp_good": 25.0,
"ams_temp_fair": 32.0,
}
)
assert response.status_code == 200
result = response.json()
assert result["ams_humidity_good"] == 35
assert result["ams_humidity_fair"] == 55
assert result["ams_temp_good"] == 25.0
assert result["ams_temp_fair"] == 32.0
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_notification_language(self, async_client: AsyncClient):
"""Verify notification language can be updated."""
response = await async_client.put(
"/api/v1/settings/",
json={"notification_language": "de"}
)
assert response.status_code == 200
assert response.json()["notification_language"] == "de"
# ========================================================================
# Settings persistence tests
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_settings_persist_after_update(self, async_client: AsyncClient):
"""CRITICAL: Verify settings changes persist across requests."""
# Update settings
await async_client.put(
"/api/v1/settings/",
json={"currency": "JPY", "check_updates": False}
)
# Verify persistence in new request
response = await async_client.get("/api/v1/settings/")
result = response.json()
assert result["currency"] == "JPY"
assert result["check_updates"] is False
@@ -0,0 +1,388 @@
"""Integration tests for Smart Plugs API endpoints.
Tests the full request/response cycle for /api/v1/smart-plugs/ endpoints.
"""
import pytest
from httpx import AsyncClient
class TestSmartPlugsAPI:
"""Integration tests for /api/v1/smart-plugs/ endpoints."""
# ========================================================================
# List endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_smart_plugs_empty(self, async_client: AsyncClient):
"""Verify empty list is returned when no plugs exist."""
response = await async_client.get("/api/v1/smart-plugs/")
assert response.status_code == 200
assert response.json() == []
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_smart_plugs_with_data(
self, async_client: AsyncClient, smart_plug_factory, db_session
):
"""Verify list returns existing plugs."""
plug = await smart_plug_factory(name="Test Plug 1")
response = await async_client.get("/api/v1/smart-plugs/")
assert response.status_code == 200
data = response.json()
assert len(data) >= 1
assert any(p["name"] == "Test Plug 1" for p in data)
# ========================================================================
# Create endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_smart_plug(self, async_client: AsyncClient):
"""Verify smart plug can be created."""
data = {
"name": "New Plug",
"ip_address": "192.168.1.100",
"enabled": True,
"auto_on": True,
"auto_off": False,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "New Plug"
assert result["ip_address"] == "192.168.1.100"
assert result["auto_off"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_smart_plug_with_printer(
self, async_client: AsyncClient, printer_factory, db_session
):
"""Verify smart plug can be linked to a printer."""
printer = await printer_factory(name="Test Printer")
data = {
"name": "Printer Plug",
"ip_address": "192.168.1.101",
"printer_id": printer.id,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["printer_id"] == printer.id
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_plug_with_invalid_printer_id(
self, async_client: AsyncClient
):
"""Verify creating plug with non-existent printer fails."""
data = {
"name": "Test Plug",
"ip_address": "192.168.1.100",
"printer_id": 9999,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 400
assert "Printer not found" in response.json()["detail"]
# ========================================================================
# Get single endpoint
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_smart_plug(
self, async_client: AsyncClient, smart_plug_factory, db_session
):
"""Verify single plug can be retrieved."""
plug = await smart_plug_factory(name="Get Test Plug")
response = await async_client.get(f"/api/v1/smart-plugs/{plug.id}")
assert response.status_code == 200
result = response.json()
assert result["id"] == plug.id
assert result["name"] == "Get Test Plug"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_smart_plug_not_found(self, async_client: AsyncClient):
"""Verify 404 for non-existent plug."""
response = await async_client.get("/api/v1/smart-plugs/9999")
assert response.status_code == 404
# ========================================================================
# Update endpoints (CRITICAL - toggle persistence)
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_auto_off_toggle(
self, async_client: AsyncClient, smart_plug_factory, db_session
):
"""CRITICAL: Verify auto_off toggle persists correctly.
This tests the regression scenario where toggling auto_off
wasn't being saved properly.
"""
# Create plug with auto_off=True
plug = await smart_plug_factory(auto_off=True)
# Verify initial state
response = await async_client.get(f"/api/v1/smart-plugs/{plug.id}")
assert response.status_code == 200
assert response.json()["auto_off"] is True
# Toggle auto_off to False
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}",
json={"auto_off": False}
)
assert response.status_code == 200
assert response.json()["auto_off"] is False
# Verify change persisted by fetching again
response = await async_client.get(f"/api/v1/smart-plugs/{plug.id}")
assert response.json()["auto_off"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_auto_on_toggle(
self, async_client: AsyncClient, smart_plug_factory, db_session
):
"""Verify auto_on toggle persists correctly."""
plug = await smart_plug_factory(auto_on=True)
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}",
json={"auto_on": False}
)
assert response.status_code == 200
assert response.json()["auto_on"] is False
# Verify persistence
response = await async_client.get(f"/api/v1/smart-plugs/{plug.id}")
assert response.json()["auto_on"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_enabled_toggle(
self, async_client: AsyncClient, smart_plug_factory, db_session
):
"""Verify enabled toggle persists correctly."""
plug = await smart_plug_factory(enabled=True)
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}",
json={"enabled": False}
)
assert response.status_code == 200
assert response.json()["enabled"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_off_delay_mode(
self, async_client: AsyncClient, smart_plug_factory, db_session
):
"""Verify off_delay_mode can be changed."""
plug = await smart_plug_factory(off_delay_mode="time")
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}",
json={"off_delay_mode": "temperature", "off_temp_threshold": 50}
)
assert response.status_code == 200
result = response.json()
assert result["off_delay_mode"] == "temperature"
assert result["off_temp_threshold"] == 50
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_schedule_settings(
self, async_client: AsyncClient, smart_plug_factory, db_session
):
"""Verify schedule settings can be updated."""
plug = await smart_plug_factory(schedule_enabled=False)
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}",
json={
"schedule_enabled": True,
"schedule_on_time": "08:00",
"schedule_off_time": "22:00",
}
)
assert response.status_code == 200
result = response.json()
assert result["schedule_enabled"] is True
assert result["schedule_on_time"] == "08:00"
assert result["schedule_off_time"] == "22:00"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_multiple_fields(
self, async_client: AsyncClient, smart_plug_factory, db_session
):
"""Verify multiple fields can be updated at once."""
plug = await smart_plug_factory(
name="Old Name",
auto_on=True,
auto_off=True,
)
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}",
json={
"name": "New Name",
"auto_on": False,
"auto_off": False,
}
)
assert response.status_code == 200
result = response.json()
assert result["name"] == "New Name"
assert result["auto_on"] is False
assert result["auto_off"] is False
# ========================================================================
# Control endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_control_smart_plug_on(
self, async_client: AsyncClient, smart_plug_factory, mock_tasmota_service, db_session
):
"""Verify smart plug can be turned on."""
plug = await smart_plug_factory()
response = await async_client.post(
f"/api/v1/smart-plugs/{plug.id}/control",
json={"action": "on"}
)
assert response.status_code == 200
result = response.json()
assert result["success"] is True
assert result["action"] == "on"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_control_smart_plug_off(
self, async_client: AsyncClient, smart_plug_factory, mock_tasmota_service, db_session
):
"""Verify smart plug can be turned off."""
plug = await smart_plug_factory()
response = await async_client.post(
f"/api/v1/smart-plugs/{plug.id}/control",
json={"action": "off"}
)
assert response.status_code == 200
result = response.json()
assert result["success"] is True
assert result["action"] == "off"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_control_smart_plug_toggle(
self, async_client: AsyncClient, smart_plug_factory, mock_tasmota_service, db_session
):
"""Verify smart plug can be toggled."""
plug = await smart_plug_factory()
response = await async_client.post(
f"/api/v1/smart-plugs/{plug.id}/control",
json={"action": "toggle"}
)
assert response.status_code == 200
result = response.json()
assert result["success"] is True
assert result["action"] == "toggle"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_control_invalid_action(
self, async_client: AsyncClient, smart_plug_factory, db_session
):
"""Verify invalid action returns error."""
plug = await smart_plug_factory()
response = await async_client.post(
f"/api/v1/smart-plugs/{plug.id}/control",
json={"action": "invalid"}
)
# FastAPI returns 422 for pydantic validation errors
assert response.status_code == 422
# ========================================================================
# Status endpoint
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_smart_plug_status(
self, async_client: AsyncClient, smart_plug_factory, mock_tasmota_service, db_session
):
"""Verify smart plug status can be retrieved."""
plug = await smart_plug_factory()
response = await async_client.get(f"/api/v1/smart-plugs/{plug.id}/status")
assert response.status_code == 200
result = response.json()
assert result["state"] == "ON"
assert result["reachable"] is True
# ========================================================================
# Delete endpoint
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_smart_plug(
self, async_client: AsyncClient, smart_plug_factory, db_session
):
"""Verify smart plug can be deleted."""
plug = await smart_plug_factory()
plug_id = plug.id
response = await async_client.delete(f"/api/v1/smart-plugs/{plug_id}")
assert response.status_code == 200
# Verify deleted
response = await async_client.get(f"/api/v1/smart-plugs/{plug_id}")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_nonexistent_plug(self, async_client: AsyncClient):
"""Verify deleting non-existent plug returns 404."""
response = await async_client.delete("/api/v1/smart-plugs/9999")
assert response.status_code == 404
+14
View File
@@ -0,0 +1,14 @@
[pytest]
testpaths = .
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function
filterwarnings =
ignore::DeprecationWarning
ignore::sqlalchemy.exc.SAWarning
# Filter warnings from async mocks - coroutines created by mocks that are
# intentionally not awaited (expected behavior in unit tests)
ignore:coroutine.*was never awaited:RuntimeWarning
markers =
unit: Unit tests (fast, no external deps)
integration: Integration tests (slower, test full API)
slow: Slow tests (skip with -m "not slow")
+1
View File
@@ -0,0 +1 @@
"""Unit tests for BamBuddy backend services."""
+1
View File
@@ -0,0 +1 @@
"""Unit tests for BamBuddy backend services layer."""
@@ -0,0 +1,215 @@
"""Unit tests for the archive service."""
import pytest
from datetime import datetime
from unittest.mock import MagicMock, AsyncMock, patch
class TestArchiveServiceHelpers:
"""Tests for archive service helper functions."""
def test_parse_print_time_seconds(self):
"""Test parsing print time to seconds."""
# Import the actual function if available, otherwise test the logic
# 2h 30m 15s = 2*3600 + 30*60 + 15 = 9015 seconds
time_str = "2h 30m 15s"
# Parse hours
hours = 2
minutes = 30
seconds = 15
total = hours * 3600 + minutes * 60 + seconds
assert total == 9015
def test_parse_filament_grams(self):
"""Test parsing filament usage to grams."""
# Example: "150.5g" -> 150.5
filament_str = "150.5g"
grams = float(filament_str.replace('g', ''))
assert grams == 150.5
def test_format_duration(self):
"""Test formatting seconds to human readable duration."""
# 3661 seconds = 1h 1m 1s
seconds = 3661
hours = seconds // 3600
minutes = (seconds % 3600) // 60
secs = seconds % 60
assert hours == 1
assert minutes == 1
assert secs == 1
class TestArchiveDataParsing:
"""Tests for parsing archive data from MQTT messages."""
def test_parse_gcode_state(self):
"""Test parsing gcode state."""
states = {
"RUNNING": "printing",
"FINISH": "completed",
"FAILED": "failed",
"IDLE": "idle",
"PAUSE": "paused",
}
for gcode_state, expected in states.items():
# Simple state mapping
mapped = gcode_state.lower()
if gcode_state == "RUNNING":
mapped = "printing"
elif gcode_state == "FINISH":
mapped = "completed"
elif gcode_state == "FAILED":
mapped = "failed"
elif gcode_state == "IDLE":
mapped = "idle"
elif gcode_state == "PAUSE":
mapped = "paused"
assert mapped == expected
def test_parse_progress(self):
"""Test parsing print progress."""
# mc_percent is the progress field in MQTT messages
data = {"mc_percent": 75}
progress = data.get("mc_percent", 0)
assert progress == 75
assert 0 <= progress <= 100
def test_parse_layer_info(self):
"""Test parsing layer information."""
data = {
"layer_num": 50,
"total_layers": 200,
}
current_layer = data.get("layer_num", 0)
total_layers = data.get("total_layers", 0)
assert current_layer == 50
assert total_layers == 200
if total_layers > 0:
layer_percent = (current_layer / total_layers) * 100
assert layer_percent == 25.0
class TestArchiveFilePaths:
"""Tests for archive file path handling."""
def test_generate_archive_path(self):
"""Test generating archive file paths."""
printer_name = "X1C_01"
print_name = "benchy"
timestamp = datetime(2024, 1, 15, 14, 30, 0)
# Expected pattern: archives/{printer}/{year}/{month}/{filename}
year = timestamp.year
month = f"{timestamp.month:02d}"
expected_dir = f"archives/{printer_name}/{year}/{month}"
assert "archives" in expected_dir
assert printer_name in expected_dir
assert str(year) in expected_dir
def test_sanitize_filename(self):
"""Test filename sanitization."""
# Characters to remove: / \ : * ? " < > |
dirty_name = 'test:file<name>.3mf'
# Simple sanitization
safe_chars = []
for c in dirty_name:
if c not in '\\/:*?"<>|':
safe_chars.append(c)
clean_name = ''.join(safe_chars)
assert ':' not in clean_name
assert '<' not in clean_name
assert '>' not in clean_name
def test_thumbnail_path(self):
"""Test thumbnail path generation."""
archive_path = "archives/X1C_01/2024/01/benchy.3mf"
# Thumbnail typically has same path with _thumb.png suffix
base_path = archive_path.rsplit('.', 1)[0]
thumbnail_path = f"{base_path}_thumb.png"
assert thumbnail_path.endswith('_thumb.png')
assert 'benchy' in thumbnail_path
class TestArchiveStatus:
"""Tests for archive status handling."""
def test_valid_status_values(self):
"""Test valid archive status values."""
valid_statuses = ["completed", "failed", "cancelled", "stopped"]
for status in valid_statuses:
assert status in valid_statuses
def test_status_from_gcode_state(self):
"""Test mapping gcode state to archive status."""
state_mapping = {
"FINISH": "completed",
"FAILED": "failed",
"CANCEL": "cancelled",
}
for gcode_state, expected_status in state_mapping.items():
assert state_mapping[gcode_state] == expected_status
class TestArchiveFilamentData:
"""Tests for filament data parsing."""
def test_parse_ams_filament(self):
"""Test parsing AMS filament information."""
ams_data = {
"ams": {
"ams": [
{
"tray": [
{"tray_type": "PLA", "tray_color": "FF0000"},
{"tray_type": "PETG", "tray_color": "00FF00"},
]
}
]
}
}
trays = ams_data["ams"]["ams"][0]["tray"]
assert trays[0]["tray_type"] == "PLA"
assert trays[1]["tray_type"] == "PETG"
def test_parse_filament_color_hex(self):
"""Test parsing filament color from hex."""
color_hex = "FF5500"
# Should be valid hex
assert len(color_hex) == 6
r = int(color_hex[0:2], 16)
g = int(color_hex[2:4], 16)
b = int(color_hex[4:6], 16)
assert r == 255
assert g == 85
assert b == 0
def test_calculate_filament_cost(self):
"""Test calculating filament cost."""
grams_used = 150.0
cost_per_kg = 25.0 # $25 per kg
cost = (grams_used / 1000) * cost_per_kg
assert cost == 3.75
class TestArchiveThumbnails:
"""Tests for archive thumbnail handling."""
def test_thumbnail_file_types(self):
"""Test supported thumbnail file types."""
supported_types = [".png", ".jpg", ".jpeg"]
for ext in supported_types:
assert ext.startswith('.')
assert ext.lower() in [".png", ".jpg", ".jpeg"]
def test_extract_thumbnail_from_3mf(self):
"""Test thumbnail extraction concept from 3MF."""
# 3MF files are ZIP archives containing:
# - Metadata/thumbnail.png
# - 3D/3dmodel.model
expected_thumbnail_paths = [
"Metadata/thumbnail.png",
"Metadata/plate_1.png",
]
for path in expected_thumbnail_paths:
assert "png" in path.lower()
@@ -0,0 +1,690 @@
"""Unit tests for NotificationService.
Tests event-based notifications and toggle behavior.
"""
import pytest
import json
from datetime import datetime
from unittest.mock import AsyncMock, MagicMock, patch
from backend.app.services.notification_service import NotificationService
class TestNotificationService:
"""Tests for NotificationService class."""
@pytest.fixture
def service(self):
"""Create a fresh NotificationService instance."""
return NotificationService()
@pytest.fixture
def mock_provider(self):
"""Create a mock notification provider."""
provider = MagicMock()
provider.id = 1
provider.name = "Test Provider"
provider.provider_type = "webhook"
provider.enabled = True
provider.config = json.dumps({"webhook_url": "http://test.local/webhook"})
provider.on_print_start = True
provider.on_print_complete = True
provider.on_print_failed = True
provider.on_print_stopped = False
provider.on_print_progress = False
provider.on_printer_offline = False
provider.on_printer_error = False
provider.on_filament_low = False
provider.on_maintenance_due = False
provider.on_ams_humidity_high = False
provider.on_ams_temperature_high = False
provider.quiet_hours_enabled = False
provider.quiet_hours_start = None
provider.quiet_hours_end = None
provider.daily_digest_enabled = False
provider.daily_digest_time = None
provider.printer_id = None
return provider
@pytest.fixture
def mock_db(self):
"""Create a mock database session."""
db = AsyncMock()
db.commit = AsyncMock()
return db
# ========================================================================
# Tests for on_print_start
# ========================================================================
@pytest.mark.asyncio
async def test_on_print_start_sends_notification(
self, service, mock_provider, mock_db
):
"""Verify notification is sent when print starts."""
with patch.object(
service, '_get_providers_for_event', new_callable=AsyncMock
) as mock_get, \
patch.object(
service, '_send_to_providers', new_callable=AsyncMock
) as mock_send, \
patch.object(
service, '_build_message_from_template', new_callable=AsyncMock
) as mock_build:
mock_get.return_value = [mock_provider]
mock_build.return_value = ("Print Started", "Test Printer: test.3mf")
await service.on_print_start(
printer_id=1,
printer_name="Test Printer",
data={"filename": "test.3mf", "subtask_name": "test"},
db=mock_db,
)
mock_get.assert_called_once()
mock_send.assert_called_once()
@pytest.mark.asyncio
async def test_on_print_start_skipped_when_no_providers(
self, service, mock_db
):
"""Verify no error when no providers are configured for event."""
with patch.object(
service, '_get_providers_for_event', new_callable=AsyncMock
) as mock_get, \
patch.object(
service, '_send_to_providers', new_callable=AsyncMock
) as mock_send:
mock_get.return_value = []
await service.on_print_start(
printer_id=1,
printer_name="Test Printer",
data={},
db=mock_db,
)
mock_send.assert_not_called()
# ========================================================================
# Tests for on_print_complete (status routing)
# ========================================================================
@pytest.mark.asyncio
async def test_on_print_complete_routes_completed_status(
self, service, mock_provider, mock_db
):
"""Verify completed status uses on_print_complete field."""
with patch.object(
service, '_get_providers_for_event', new_callable=AsyncMock
) as mock_get, \
patch.object(
service, '_send_to_providers', new_callable=AsyncMock
), \
patch.object(
service, '_build_message_from_template', new_callable=AsyncMock
) as mock_build:
mock_get.return_value = [mock_provider]
mock_build.return_value = ("Test", "Test")
await service.on_print_complete(
printer_id=1,
printer_name="Test",
status="completed",
data={},
db=mock_db,
)
# Verify the correct event field was queried
call_args = mock_get.call_args
assert call_args[0][1] == "on_print_complete"
@pytest.mark.asyncio
async def test_on_print_complete_routes_failed_status(
self, service, mock_provider, mock_db
):
"""Verify failed status uses on_print_failed field."""
with patch.object(
service, '_get_providers_for_event', new_callable=AsyncMock
) as mock_get, \
patch.object(
service, '_send_to_providers', new_callable=AsyncMock
), \
patch.object(
service, '_build_message_from_template', new_callable=AsyncMock
) as mock_build:
mock_get.return_value = [mock_provider]
mock_build.return_value = ("Test", "Test")
await service.on_print_complete(
printer_id=1,
printer_name="Test",
status="failed",
data={},
db=mock_db,
)
call_args = mock_get.call_args
assert call_args[0][1] == "on_print_failed"
@pytest.mark.asyncio
async def test_on_print_complete_routes_stopped_status(
self, service, mock_provider, mock_db
):
"""Verify stopped status uses on_print_stopped field."""
with patch.object(
service, '_get_providers_for_event', new_callable=AsyncMock
) as mock_get, \
patch.object(
service, '_send_to_providers', new_callable=AsyncMock
), \
patch.object(
service, '_build_message_from_template', new_callable=AsyncMock
) as mock_build:
mock_get.return_value = [mock_provider]
mock_build.return_value = ("Test", "Test")
await service.on_print_complete(
printer_id=1,
printer_name="Test",
status="stopped",
data={},
db=mock_db,
)
call_args = mock_get.call_args
assert call_args[0][1] == "on_print_stopped"
@pytest.mark.asyncio
async def test_on_print_complete_routes_aborted_status(
self, service, mock_provider, mock_db
):
"""Verify aborted status uses on_print_stopped field."""
with patch.object(
service, '_get_providers_for_event', new_callable=AsyncMock
) as mock_get, \
patch.object(
service, '_send_to_providers', new_callable=AsyncMock
), \
patch.object(
service, '_build_message_from_template', new_callable=AsyncMock
) as mock_build:
mock_get.return_value = [mock_provider]
mock_build.return_value = ("Test", "Test")
await service.on_print_complete(
printer_id=1,
printer_name="Test",
status="aborted",
data={},
db=mock_db,
)
call_args = mock_get.call_args
assert call_args[0][1] == "on_print_stopped"
# ========================================================================
# Tests for provider filtering
# ========================================================================
@pytest.mark.asyncio
async def test_disabled_provider_not_returned(self, service, mock_provider, mock_db):
"""CRITICAL: Verify disabled providers don't receive notifications."""
mock_provider.enabled = False
# The actual filtering happens in _get_providers_for_event
# which queries only enabled providers
with patch.object(
service, '_get_providers_for_event', new_callable=AsyncMock
) as mock_get:
# Simulate the query filtering out disabled providers
mock_get.return_value = []
result = await service._get_providers_for_event(
mock_db, "on_print_start", printer_id=1
)
assert len(result) == 0
@pytest.mark.asyncio
async def test_provider_filtered_by_printer_id(self, service, mock_provider, mock_db):
"""Verify providers can be filtered by specific printer."""
mock_provider.printer_id = 2 # Linked to printer 2
with patch.object(
service, '_get_providers_for_event', new_callable=AsyncMock
) as mock_get:
# When querying for printer 1, provider linked to printer 2 is excluded
mock_get.return_value = []
result = await service._get_providers_for_event(
mock_db, "on_print_start", printer_id=1
)
assert len(result) == 0
# ========================================================================
# Tests for quiet hours
# ========================================================================
def test_is_in_quiet_hours_during_quiet_period(self, service, mock_provider):
"""Verify notifications are blocked during quiet hours."""
mock_provider.quiet_hours_enabled = True
mock_provider.quiet_hours_start = "22:00"
mock_provider.quiet_hours_end = "07:00"
with patch(
'backend.app.services.notification_service.datetime'
) as mock_datetime:
# Test during quiet hours (23:00)
mock_now = MagicMock()
mock_now.hour = 23
mock_now.minute = 0
mock_datetime.now.return_value = mock_now
result = service._is_in_quiet_hours(mock_provider)
assert result is True
def test_is_in_quiet_hours_outside_quiet_period(self, service, mock_provider):
"""Verify notifications are allowed outside quiet hours."""
mock_provider.quiet_hours_enabled = True
mock_provider.quiet_hours_start = "22:00"
mock_provider.quiet_hours_end = "07:00"
with patch(
'backend.app.services.notification_service.datetime'
) as mock_datetime:
# Test outside quiet hours (12:00)
mock_now = MagicMock()
mock_now.hour = 12
mock_now.minute = 0
mock_datetime.now.return_value = mock_now
result = service._is_in_quiet_hours(mock_provider)
assert result is False
def test_is_in_quiet_hours_disabled(self, service, mock_provider):
"""Verify quiet hours check returns False when disabled."""
mock_provider.quiet_hours_enabled = False
result = service._is_in_quiet_hours(mock_provider)
assert result is False
def test_is_in_quiet_hours_early_morning(self, service, mock_provider):
"""Verify quiet hours work across midnight (early morning)."""
mock_provider.quiet_hours_enabled = True
mock_provider.quiet_hours_start = "22:00"
mock_provider.quiet_hours_end = "07:00"
with patch(
'backend.app.services.notification_service.datetime'
) as mock_datetime:
# Test early morning (03:00) - should be in quiet hours
mock_now = MagicMock()
mock_now.hour = 3
mock_now.minute = 0
mock_datetime.now.return_value = mock_now
result = service._is_in_quiet_hours(mock_provider)
assert result is True
# ========================================================================
# Tests for AMS alarms
# ========================================================================
@pytest.mark.asyncio
async def test_on_ams_humidity_high_sends_notification(
self, service, mock_provider, mock_db
):
"""Verify AMS humidity alarm sends notification."""
mock_provider.on_ams_humidity_high = True
with patch.object(
service, '_get_providers_for_event', new_callable=AsyncMock
) as mock_get, \
patch.object(
service, '_send_to_providers', new_callable=AsyncMock
) as mock_send, \
patch.object(
service, '_build_message_from_template', new_callable=AsyncMock
) as mock_build:
mock_get.return_value = [mock_provider]
mock_build.return_value = ("AMS Humidity Alert", "High humidity detected")
await service.on_ams_humidity_high(
printer_id=1,
printer_name="Test Printer",
ams_label="AMS-A",
humidity=75.0,
threshold=60.0,
db=mock_db,
)
mock_send.assert_called_once()
# Verify force_immediate is True for alarms
call_kwargs = mock_send.call_args[1]
assert call_kwargs.get('force_immediate') is True
@pytest.mark.asyncio
async def test_on_ams_temperature_high_sends_notification(
self, service, mock_provider, mock_db
):
"""Verify AMS temperature alarm sends notification."""
mock_provider.on_ams_temperature_high = True
with patch.object(
service, '_get_providers_for_event', new_callable=AsyncMock
) as mock_get, \
patch.object(
service, '_send_to_providers', new_callable=AsyncMock
) as mock_send, \
patch.object(
service, '_build_message_from_template', new_callable=AsyncMock
) as mock_build:
mock_get.return_value = [mock_provider]
mock_build.return_value = ("AMS Temperature Alert", "High temp detected")
await service.on_ams_temperature_high(
printer_id=1,
printer_name="Test Printer",
ams_label="AMS-A",
temperature=40.0,
threshold=35.0,
db=mock_db,
)
mock_send.assert_called_once()
# Verify force_immediate is True for alarms
call_kwargs = mock_send.call_args[1]
assert call_kwargs.get('force_immediate') is True
@pytest.mark.asyncio
async def test_ams_alarm_skipped_when_toggle_disabled(
self, service, mock_provider, mock_db
):
"""CRITICAL: Verify AMS alarms respect toggle setting."""
mock_provider.on_ams_humidity_high = False
with patch.object(
service, '_get_providers_for_event', new_callable=AsyncMock
) as mock_get, \
patch.object(
service, '_send_to_providers', new_callable=AsyncMock
) as mock_send:
# Provider with toggle disabled won't be returned
mock_get.return_value = []
await service.on_ams_humidity_high(
printer_id=1,
printer_name="Test",
ams_label="AMS-A",
humidity=75.0,
threshold=60.0,
db=mock_db,
)
mock_send.assert_not_called()
# ========================================================================
# Tests for daily digest
# ========================================================================
@pytest.mark.asyncio
async def test_daily_digest_queues_notification(
self, service, mock_provider, mock_db
):
"""Verify notifications are queued when digest mode is enabled."""
mock_provider.daily_digest_enabled = True
mock_provider.daily_digest_time = "09:00"
with patch.object(
service, '_get_providers_for_event', new_callable=AsyncMock
) as mock_get, \
patch.object(
service, '_send_to_providers', new_callable=AsyncMock
) as mock_send, \
patch.object(
service, '_build_message_from_template', new_callable=AsyncMock
) as mock_build:
mock_get.return_value = [mock_provider]
mock_build.return_value = ("Test", "Test")
await service.on_print_complete(
printer_id=1,
printer_name="Test",
status="completed",
data={},
db=mock_db,
)
# When digest is enabled, _send_to_providers should still be called
# but internally it will queue instead of send immediately
mock_send.assert_called_once()
@pytest.mark.asyncio
async def test_force_immediate_bypasses_digest(
self, service, mock_provider, mock_db
):
"""Verify force_immediate=True bypasses digest mode."""
mock_provider.daily_digest_enabled = True
mock_provider.on_ams_humidity_high = True
with patch.object(
service, '_get_providers_for_event', new_callable=AsyncMock
) as mock_get, \
patch.object(
service, '_send_to_providers', new_callable=AsyncMock
) as mock_send, \
patch.object(
service, '_build_message_from_template', new_callable=AsyncMock
) as mock_build:
mock_get.return_value = [mock_provider]
mock_build.return_value = ("Alert", "Alert message")
await service.on_ams_humidity_high(
printer_id=1,
printer_name="Test",
ams_label="AMS-A",
humidity=75.0,
threshold=60.0,
db=mock_db,
)
# Verify force_immediate is passed
call_kwargs = mock_send.call_args[1]
assert call_kwargs.get('force_immediate') is True
class TestDigestModeAlwaysSendsImmediately:
"""CRITICAL: Tests that notifications always send immediately regardless of digest setting."""
@pytest.fixture
def service(self):
return NotificationService()
@pytest.mark.asyncio
async def test_notification_sends_immediately_even_with_digest_enabled(self, service):
"""CRITICAL: All notifications must be sent immediately, digest is just a summary."""
# Create a mock provider with digest enabled
mock_provider = MagicMock()
mock_provider.id = 1
mock_provider.name = "Test Provider"
mock_provider.provider_type = "ntfy"
mock_provider.enabled = True
mock_provider.daily_digest_enabled = True # Digest enabled
mock_provider.daily_digest_time = "23:59"
mock_provider.config = '{"server": "https://ntfy.sh", "topic": "test"}'
mock_db = AsyncMock()
# Mock the _send_to_provider method
with patch.object(service, '_send_to_provider', new_callable=AsyncMock) as mock_send:
mock_send.return_value = (True, None)
with patch.object(service, '_queue_for_digest', new_callable=AsyncMock) as mock_queue:
with patch.object(service, '_update_provider_status', new_callable=AsyncMock):
with patch.object(service, '_log_notification', new_callable=AsyncMock):
await service._send_to_providers(
providers=[mock_provider],
title="Print Started",
message="Your print has started",
db=mock_db,
event_type="print_start",
)
# CRITICAL: _send_to_provider MUST be called (immediate send)
mock_send.assert_called_once()
# Digest queue should also be called (for daily summary)
mock_queue.assert_called_once()
@pytest.mark.asyncio
async def test_notification_sends_without_digest_queue_when_disabled(self, service):
"""When digest is disabled, notification sends but no digest queue."""
mock_provider = MagicMock()
mock_provider.id = 1
mock_provider.name = "Test Provider"
mock_provider.provider_type = "ntfy"
mock_provider.enabled = True
mock_provider.daily_digest_enabled = False # Digest disabled
mock_provider.daily_digest_time = None
mock_provider.config = '{"server": "https://ntfy.sh", "topic": "test"}'
mock_db = AsyncMock()
with patch.object(service, '_send_to_provider', new_callable=AsyncMock) as mock_send:
mock_send.return_value = (True, None)
with patch.object(service, '_queue_for_digest', new_callable=AsyncMock) as mock_queue:
with patch.object(service, '_update_provider_status', new_callable=AsyncMock):
with patch.object(service, '_log_notification', new_callable=AsyncMock):
await service._send_to_providers(
providers=[mock_provider],
title="Print Started",
message="Your print has started",
db=mock_db,
event_type="print_start",
)
# Notification must still be sent immediately
mock_send.assert_called_once()
# Digest queue should NOT be called when digest is disabled
mock_queue.assert_not_called()
class TestNotificationProviderTypes:
"""Tests for different notification provider types."""
@pytest.fixture
def service(self):
return NotificationService()
@pytest.mark.asyncio
async def test_webhook_provider_sends_request(self, service):
"""Verify webhook provider sends HTTP request."""
config = {
"webhook_url": "http://test.local/webhook",
"field_title": "title",
"field_message": "message",
}
# Create a mock response
mock_response = MagicMock()
mock_response.status_code = 200
# Mock the _get_client method
mock_client = AsyncMock()
mock_client.post = AsyncMock(return_value=mock_response)
with patch.object(service, '_get_client', new_callable=AsyncMock) as mock_get_client:
mock_get_client.return_value = mock_client
success, message = await service._send_webhook(
config, "Test Title", "Test Message"
)
assert success is True
mock_client.post.assert_called_once()
@pytest.mark.asyncio
async def test_webhook_handles_failure(self, service):
"""Verify webhook gracefully handles HTTP errors."""
config = {
"webhook_url": "http://test.local/webhook",
}
with patch('httpx.AsyncClient') as mock_client_class:
mock_instance = AsyncMock()
mock_instance.post.side_effect = Exception("Connection failed")
mock_client_class.return_value.__aenter__ = AsyncMock(
return_value=mock_instance
)
mock_client_class.return_value.__aexit__ = AsyncMock()
success, message = await service._send_webhook(
config, "Test", "Test"
)
assert success is False
assert "Connection failed" in message or "error" in message.lower()
class TestNotificationTemplates:
"""Tests for notification message template rendering."""
@pytest.fixture
def service(self):
return NotificationService()
@pytest.mark.asyncio
async def test_template_renders_variables(self, service):
"""Verify template variables are replaced correctly."""
template_title = "Print {progress}% Complete"
template_body = "{printer}: {filename}\nRemaining: {remaining_time}"
variables = {
"printer": "Test Printer",
"filename": "test.3mf",
"progress": "50",
"remaining_time": "1h 30m",
}
title = template_title.format(**variables)
body = template_body.format(**variables)
assert title == "Print 50% Complete"
assert "Test Printer" in body
assert "test.3mf" in body
assert "1h 30m" in body
@pytest.mark.asyncio
async def test_template_handles_missing_variables(self, service):
"""Verify missing template variables don't cause crashes."""
template = "{printer}: {unknown_var}"
variables = {"printer": "Test"}
# Should handle gracefully - either leave placeholder or skip
try:
result = template.format_map(
{**variables, "unknown_var": "{unknown_var}"}
)
assert "Test" in result
except KeyError:
pytest.fail("Template should handle missing variables gracefully")
@@ -0,0 +1,778 @@
"""Unit tests for PrinterManager service.
Tests printer connection management, status tracking, and print control.
"""
import asyncio
import pytest
from unittest.mock import MagicMock, AsyncMock, patch, PropertyMock
from datetime import datetime
from backend.app.services.printer_manager import (
PrinterManager,
printer_state_to_dict,
init_printer_connections,
)
class TestPrinterManager:
"""Tests for PrinterManager class."""
@pytest.fixture
def manager(self):
"""Create a fresh PrinterManager instance."""
return PrinterManager()
@pytest.fixture
def mock_printer(self):
"""Create a mock Printer object."""
printer = MagicMock()
printer.id = 1
printer.ip_address = "192.168.1.100"
printer.serial_number = "00M09A123456789"
printer.access_code = "12345678"
printer.is_active = True
return printer
@pytest.fixture
def mock_client(self):
"""Create a mock BambuMQTTClient."""
client = MagicMock()
client.state = MagicMock()
client.state.connected = True
client.state.state = "IDLE"
client.state.progress = 0
client.state.temperatures = {"nozzle": 25, "bed": 25}
client.state.raw_data = {}
client.logging_enabled = False
return client
# ========================================================================
# Tests for initialization
# ========================================================================
def test_init_creates_empty_clients_dict(self, manager):
"""Verify manager initializes with empty clients dict."""
assert manager._clients == {}
def test_init_callbacks_are_none(self, manager):
"""Verify all callbacks are initially None."""
assert manager._on_print_start is None
assert manager._on_print_complete is None
assert manager._on_status_change is None
assert manager._on_ams_change is None
def test_init_loop_is_none(self, manager):
"""Verify event loop is initially None."""
assert manager._loop is None
# ========================================================================
# Tests for callback setters
# ========================================================================
def test_set_event_loop(self, manager):
"""Verify event loop can be set."""
mock_loop = MagicMock()
manager.set_event_loop(mock_loop)
assert manager._loop == mock_loop
def test_set_print_start_callback(self, manager):
"""Verify print start callback can be set."""
callback = MagicMock()
manager.set_print_start_callback(callback)
assert manager._on_print_start == callback
def test_set_print_complete_callback(self, manager):
"""Verify print complete callback can be set."""
callback = MagicMock()
manager.set_print_complete_callback(callback)
assert manager._on_print_complete == callback
def test_set_status_change_callback(self, manager):
"""Verify status change callback can be set."""
callback = MagicMock()
manager.set_status_change_callback(callback)
assert manager._on_status_change == callback
def test_set_ams_change_callback(self, manager):
"""Verify AMS change callback can be set."""
callback = MagicMock()
manager.set_ams_change_callback(callback)
assert manager._on_ams_change == callback
# ========================================================================
# Tests for _schedule_async
# ========================================================================
def test_schedule_async_with_running_loop(self, manager):
"""Verify async coroutine is scheduled when loop is running."""
mock_loop = MagicMock()
mock_loop.is_running.return_value = True
manager._loop = mock_loop
async def dummy_coro():
pass
coro = dummy_coro()
manager._schedule_async(coro)
mock_loop.is_running.assert_called_once()
# Clean up the coroutine
coro.close()
def test_schedule_async_without_loop(self, manager):
"""Verify nothing happens when no loop is set."""
async def dummy_coro():
pass
coro = dummy_coro()
# Should not raise
manager._schedule_async(coro)
coro.close()
def test_schedule_async_with_stopped_loop(self, manager):
"""Verify nothing happens when loop is not running."""
mock_loop = MagicMock()
mock_loop.is_running.return_value = False
manager._loop = mock_loop
async def dummy_coro():
pass
coro = dummy_coro()
manager._schedule_async(coro)
coro.close()
# ========================================================================
# Tests for connect_printer
# ========================================================================
@pytest.mark.asyncio
async def test_connect_printer_creates_client(self, manager, mock_printer):
"""Verify connecting creates an MQTT client."""
with patch(
'backend.app.services.printer_manager.BambuMQTTClient'
) as MockClient:
mock_instance = MagicMock()
mock_instance.state = MagicMock()
mock_instance.state.connected = True
MockClient.return_value = mock_instance
result = await manager.connect_printer(mock_printer)
MockClient.assert_called_once()
mock_instance.connect.assert_called_once()
assert mock_printer.id in manager._clients
assert result is True
@pytest.mark.asyncio
async def test_connect_printer_disconnects_existing(self, manager, mock_printer, mock_client):
"""Verify connecting disconnects existing client first."""
manager._clients[mock_printer.id] = mock_client
with patch(
'backend.app.services.printer_manager.BambuMQTTClient'
) as MockClient:
new_client = MagicMock()
new_client.state = MagicMock()
new_client.state.connected = True
MockClient.return_value = new_client
await manager.connect_printer(mock_printer)
mock_client.disconnect.assert_called_once()
@pytest.mark.asyncio
async def test_connect_printer_returns_false_on_failure(self, manager, mock_printer):
"""Verify returns False when connection fails."""
with patch(
'backend.app.services.printer_manager.BambuMQTTClient'
) as MockClient:
mock_instance = MagicMock()
mock_instance.state = MagicMock()
mock_instance.state.connected = False
MockClient.return_value = mock_instance
result = await manager.connect_printer(mock_printer)
assert result is False
# ========================================================================
# Tests for disconnect_printer
# ========================================================================
def test_disconnect_printer_removes_client(self, manager, mock_client):
"""Verify disconnecting removes and disconnects client."""
manager._clients[1] = mock_client
manager.disconnect_printer(1)
mock_client.disconnect.assert_called_once()
assert 1 not in manager._clients
def test_disconnect_printer_handles_missing(self, manager):
"""Verify disconnecting non-existent printer doesn't raise."""
manager.disconnect_printer(999) # Should not raise
# ========================================================================
# Tests for disconnect_all
# ========================================================================
def test_disconnect_all_disconnects_all_clients(self, manager):
"""Verify all clients are disconnected."""
client1 = MagicMock()
client2 = MagicMock()
manager._clients[1] = client1
manager._clients[2] = client2
manager.disconnect_all()
client1.disconnect.assert_called_once()
client2.disconnect.assert_called_once()
assert len(manager._clients) == 0
# ========================================================================
# Tests for get_status
# ========================================================================
def test_get_status_returns_state(self, manager, mock_client):
"""Verify get_status returns client state."""
manager._clients[1] = mock_client
result = manager.get_status(1)
mock_client.check_staleness.assert_called_once()
assert result == mock_client.state
def test_get_status_returns_none_for_unknown(self, manager):
"""Verify get_status returns None for unknown printer."""
result = manager.get_status(999)
assert result is None
# ========================================================================
# Tests for get_all_statuses
# ========================================================================
def test_get_all_statuses_returns_all(self, manager):
"""Verify all statuses are returned."""
client1 = MagicMock()
client1.state = MagicMock(connected=True)
client2 = MagicMock()
client2.state = MagicMock(connected=False)
manager._clients[1] = client1
manager._clients[2] = client2
result = manager.get_all_statuses()
assert len(result) == 2
assert 1 in result
assert 2 in result
client1.check_staleness.assert_called_once()
client2.check_staleness.assert_called_once()
# ========================================================================
# Tests for is_connected
# ========================================================================
def test_is_connected_returns_true(self, manager, mock_client):
"""Verify is_connected returns True for connected printer."""
mock_client.check_staleness.return_value = True
manager._clients[1] = mock_client
result = manager.is_connected(1)
assert result is True
def test_is_connected_returns_false_for_unknown(self, manager):
"""Verify is_connected returns False for unknown printer."""
result = manager.is_connected(999)
assert result is False
# ========================================================================
# Tests for get_client
# ========================================================================
def test_get_client_returns_client(self, manager, mock_client):
"""Verify get_client returns the client."""
manager._clients[1] = mock_client
result = manager.get_client(1)
assert result == mock_client
def test_get_client_returns_none_for_unknown(self, manager):
"""Verify get_client returns None for unknown printer."""
result = manager.get_client(999)
assert result is None
# ========================================================================
# Tests for mark_printer_offline
# ========================================================================
def test_mark_printer_offline_updates_state(self, manager, mock_client):
"""Verify mark_printer_offline updates client state."""
mock_client.state.connected = True
manager._clients[1] = mock_client
manager.mark_printer_offline(1)
assert mock_client.state.connected is False
assert mock_client.state.state == "unknown"
def test_mark_printer_offline_triggers_callback(self, manager, mock_client):
"""Verify mark_printer_offline triggers status callback."""
mock_client.state.connected = True
manager._clients[1] = mock_client
# Callback must return a coroutine
async def async_callback(printer_id, state):
pass
manager._on_status_change = async_callback
# Need a running loop for callback
mock_loop = MagicMock()
mock_loop.is_running.return_value = True
manager._loop = mock_loop
manager.mark_printer_offline(1)
# Callback should be scheduled via run_coroutine_threadsafe
mock_loop.is_running.assert_called()
# State should be updated
assert mock_client.state.connected is False
def test_mark_printer_offline_handles_unknown(self, manager):
"""Verify mark_printer_offline handles unknown printer."""
manager.mark_printer_offline(999) # Should not raise
def test_mark_printer_offline_skips_already_offline(self, manager, mock_client):
"""Verify mark_printer_offline skips already offline printer."""
mock_client.state.connected = False
manager._clients[1] = mock_client
manager.mark_printer_offline(1)
# State should remain unchanged
assert mock_client.state.connected is False
# ========================================================================
# Tests for start_print
# ========================================================================
def test_start_print_calls_client(self, manager, mock_client):
"""Verify start_print calls client method."""
mock_client.start_print.return_value = True
manager._clients[1] = mock_client
result = manager.start_print(1, "test.gcode")
mock_client.start_print.assert_called_once_with("test.gcode")
assert result is True
def test_start_print_returns_false_for_unknown(self, manager):
"""Verify start_print returns False for unknown printer."""
result = manager.start_print(999, "test.gcode")
assert result is False
# ========================================================================
# Tests for stop_print
# ========================================================================
def test_stop_print_calls_client(self, manager, mock_client):
"""Verify stop_print calls client method."""
mock_client.stop_print.return_value = True
manager._clients[1] = mock_client
result = manager.stop_print(1)
mock_client.stop_print.assert_called_once()
assert result is True
def test_stop_print_returns_false_for_unknown(self, manager):
"""Verify stop_print returns False for unknown printer."""
result = manager.stop_print(999)
assert result is False
# ========================================================================
# Tests for wait_for_cooldown
# ========================================================================
@pytest.mark.asyncio
async def test_wait_for_cooldown_returns_true_when_cool(self, manager, mock_client):
"""Verify wait_for_cooldown returns True when printer is cool."""
mock_client.state.connected = True
mock_client.state.temperatures = {"nozzle": 40, "bed": 30}
mock_client.check_staleness.return_value = True
manager._clients[1] = mock_client
result = await manager.wait_for_cooldown(1, target_temp=50)
assert result is True
@pytest.mark.asyncio
async def test_wait_for_cooldown_returns_false_on_disconnect(self, manager, mock_client):
"""Verify wait_for_cooldown returns False when printer disconnects."""
mock_client.state.connected = False
mock_client.check_staleness.return_value = False
manager._clients[1] = mock_client
result = await manager.wait_for_cooldown(1, target_temp=50, timeout=1)
assert result is False
@pytest.mark.asyncio
async def test_wait_for_cooldown_returns_false_for_unknown(self, manager):
"""Verify wait_for_cooldown returns False for unknown printer."""
result = await manager.wait_for_cooldown(999, target_temp=50, timeout=1)
assert result is False
@pytest.mark.asyncio
async def test_wait_for_cooldown_checks_both_nozzles(self, manager, mock_client):
"""Verify wait_for_cooldown checks both nozzles for dual extruders."""
mock_client.state.connected = True
mock_client.state.temperatures = {"nozzle": 40, "nozzle_2": 45, "bed": 30}
mock_client.check_staleness.return_value = True
manager._clients[1] = mock_client
result = await manager.wait_for_cooldown(1, target_temp=50)
assert result is True
# ========================================================================
# Tests for logging methods
# ========================================================================
def test_enable_logging_calls_client(self, manager, mock_client):
"""Verify enable_logging calls client method."""
manager._clients[1] = mock_client
result = manager.enable_logging(1, True)
mock_client.enable_logging.assert_called_once_with(True)
assert result is True
def test_enable_logging_returns_false_for_unknown(self, manager):
"""Verify enable_logging returns False for unknown printer."""
result = manager.enable_logging(999, True)
assert result is False
def test_get_logs_returns_logs(self, manager, mock_client):
"""Verify get_logs returns client logs."""
mock_logs = [MagicMock(), MagicMock()]
mock_client.get_logs.return_value = mock_logs
manager._clients[1] = mock_client
result = manager.get_logs(1)
assert result == mock_logs
def test_get_logs_returns_empty_for_unknown(self, manager):
"""Verify get_logs returns empty list for unknown printer."""
result = manager.get_logs(999)
assert result == []
def test_clear_logs_calls_client(self, manager, mock_client):
"""Verify clear_logs calls client method."""
manager._clients[1] = mock_client
result = manager.clear_logs(1)
mock_client.clear_logs.assert_called_once()
assert result is True
def test_clear_logs_returns_false_for_unknown(self, manager):
"""Verify clear_logs returns False for unknown printer."""
result = manager.clear_logs(999)
assert result is False
def test_is_logging_enabled_returns_status(self, manager, mock_client):
"""Verify is_logging_enabled returns client status."""
mock_client.logging_enabled = True
manager._clients[1] = mock_client
result = manager.is_logging_enabled(1)
assert result is True
def test_is_logging_enabled_returns_false_for_unknown(self, manager):
"""Verify is_logging_enabled returns False for unknown printer."""
result = manager.is_logging_enabled(999)
assert result is False
# ========================================================================
# Tests for request_status_update
# ========================================================================
def test_request_status_update_calls_client(self, manager, mock_client):
"""Verify request_status_update calls client method."""
mock_client.request_status_update.return_value = True
manager._clients[1] = mock_client
result = manager.request_status_update(1)
mock_client.request_status_update.assert_called_once()
assert result is True
def test_request_status_update_returns_false_for_unknown(self, manager):
"""Verify request_status_update returns False for unknown printer."""
result = manager.request_status_update(999)
assert result is False
# ========================================================================
# Tests for test_connection
# ========================================================================
@pytest.mark.asyncio
async def test_test_connection_success(self, manager):
"""Verify test_connection returns success on connection."""
with patch(
'backend.app.services.printer_manager.BambuMQTTClient'
) as MockClient:
mock_instance = MagicMock()
mock_instance.state = MagicMock()
mock_instance.state.connected = True
mock_instance.state.state = "IDLE"
mock_instance.state.raw_data = {"device_model": "X1C"}
MockClient.return_value = mock_instance
result = await manager.test_connection(
"192.168.1.100", "00M09A123456789", "12345678"
)
assert result["success"] is True
assert result["state"] == "IDLE"
assert result["model"] == "X1C"
mock_instance.disconnect.assert_called_once()
@pytest.mark.asyncio
async def test_test_connection_failure(self, manager):
"""Verify test_connection returns failure on connection error."""
with patch(
'backend.app.services.printer_manager.BambuMQTTClient'
) as MockClient:
mock_instance = MagicMock()
mock_instance.state = MagicMock()
mock_instance.state.connected = False
MockClient.return_value = mock_instance
result = await manager.test_connection(
"192.168.1.100", "00M09A123456789", "12345678"
)
assert result["success"] is False
assert result["state"] is None
mock_instance.disconnect.assert_called_once()
class TestPrinterStateToDict:
"""Tests for printer_state_to_dict helper function."""
@pytest.fixture
def mock_state(self):
"""Create a mock PrinterState."""
state = MagicMock()
state.connected = True
state.state = "RUNNING"
state.current_print = "test.3mf"
state.subtask_name = "Test Print"
state.gcode_file = "/sdcard/test.gcode"
state.progress = 50
state.remaining_time = 3600
state.layer_num = 10
state.total_layers = 20
state.temperatures = {"nozzle": 200, "bed": 60}
state.hms_errors = []
state.ams_status_main = 0
state.ams_status_sub = 0
state.tray_now = "1"
state.wifi_signal = -50
state.raw_data = {}
return state
def test_basic_conversion(self, mock_state):
"""Verify basic state fields are converted."""
result = printer_state_to_dict(mock_state)
assert result["connected"] is True
assert result["state"] == "RUNNING"
assert result["progress"] == 50
assert result["temperatures"] == {"nozzle": 200, "bed": 60}
def test_ams_data_parsing(self, mock_state):
"""Verify AMS data is parsed correctly."""
mock_state.raw_data = {
"ams": [{
"id": 0,
"humidity_raw": 45,
"temp": 25,
"tray": [
{
"id": 0,
"tray_color": "FF0000",
"tray_type": "PLA",
"tray_sub_brands": "Generic",
"remain": 80,
"k": 0.5,
"tag_uid": "ABC123",
"tray_uuid": "uuid-123",
}
]
}]
}
result = printer_state_to_dict(mock_state)
assert result["ams"] is not None
assert len(result["ams"]) == 1
assert result["ams"][0]["humidity"] == 45
assert len(result["ams"][0]["tray"]) == 1
assert result["ams"][0]["tray"][0]["tray_color"] == "FF0000"
def test_empty_tag_uid_becomes_none(self, mock_state):
"""Verify empty tag_uid is converted to None."""
mock_state.raw_data = {
"ams": [{
"id": 0,
"tray": [{
"id": 0,
"tag_uid": "",
"tray_uuid": "00000000000000000000000000000000",
}]
}]
}
result = printer_state_to_dict(mock_state)
assert result["ams"][0]["tray"][0]["tag_uid"] is None
assert result["ams"][0]["tray"][0]["tray_uuid"] is None
def test_zero_tag_uid_becomes_none(self, mock_state):
"""Verify zero tag_uid is converted to None."""
mock_state.raw_data = {
"ams": [{
"id": 0,
"tray": [{
"id": 0,
"tag_uid": "0000000000000000",
}]
}]
}
result = printer_state_to_dict(mock_state)
assert result["ams"][0]["tray"][0]["tag_uid"] is None
def test_vt_tray_parsing(self, mock_state):
"""Verify virtual tray is parsed correctly."""
mock_state.raw_data = {
"vt_tray": {
"tray_color": "00FF00",
"tray_type": "PETG",
"tray_sub_brands": "Generic",
"remain": 60,
"tag_uid": "VT123",
}
}
result = printer_state_to_dict(mock_state)
assert result["vt_tray"] is not None
assert result["vt_tray"]["id"] == 254
assert result["vt_tray"]["tray_color"] == "00FF00"
assert result["vt_tray"]["tray_type"] == "PETG"
def test_hms_errors_conversion(self, mock_state):
"""Verify HMS errors are converted correctly."""
error = MagicMock()
error.code = "0700_0100"
error.attr = 1
error.module = "AMS"
error.severity = 2
mock_state.hms_errors = [error]
result = printer_state_to_dict(mock_state)
assert len(result["hms_errors"]) == 1
assert result["hms_errors"][0]["code"] == "0700_0100"
assert result["hms_errors"][0]["module"] == "AMS"
def test_cover_url_added_for_running_print(self, mock_state):
"""Verify cover_url is added for running prints."""
result = printer_state_to_dict(mock_state, printer_id=1)
assert result["cover_url"] == "/api/v1/printers/1/cover"
def test_cover_url_none_when_not_running(self, mock_state):
"""Verify cover_url is None when not printing."""
mock_state.state = "IDLE"
result = printer_state_to_dict(mock_state, printer_id=1)
assert result["cover_url"] is None
def test_ams_ht_detection(self, mock_state):
"""Verify AMS-HT is detected (1 tray vs 4)."""
mock_state.raw_data = {
"ams": [{
"id": 0,
"tray": [{"id": 0}] # Only 1 tray = AMS-HT
}]
}
result = printer_state_to_dict(mock_state)
assert result["ams"][0]["is_ams_ht"] is True
def test_regular_ams_detection(self, mock_state):
"""Verify regular AMS is detected (4 trays)."""
mock_state.raw_data = {
"ams": [{
"id": 0,
"tray": [{"id": 0}, {"id": 1}, {"id": 2}, {"id": 3}]
}]
}
result = printer_state_to_dict(mock_state)
assert result["ams"][0]["is_ams_ht"] is False
class TestInitPrinterConnections:
"""Tests for init_printer_connections function."""
@pytest.mark.asyncio
async def test_connects_all_active_printers(self):
"""Verify all active printers are connected."""
mock_db = AsyncMock()
mock_printer1 = MagicMock(id=1, is_active=True)
mock_printer2 = MagicMock(id=2, is_active=True)
mock_result = MagicMock()
mock_result.scalars.return_value.all.return_value = [mock_printer1, mock_printer2]
mock_db.execute.return_value = mock_result
with patch(
'backend.app.services.printer_manager.printer_manager'
) as mock_manager:
mock_manager.connect_printer = AsyncMock()
await init_printer_connections(mock_db)
assert mock_manager.connect_printer.call_count == 2
@pytest.mark.asyncio
async def test_handles_empty_printer_list(self):
"""Verify empty printer list is handled."""
mock_db = AsyncMock()
mock_result = MagicMock()
mock_result.scalars.return_value.all.return_value = []
mock_db.execute.return_value = mock_result
with patch(
'backend.app.services.printer_manager.printer_manager'
) as mock_manager:
mock_manager.connect_printer = AsyncMock()
await init_printer_connections(mock_db)
mock_manager.connect_printer.assert_not_called()
@@ -0,0 +1,603 @@
"""Unit tests for SmartPlugManager service.
These tests specifically target the auto-off behavior and toggle functionality
that were identified as common regression points.
"""
import pytest
import asyncio
from datetime import datetime
from unittest.mock import AsyncMock, MagicMock, patch
from backend.app.services.smart_plug_manager import SmartPlugManager
class TestSmartPlugManager:
"""Tests for SmartPlugManager class."""
@pytest.fixture
def manager(self):
"""Create a fresh SmartPlugManager instance."""
return SmartPlugManager()
@pytest.fixture
def mock_plug(self):
"""Create a mock SmartPlug object."""
plug = MagicMock()
plug.id = 1
plug.name = "Test Plug"
plug.ip_address = "192.168.1.100"
plug.username = None
plug.password = None
plug.enabled = True
plug.auto_on = True
plug.auto_off = True
plug.off_delay_mode = "time"
plug.off_delay_minutes = 5
plug.off_temp_threshold = 70
plug.printer_id = 1
plug.auto_off_executed = False
plug.auto_off_pending = False
plug.last_state = "ON"
plug.last_checked = None
return plug
@pytest.fixture
def mock_db(self):
"""Create a mock database session."""
db = AsyncMock()
db.commit = AsyncMock()
db.refresh = AsyncMock()
return db
# ========================================================================
# Tests for on_print_start
# ========================================================================
@pytest.mark.asyncio
async def test_on_print_start_turns_on_plug(self, manager, mock_plug, mock_db):
"""Verify plug is turned ON when print starts with auto_on enabled."""
with patch.object(
manager, '_get_plug_for_printer', new_callable=AsyncMock
) as mock_get_plug, \
patch('backend.app.services.smart_plug_manager.tasmota_service') as mock_tasmota:
mock_get_plug.return_value = mock_plug
mock_tasmota.turn_on = AsyncMock(return_value=True)
await manager.on_print_start(printer_id=1, db=mock_db)
mock_tasmota.turn_on.assert_called_once_with(mock_plug)
@pytest.mark.asyncio
async def test_on_print_start_skipped_when_auto_on_disabled(
self, manager, mock_plug, mock_db
):
"""Verify plug is NOT turned on when auto_on is disabled."""
mock_plug.auto_on = False
with patch.object(
manager, '_get_plug_for_printer', new_callable=AsyncMock
) as mock_get_plug, \
patch('backend.app.services.smart_plug_manager.tasmota_service') as mock_tasmota:
mock_get_plug.return_value = mock_plug
mock_tasmota.turn_on = AsyncMock()
await manager.on_print_start(printer_id=1, db=mock_db)
mock_tasmota.turn_on.assert_not_called()
@pytest.mark.asyncio
async def test_on_print_start_skipped_when_plug_disabled(
self, manager, mock_plug, mock_db
):
"""Verify plug is NOT turned on when plug.enabled is False."""
mock_plug.enabled = False
with patch.object(
manager, '_get_plug_for_printer', new_callable=AsyncMock
) as mock_get_plug, \
patch('backend.app.services.smart_plug_manager.tasmota_service') as mock_tasmota:
mock_get_plug.return_value = mock_plug
mock_tasmota.turn_on = AsyncMock()
await manager.on_print_start(printer_id=1, db=mock_db)
mock_tasmota.turn_on.assert_not_called()
@pytest.mark.asyncio
async def test_on_print_start_skipped_when_no_plug_found(
self, manager, mock_db
):
"""Verify graceful handling when no plug is linked to printer."""
with patch.object(
manager, '_get_plug_for_printer', new_callable=AsyncMock
) as mock_get_plug, \
patch('backend.app.services.smart_plug_manager.tasmota_service') as mock_tasmota:
mock_get_plug.return_value = None
mock_tasmota.turn_on = AsyncMock()
# Should not raise any exception
await manager.on_print_start(printer_id=999, db=mock_db)
mock_tasmota.turn_on.assert_not_called()
@pytest.mark.asyncio
async def test_on_print_start_cancels_pending_off(
self, manager, mock_plug, mock_db
):
"""Verify starting a new print cancels any pending auto-off."""
# Set up a pending task
mock_task = MagicMock()
manager._pending_off[mock_plug.id] = mock_task
with patch.object(
manager, '_get_plug_for_printer', new_callable=AsyncMock
) as mock_get_plug, \
patch.object(
manager, '_mark_auto_off_pending', new_callable=AsyncMock
), \
patch('backend.app.services.smart_plug_manager.tasmota_service') as mock_tasmota:
mock_get_plug.return_value = mock_plug
mock_tasmota.turn_on = AsyncMock(return_value=True)
await manager.on_print_start(printer_id=1, db=mock_db)
mock_task.cancel.assert_called_once()
assert mock_plug.id not in manager._pending_off
@pytest.mark.asyncio
async def test_on_print_start_resets_auto_off_executed_flag(
self, manager, mock_plug, mock_db
):
"""Verify auto_off_executed flag is reset when turning on."""
mock_plug.auto_off_executed = True
with patch.object(
manager, '_get_plug_for_printer', new_callable=AsyncMock
) as mock_get_plug, \
patch('backend.app.services.smart_plug_manager.tasmota_service') as mock_tasmota:
mock_get_plug.return_value = mock_plug
mock_tasmota.turn_on = AsyncMock(return_value=True)
await manager.on_print_start(printer_id=1, db=mock_db)
assert mock_plug.auto_off_executed is False
# ========================================================================
# Tests for on_print_complete
# ========================================================================
@pytest.mark.asyncio
async def test_on_print_complete_schedules_time_based_off(
self, manager, mock_plug, mock_db
):
"""Verify time-based auto-off is scheduled when print completes."""
mock_plug.off_delay_mode = "time"
mock_plug.off_delay_minutes = 5
with patch.object(
manager, '_get_plug_for_printer', new_callable=AsyncMock
) as mock_get_plug, \
patch.object(manager, '_schedule_delayed_off') as mock_schedule:
mock_get_plug.return_value = mock_plug
await manager.on_print_complete(
printer_id=1, status="completed", db=mock_db
)
mock_schedule.assert_called_once_with(mock_plug, 1, 300) # 5 min * 60 sec
@pytest.mark.asyncio
async def test_on_print_complete_schedules_temp_based_off(
self, manager, mock_plug, mock_db
):
"""Verify temperature-based auto-off is scheduled when print completes."""
mock_plug.off_delay_mode = "temperature"
mock_plug.off_temp_threshold = 70
with patch.object(
manager, '_get_plug_for_printer', new_callable=AsyncMock
) as mock_get_plug, \
patch.object(manager, '_schedule_temp_based_off') as mock_schedule:
mock_get_plug.return_value = mock_plug
await manager.on_print_complete(
printer_id=1, status="completed", db=mock_db
)
mock_schedule.assert_called_once_with(mock_plug, 1, 70)
@pytest.mark.asyncio
async def test_on_print_complete_skipped_when_auto_off_disabled(
self, manager, mock_plug, mock_db
):
"""CRITICAL: Verify auto-off does NOT trigger when auto_off is False.
This is a key regression test - the toggle must respect the setting.
"""
mock_plug.auto_off = False
with patch.object(
manager, '_get_plug_for_printer', new_callable=AsyncMock
) as mock_get_plug, \
patch.object(manager, '_schedule_delayed_off') as mock_schedule, \
patch.object(manager, '_schedule_temp_based_off') as mock_temp:
mock_get_plug.return_value = mock_plug
await manager.on_print_complete(
printer_id=1, status="completed", db=mock_db
)
mock_schedule.assert_not_called()
mock_temp.assert_not_called()
@pytest.mark.asyncio
async def test_on_print_complete_skipped_when_plug_disabled(
self, manager, mock_plug, mock_db
):
"""Verify auto-off does NOT trigger when plug is disabled."""
mock_plug.enabled = False
with patch.object(
manager, '_get_plug_for_printer', new_callable=AsyncMock
) as mock_get_plug, \
patch.object(manager, '_schedule_delayed_off') as mock_schedule:
mock_get_plug.return_value = mock_plug
await manager.on_print_complete(
printer_id=1, status="completed", db=mock_db
)
mock_schedule.assert_not_called()
@pytest.mark.asyncio
async def test_on_print_complete_skipped_on_failed_print(
self, manager, mock_plug, mock_db
):
"""Verify auto-off does NOT trigger on failed prints for investigation."""
with patch.object(
manager, '_get_plug_for_printer', new_callable=AsyncMock
) as mock_get_plug, \
patch.object(manager, '_schedule_delayed_off') as mock_schedule:
mock_get_plug.return_value = mock_plug
await manager.on_print_complete(
printer_id=1, status="failed", db=mock_db
)
mock_schedule.assert_not_called()
@pytest.mark.asyncio
async def test_on_print_complete_skipped_on_aborted_print(
self, manager, mock_plug, mock_db
):
"""Verify auto-off does NOT trigger on aborted prints."""
with patch.object(
manager, '_get_plug_for_printer', new_callable=AsyncMock
) as mock_get_plug, \
patch.object(manager, '_schedule_delayed_off') as mock_schedule:
mock_get_plug.return_value = mock_plug
await manager.on_print_complete(
printer_id=1, status="aborted", db=mock_db
)
mock_schedule.assert_not_called()
# ========================================================================
# Tests for _cancel_pending_off
# ========================================================================
@pytest.mark.asyncio
async def test_cancel_pending_off_removes_task(self, manager, mock_plug):
"""Verify pending off tasks can be cancelled."""
mock_task = MagicMock()
manager._pending_off[mock_plug.id] = mock_task
with patch.object(
manager, '_mark_auto_off_pending', new_callable=AsyncMock
):
manager._cancel_pending_off(mock_plug.id)
assert mock_plug.id not in manager._pending_off
mock_task.cancel.assert_called_once()
@pytest.mark.asyncio
async def test_cancel_pending_off_handles_missing_task(self, manager):
"""Verify no error when cancelling non-existent task."""
# Should not raise any exception
with patch.object(
manager, '_mark_auto_off_pending', new_callable=AsyncMock
):
manager._cancel_pending_off(999) # Non-existent plug ID
@pytest.mark.asyncio
async def test_cancel_all_pending(self, manager, mock_plug):
"""Verify all pending tasks can be cancelled."""
mock_task1 = MagicMock()
mock_task2 = MagicMock()
manager._pending_off[1] = mock_task1
manager._pending_off[2] = mock_task2
with patch('asyncio.create_task') as mock_create:
manager.cancel_all_pending()
assert len(manager._pending_off) == 0
mock_task1.cancel.assert_called_once()
mock_task2.cancel.assert_called_once()
# ========================================================================
# Tests for scheduler
# ========================================================================
def test_start_scheduler(self, manager):
"""Verify scheduler can be started."""
assert manager._scheduler_task is None
# Mock _schedule_loop to return a mock coroutine to avoid unawaited coroutine warning
with patch.object(manager, '_schedule_loop') as mock_loop, \
patch('asyncio.create_task') as mock_create:
mock_create.return_value = MagicMock()
manager.start_scheduler()
assert manager._scheduler_task is not None
mock_loop.assert_called_once()
def test_stop_scheduler(self, manager):
"""Verify scheduler can be stopped."""
mock_task = MagicMock()
manager._scheduler_task = mock_task
manager.stop_scheduler()
mock_task.cancel.assert_called_once()
assert manager._scheduler_task is None
def test_start_scheduler_idempotent(self, manager):
"""Verify starting scheduler twice doesn't create multiple tasks."""
mock_task = MagicMock()
manager._scheduler_task = mock_task
# Mock _schedule_loop to avoid unawaited coroutine warning (in case it's called)
with patch.object(manager, '_schedule_loop') as mock_loop, \
patch('asyncio.create_task') as mock_create:
manager.start_scheduler()
mock_create.assert_not_called() # Should not create new task
mock_loop.assert_not_called() # Should not call _schedule_loop
class TestScheduleLoop:
"""Tests for the schedule-based plug control."""
@pytest.fixture
def manager(self):
return SmartPlugManager()
@pytest.mark.asyncio
async def test_check_schedules_turns_on_at_scheduled_time(self, manager):
"""Verify scheduled on-time turns plug on."""
mock_plug = MagicMock()
mock_plug.id = 1
mock_plug.name = "Test Plug"
mock_plug.enabled = True
mock_plug.schedule_enabled = True
mock_plug.schedule_on_time = "08:00"
mock_plug.schedule_off_time = "22:00"
mock_plug.printer_id = None
mock_plug.last_state = "OFF"
with patch(
'backend.app.services.smart_plug_manager.datetime'
) as mock_datetime, \
patch(
'backend.app.core.database.async_session'
) as mock_session_ctx, \
patch(
'backend.app.services.smart_plug_manager.tasmota_service'
) as mock_tasmota:
# Set current time to 08:00
mock_now = MagicMock()
mock_now.strftime.return_value = "08:00"
mock_datetime.now.return_value = mock_now
mock_datetime.utcnow.return_value = datetime.utcnow()
# Set up async session mock
mock_db = AsyncMock()
mock_result = MagicMock()
mock_result.scalars.return_value.all.return_value = [mock_plug]
mock_db.execute = AsyncMock(return_value=mock_result)
mock_db.commit = AsyncMock()
mock_session_ctx.return_value.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.return_value.__aexit__ = AsyncMock()
mock_tasmota.turn_on = AsyncMock(return_value=True)
await manager._check_schedules()
mock_tasmota.turn_on.assert_called_once_with(mock_plug)
@pytest.mark.asyncio
async def test_check_schedules_turns_off_at_scheduled_time(self, manager):
"""Verify scheduled off-time turns plug off."""
mock_plug = MagicMock()
mock_plug.id = 1
mock_plug.name = "Test Plug"
mock_plug.enabled = True
mock_plug.schedule_enabled = True
mock_plug.schedule_on_time = "08:00"
mock_plug.schedule_off_time = "22:00"
mock_plug.printer_id = 1
mock_plug.last_state = "ON"
with patch(
'backend.app.services.smart_plug_manager.datetime'
) as mock_datetime, \
patch(
'backend.app.core.database.async_session'
) as mock_session_ctx, \
patch(
'backend.app.services.smart_plug_manager.tasmota_service'
) as mock_tasmota, \
patch(
'backend.app.services.smart_plug_manager.printer_manager'
) as mock_pm:
# Set current time to 22:00
mock_now = MagicMock()
mock_now.strftime.return_value = "22:00"
mock_datetime.now.return_value = mock_now
mock_datetime.utcnow.return_value = datetime.utcnow()
# Set up async session mock
mock_db = AsyncMock()
mock_result = MagicMock()
mock_result.scalars.return_value.all.return_value = [mock_plug]
mock_db.execute = AsyncMock(return_value=mock_result)
mock_db.commit = AsyncMock()
mock_session_ctx.return_value.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.return_value.__aexit__ = AsyncMock()
mock_tasmota.turn_off = AsyncMock(return_value=True)
mock_pm.mark_printer_offline = MagicMock()
await manager._check_schedules()
mock_tasmota.turn_off.assert_called_once_with(mock_plug)
@pytest.mark.asyncio
async def test_check_schedules_skipped_when_disabled(self, manager):
"""Verify schedule is skipped when schedule_enabled is False."""
mock_plug = MagicMock()
mock_plug.id = 1
mock_plug.enabled = True
mock_plug.schedule_enabled = False # Disabled
with patch(
'backend.app.services.smart_plug_manager.datetime'
) as mock_datetime, \
patch(
'backend.app.core.database.async_session'
) as mock_session_ctx, \
patch(
'backend.app.services.smart_plug_manager.tasmota_service'
) as mock_tasmota:
mock_now = MagicMock()
mock_now.strftime.return_value = "08:00"
mock_datetime.now.return_value = mock_now
# Set up async session mock - returns no plugs (filtered by schedule_enabled)
mock_db = AsyncMock()
mock_result = MagicMock()
mock_result.scalars.return_value.all.return_value = []
mock_db.execute = AsyncMock(return_value=mock_result)
mock_db.commit = AsyncMock()
mock_session_ctx.return_value.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.return_value.__aexit__ = AsyncMock()
mock_tasmota.turn_on = AsyncMock()
await manager._check_schedules()
mock_tasmota.turn_on.assert_not_called()
class TestPendingAutoOffPersistence:
"""Tests for auto-off pending state persistence (restart recovery)."""
@pytest.fixture
def manager(self):
return SmartPlugManager()
@pytest.mark.asyncio
async def test_resume_pending_auto_offs_temperature_mode(self, manager):
"""Verify temperature-based pending auto-offs are resumed on startup."""
mock_plug = MagicMock()
mock_plug.id = 1
mock_plug.name = "Test Plug"
mock_plug.ip_address = "192.168.1.100"
mock_plug.username = None
mock_plug.password = None
mock_plug.printer_id = 1
mock_plug.auto_off_pending = True
mock_plug.auto_off_pending_since = datetime.utcnow()
mock_plug.off_delay_mode = "temperature"
mock_plug.off_temp_threshold = 70
with patch(
'backend.app.core.database.async_session'
) as mock_session_ctx, \
patch.object(
manager, '_schedule_temp_based_off'
) as mock_schedule:
mock_db = AsyncMock()
mock_result = MagicMock()
mock_result.scalars.return_value.all.return_value = [mock_plug]
mock_db.execute = AsyncMock(return_value=mock_result)
mock_session_ctx.return_value.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.return_value.__aexit__ = AsyncMock()
await manager.resume_pending_auto_offs()
mock_schedule.assert_called_once_with(mock_plug, 1, 70)
@pytest.mark.asyncio
async def test_resume_pending_auto_offs_time_mode_immediate_off(self, manager):
"""Verify time-based pending auto-offs turn off immediately on resume."""
mock_plug = MagicMock()
mock_plug.id = 1
mock_plug.name = "Test Plug"
mock_plug.ip_address = "192.168.1.100"
mock_plug.username = None
mock_plug.password = None
mock_plug.printer_id = 1
mock_plug.auto_off_pending = True
mock_plug.auto_off_pending_since = datetime.utcnow()
mock_plug.off_delay_mode = "time"
with patch(
'backend.app.core.database.async_session'
) as mock_session_ctx, \
patch(
'backend.app.services.smart_plug_manager.tasmota_service'
) as mock_tasmota, \
patch.object(
manager, '_mark_auto_off_executed', new_callable=AsyncMock
) as mock_mark, \
patch(
'backend.app.services.smart_plug_manager.printer_manager'
) as mock_pm:
mock_db = AsyncMock()
mock_result = MagicMock()
mock_result.scalars.return_value.all.return_value = [mock_plug]
mock_db.execute = AsyncMock(return_value=mock_result)
mock_session_ctx.return_value.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.return_value.__aexit__ = AsyncMock()
mock_tasmota.turn_off = AsyncMock(return_value=True)
await manager.resume_pending_auto_offs()
mock_tasmota.turn_off.assert_called_once()
mock_mark.assert_called_once_with(1)
+383
View File
@@ -0,0 +1,383 @@
"""Unit tests for TasmotaService.
Tests smart plug HTTP communication and error handling.
"""
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
from backend.app.services.tasmota import TasmotaService
class TestTasmotaService:
"""Tests for TasmotaService class."""
@pytest.fixture
def service(self):
"""Create a TasmotaService instance."""
return TasmotaService(timeout=5.0)
@pytest.fixture
def mock_plug(self):
"""Create a mock SmartPlug object."""
plug = MagicMock()
plug.ip_address = "192.168.1.100"
plug.username = None
plug.password = None
plug.name = "Test Plug"
return plug
# ========================================================================
# Tests for URL building
# ========================================================================
def test_build_url_without_auth(self, service):
"""Verify URL is built correctly without auth."""
url = service._build_url("192.168.1.100", "Power On")
assert url == "http://192.168.1.100/cm?cmnd=Power%20On"
def test_build_url_with_auth(self, service):
"""Verify URL includes credentials when provided."""
url = service._build_url(
"192.168.1.100", "Power On", username="admin", password="secret"
)
assert url == "http://admin:secret@192.168.1.100/cm?cmnd=Power%20On"
def test_build_url_encodes_special_characters(self, service):
"""Verify special characters in commands are encoded."""
url = service._build_url("192.168.1.100", "Backlog Power On; Delay 100")
assert "Backlog%20Power%20On" in url
# ========================================================================
# Tests for turn_on
# ========================================================================
@pytest.mark.asyncio
async def test_turn_on_success(self, service, mock_plug):
"""Verify turn_on returns True on success."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
mock_send.return_value = {"POWER": "ON"}
result = await service.turn_on(mock_plug)
assert result is True
mock_send.assert_called_once_with(
"192.168.1.100", "Power On", None, None
)
@pytest.mark.asyncio
async def test_turn_on_failure(self, service, mock_plug):
"""Verify turn_on returns False on failure."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
mock_send.return_value = None
result = await service.turn_on(mock_plug)
assert result is False
@pytest.mark.asyncio
async def test_turn_on_with_auth(self, service, mock_plug):
"""Verify turn_on passes credentials when provided."""
mock_plug.username = "admin"
mock_plug.password = "secret"
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
mock_send.return_value = {"POWER": "ON"}
await service.turn_on(mock_plug)
mock_send.assert_called_once_with(
"192.168.1.100", "Power On", "admin", "secret"
)
# ========================================================================
# Tests for turn_off
# ========================================================================
@pytest.mark.asyncio
async def test_turn_off_success(self, service, mock_plug):
"""Verify turn_off returns True on success."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
mock_send.return_value = {"POWER": "OFF"}
result = await service.turn_off(mock_plug)
assert result is True
@pytest.mark.asyncio
async def test_turn_off_failure(self, service, mock_plug):
"""Verify turn_off returns False on failure."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
mock_send.return_value = None
result = await service.turn_off(mock_plug)
assert result is False
# ========================================================================
# Tests for toggle
# ========================================================================
@pytest.mark.asyncio
async def test_toggle_success(self, service, mock_plug):
"""Verify toggle returns True on success."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
mock_send.return_value = {"POWER": "ON"}
result = await service.toggle(mock_plug)
assert result is True
mock_send.assert_called_once_with(
"192.168.1.100", "Power Toggle", None, None
)
# ========================================================================
# Tests for get_status
# ========================================================================
@pytest.mark.asyncio
async def test_get_status_returns_on(self, service, mock_plug):
"""Verify get_status returns correct state when ON."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
# Tasmota returns {"POWER": "ON"} for Power command
mock_send.return_value = {"POWER": "ON"}
result = await service.get_status(mock_plug)
assert result is not None
assert result["state"] == "ON"
assert result["reachable"] is True
@pytest.mark.asyncio
async def test_get_status_returns_off(self, service, mock_plug):
"""Verify get_status returns correct state when OFF."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
# Tasmota returns {"POWER": "OFF"} for Power command
mock_send.return_value = {"POWER": "OFF"}
result = await service.get_status(mock_plug)
assert result is not None
assert result["state"] == "OFF"
@pytest.mark.asyncio
async def test_get_status_unreachable(self, service, mock_plug):
"""Verify get_status handles unreachable device."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
mock_send.return_value = None
result = await service.get_status(mock_plug)
assert result is not None
assert result["reachable"] is False
# ========================================================================
# Tests for get_energy
# ========================================================================
@pytest.mark.asyncio
async def test_get_energy_returns_data(self, service, mock_plug):
"""Verify get_energy parses energy data correctly."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
mock_send.return_value = {
"StatusSNS": {
"ENERGY": {
"Power": 150.5,
"Voltage": 120.0,
"Current": 1.25,
"Today": 2.5,
"Total": 100.0,
"Factor": 0.95,
}
}
}
result = await service.get_energy(mock_plug)
assert result is not None
assert result["power"] == 150.5
assert result["voltage"] == 120.0
assert result["current"] == 1.25
assert result["today"] == 2.5
assert result["total"] == 100.0
assert result["factor"] == 0.95
@pytest.mark.asyncio
async def test_get_energy_handles_missing_data(self, service, mock_plug):
"""Verify get_energy handles devices without energy monitoring."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
mock_send.return_value = {"StatusSNS": {}}
result = await service.get_energy(mock_plug)
assert result is None
@pytest.mark.asyncio
async def test_get_energy_handles_unreachable(self, service, mock_plug):
"""Verify get_energy handles unreachable device."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
mock_send.return_value = None
result = await service.get_energy(mock_plug)
assert result is None
@pytest.mark.asyncio
async def test_get_energy_handles_partial_data(self, service, mock_plug):
"""Verify get_energy handles partial energy data."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
mock_send.return_value = {
"StatusSNS": {
"ENERGY": {
"Power": 150.5,
# Missing other fields
}
}
}
result = await service.get_energy(mock_plug)
assert result is not None
assert result["power"] == 150.5
# Missing fields should be None or 0
assert result.get("voltage") is None or result.get("voltage") == 0
# ========================================================================
# Tests for test_connection
# ========================================================================
@pytest.mark.asyncio
async def test_test_connection_success(self, service):
"""Verify test_connection returns success on reachable device."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
# First call (Power) returns state, second call (Status 0) returns device info
mock_send.side_effect = [
{"POWER": "ON"}, # Power command response
{"Status": {"DeviceName": "Test Plug"}} # Status 0 response
]
result = await service.test_connection("192.168.1.100")
assert result["success"] is True
assert result["state"] == "ON"
assert result["device_name"] == "Test Plug"
@pytest.mark.asyncio
async def test_test_connection_failure(self, service):
"""Verify test_connection returns failure on unreachable device."""
with patch.object(
service, '_send_command', new_callable=AsyncMock
) as mock_send:
mock_send.return_value = None
result = await service.test_connection("192.168.1.100")
assert result["success"] is False
# ========================================================================
# Tests for _send_command
# ========================================================================
@pytest.mark.asyncio
async def test_send_command_handles_timeout(self, service):
"""Verify timeout is handled gracefully."""
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
mock_client.get.side_effect = httpx.TimeoutException("Timeout")
mock_client_class.return_value.__aenter__ = AsyncMock(
return_value=mock_client
)
mock_client_class.return_value.__aexit__ = AsyncMock()
result = await service._send_command("192.168.1.100", "Power")
assert result is None
@pytest.mark.asyncio
async def test_send_command_handles_connection_error(self, service):
"""Verify connection error is handled gracefully."""
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
mock_client.get.side_effect = httpx.ConnectError("Connection refused")
mock_client_class.return_value.__aenter__ = AsyncMock(
return_value=mock_client
)
mock_client_class.return_value.__aexit__ = AsyncMock()
result = await service._send_command("192.168.1.100", "Power")
assert result is None
@pytest.mark.asyncio
async def test_send_command_handles_invalid_json(self, service):
"""Verify invalid JSON response is handled gracefully."""
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
mock_response = MagicMock()
mock_response.json.side_effect = ValueError("Invalid JSON")
mock_client.get.return_value = mock_response
mock_client_class.return_value.__aenter__ = AsyncMock(
return_value=mock_client
)
mock_client_class.return_value.__aexit__ = AsyncMock()
result = await service._send_command("192.168.1.100", "Power")
assert result is None
@pytest.mark.asyncio
async def test_send_command_success(self, service):
"""Verify successful command returns parsed JSON."""
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
mock_response = MagicMock()
mock_response.json.return_value = {"POWER": "ON"}
mock_client.get.return_value = mock_response
mock_client_class.return_value.__aenter__ = AsyncMock(
return_value=mock_client
)
mock_client_class.return_value.__aexit__ = AsyncMock()
result = await service._send_command("192.168.1.100", "Power")
assert result == {"POWER": "ON"}
class TestTasmotaServiceSingleton:
"""Tests for the global tasmota_service singleton."""
def test_singleton_exists(self):
"""Verify global tasmota_service instance exists."""
from backend.app.services.tasmota import tasmota_service
assert tasmota_service is not None
assert isinstance(tasmota_service, TasmotaService)