mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
0cc7cf0f36
A PUT /settings with null for a number setting stored the literal "None". From then on int()/float() raised inside the response builder, and every settings read returned 503 until the row was fixed by hand. Explicit null for any boolean or numeric setting is now refused with a 422 that names the keys, and nothing in that request is saved. A numeric row that does not parse reads back as its default with a warning, so an install that already stored one recovers. The typed-key lists moved to module constants so the save check and the read path share them.
1058 lines
42 KiB
Python
1058 lines
42 KiB
Python
"""Integration tests for Settings API endpoints.
|
|
|
|
Tests the full request/response cycle for /api/v1/settings/ endpoints.
|
|
"""
|
|
|
|
import os
|
|
|
|
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)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_unset_temp_alarm_reads_back_as_null(self, async_client: AsyncClient, db_session):
|
|
"""#2905: ams_temp_alarm is nullable, and settings storage stringifies
|
|
None to the literal "None".
|
|
|
|
Putting it in the plain float-cast list would make float("None") raise
|
|
inside the response builder and take the whole settings response with it
|
|
— every unrelated setting on the page included.
|
|
"""
|
|
from backend.app.models.settings import Settings
|
|
|
|
db_session.add(Settings(key="ams_temp_alarm", value="None"))
|
|
await db_session.commit()
|
|
|
|
response = await async_client.get("/api/v1/settings/")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["ams_temp_alarm"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
@pytest.mark.parametrize(
|
|
("key", "default"),
|
|
[
|
|
("ambient_drying_sustained_minutes", 0),
|
|
("queue_keep_warm_bed_temp", 90),
|
|
("default_filament_cost", 25.0),
|
|
],
|
|
)
|
|
async def test_unparseable_number_setting_reads_back_as_the_default(
|
|
self, async_client: AsyncClient, db_session, key, default
|
|
):
|
|
"""A numeric row holding "None" (an old null save) or any other
|
|
unparseable value must fall back to the default, not make int()/float()
|
|
raise inside the response builder and take every setting with it."""
|
|
from backend.app.models.settings import Settings
|
|
|
|
db_session.add(Settings(key=key, value="None"))
|
|
await db_session.commit()
|
|
|
|
response = await async_client.get("/api/v1/settings/")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()[key] == default
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
@pytest.mark.parametrize("key", ["ambient_drying_sustained_minutes", "default_filament_cost", "auto_archive"])
|
|
async def test_null_for_a_typed_setting_is_refused_and_not_stored(self, async_client: AsyncClient, key):
|
|
"""A null for a boolean or numeric setting would be stored as the literal
|
|
"None". The whole request is refused, so a valid field sent alongside it
|
|
is not half-applied either."""
|
|
before = (await async_client.get("/api/v1/settings/")).json()
|
|
assert before["currency"] != "EUR"
|
|
|
|
response = await async_client.put("/api/v1/settings/", json={key: None, "currency": "EUR"})
|
|
|
|
assert response.status_code == 422
|
|
assert key in response.json()["detail"]
|
|
after = (await async_client.get("/api/v1/settings/")).json()
|
|
assert after[key] == before[key]
|
|
assert after["currency"] == before["currency"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_a_set_temp_alarm_reads_back_as_a_float(self, async_client: AsyncClient, db_session):
|
|
from backend.app.models.settings import Settings
|
|
|
|
db_session.add(Settings(key="ams_temp_alarm", value="45"))
|
|
await db_session.commit()
|
|
|
|
response = await async_client.get("/api/v1/settings/")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["ams_temp_alarm"] == 45.0
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_a_malformed_temp_alarm_does_not_break_the_response(self, async_client: AsyncClient, db_session):
|
|
"""A hand-edited or half-written value must degrade to "unset" rather
|
|
than making the settings page unreachable."""
|
|
from backend.app.models.settings import Settings
|
|
|
|
db_session.add(Settings(key="ams_temp_alarm", value="warm"))
|
|
await db_session.commit()
|
|
|
|
response = await async_client.get("/api/v1/settings/")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["ams_temp_alarm"] is None
|
|
assert "currency" in response.json(), "the rest of the page still renders"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_temp_alarm_survives_a_set_then_clear_round_trip(self, async_client: AsyncClient):
|
|
"""The path the settings page actually takes, end to end (#2905).
|
|
|
|
The tests above seed rows directly, which pins the read but not the
|
|
convention the whole design rests on: clearing the field sends an
|
|
explicit ``null``, ``update_settings`` stores that as the literal
|
|
string ``"None"``, and the response builder has to turn it back into
|
|
``None``. A regression anywhere along that chain would leave a cleared
|
|
threshold reading back as the old number, and no test above would fail.
|
|
"""
|
|
from sqlalchemy import select
|
|
|
|
# Imported here, not at module scope: the async_client fixture patches
|
|
# core.database.async_session onto the test engine, so a name bound at
|
|
# import time would point at the real app database and find nothing.
|
|
from backend.app.core.database import async_session
|
|
from backend.app.models.settings import Settings
|
|
|
|
response = await async_client.put("/api/v1/settings/", json={"ams_temp_alarm": 45})
|
|
assert response.status_code == 200
|
|
assert response.json()["ams_temp_alarm"] == 45.0
|
|
assert (await async_client.get("/api/v1/settings/")).json()["ams_temp_alarm"] == 45.0
|
|
|
|
response = await async_client.put("/api/v1/settings/", json={"ams_temp_alarm": None})
|
|
assert response.status_code == 200
|
|
assert response.json()["ams_temp_alarm"] is None
|
|
assert (await async_client.get("/api/v1/settings/")).json()["ams_temp_alarm"] is None
|
|
|
|
# Pin the stored form too — the fallback in _resolve_temp_alarm_threshold
|
|
# is written against this exact string, so a storage change that silently
|
|
# switched to "" or NULL would break the alarm rather than this test.
|
|
async with async_session() as db:
|
|
row = (await db.execute(select(Settings).where(Settings.key == "ams_temp_alarm"))).scalar_one()
|
|
assert row.value == "None"
|
|
|
|
# ========================================================================
|
|
# 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_low_stock_threshold(self, async_client: AsyncClient):
|
|
"""Verify low stock threshold setting can be updated."""
|
|
# Get default value
|
|
response = await async_client.get("/api/v1/settings/")
|
|
assert response.status_code == 200
|
|
assert response.json()["low_stock_threshold"] == 20.0
|
|
|
|
# Update to custom value
|
|
response = await async_client.put("/api/v1/settings/", json={"low_stock_threshold": 15.5})
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["low_stock_threshold"] == 15.5
|
|
|
|
# Verify persistence
|
|
response = await async_client.get("/api/v1/settings/")
|
|
assert response.status_code == 200
|
|
assert response.json()["low_stock_threshold"] == 15.5
|
|
|
|
@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_update_theme_settings(self, async_client: AsyncClient):
|
|
"""Verify theme settings can be updated."""
|
|
response = await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={
|
|
"dark_style": "glow",
|
|
"dark_background": "forest",
|
|
"dark_accent": "teal",
|
|
"light_style": "vibrant",
|
|
"light_background": "warm",
|
|
"light_accent": "blue",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["dark_style"] == "glow"
|
|
assert result["dark_background"] == "forest"
|
|
assert result["dark_accent"] == "teal"
|
|
assert result["light_style"] == "vibrant"
|
|
assert result["light_background"] == "warm"
|
|
assert result["light_accent"] == "blue"
|
|
|
|
@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
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_check_printer_firmware(self, async_client: AsyncClient):
|
|
"""Verify check_printer_firmware can be updated."""
|
|
# Default should be True
|
|
response = await async_client.get("/api/v1/settings/")
|
|
assert response.json()["check_printer_firmware"] is True
|
|
|
|
# Update to False
|
|
response = await async_client.put("/api/v1/settings/", json={"check_printer_firmware": False})
|
|
assert response.status_code == 200
|
|
assert response.json()["check_printer_firmware"] is False
|
|
|
|
# Verify persistence
|
|
response = await async_client.get("/api/v1/settings/")
|
|
assert response.json()["check_printer_firmware"] is False
|
|
|
|
# Update back to True
|
|
response = await async_client.put("/api/v1/settings/", json={"check_printer_firmware": True})
|
|
assert response.status_code == 200
|
|
assert response.json()["check_printer_firmware"] is True
|
|
|
|
# ========================================================================
|
|
# MQTT settings tests
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_mqtt_settings(self, async_client: AsyncClient):
|
|
"""Verify MQTT settings can be updated."""
|
|
response = await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={
|
|
"mqtt_enabled": True,
|
|
"mqtt_broker": "mqtt.example.com",
|
|
"mqtt_port": 8883,
|
|
"mqtt_username": "testuser",
|
|
"mqtt_password": "testpass",
|
|
"mqtt_topic_prefix": "myprefix",
|
|
"mqtt_use_tls": True,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["mqtt_enabled"] is True
|
|
assert result["mqtt_broker"] == "mqtt.example.com"
|
|
assert result["mqtt_port"] == 8883
|
|
assert result["mqtt_username"] == "testuser"
|
|
assert result["mqtt_password"] == "testpass"
|
|
assert result["mqtt_topic_prefix"] == "myprefix"
|
|
assert result["mqtt_use_tls"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_mqtt_status_endpoint(self, async_client: AsyncClient):
|
|
"""Verify MQTT status endpoint returns expected fields."""
|
|
response = await async_client.get("/api/v1/settings/mqtt/status")
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert "enabled" in result
|
|
assert "connected" in result
|
|
assert "broker" in result
|
|
assert "port" in result
|
|
assert "topic_prefix" in result
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_mqtt_defaults(self, async_client: AsyncClient):
|
|
"""Verify MQTT has correct default values."""
|
|
# Reset MQTT settings to defaults
|
|
await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={
|
|
"mqtt_enabled": False,
|
|
"mqtt_broker": "",
|
|
"mqtt_port": 1883,
|
|
"mqtt_username": "",
|
|
"mqtt_password": "",
|
|
"mqtt_topic_prefix": "bambuddy",
|
|
"mqtt_use_tls": False,
|
|
},
|
|
)
|
|
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
|
|
assert result["mqtt_enabled"] is False
|
|
assert result["mqtt_port"] == 1883
|
|
assert result["mqtt_topic_prefix"] == "bambuddy"
|
|
assert result["mqtt_use_tls"] is False
|
|
|
|
# ========================================================================
|
|
# Camera settings tests
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_camera_view_mode(self, async_client: AsyncClient):
|
|
"""Verify camera view mode can be updated."""
|
|
response = await async_client.put("/api/v1/settings/", json={"camera_view_mode": "embedded"})
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["camera_view_mode"] == "embedded"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_camera_view_mode_persists(self, async_client: AsyncClient):
|
|
"""CRITICAL: Verify camera view mode persists after update."""
|
|
# Update to embedded
|
|
await async_client.put("/api/v1/settings/", json={"camera_view_mode": "embedded"})
|
|
|
|
# Verify persistence in new request
|
|
response = await async_client.get("/api/v1/settings/")
|
|
assert response.json()["camera_view_mode"] == "embedded"
|
|
|
|
# Update back to window
|
|
await async_client.put("/api/v1/settings/", json={"camera_view_mode": "window"})
|
|
|
|
# Verify persistence
|
|
response = await async_client.get("/api/v1/settings/")
|
|
assert response.json()["camera_view_mode"] == "window"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_camera_view_mode_default(self, async_client: AsyncClient):
|
|
"""Verify camera view mode has correct default value."""
|
|
# Reset by requesting settings (default should be 'window')
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
|
|
assert "camera_view_mode" in result
|
|
# Default is 'window' as defined in schema
|
|
assert result["camera_view_mode"] in ["window", "embedded"]
|
|
|
|
# ========================================================================
|
|
# Per-printer mapping settings tests
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_per_printer_mapping_expanded(self, async_client: AsyncClient):
|
|
"""Verify per_printer_mapping_expanded can be updated."""
|
|
response = await async_client.put("/api/v1/settings/", json={"per_printer_mapping_expanded": True})
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["per_printer_mapping_expanded"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_per_printer_mapping_expanded_persists(self, async_client: AsyncClient):
|
|
"""CRITICAL: Verify per_printer_mapping_expanded persists after update."""
|
|
# Update to True
|
|
await async_client.put("/api/v1/settings/", json={"per_printer_mapping_expanded": True})
|
|
|
|
# Verify persistence in new request
|
|
response = await async_client.get("/api/v1/settings/")
|
|
assert response.json()["per_printer_mapping_expanded"] is True
|
|
|
|
# Update back to False
|
|
await async_client.put("/api/v1/settings/", json={"per_printer_mapping_expanded": False})
|
|
|
|
# Verify persistence
|
|
response = await async_client.get("/api/v1/settings/")
|
|
assert response.json()["per_printer_mapping_expanded"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_per_printer_mapping_expanded_default(self, async_client: AsyncClient):
|
|
"""Verify per_printer_mapping_expanded has correct default value."""
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
|
|
assert "per_printer_mapping_expanded" in result
|
|
# Default is False as defined in schema
|
|
assert isinstance(result["per_printer_mapping_expanded"], bool)
|
|
|
|
# ========================================================================
|
|
# Stagger settings tests
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_stagger_settings_defaults(self, async_client: AsyncClient):
|
|
"""Verify stagger settings have correct defaults."""
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
|
|
assert result["stagger_group_size"] == 2
|
|
assert result["stagger_interval_minutes"] == 5
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_stagger_settings(self, async_client: AsyncClient):
|
|
"""Verify stagger settings can be updated."""
|
|
response = await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={"stagger_group_size": 3, "stagger_interval_minutes": 10},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["stagger_group_size"] == 3
|
|
assert result["stagger_interval_minutes"] == 10
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_stagger_settings_persist(self, async_client: AsyncClient):
|
|
"""Verify stagger settings persist after update."""
|
|
await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={"stagger_group_size": 4, "stagger_interval_minutes": 15},
|
|
)
|
|
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
assert result["stagger_group_size"] == 4
|
|
assert result["stagger_interval_minutes"] == 15
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_stagger_settings_validation(self, async_client: AsyncClient):
|
|
"""Verify stagger settings reject out-of-range values."""
|
|
response = await async_client.put("/api/v1/settings/", json={"stagger_group_size": 0})
|
|
assert response.status_code == 422
|
|
|
|
response = await async_client.put("/api/v1/settings/", json={"stagger_group_size": 51})
|
|
assert response.status_code == 422
|
|
|
|
response = await async_client.put("/api/v1/settings/", json={"stagger_interval_minutes": 0})
|
|
assert response.status_code == 422
|
|
|
|
response = await async_client.put("/api/v1/settings/", json={"stagger_interval_minutes": 61})
|
|
assert response.status_code == 422
|
|
|
|
# ========================================================================
|
|
# Default print options tests
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_default_print_options_defaults(self, async_client: AsyncClient):
|
|
"""Verify default print options have correct defaults."""
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
|
|
# bed_levelling / flow_cali are tri-state, defaulting to "auto".
|
|
assert result["default_bed_levelling"] == "auto"
|
|
assert result["default_flow_cali"] == "auto"
|
|
assert result["default_vibration_cali"] is True
|
|
assert result["default_layer_inspect"] is False
|
|
assert result["default_timelapse"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_default_print_options(self, async_client: AsyncClient):
|
|
"""Verify default print options can be updated (tri-state + booleans)."""
|
|
response = await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={
|
|
"default_bed_levelling": "off",
|
|
"default_flow_cali": "on",
|
|
"default_vibration_cali": False,
|
|
"default_layer_inspect": True,
|
|
"default_timelapse": True,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["default_bed_levelling"] == "off"
|
|
assert result["default_flow_cali"] == "on"
|
|
assert result["default_vibration_cali"] is False
|
|
assert result["default_layer_inspect"] is True
|
|
assert result["default_timelapse"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_default_print_options_legacy_bool_coerced(self, async_client: AsyncClient):
|
|
"""Old clients sending booleans for the tri-state options still work.
|
|
|
|
The TriState validator maps true->"on", false->"off" on input so a
|
|
pre-upgrade frontend never writes an invalid value.
|
|
"""
|
|
response = await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={"default_bed_levelling": False, "default_flow_cali": True},
|
|
)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["default_bed_levelling"] == "off"
|
|
assert result["default_flow_cali"] == "on"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_default_print_options_persist(self, async_client: AsyncClient):
|
|
"""CRITICAL: Verify default print options persist after update."""
|
|
await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={
|
|
"default_bed_levelling": "on",
|
|
"default_timelapse": True,
|
|
},
|
|
)
|
|
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
assert result["default_bed_levelling"] == "on"
|
|
assert result["default_timelapse"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_default_print_options_partial_update(self, async_client: AsyncClient):
|
|
"""Verify partial updates don't affect other default print options."""
|
|
# Set all to non-default
|
|
await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={
|
|
"default_bed_levelling": "off",
|
|
"default_flow_cali": "on",
|
|
},
|
|
)
|
|
|
|
# Update only one
|
|
response = await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={"default_bed_levelling": "auto"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["default_bed_levelling"] == "auto"
|
|
assert result["default_flow_cali"] == "on" # Should remain from previous update
|
|
|
|
# ========================================================================
|
|
# Home Assistant environment variable tests
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_ha_settings_default_no_env_vars(self, async_client: AsyncClient):
|
|
"""Verify HA settings work without environment variables (default behavior)."""
|
|
# Ensure no env vars are set
|
|
os.environ.pop("HA_URL", None)
|
|
os.environ.pop("HA_TOKEN", None)
|
|
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
|
|
assert response.status_code == 200
|
|
assert "ha_enabled" in result
|
|
assert "ha_url" in result
|
|
assert "ha_token" in result
|
|
assert "ha_url_from_env" in result
|
|
assert "ha_token_from_env" in result
|
|
assert "ha_env_managed" in result
|
|
|
|
# Default values without env vars
|
|
assert result["ha_url_from_env"] is False
|
|
assert result["ha_token_from_env"] is False
|
|
assert result["ha_env_managed"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_ha_settings_with_both_env_vars(self, async_client: AsyncClient):
|
|
"""Verify HA settings are overridden when both env vars are set."""
|
|
# Set environment variables
|
|
os.environ["HA_URL"] = "http://supervisor/core"
|
|
os.environ["HA_TOKEN"] = "test-token-12345"
|
|
|
|
try:
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
|
|
assert response.status_code == 200
|
|
|
|
# Verify env var values are used
|
|
assert result["ha_url"] == "http://supervisor/core"
|
|
assert result["ha_token"] == "test-token-12345"
|
|
|
|
# Verify metadata fields
|
|
assert result["ha_url_from_env"] is True
|
|
assert result["ha_token_from_env"] is True
|
|
assert result["ha_env_managed"] is True
|
|
|
|
# Verify auto-enable behavior
|
|
assert result["ha_enabled"] is True
|
|
|
|
finally:
|
|
# Clean up
|
|
os.environ.pop("HA_URL", None)
|
|
os.environ.pop("HA_TOKEN", None)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_ha_settings_with_only_url_env_var(self, async_client: AsyncClient):
|
|
"""Verify partial configuration when only HA_URL is set."""
|
|
# Set only URL env var
|
|
os.environ["HA_URL"] = "http://supervisor/core"
|
|
os.environ.pop("HA_TOKEN", None)
|
|
|
|
try:
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
|
|
assert response.status_code == 200
|
|
|
|
# Verify URL is from env, token is from database
|
|
assert result["ha_url"] == "http://supervisor/core"
|
|
assert result["ha_url_from_env"] is True
|
|
assert result["ha_token_from_env"] is False
|
|
assert result["ha_env_managed"] is False
|
|
|
|
# No auto-enable with partial config
|
|
assert result["ha_enabled"] is False # Database default
|
|
|
|
finally:
|
|
os.environ.pop("HA_URL", None)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_ha_settings_with_only_token_env_var(self, async_client: AsyncClient):
|
|
"""Verify partial configuration when only HA_TOKEN is set."""
|
|
# Set only token env var
|
|
os.environ.pop("HA_URL", None)
|
|
os.environ["HA_TOKEN"] = "test-token-12345"
|
|
|
|
try:
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
|
|
assert response.status_code == 200
|
|
|
|
# Verify token is from env, URL is from database
|
|
assert result["ha_token"] == "test-token-12345"
|
|
assert result["ha_url_from_env"] is False
|
|
assert result["ha_token_from_env"] is True
|
|
assert result["ha_env_managed"] is False
|
|
|
|
# No auto-enable with partial config
|
|
assert result["ha_enabled"] is False # Database default
|
|
|
|
finally:
|
|
os.environ.pop("HA_TOKEN", None)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_ha_settings_env_vars_override_database(self, async_client: AsyncClient):
|
|
"""Verify environment variables take precedence over database values."""
|
|
# First, set database values
|
|
await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={
|
|
"ha_enabled": True,
|
|
"ha_url": "http://database-url:8123",
|
|
"ha_token": "database-token",
|
|
},
|
|
)
|
|
|
|
# Verify database values are set
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
assert result["ha_url"] == "http://database-url:8123"
|
|
assert result["ha_token"] == "database-token"
|
|
|
|
# Now set environment variables
|
|
os.environ["HA_URL"] = "http://env-url/core"
|
|
os.environ["HA_TOKEN"] = "env-token-xyz"
|
|
|
|
try:
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
|
|
# Verify env vars override database
|
|
assert result["ha_url"] == "http://env-url/core"
|
|
assert result["ha_token"] == "env-token-xyz"
|
|
assert result["ha_url_from_env"] is True
|
|
assert result["ha_token_from_env"] is True
|
|
assert result["ha_env_managed"] is True
|
|
assert result["ha_enabled"] is True
|
|
|
|
finally:
|
|
os.environ.pop("HA_URL", None)
|
|
os.environ.pop("HA_TOKEN", None)
|
|
|
|
# Verify database values are still there after removing env vars
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
assert result["ha_url"] == "http://database-url:8123"
|
|
assert result["ha_token"] == "database-token"
|
|
assert result["ha_url_from_env"] is False
|
|
assert result["ha_token_from_env"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_ha_settings_database_updates_accepted_but_ignored(self, async_client: AsyncClient):
|
|
"""Verify database updates are accepted but have no effect when env vars are set."""
|
|
# Set environment variables
|
|
os.environ["HA_URL"] = "http://supervisor/core"
|
|
os.environ["HA_TOKEN"] = "env-token"
|
|
|
|
try:
|
|
# Attempt to update via API
|
|
response = await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={
|
|
"ha_url": "http://different-url:8123",
|
|
"ha_token": "different-token",
|
|
},
|
|
)
|
|
|
|
# Update should succeed
|
|
assert response.status_code == 200
|
|
|
|
# But values should still be from env vars
|
|
result = response.json()
|
|
assert result["ha_url"] == "http://supervisor/core"
|
|
assert result["ha_token"] == "env-token"
|
|
assert result["ha_url_from_env"] is True
|
|
assert result["ha_token_from_env"] is True
|
|
|
|
finally:
|
|
os.environ.pop("HA_URL", None)
|
|
os.environ.pop("HA_TOKEN", None)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_ha_settings_empty_env_vars_treated_as_not_set(self, async_client: AsyncClient):
|
|
"""Verify empty environment variables are treated as not set."""
|
|
# Set empty env vars
|
|
os.environ["HA_URL"] = ""
|
|
os.environ["HA_TOKEN"] = ""
|
|
|
|
try:
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
|
|
# Empty env vars should be treated as not set
|
|
assert result["ha_url_from_env"] is False
|
|
assert result["ha_token_from_env"] is False
|
|
assert result["ha_env_managed"] is False
|
|
|
|
finally:
|
|
os.environ.pop("HA_URL", None)
|
|
os.environ.pop("HA_TOKEN", None)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_ha_settings_can_be_updated_normally_without_env_vars(self, async_client: AsyncClient):
|
|
"""Verify HA settings can be updated normally when env vars are not set."""
|
|
# Ensure no env vars
|
|
os.environ.pop("HA_URL", None)
|
|
os.environ.pop("HA_TOKEN", None)
|
|
|
|
# Update HA settings
|
|
response = await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={
|
|
"ha_enabled": True,
|
|
"ha_url": "http://192.168.1.100:8123",
|
|
"ha_token": "my-long-lived-token",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["ha_enabled"] is True
|
|
assert result["ha_url"] == "http://192.168.1.100:8123"
|
|
assert result["ha_token"] == "my-long-lived-token"
|
|
assert result["ha_url_from_env"] is False
|
|
assert result["ha_token_from_env"] is False
|
|
assert result["ha_env_managed"] is False
|
|
|
|
# Verify persistence
|
|
response = await async_client.get("/api/v1/settings/")
|
|
result = response.json()
|
|
assert result["ha_enabled"] is True
|
|
assert result["ha_url"] == "http://192.168.1.100:8123"
|
|
assert result["ha_token"] == "my-long-lived-token"
|
|
|
|
|
|
class TestOpenInSlicerOverride:
|
|
"""Per #1329, the desktop 'Open in Slicer' target can diverge from the API
|
|
sidecar slicer. The new `open_in_slicer` setting is None by default (frontend
|
|
inherits from `preferred_slicer`); setting it to 'orcaslicer' or 'bambu_studio'
|
|
overrides only the desktop URI handoff, not the in-app SliceModal."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_open_in_slicer_default_is_null(self, async_client: AsyncClient):
|
|
response = await async_client.get("/api/v1/settings/")
|
|
assert response.status_code == 200
|
|
# Default null so existing installs behave identically — the frontend
|
|
# then falls back to preferred_slicer.
|
|
assert response.json()["open_in_slicer"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_open_in_slicer_override_persists(self, async_client: AsyncClient):
|
|
# Set preferred_slicer=bambu_studio (API sidecar) but
|
|
# open_in_slicer=orcaslicer (desktop). Exactly the reporter's case:
|
|
# slice via Bambu Studio sidecar, open files locally in OrcaSlicer.
|
|
response = await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={"preferred_slicer": "bambu_studio", "open_in_slicer": "orcaslicer"},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["preferred_slicer"] == "bambu_studio"
|
|
assert body["open_in_slicer"] == "orcaslicer"
|
|
|
|
# Persisted across a fresh GET.
|
|
get_resp = await async_client.get("/api/v1/settings/")
|
|
assert get_resp.json()["preferred_slicer"] == "bambu_studio"
|
|
assert get_resp.json()["open_in_slicer"] == "orcaslicer"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_open_in_slicer_can_be_cleared_to_null(self, async_client: AsyncClient):
|
|
# Reset path: user picks an override, then later goes back to "Same as
|
|
# API slicer". The literal string "None" the PUT path writes for a
|
|
# None value must be normalized back to a real null on GET — otherwise
|
|
# the frontend can't distinguish "explicit override absent" from
|
|
# "explicit override set to a bogus value".
|
|
await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={"open_in_slicer": "orcaslicer"},
|
|
)
|
|
response = await async_client.put(
|
|
"/api/v1/settings/",
|
|
json={"open_in_slicer": None},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["open_in_slicer"] is None
|
|
|
|
# And a fresh GET also sees it as null, not the literal string "None".
|
|
get_resp = await async_client.get("/api/v1/settings/")
|
|
assert get_resp.json()["open_in_slicer"] is None
|
|
|
|
|
|
class TestSimplifiedBackupRestore:
|
|
"""Integration tests for the simplified backup/restore endpoints (ZIP-based).
|
|
|
|
Note: Tests that require actual file operations (backup creation) are skipped
|
|
because the test suite uses an in-memory database. These tests focus on
|
|
validation and error handling which don't require file I/O.
|
|
"""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_restore_requires_zip_file(self, async_client: AsyncClient):
|
|
"""Verify restore rejects non-ZIP files."""
|
|
files = {"file": ("backup.txt", b"not a zip file", "text/plain")}
|
|
response = await async_client.post("/api/v1/settings/restore", files=files)
|
|
|
|
assert response.status_code == 400
|
|
assert "zip" in response.json()["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_restore_requires_database_in_zip(self, async_client: AsyncClient):
|
|
"""Verify restore rejects ZIP without database file."""
|
|
import io
|
|
import zipfile
|
|
|
|
# Create a ZIP without bambuddy.db
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr("dummy.txt", "dummy content")
|
|
zip_buffer.seek(0)
|
|
|
|
files = {"file": ("backup.zip", zip_buffer.read(), "application/zip")}
|
|
response = await async_client.post("/api/v1/settings/restore", files=files)
|
|
|
|
assert response.status_code == 400
|
|
assert "missing bambuddy.db" in response.json()["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_restore_invalid_zip(self, async_client: AsyncClient):
|
|
"""Verify restore rejects corrupted ZIP files."""
|
|
files = {"file": ("backup.zip", b"not valid zip content", "application/zip")}
|
|
response = await async_client.post("/api/v1/settings/restore", files=files)
|
|
|
|
assert response.status_code == 400
|
|
assert "not a valid zip" in response.json()["detail"].lower()
|