mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
c457cf54bf
A bundle described everything except the process it runs in. So a report of memory climbing over days until the OOM killer fires arrives with no way to act on it: the numbers that name the mechanism only exist while it is happening, and by the time anyone asks, the container has been restarted. The new `process` section carries what actually separates the candidates. Resident against virtual memory: 650MB RSS with 12.9GB VMS is address space — thread stacks or allocator arenas — not a heap full of live data, and that reading is the opposite of the one the reporter drew from the same figures. Thread count and child-process count then split those two apart, and a census of live objects by type names what a growing heap is filling up with. Open files, sockets and uptime round it out. Three constraints worth keeping: The heap census is skipped above 2GB. gc.get_objects() materialises every tracked object, so it costs most on exactly the process that can least afford it — a bundle generated to diagnose runaway memory must not be the allocation that tips the host over. Everything else is still collected, and the skip is recorded with its reason rather than silently omitted. Children are recorded by executable name only. An ffmpeg command line carries the camera URL, and with it the camera's password. Collection runs off the event loop and every metric is independently best-effort. psutil raises on hardened kernels and in restricted containers, and the bundle is how someone reports a problem in the first place — it has to be produced even when half the numbers are unavailable. This does not fix #2734, and nothing here should be read as having found its cause. The bundle's own evidence contradicts both proposed causes: the orphan janitor ran 7 times in 26 days over 725 stream-ends and killed no orphaned ffmpeg, which is not the #776 signature; and the 5 "database is locked" errors all fall between two OOM kills, making them a symptom of the memory pressure rather than a source of it.
1576 lines
65 KiB
Python
1576 lines
65 KiB
Python
"""Unit tests for support module helper functions.
|
|
|
|
Tests _anonymize_mqtt_broker, _check_port, _get_container_memory_limit,
|
|
_format_bytes, and _collect_support_info diagnostic sections.
|
|
"""
|
|
|
|
import asyncio
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
class TestApplyLogLevel:
|
|
"""Tests for _apply_log_level() debug noise suppression."""
|
|
|
|
def test_debug_mode_suppresses_sqlalchemy_to_warning(self):
|
|
"""Verify sqlalchemy.engine is set to WARNING (not INFO) in debug mode."""
|
|
import logging
|
|
|
|
from backend.app.api.routes.support import _apply_log_level
|
|
|
|
_apply_log_level(True)
|
|
|
|
assert logging.getLogger("sqlalchemy.engine").level == logging.WARNING
|
|
|
|
def test_debug_mode_suppresses_aiosqlite(self):
|
|
"""Verify aiosqlite is set to WARNING in debug mode to prevent cursor noise."""
|
|
import logging
|
|
|
|
from backend.app.api.routes.support import _apply_log_level
|
|
|
|
_apply_log_level(True)
|
|
|
|
assert logging.getLogger("aiosqlite").level == logging.WARNING
|
|
|
|
def test_debug_mode_keeps_httpx_pinned_to_warning(self):
|
|
"""httpx/httpcore must stay at WARNING even in debug mode — at INFO/DEBUG
|
|
they log full request URLs, leaking webhook tokens (Discord etc.)."""
|
|
import logging
|
|
|
|
from backend.app.api.routes.support import _apply_log_level
|
|
|
|
_apply_log_level(True)
|
|
|
|
assert logging.getLogger("httpcore").level == logging.WARNING
|
|
assert logging.getLogger("httpx").level == logging.WARNING
|
|
|
|
def test_non_debug_mode_suppresses_all_noisy_loggers(self):
|
|
"""Verify all noisy loggers are set to WARNING in non-debug mode."""
|
|
import logging
|
|
|
|
from backend.app.api.routes.support import _apply_log_level
|
|
|
|
_apply_log_level(False)
|
|
|
|
assert logging.getLogger("sqlalchemy.engine").level == logging.WARNING
|
|
assert logging.getLogger("httpcore").level == logging.WARNING
|
|
assert logging.getLogger("httpx").level == logging.WARNING
|
|
assert logging.getLogger("paho.mqtt").level == logging.WARNING
|
|
|
|
|
|
class TestAnonymizeMqttBroker:
|
|
"""Tests for _anonymize_mqtt_broker()."""
|
|
|
|
def test_empty_string(self):
|
|
from backend.app.api.routes.support import _anonymize_mqtt_broker
|
|
|
|
assert _anonymize_mqtt_broker("") == ""
|
|
|
|
def test_ipv4_address(self):
|
|
from backend.app.api.routes.support import _anonymize_mqtt_broker
|
|
|
|
assert _anonymize_mqtt_broker("192.168.1.100") == "[IP]"
|
|
|
|
def test_ipv6_address(self):
|
|
from backend.app.api.routes.support import _anonymize_mqtt_broker
|
|
|
|
assert _anonymize_mqtt_broker("::1") == "[IP]"
|
|
|
|
def test_hostname_with_domain(self):
|
|
from backend.app.api.routes.support import _anonymize_mqtt_broker
|
|
|
|
assert _anonymize_mqtt_broker("mqtt.example.com") == "*.example.com"
|
|
|
|
def test_hostname_with_subdomain(self):
|
|
from backend.app.api.routes.support import _anonymize_mqtt_broker
|
|
|
|
assert _anonymize_mqtt_broker("broker.mqtt.example.com") == "*.example.com"
|
|
|
|
def test_single_part_hostname(self):
|
|
from backend.app.api.routes.support import _anonymize_mqtt_broker
|
|
|
|
assert _anonymize_mqtt_broker("localhost") == "localhost"
|
|
|
|
|
|
class TestCheckPort:
|
|
"""Tests for _check_port()."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.unit
|
|
async def test_reachable_port(self):
|
|
from backend.app.api.routes.support import _check_port
|
|
|
|
# Mock a successful connection
|
|
mock_writer = AsyncMock()
|
|
mock_writer.close = MagicMock()
|
|
mock_writer.wait_closed = AsyncMock()
|
|
|
|
with patch("backend.app.api.routes.support.asyncio.open_connection", return_value=(AsyncMock(), mock_writer)):
|
|
result = await _check_port("192.168.1.1", 8883, timeout=1.0)
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.unit
|
|
async def test_unreachable_port(self):
|
|
from backend.app.api.routes.support import _check_port
|
|
|
|
with (
|
|
patch(
|
|
"backend.app.api.routes.support.asyncio.open_connection",
|
|
side_effect=ConnectionRefusedError,
|
|
),
|
|
patch(
|
|
"backend.app.api.routes.support.asyncio.wait_for",
|
|
side_effect=ConnectionRefusedError,
|
|
),
|
|
):
|
|
result = await _check_port("192.168.1.1", 8883, timeout=1.0)
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.unit
|
|
async def test_timeout(self):
|
|
from backend.app.api.routes.support import _check_port
|
|
|
|
with patch(
|
|
"backend.app.api.routes.support.asyncio.wait_for",
|
|
side_effect=asyncio.TimeoutError,
|
|
):
|
|
result = await _check_port("192.168.1.1", 8883, timeout=0.1)
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestGetContainerMemoryLimit:
|
|
"""Tests for _get_container_memory_limit()."""
|
|
|
|
def test_cgroup_v2_with_limit(self):
|
|
from backend.app.api.routes.support import _get_container_memory_limit
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
v2_path = Path(tmpdir) / "memory.max"
|
|
v2_path.write_text("1073741824\n")
|
|
|
|
with patch("backend.app.api.routes.support.Path") as mock_path:
|
|
# v2 path exists with value
|
|
v2_mock = MagicMock()
|
|
v2_mock.exists.return_value = True
|
|
v2_mock.read_text.return_value = "1073741824\n"
|
|
|
|
v1_mock = MagicMock()
|
|
v1_mock.exists.return_value = False
|
|
|
|
mock_path.side_effect = lambda p: v2_mock if "memory.max" in p else v1_mock
|
|
|
|
result = _get_container_memory_limit()
|
|
|
|
assert result == 1073741824
|
|
|
|
def test_cgroup_v2_unlimited(self):
|
|
from backend.app.api.routes.support import _get_container_memory_limit
|
|
|
|
with patch("backend.app.api.routes.support.Path") as mock_path:
|
|
v2_mock = MagicMock()
|
|
v2_mock.exists.return_value = True
|
|
v2_mock.read_text.return_value = "max\n"
|
|
|
|
v1_mock = MagicMock()
|
|
v1_mock.exists.return_value = False
|
|
|
|
mock_path.side_effect = lambda p: v2_mock if "memory.max" in p else v1_mock
|
|
|
|
result = _get_container_memory_limit()
|
|
|
|
assert result is None
|
|
|
|
def test_no_cgroup_files(self):
|
|
from backend.app.api.routes.support import _get_container_memory_limit
|
|
|
|
with patch("backend.app.api.routes.support.Path") as mock_path:
|
|
mock_instance = MagicMock()
|
|
mock_instance.exists.return_value = False
|
|
mock_path.return_value = mock_instance
|
|
|
|
result = _get_container_memory_limit()
|
|
|
|
assert result is None
|
|
|
|
|
|
class TestFormatBytes:
|
|
"""Tests for _format_bytes()."""
|
|
|
|
def test_bytes(self):
|
|
from backend.app.api.routes.support import _format_bytes
|
|
|
|
assert _format_bytes(500) == "500 B"
|
|
|
|
def test_kilobytes(self):
|
|
from backend.app.api.routes.support import _format_bytes
|
|
|
|
assert _format_bytes(2048) == "2.0 KB"
|
|
|
|
def test_megabytes(self):
|
|
from backend.app.api.routes.support import _format_bytes
|
|
|
|
assert _format_bytes(10 * 1024 * 1024) == "10.0 MB"
|
|
|
|
def test_gigabytes(self):
|
|
from backend.app.api.routes.support import _format_bytes
|
|
|
|
assert _format_bytes(2 * 1024 * 1024 * 1024) == "2.00 GB"
|
|
|
|
def test_zero(self):
|
|
from backend.app.api.routes.support import _format_bytes
|
|
|
|
assert _format_bytes(0) == "0 B"
|
|
|
|
|
|
class TestSanitizeLogContent:
|
|
"""Tests for _sanitize_log_content() redaction."""
|
|
|
|
def test_ipv4_addresses_redacted(self):
|
|
"""IPv4 addresses in log lines are replaced with [IP]."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "2024-01-15 Connected to printer at 192.168.1.100 on port 8883"
|
|
result = _sanitize_log_content(content)
|
|
assert "192.168.1.100" not in result
|
|
assert "[IP]" in result
|
|
assert "on port 8883" in result
|
|
|
|
def test_multiple_ipv4_addresses_redacted(self):
|
|
"""Multiple different IPs in the same line are all redacted."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "Proxy 10.0.0.1 -> 192.168.1.50"
|
|
result = _sanitize_log_content(content)
|
|
assert result == "Proxy [IP] -> [IP]"
|
|
|
|
def test_firmware_versions_with_leading_zeros_preserved(self):
|
|
"""Firmware versions like 01.09.01.00 have leading zeros and should NOT be redacted."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "Firmware version: 01.09.01.00"
|
|
result = _sanitize_log_content(content)
|
|
assert "01.09.01.00" in result
|
|
|
|
def test_firmware_version_mixed_with_ip(self):
|
|
"""Firmware versions preserved while real IPs are redacted in the same line."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "Printer at 192.168.1.5 running firmware 01.07.02.00"
|
|
result = _sanitize_log_content(content)
|
|
assert "192.168.1.5" not in result
|
|
assert "01.07.02.00" in result
|
|
assert "[IP] running firmware 01.07.02.00" in result
|
|
|
|
def test_printer_ip_from_sensitive_strings(self):
|
|
"""Printer IPs in sensitive_strings are replaced before regex pass."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "Connecting to 192.168.1.100"
|
|
result = _sanitize_log_content(content, sensitive_strings={"192.168.1.100": "[IP]"})
|
|
assert result == "Connecting to [IP]"
|
|
|
|
def test_edge_case_zero_ip(self):
|
|
"""0.0.0.0 is a valid IP and should be redacted."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "Binding to 0.0.0.0"
|
|
result = _sanitize_log_content(content)
|
|
assert result == "Binding to [IP]"
|
|
|
|
def test_edge_case_broadcast_ip(self):
|
|
"""255.255.255.255 is a valid IP and should be redacted."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "Broadcast to 255.255.255.255"
|
|
result = _sanitize_log_content(content)
|
|
assert result == "Broadcast to [IP]"
|
|
|
|
def test_invalid_octet_not_redacted(self):
|
|
"""Octets >255 are not valid IPs and should not be redacted."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "Value 999.999.999.999"
|
|
result = _sanitize_log_content(content)
|
|
assert "999.999.999.999" in result
|
|
|
|
def test_existing_serial_redaction_still_works(self):
|
|
"""Serial number redaction still functions alongside IP redaction."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "Printer 01SABCDEF1234 at 10.0.0.5"
|
|
result = _sanitize_log_content(content)
|
|
assert "[SERIAL]" in result
|
|
assert "[IP]" in result
|
|
assert "01SABCDEF1234" not in result
|
|
assert "10.0.0.5" not in result
|
|
|
|
def test_existing_email_redaction_still_works(self):
|
|
"""Email redaction still functions alongside IP redaction."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "User user@example.com from 172.16.0.1"
|
|
result = _sanitize_log_content(content)
|
|
assert "[EMAIL]" in result
|
|
assert "[IP]" in result
|
|
|
|
def test_existing_path_redaction_still_works(self):
|
|
"""Path redaction still functions alongside IP redaction."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "Config at /home/john/config.yaml from 192.168.0.1"
|
|
result = _sanitize_log_content(content)
|
|
assert "/home/[user]/" in result
|
|
assert "[IP]" in result
|
|
|
|
def test_ldap_dn_redacted_reporter_line(self):
|
|
"""#2681: the exact reporter line — the CN (real name) must not survive."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = (
|
|
"LDAP authentication successful for user: jschmoe "
|
|
"(DN: CN=Joe Schmoe,CN=Users,DC=ad,DC=example,DC=com, groups: 4)"
|
|
)
|
|
result = _sanitize_log_content(content)
|
|
assert "Joe Schmoe" not in result
|
|
assert "DC=example" not in result
|
|
assert result == "LDAP authentication successful for user: jschmoe (DN: [DN], groups: 4)"
|
|
|
|
def test_ldap_dn_redacted_in_exception_string(self):
|
|
"""DNs that leak indirectly via ldap3 exception text are caught too."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "LDAP bind failed for user jschmoe: invalidCredentials at uid=jschmoe,ou=people,dc=example,dc=org"
|
|
result = _sanitize_log_content(content)
|
|
assert "uid=jschmoe" not in result
|
|
assert "[DN]" in result
|
|
|
|
def test_ldap_group_dn_redacted(self):
|
|
"""Group DNs (from group-mapping logs) are PII-bearing and redacted."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "Mapped CN=Admins,OU=Groups,DC=corp,DC=local -> Administrators"
|
|
result = _sanitize_log_content(content)
|
|
assert "CN=Admins" not in result
|
|
assert "DC=corp" not in result
|
|
assert "[DN]" in result
|
|
assert "Administrators" in result # the non-PII target group name survives
|
|
|
|
def test_non_dn_key_value_line_not_clobbered(self):
|
|
"""An ordinary key=value log line must not be mistaken for a DN."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "Dispatch decision: mode=queue, state=FINISH, printer=1"
|
|
result = _sanitize_log_content(content)
|
|
assert result == content
|
|
|
|
def test_single_rdn_not_redacted(self):
|
|
"""A lone attr=value (not a multi-component DN) is left alone."""
|
|
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
|
|
|
|
content = "Country C=US selected"
|
|
result = _sanitize_log_content(content)
|
|
assert result == content
|
|
|
|
|
|
class TestCollectSupportInfo:
|
|
"""Tests for _collect_support_info() new diagnostic sections."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.unit
|
|
async def test_environment_has_timezone(self):
|
|
"""Verify environment section includes timezone."""
|
|
from backend.app.api.routes.support import _collect_support_info
|
|
|
|
with (
|
|
patch("backend.app.api.routes.support.is_running_in_docker", return_value=False),
|
|
patch("backend.app.api.routes.support.async_session") as mock_session_ctx,
|
|
patch("backend.app.api.routes.support.printer_manager") as mock_pm,
|
|
patch("backend.app.api.routes.support.get_network_interfaces", return_value=[]),
|
|
patch("backend.app.api.routes.support.ws_manager") as mock_ws,
|
|
patch.dict("os.environ", {"TZ": "America/New_York"}),
|
|
):
|
|
mock_pm.get_all_statuses.return_value = {}
|
|
mock_ws.active_connections = []
|
|
|
|
mock_db = AsyncMock()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar.return_value = 0
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
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(return_value=False)
|
|
|
|
info = await _collect_support_info()
|
|
|
|
assert info["environment"]["timezone"] == "America/New_York"
|
|
assert info["environment"]["docker"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.unit
|
|
async def test_docker_section_present_when_in_docker(self):
|
|
"""Verify docker section is added when running in Docker."""
|
|
from backend.app.api.routes.support import _collect_support_info
|
|
|
|
with (
|
|
patch("backend.app.api.routes.support.is_running_in_docker", return_value=True),
|
|
patch("backend.app.api.routes.support._get_container_memory_limit", return_value=1073741824),
|
|
patch("backend.app.api.routes.support._detect_docker_network_mode", return_value="bridge"),
|
|
patch("backend.app.api.routes.support.async_session") as mock_session_ctx,
|
|
patch("backend.app.api.routes.support.printer_manager") as mock_pm,
|
|
patch(
|
|
"backend.app.api.routes.support.get_network_interfaces",
|
|
return_value=[{"name": "eth0", "subnet": "172.17.0.0/16"}],
|
|
),
|
|
patch("backend.app.api.routes.support.ws_manager") as mock_ws,
|
|
):
|
|
mock_pm.get_all_statuses.return_value = {}
|
|
mock_ws.active_connections = []
|
|
|
|
mock_db = AsyncMock()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar.return_value = 0
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
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(return_value=False)
|
|
|
|
info = await _collect_support_info()
|
|
|
|
assert "docker" in info
|
|
assert info["docker"]["container_memory_limit_bytes"] == 1073741824
|
|
assert info["docker"]["container_memory_limit_formatted"] == "1.00 GB"
|
|
assert info["docker"]["network_mode_hint"] == "bridge"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.unit
|
|
async def test_docker_section_absent_when_not_docker(self):
|
|
"""Verify docker section is absent when not in Docker."""
|
|
from backend.app.api.routes.support import _collect_support_info
|
|
|
|
with (
|
|
patch("backend.app.api.routes.support.is_running_in_docker", return_value=False),
|
|
patch("backend.app.api.routes.support.async_session") as mock_session_ctx,
|
|
patch("backend.app.api.routes.support.printer_manager") as mock_pm,
|
|
patch("backend.app.api.routes.support.get_network_interfaces", return_value=[]),
|
|
patch("backend.app.api.routes.support.ws_manager") as mock_ws,
|
|
):
|
|
mock_pm.get_all_statuses.return_value = {}
|
|
mock_ws.active_connections = []
|
|
|
|
mock_db = AsyncMock()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar.return_value = 0
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
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(return_value=False)
|
|
|
|
info = await _collect_support_info()
|
|
|
|
assert "docker" not in info
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.unit
|
|
async def test_dependencies_section(self):
|
|
"""Verify dependencies section lists package versions."""
|
|
from backend.app.api.routes.support import _collect_support_info
|
|
|
|
with (
|
|
patch("backend.app.api.routes.support.is_running_in_docker", return_value=False),
|
|
patch("backend.app.api.routes.support.async_session") as mock_session_ctx,
|
|
patch("backend.app.api.routes.support.printer_manager") as mock_pm,
|
|
patch("backend.app.api.routes.support.get_network_interfaces", return_value=[]),
|
|
patch("backend.app.api.routes.support.ws_manager") as mock_ws,
|
|
):
|
|
mock_pm.get_all_statuses.return_value = {}
|
|
mock_ws.active_connections = []
|
|
|
|
mock_db = AsyncMock()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar.return_value = 0
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
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(return_value=False)
|
|
|
|
info = await _collect_support_info()
|
|
|
|
assert "dependencies" in info
|
|
# fastapi should be installed in test environment
|
|
assert "fastapi" in info["dependencies"]
|
|
assert info["dependencies"]["fastapi"] is not None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.unit
|
|
async def test_websockets_section(self):
|
|
"""Verify websockets section shows connection count."""
|
|
from backend.app.api.routes.support import _collect_support_info
|
|
|
|
with (
|
|
patch("backend.app.api.routes.support.is_running_in_docker", return_value=False),
|
|
patch("backend.app.api.routes.support.async_session") as mock_session_ctx,
|
|
patch("backend.app.api.routes.support.printer_manager") as mock_pm,
|
|
patch("backend.app.api.routes.support.get_network_interfaces", return_value=[]),
|
|
patch("backend.app.api.routes.support.ws_manager") as mock_ws,
|
|
):
|
|
mock_pm.get_all_statuses.return_value = {}
|
|
mock_ws.active_connections = ["conn1", "conn2"]
|
|
|
|
mock_db = AsyncMock()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar.return_value = 0
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
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(return_value=False)
|
|
|
|
info = await _collect_support_info()
|
|
|
|
assert info["websockets"]["active_connections"] == 2
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.unit
|
|
async def test_network_section(self):
|
|
"""Verify network section shows interface subnets."""
|
|
from backend.app.api.routes.support import _collect_support_info
|
|
|
|
mock_interfaces = [
|
|
{"name": "eth0", "ip": "192.168.1.100", "netmask": "255.255.255.0", "subnet": "192.168.1.0/24"},
|
|
{"name": "wlan0", "ip": "10.0.0.50", "netmask": "255.255.255.0", "subnet": "10.0.0.0/24"},
|
|
]
|
|
|
|
with (
|
|
patch("backend.app.api.routes.support.is_running_in_docker", return_value=False),
|
|
patch("backend.app.api.routes.support.async_session") as mock_session_ctx,
|
|
patch("backend.app.api.routes.support.printer_manager") as mock_pm,
|
|
patch("backend.app.api.routes.support.get_network_interfaces", return_value=mock_interfaces),
|
|
patch("backend.app.api.routes.support.ws_manager") as mock_ws,
|
|
):
|
|
mock_pm.get_all_statuses.return_value = {}
|
|
mock_ws.active_connections = []
|
|
|
|
mock_db = AsyncMock()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar.return_value = 0
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
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(return_value=False)
|
|
|
|
info = await _collect_support_info()
|
|
|
|
assert info["network"]["interface_count"] == 2
|
|
assert info["network"]["interfaces"][0]["name"] == "eth0"
|
|
assert info["network"]["interfaces"][0]["subnet"] == "x.x.1.0/24"
|
|
# Verify IP addresses are NOT included (first two octets masked)
|
|
for iface in info["network"]["interfaces"]:
|
|
assert "ip" not in iface
|
|
assert iface["subnet"].startswith("x.x.")
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.unit
|
|
async def test_log_file_section(self):
|
|
"""Verify log file section shows size info."""
|
|
from backend.app.api.routes.support import _collect_support_info
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
log_dir = Path(tmpdir)
|
|
log_file = log_dir / "bambuddy.log"
|
|
log_file.write_text("some log content\n" * 100)
|
|
|
|
with (
|
|
patch("backend.app.api.routes.support.is_running_in_docker", return_value=False),
|
|
patch("backend.app.api.routes.support.async_session") as mock_session_ctx,
|
|
patch("backend.app.api.routes.support.printer_manager") as mock_pm,
|
|
patch("backend.app.api.routes.support.get_network_interfaces", return_value=[]),
|
|
patch("backend.app.api.routes.support.ws_manager") as mock_ws,
|
|
patch("backend.app.api.routes.support.settings") as mock_settings,
|
|
):
|
|
mock_settings.base_dir = Path(tmpdir)
|
|
mock_settings.log_dir = log_dir
|
|
mock_settings.debug = False
|
|
mock_pm.get_all_statuses.return_value = {}
|
|
mock_ws.active_connections = []
|
|
|
|
mock_db = AsyncMock()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar.return_value = 0
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
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(return_value=False)
|
|
|
|
info = await _collect_support_info()
|
|
|
|
assert "log_file" in info
|
|
assert info["log_file"]["size_bytes"] > 0
|
|
assert "B" in info["log_file"]["size_formatted"] or "KB" in info["log_file"]["size_formatted"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.unit
|
|
async def test_settings_include_all_keys_with_sensitive_redacted(self):
|
|
"""All settings keys must appear in output; sensitive values are replaced with [REDACTED]."""
|
|
from backend.app.api.routes.support import _collect_support_info
|
|
|
|
fake_settings = [
|
|
MagicMock(key="benign_flag", value="true"),
|
|
MagicMock(key="bambu_cloud_token", value="super-secret"),
|
|
MagicMock(key="github_webhook", value="https://hooks.example/abc"),
|
|
MagicMock(key="empty_password", value=""),
|
|
MagicMock(key="local_backup_path", value="/data/backups"),
|
|
# Regression: setting was leaking before the `broker` keyword was added.
|
|
MagicMock(key="mqtt_broker", value="192.168.255.16"),
|
|
# Regression: setting was leaking before the `auth_key` keyword was
|
|
# added — and a value-prefix safety net (`tskey-`) was introduced
|
|
# so future Tailscale settings auto-redact even if we forget the key.
|
|
MagicMock(key="virtual_printer_tailscale_auth_key", value="tskey-auth-secrettokenhere"),
|
|
# Value-prefix safety net standalone: a hypothetical future setting
|
|
# named without "auth_key" but whose value starts with the Tailscale
|
|
# prefix must still redact.
|
|
MagicMock(key="some_future_ts_setting", value="tskey-other-secret"),
|
|
]
|
|
|
|
def make_result(rows=None):
|
|
r = MagicMock()
|
|
r.scalar.return_value = 0
|
|
r.scalar_one_or_none.return_value = None
|
|
r.scalars.return_value.all.return_value = rows or []
|
|
r.all.return_value = []
|
|
return r
|
|
|
|
async def fake_execute(stmt, *_a, **_kw):
|
|
sql = str(stmt).lower()
|
|
# Route by table name in the compiled SQL
|
|
if "from settings" in sql or "settings.key" in sql:
|
|
return make_result(fake_settings)
|
|
return make_result([])
|
|
|
|
with (
|
|
tempfile.TemporaryDirectory() as tmpdir,
|
|
patch("backend.app.api.routes.support.is_running_in_docker", return_value=False),
|
|
patch("backend.app.api.routes.support.async_session") as mock_session_ctx,
|
|
patch("backend.app.api.routes.support.printer_manager") as mock_pm,
|
|
patch("backend.app.api.routes.support.get_network_interfaces", return_value=[]),
|
|
patch("backend.app.api.routes.support.ws_manager") as mock_ws,
|
|
patch("backend.app.api.routes.support.settings") as mock_settings,
|
|
):
|
|
mock_settings.base_dir = Path(tmpdir)
|
|
mock_settings.log_dir = Path(tmpdir)
|
|
mock_settings.debug = False
|
|
mock_pm.get_all_statuses.return_value = {}
|
|
mock_ws.active_connections = []
|
|
|
|
mock_db = AsyncMock()
|
|
mock_db.execute = fake_execute
|
|
mock_session_ctx.return_value.__aenter__ = AsyncMock(return_value=mock_db)
|
|
mock_session_ctx.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
info = await _collect_support_info()
|
|
|
|
s = info["settings"]
|
|
assert s.get("bambu_cloud_token") == "[REDACTED]"
|
|
assert s.get("github_webhook") == "[REDACTED]"
|
|
assert s.get("local_backup_path") == "[REDACTED]"
|
|
assert s.get("empty_password") == ""
|
|
assert s.get("benign_flag") == "true"
|
|
assert s.get("mqtt_broker") == "[REDACTED]"
|
|
assert s.get("virtual_printer_tailscale_auth_key") == "[REDACTED]"
|
|
assert s.get("some_future_ts_setting") == "[REDACTED]"
|
|
|
|
|
|
class TestParseObicoEnabledPrinters:
|
|
"""Tests for the per-printer obico flag parser used by the bundle.
|
|
|
|
The setting is written by the settings UI as a JSON array and read by
|
|
ObicoDetectionService._load_settings as one; the bundle used to split it on
|
|
commas and call empty "no printers", so a default Obico setup was reported
|
|
as monitoring nothing while it was in fact monitoring everything (#2733).
|
|
"""
|
|
|
|
def test_empty_means_all_printers(self):
|
|
from backend.app.api.routes.support import _parse_obico_enabled_printers
|
|
|
|
# None (not "no printers") — the same convention _load_settings uses.
|
|
assert _parse_obico_enabled_printers("") is None
|
|
assert _parse_obico_enabled_printers(" ") is None
|
|
assert _parse_obico_enabled_printers(None) is None
|
|
|
|
def test_json_array_is_the_stored_shape(self):
|
|
from backend.app.api.routes.support import _parse_obico_enabled_printers
|
|
|
|
assert _parse_obico_enabled_printers("[1, 2, 3]") == {1, 2, 3}
|
|
assert _parse_obico_enabled_printers("[]") == set()
|
|
|
|
def test_comma_separated_ids_still_parse(self):
|
|
# Legacy fallback for any install that stored the old shape.
|
|
from backend.app.api.routes.support import _parse_obico_enabled_printers
|
|
|
|
assert _parse_obico_enabled_printers("1,2,3") == {1, 2, 3}
|
|
assert _parse_obico_enabled_printers("1, 2 ,3") == {1, 2, 3}
|
|
|
|
def test_non_integer_tokens_are_skipped(self):
|
|
# Defensive against legacy/manually-edited setting values.
|
|
from backend.app.api.routes.support import _parse_obico_enabled_printers
|
|
|
|
assert _parse_obico_enabled_printers("1,abc,2") == {1, 2}
|
|
assert _parse_obico_enabled_printers(",,1,") == {1}
|
|
assert _parse_obico_enabled_printers('[1, "two", 3]') == {1, 3}
|
|
|
|
def test_json_object_is_not_a_printer_list(self):
|
|
from backend.app.api.routes.support import _parse_obico_enabled_printers
|
|
|
|
# Falls through to the comma parser, which finds no integers.
|
|
assert _parse_obico_enabled_printers('{"1": true}') == set()
|
|
|
|
|
|
class TestCheckUrlReachable:
|
|
"""Tests for the slicer-API reachability ping."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_url_returns_none(self):
|
|
from backend.app.api.routes.support import _check_url_reachable
|
|
|
|
assert await _check_url_reachable("") is None
|
|
assert await _check_url_reachable(" ") is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_successful_response_is_reachable_even_on_404(self):
|
|
# A 404 means the API is up; we want to separate network failure from
|
|
# configuration mistakes, so non-empty status counts as reachable.
|
|
from backend.app.api.routes.support import _check_url_reachable
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client_cls.return_value.__aenter__.return_value = mock_client
|
|
mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 404
|
|
mock_client.get = AsyncMock(return_value=mock_response)
|
|
|
|
result = await _check_url_reachable("http://localhost:3001/api")
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connection_error_returns_false(self):
|
|
from backend.app.api.routes.support import _check_url_reachable
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_cls:
|
|
mock_client_cls.return_value.__aenter__.side_effect = ConnectionError("boom")
|
|
|
|
result = await _check_url_reachable("http://nowhere:9999")
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestFetchSlicerHealth:
|
|
"""Tests for the slicer-API health probe that extracts the bundled CLI
|
|
version. Knowing the version in the support bundle lets the reviewer
|
|
confirm the user is running the image they think they are — exactly the
|
|
diagnostic that was missing when issue #1312 surfaced."""
|
|
|
|
def _mock_httpx(self, status_code: int, body):
|
|
"""Construct a patched httpx.AsyncClient that returns a fixed response."""
|
|
mock_client_cls = MagicMock()
|
|
mock_client = AsyncMock()
|
|
mock_client_cls.return_value.__aenter__.return_value = mock_client
|
|
mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = status_code
|
|
if isinstance(body, Exception):
|
|
mock_response.json.side_effect = body
|
|
else:
|
|
mock_response.json.return_value = body
|
|
mock_client.get = AsyncMock(return_value=mock_response)
|
|
return mock_client_cls, mock_client
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_url_returns_none(self):
|
|
from backend.app.api.routes.support import _fetch_slicer_health
|
|
|
|
assert await _fetch_slicer_health("") is None
|
|
assert await _fetch_slicer_health(" ") is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parses_version_from_orcaslicer_field(self):
|
|
"""The default sidecar wrapper labels both orca and bambu CLIs under
|
|
``checks.orcaslicer``. The probe must read whichever non-dataPath child
|
|
carries a ``version`` field instead of hardcoding the field name."""
|
|
from backend.app.api.routes.support import _fetch_slicer_health
|
|
|
|
body = {
|
|
"status": "healthy",
|
|
"checks": {
|
|
"orcaslicer": {"available": True, "version": "2.3.2"},
|
|
"dataPath": {"accessible": True},
|
|
},
|
|
}
|
|
mock_client_cls, mock_client = self._mock_httpx(200, body)
|
|
with patch("httpx.AsyncClient", mock_client_cls):
|
|
result = await _fetch_slicer_health("http://orca:3003")
|
|
|
|
assert result == {"reachable": True, "version": "2.3.2"}
|
|
# And the URL was actually composed as /health.
|
|
mock_client.get.assert_awaited_once()
|
|
assert mock_client.get.await_args[0][0] == "http://orca:3003/health"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parses_version_when_wrapper_uses_bambustudio_field(self):
|
|
"""Future-proofing: if the wrapper is ever fixed to label the bambu CLI
|
|
as ``bambustudio``, the probe must still pick up the version. The probe
|
|
walks every non-dataPath key looking for a ``version`` field rather
|
|
than hardcoding the slicer name."""
|
|
from backend.app.api.routes.support import _fetch_slicer_health
|
|
|
|
body = {
|
|
"status": "healthy",
|
|
"checks": {
|
|
"bambustudio": {"available": True, "version": "02.06.00.51"},
|
|
"dataPath": {"accessible": True},
|
|
},
|
|
}
|
|
mock_client_cls, _ = self._mock_httpx(200, body)
|
|
with patch("httpx.AsyncClient", mock_client_cls):
|
|
result = await _fetch_slicer_health("http://bs:3001")
|
|
|
|
assert result == {"reachable": True, "version": "02.06.00.51"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_version_unknown_propagates_as_string(self):
|
|
"""The wrapper emits literal ``"unknown"`` when it can't parse the
|
|
slicer's --help output. We surface that as-is — it's diagnostic on
|
|
its own (tells the reviewer the regex didn't match)."""
|
|
from backend.app.api.routes.support import _fetch_slicer_health
|
|
|
|
body = {
|
|
"status": "healthy",
|
|
"checks": {
|
|
"orcaslicer": {"available": True, "version": "unknown"},
|
|
"dataPath": {"accessible": True},
|
|
},
|
|
}
|
|
mock_client_cls, _ = self._mock_httpx(200, body)
|
|
with patch("httpx.AsyncClient", mock_client_cls):
|
|
result = await _fetch_slicer_health("http://bs:3001")
|
|
|
|
assert result == {"reachable": True, "version": "unknown"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_200_status_is_reachable_but_no_version(self):
|
|
"""If the URL responds with a non-200, the host is up but the endpoint
|
|
isn't the expected one — surface reachable=True so the reviewer can
|
|
spot misconfiguration without conflating it with a network failure."""
|
|
from backend.app.api.routes.support import _fetch_slicer_health
|
|
|
|
mock_client_cls, _ = self._mock_httpx(404, {})
|
|
with patch("httpx.AsyncClient", mock_client_cls):
|
|
result = await _fetch_slicer_health("http://bs:3001")
|
|
|
|
assert result == {"reachable": True, "version": None}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_malformed_json_returns_reachable_no_version(self):
|
|
from backend.app.api.routes.support import _fetch_slicer_health
|
|
|
|
mock_client_cls, _ = self._mock_httpx(200, ValueError("not json"))
|
|
with patch("httpx.AsyncClient", mock_client_cls):
|
|
result = await _fetch_slicer_health("http://bs:3001")
|
|
|
|
assert result == {"reachable": True, "version": None}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_checks_block_returns_no_version(self):
|
|
from backend.app.api.routes.support import _fetch_slicer_health
|
|
|
|
mock_client_cls, _ = self._mock_httpx(200, {"status": "healthy"})
|
|
with patch("httpx.AsyncClient", mock_client_cls):
|
|
result = await _fetch_slicer_health("http://bs:3001")
|
|
|
|
assert result == {"reachable": True, "version": None}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connection_error_returns_unreachable(self):
|
|
from backend.app.api.routes.support import _fetch_slicer_health
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_cls:
|
|
mock_client_cls.return_value.__aenter__.side_effect = ConnectionError("boom")
|
|
|
|
result = await _fetch_slicer_health("http://nowhere:9999")
|
|
|
|
assert result == {"reachable": False, "version": None}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_strips_trailing_slash_before_appending_health(self):
|
|
"""Defensive: URLs entered with trailing slashes in Settings should
|
|
still produce a well-formed /health URL (no double-slash)."""
|
|
from backend.app.api.routes.support import _fetch_slicer_health
|
|
|
|
body = {"status": "healthy", "checks": {"orcaslicer": {"available": True, "version": "2.3.2"}}}
|
|
mock_client_cls, mock_client = self._mock_httpx(200, body)
|
|
with patch("httpx.AsyncClient", mock_client_cls):
|
|
await _fetch_slicer_health("http://bs:3001/")
|
|
|
|
assert mock_client.get.await_args[0][0] == "http://bs:3001/health"
|
|
|
|
|
|
class TestCollectSlicerApiInfo:
|
|
"""Tests for the slicer-API info block (configured URLs + reachability).
|
|
|
|
The collector reads URLs DIRECTLY from the DB rather than from the
|
|
already-redacted ``info["settings"]`` dict — the previous version was
|
|
pinging the literal string "[REDACTED]" (which httpx rejects) and getting
|
|
``False`` for any installation that actually had a slicer-API configured.
|
|
These tests inject the raw URLs via a mocked `async_session` so the
|
|
collector sees them as if they came from the unredacted Settings table.
|
|
"""
|
|
|
|
def _make_settings_session(self, settings_dict):
|
|
rows = [MagicMock(key=k, value=v) for k, v in settings_dict.items()]
|
|
result = MagicMock()
|
|
result.scalars.return_value.all.return_value = rows
|
|
mock_db = AsyncMock()
|
|
mock_db.execute = AsyncMock(return_value=result)
|
|
ctx = MagicMock()
|
|
ctx.return_value.__aenter__ = AsyncMock(return_value=mock_db)
|
|
ctx.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
return ctx
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_does_not_run_reachability_check(self):
|
|
from backend.app.api.routes.support import _collect_slicer_api_info
|
|
|
|
session_ctx = self._make_settings_session({"use_slicer_api": "false", "preferred_slicer": "bambu_studio"})
|
|
with (
|
|
patch("backend.app.api.routes.support.async_session", session_ctx),
|
|
patch("backend.app.api.routes.support._fetch_slicer_health") as mock_health,
|
|
):
|
|
info = await _collect_slicer_api_info()
|
|
|
|
mock_health.assert_not_called()
|
|
assert info["enabled"] is False
|
|
assert info["preferred"] == "bambu_studio"
|
|
assert info["bambu_studio_url_set_in_db"] is False
|
|
assert info["orcaslicer_url_set_in_db"] is False
|
|
assert "bambu_studio_reachable" not in info
|
|
assert "orcaslicer_reachable" not in info
|
|
assert "bambu_studio_version" not in info
|
|
assert "orcaslicer_version" not in info
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enabled_runs_reachability_check_for_both_urls(self):
|
|
from backend.app.api.routes.support import _collect_slicer_api_info
|
|
|
|
async def fake_health(url, timeout=2.0):
|
|
if "orca" in url:
|
|
return {"reachable": True, "version": "2.3.2"}
|
|
return {"reachable": False, "version": None}
|
|
|
|
session_ctx = self._make_settings_session(
|
|
{
|
|
"use_slicer_api": "true",
|
|
"preferred_slicer": "orcaslicer",
|
|
"bambu_studio_api_url": "http://bs:3001",
|
|
"orcaslicer_api_url": "http://orca:3003",
|
|
}
|
|
)
|
|
with (
|
|
patch("backend.app.api.routes.support.async_session", session_ctx),
|
|
patch("backend.app.api.routes.support._fetch_slicer_health", side_effect=fake_health),
|
|
):
|
|
info = await _collect_slicer_api_info()
|
|
|
|
assert info["enabled"] is True
|
|
assert info["bambu_studio_url_set_in_db"] is True
|
|
assert info["orcaslicer_url_set_in_db"] is True
|
|
assert info["bambu_studio_url_source"] == "db"
|
|
assert info["orcaslicer_url_source"] == "db"
|
|
assert info["bambu_studio_reachable"] is False
|
|
assert info["orcaslicer_reachable"] is True
|
|
assert info["bambu_studio_version"] is None
|
|
assert info["orcaslicer_version"] == "2.3.2"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_env_var_fallback_url_pinged_when_db_setting_empty(self):
|
|
"""Regression for the second pass on #support-bundle audit: the
|
|
previous version returned `null` for `bambu_studio_reachable` on every
|
|
installation that ran the sidecar via env var rather than via the DB
|
|
setting (the common case for the default `http://localhost:3001`).
|
|
The resolver now mirrors the precedence used by `archives.py:3174-3180`
|
|
— DB setting first, then `app_settings.bambu_studio_api_url` (which
|
|
reads the `BAMBU_STUDIO_API_URL` env var or the built-in default).
|
|
"""
|
|
from backend.app.api.routes.support import _collect_slicer_api_info
|
|
|
|
seen_urls: list[str] = []
|
|
|
|
async def fake_health(url, timeout=2.0):
|
|
seen_urls.append(url)
|
|
return {"reachable": True, "version": "02.06.00.51"}
|
|
|
|
# DB has use_slicer_api=true but NO bambu_studio_api_url row, simulating
|
|
# a user who set the URL via the BAMBU_STUDIO_API_URL env var.
|
|
session_ctx = self._make_settings_session({"use_slicer_api": "true", "preferred_slicer": "bambu_studio"})
|
|
with (
|
|
patch("backend.app.api.routes.support.async_session", session_ctx),
|
|
patch("backend.app.api.routes.support._fetch_slicer_health", side_effect=fake_health),
|
|
patch("backend.app.api.routes.support.settings") as mock_app_settings,
|
|
):
|
|
# Pydantic-settings would normally do this for us when reading the
|
|
# env var — we mock the resolved value directly.
|
|
mock_app_settings.bambu_studio_api_url = "http://my-sidecar:3001"
|
|
mock_app_settings.slicer_api_url = "http://localhost:3003"
|
|
|
|
info = await _collect_slicer_api_info()
|
|
|
|
# The env-var URL was the one actually pinged.
|
|
assert "http://my-sidecar:3001" in seen_urls
|
|
# And the source-tracking field shows we fell back from the DB to env.
|
|
assert info["bambu_studio_url_set_in_db"] is False
|
|
assert info["bambu_studio_url_source"] == "env_or_default"
|
|
assert info["bambu_studio_reachable"] is True
|
|
assert info["bambu_studio_version"] == "02.06.00.51"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reachability_uses_unredacted_url(self):
|
|
"""Regression: the collector previously pinged the literal '[REDACTED]'
|
|
from the already-sanitized info["settings"] dict and always returned
|
|
False. The collector must read the un-redacted URL fresh from the DB.
|
|
"""
|
|
from backend.app.api.routes.support import _collect_slicer_api_info
|
|
|
|
seen_urls: list[str] = []
|
|
|
|
async def fake_health(url, timeout=2.0):
|
|
seen_urls.append(url)
|
|
return {"reachable": True, "version": "unknown"}
|
|
|
|
session_ctx = self._make_settings_session(
|
|
{
|
|
"use_slicer_api": "true",
|
|
"bambu_studio_api_url": "http://real-bs-host:3001",
|
|
"orcaslicer_api_url": "http://real-orca-host:3003",
|
|
}
|
|
)
|
|
with (
|
|
patch("backend.app.api.routes.support.async_session", session_ctx),
|
|
patch("backend.app.api.routes.support._fetch_slicer_health", side_effect=fake_health),
|
|
):
|
|
await _collect_slicer_api_info()
|
|
|
|
assert "http://real-bs-host:3001" in seen_urls
|
|
assert "http://real-orca-host:3003" in seen_urls
|
|
assert "[REDACTED]" not in seen_urls
|
|
|
|
|
|
class TestCollectAuthInfo:
|
|
"""Tests for the OIDC / 2FA / API-key / group bundle block."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_database_returns_zero_counts_and_empty_list(self):
|
|
from backend.app.api.routes.support import _collect_auth_info
|
|
|
|
def make_count(value):
|
|
r = MagicMock()
|
|
r.scalar.return_value = value
|
|
r.scalar_one_or_none.return_value = None
|
|
r.scalars.return_value.all.return_value = []
|
|
r.all.return_value = []
|
|
return r
|
|
|
|
async def fake_execute(stmt, *_a, **_kw):
|
|
return make_count(0)
|
|
|
|
db = AsyncMock()
|
|
db.execute = fake_execute
|
|
|
|
info = await _collect_auth_info(db)
|
|
|
|
assert info["oidc_providers"] == []
|
|
assert info["users_with_totp"] == 0
|
|
assert info["email_otp_codes_pending"] == 0
|
|
assert info["api_keys_total"] == 0
|
|
assert info["api_keys_enabled"] == 0
|
|
assert info["api_keys_expired"] == 0
|
|
assert info["long_lived_tokens_total"] == 0
|
|
assert info["long_lived_tokens_active"] == 0
|
|
assert info["groups_system"] == 0
|
|
assert info["groups_custom"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_oidc_provider_names_exported_in_cleartext(self):
|
|
"""Provider names are login-button labels — public, not a secret. Triage
|
|
for SSO bugs is significantly easier when the provider is identified."""
|
|
from backend.app.api.routes.support import _collect_auth_info
|
|
|
|
provider = MagicMock()
|
|
provider.id = 1
|
|
provider.name = "PocketID"
|
|
provider.is_enabled = True
|
|
provider.scopes = "openid email profile"
|
|
provider.email_claim = "email"
|
|
provider.require_email_verified = True
|
|
provider.auto_create_users = False
|
|
provider.auto_link_existing_accounts = False
|
|
provider.default_group_id = None
|
|
provider.icon_url = None
|
|
|
|
def make_result(rows=None, count=0):
|
|
r = MagicMock()
|
|
r.scalar.return_value = count
|
|
r.scalar_one_or_none.return_value = None
|
|
r.scalars.return_value.all.return_value = rows or []
|
|
r.all.return_value = []
|
|
return r
|
|
|
|
async def fake_execute(stmt, *_a, **_kw):
|
|
sql = str(stmt).lower()
|
|
if "oidc_providers" in sql and "user_oidc_link" not in sql:
|
|
return make_result([provider])
|
|
return make_result(count=0)
|
|
|
|
db = AsyncMock()
|
|
db.execute = fake_execute
|
|
|
|
info = await _collect_auth_info(db)
|
|
|
|
assert len(info["oidc_providers"]) == 1
|
|
oidc = info["oidc_providers"][0]
|
|
assert oidc["name"] == "PocketID"
|
|
# No secrets leak through — these fields don't exist on the dict.
|
|
assert "client_id" not in oidc
|
|
assert "client_secret" not in oidc
|
|
assert "issuer_url" not in oidc
|
|
|
|
|
|
class TestCollectGitHubBackupInfo:
|
|
"""Tests for the GitHub-backup provider/failure-count block."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_aggregates_providers_and_recent_failures(self):
|
|
from backend.app.api.routes.support import _collect_github_backup_info
|
|
|
|
c1 = MagicMock(provider="github", last_backup_status="success", schedule_enabled=True)
|
|
c2 = MagicMock(provider="github", last_backup_status="failed", schedule_enabled=False)
|
|
c3 = MagicMock(provider="gitea", last_backup_status="failed", schedule_enabled=True)
|
|
|
|
result = MagicMock()
|
|
result.scalars.return_value.all.return_value = [c1, c2, c3]
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(return_value=result)
|
|
|
|
info = await _collect_github_backup_info(db)
|
|
|
|
assert info["configs_total"] == 3
|
|
assert info["providers_used"] == {"github": 2, "gitea": 1}
|
|
assert info["schedule_enabled_count"] == 2
|
|
assert info["last_failure_count"] == 2
|
|
|
|
|
|
class TestRedactRawPushStatus:
|
|
"""Tests for _redact_raw_push_status() — the bundle dump scrubber."""
|
|
|
|
def test_drops_user_filename_and_cloud_ids(self):
|
|
from backend.app.api.routes.support import _redact_raw_push_status
|
|
|
|
raw = {
|
|
"subtask_name": "private_model.gcode",
|
|
"gcode_file": "Metadata/private.gcode",
|
|
"subtask_id": "1234567890",
|
|
"task_id": "9999",
|
|
"project_id": "proj-abc",
|
|
"design_id": "design-1",
|
|
"profile_id": "p-1",
|
|
"model_id": "m-1",
|
|
"gcode_state": "RUNNING",
|
|
"layer_num": 42, # control: non-sensitive sibling must survive
|
|
}
|
|
|
|
out = _redact_raw_push_status(raw)
|
|
|
|
assert "subtask_name" not in out
|
|
assert "gcode_file" not in out
|
|
assert "subtask_id" not in out
|
|
assert "task_id" not in out
|
|
assert "project_id" not in out
|
|
assert "design_id" not in out
|
|
assert "profile_id" not in out
|
|
assert "model_id" not in out
|
|
assert "gcode_state" not in out
|
|
assert out["layer_num"] == 42
|
|
|
|
def test_redacts_net_info_ip_addresses(self):
|
|
from backend.app.api.routes.support import _redact_raw_push_status
|
|
|
|
raw = {
|
|
"net": {
|
|
"conf": 1,
|
|
"info": [
|
|
{"ip": "192.168.1.42", "mask": "255.255.255.0"},
|
|
{"ip": "10.0.0.1", "mask": "255.0.0.0"},
|
|
],
|
|
},
|
|
}
|
|
|
|
out = _redact_raw_push_status(raw)
|
|
|
|
# LAN topology must be scrubbed (mirrors the #1429 VP fix).
|
|
assert out["net"]["info"][0]["ip"] == "0.0.0.0" # nosec B104 - redaction sentinel, not a bind address
|
|
assert out["net"]["info"][1]["ip"] == "0.0.0.0" # nosec B104 - redaction sentinel, not a bind address
|
|
# Non-IP siblings inside the entry survive so the shape stays
|
|
# diagnosable (interface count, mask presence, etc.).
|
|
assert out["net"]["info"][0]["mask"] == "255.255.255.0"
|
|
assert out["net"]["conf"] == 1
|
|
|
|
def test_preserves_print_cfg_and_ams_payloads(self):
|
|
"""The point of bundling raw_data is keeping these — print.cfg is what
|
|
unblocks per-model AMS Backup detection (deferred in 85fbd7fc).
|
|
"""
|
|
from backend.app.api.routes.support import _redact_raw_push_status
|
|
|
|
raw = {
|
|
"print": {
|
|
"cfg": 0x4000000, # bit-26 — the H2D AMS Backup bit
|
|
"option": 12345,
|
|
},
|
|
"ams": {
|
|
"ams": [
|
|
{
|
|
"id": "0",
|
|
"humidity": "3",
|
|
"tray": [
|
|
{"id": "0", "tray_type": "PLA", "tray_color": "FF0000FF"},
|
|
],
|
|
}
|
|
]
|
|
},
|
|
"vt_tray": {"tray_info_idx": "GFA00", "tray_type": "PLA", "tray_color": "00FF00FF"},
|
|
"vir_slot": [{"id": "0", "tray_type": "PLA"}],
|
|
"mapping": [0, 1, 2, 3],
|
|
"ams_extruder_map": {"0": 1},
|
|
}
|
|
|
|
out = _redact_raw_push_status(raw)
|
|
|
|
assert out["print"]["cfg"] == 0x4000000
|
|
assert out["print"]["option"] == 12345
|
|
assert out["ams"]["ams"][0]["tray"][0]["tray_type"] == "PLA"
|
|
assert out["vt_tray"]["tray_info_idx"] == "GFA00"
|
|
assert out["vir_slot"][0]["tray_type"] == "PLA"
|
|
assert out["mapping"] == [0, 1, 2, 3]
|
|
assert out["ams_extruder_map"] == {"0": 1}
|
|
|
|
def test_does_not_mutate_input(self):
|
|
"""Live state.raw_data must not be touched — the dispatcher reads it on
|
|
every tick, mutation would race the next push.
|
|
"""
|
|
from backend.app.api.routes.support import _redact_raw_push_status
|
|
|
|
raw = {
|
|
"subtask_name": "secret.gcode",
|
|
"net": {"info": [{"ip": "192.168.1.5"}]},
|
|
"print": {"cfg": 1},
|
|
}
|
|
original_subtask = raw["subtask_name"]
|
|
original_ip = raw["net"]["info"][0]["ip"]
|
|
|
|
_redact_raw_push_status(raw)
|
|
|
|
assert raw["subtask_name"] == original_subtask
|
|
assert raw["net"]["info"][0]["ip"] == original_ip
|
|
|
|
def test_handles_non_dict_gracefully(self):
|
|
from backend.app.api.routes.support import _redact_raw_push_status
|
|
|
|
assert _redact_raw_push_status(None) == {} # type: ignore[arg-type]
|
|
assert _redact_raw_push_status([]) == {} # type: ignore[arg-type]
|
|
assert _redact_raw_push_status("") == {} # type: ignore[arg-type]
|
|
|
|
|
|
class TestSanitizePushStatusValues:
|
|
"""The bundled push_status snapshot must stay parseable JSON.
|
|
|
|
Sanitization used to run over the *serialised* snapshot. The generic
|
|
Bambu-serial regex in ``log_reader`` (``0[0-3][A-Z0-9][A-Z0-9]{9,13}``)
|
|
matches the decimal expansion of a float as readily as a serial, so an AMS
|
|
``k`` flow factor came out as ``0.[SERIAL]`` and the whole file stopped
|
|
parsing — found in a real bundle while diagnosing #2702, which is exactly
|
|
the case the snapshot was added to serve.
|
|
"""
|
|
|
|
def test_float_that_matches_the_serial_regex_survives(self):
|
|
"""The observed reproducer, verbatim."""
|
|
import json
|
|
|
|
from backend.app.api.routes.support import _sanitize_push_status_values
|
|
|
|
raw = {"ams": [{"tray": [{"k": 0.0199999995529652}]}]}
|
|
|
|
out = _sanitize_push_status_values(raw, {})
|
|
|
|
assert json.loads(json.dumps(out)) == raw
|
|
|
|
def test_output_always_parses(self):
|
|
"""Whatever it does to values, the result must be valid JSON."""
|
|
import json
|
|
|
|
from backend.app.api.routes.support import _sanitize_push_status_values
|
|
|
|
raw = {
|
|
"k_values": [0.0199999995529652, 0.019999999552965164, 0.02],
|
|
"home_flag": 7554487,
|
|
"sdcard": True,
|
|
"resolution": "",
|
|
"nozzle": None,
|
|
}
|
|
|
|
json.loads(json.dumps(_sanitize_push_status_values(raw, {})))
|
|
|
|
def test_still_redacts_strings(self):
|
|
"""The point of the pass is not lost — string values are sanitized."""
|
|
from backend.app.api.routes.support import _sanitize_push_status_values
|
|
|
|
raw = {"tag_uid": "0123456789ABCDEF", "name": "Martin's P1S", "ip": "192.168.1.50"}
|
|
|
|
out = _sanitize_push_status_values(raw, {"Martin's P1S": "[PRINTER]"})
|
|
|
|
assert out["tag_uid"] == "[SERIAL]"
|
|
assert out["name"] == "[PRINTER]"
|
|
assert out["ip"] == "[IP]"
|
|
|
|
def test_walks_nested_containers(self):
|
|
from backend.app.api.routes.support import _sanitize_push_status_values
|
|
|
|
raw = {"ams": [{"tray": [{"tray_uuid": "0123456789ABCDEF"}]}]}
|
|
|
|
out = _sanitize_push_status_values(raw, {})
|
|
|
|
assert out["ams"][0]["tray"][0]["tray_uuid"] == "[SERIAL]"
|
|
|
|
def test_keys_are_left_alone(self):
|
|
"""Keys are structural — renaming one would break the schema."""
|
|
from backend.app.api.routes.support import _sanitize_push_status_values
|
|
|
|
raw = {"0123456789ABCDEF": 1}
|
|
|
|
assert list(_sanitize_push_status_values(raw, {})) == ["0123456789ABCDEF"]
|
|
|
|
def test_non_json_scalars_are_sanitized_not_smuggled(self):
|
|
"""``json.dumps(default=str)`` runs after this pass, so do it here."""
|
|
from datetime import datetime, timezone
|
|
|
|
from backend.app.api.routes.support import _sanitize_push_status_values
|
|
|
|
raw = {"seen_at": datetime(2026, 7, 29, 23, 12, 40, tzinfo=timezone.utc), "who": object()}
|
|
|
|
out = _sanitize_push_status_values(raw, {"2026-07-29": "[WHEN]"})
|
|
|
|
assert out["seen_at"].startswith("[WHEN]")
|
|
assert isinstance(out["who"], str)
|
|
|
|
def test_bools_stay_bools(self):
|
|
"""`isinstance(True, int)` — a bool must not fall through to str()."""
|
|
from backend.app.api.routes.support import _sanitize_push_status_values
|
|
|
|
out = _sanitize_push_status_values({"sdcard": True, "force_upgrade": False}, {})
|
|
|
|
assert out["sdcard"] is True
|
|
assert out["force_upgrade"] is False
|
|
|
|
def test_the_full_bundle_chain_on_a_real_p1s_payload(self):
|
|
"""The route's transform, end to end, on the shape from the #2702 bundle.
|
|
|
|
`_redact_raw_push_status` then `_sanitize_push_status_values` then
|
|
`json.dumps(default=str)` — the composition the bundle writer applies.
|
|
The bundle that exposed this had five `k` values corrupted, so the
|
|
snapshot could not be read at all; the field the report was about
|
|
(`total_layer_num`) was sitting in it, intact and unreachable.
|
|
"""
|
|
import json
|
|
|
|
from backend.app.api.routes.support import (
|
|
_redact_raw_push_status,
|
|
_sanitize_push_status_values,
|
|
)
|
|
|
|
raw = {
|
|
"gcode_file": "AMS_Filament_Clip_3MF.3mf",
|
|
"layer_num": 2,
|
|
"total_layer_num": 33,
|
|
"home_flag": 7554487,
|
|
"sdcard": True,
|
|
"net": {"info": [{"ip": "192.168.1.50", "mask": 0}]},
|
|
"ams": {
|
|
"ams": [
|
|
{
|
|
"id": "0",
|
|
"humidity": "5",
|
|
"tray": [
|
|
{"id": "0", "k": 0.0199999995529652, "tag_uid": "0123456789ABCDEF"},
|
|
{"id": "1", "k": 0.0209999997168779, "tag_uid": "44F782D000000100"},
|
|
],
|
|
}
|
|
]
|
|
},
|
|
}
|
|
|
|
snapshot = {
|
|
"model": "P1S",
|
|
"firmware_version": "01.10.00.00",
|
|
"raw_data": _redact_raw_push_status(raw),
|
|
}
|
|
text = json.dumps(_sanitize_push_status_values(snapshot, {}), indent=2, default=str)
|
|
|
|
parsed = json.loads(text) # used to raise "Expecting ',' delimiter"
|
|
trays = parsed["raw_data"]["ams"]["ams"][0]["tray"]
|
|
assert [t["k"] for t in trays] == [0.0199999995529652, 0.0209999997168779]
|
|
assert parsed["raw_data"]["total_layer_num"] == 33
|
|
# Redaction still did its job on both fronts.
|
|
assert "gcode_file" not in parsed["raw_data"]
|
|
assert trays[0]["tag_uid"] == "[SERIAL]"
|
|
# The structural pass replaces the printer's LAN address with the
|
|
# sentinel 0.0.0.0, which is itself an IPv4 literal, so the value pass
|
|
# then masks it to [IP]. Harmless — the real address is already gone —
|
|
# and matches what shipped in the bundle behind #2702.
|
|
assert parsed["raw_data"]["net"]["info"][0]["ip"] == "[IP]"
|
|
|
|
def test_does_not_mutate_the_live_snapshot(self):
|
|
"""`state.raw_data` is read by the dispatcher on every tick.
|
|
|
|
The bundle writer passes a redacted copy, but a walker that mutated in
|
|
place would still be one refactor away from redacting the live state.
|
|
"""
|
|
import copy
|
|
|
|
from backend.app.api.routes.support import _sanitize_push_status_values
|
|
|
|
raw = {"tag_uid": "0123456789ABCDEF", "ams": [{"tray": [{"k": 0.02, "n": "0123456789ABCDEF"}]}]}
|
|
before = copy.deepcopy(raw)
|
|
|
|
out = _sanitize_push_status_values(raw, {})
|
|
|
|
assert raw == before, "input was mutated"
|
|
assert out["tag_uid"] == "[SERIAL]" # and the copy really was redacted
|
|
|
|
|
|
class TestProcessInfo:
|
|
"""Bambuddy's own footprint in the bundle (#2734).
|
|
|
|
Bundles carried nothing about the process itself, so "memory climbs over
|
|
days until the OOM killer fires" could not be triaged from a bundle — the
|
|
reporter had to run shell commands by hand, and the numbers that would have
|
|
named the mechanism were unrecoverable afterwards.
|
|
"""
|
|
|
|
def test_reports_the_figures_that_separate_the_mechanisms(self):
|
|
"""RSS vs VMS, threads and children distinguish a heap that is growing
|
|
from address space, a thread leak, and a child-process leak."""
|
|
from backend.app.api.routes.support import _collect_process_info
|
|
|
|
info = _collect_process_info()
|
|
|
|
assert info["available"] is True
|
|
for key in ("rss_bytes", "vms_bytes", "num_threads", "children_total"):
|
|
assert isinstance(info[key], int), key
|
|
|
|
def test_children_are_named_but_never_quoted(self):
|
|
"""An ffmpeg argv carries the camera URL, and with it its password. The
|
|
count per executable is what identifies a leak; the arguments are not
|
|
needed and must not travel."""
|
|
from backend.app.api.routes.support import _collect_process_info
|
|
|
|
info = _collect_process_info()
|
|
|
|
for name in info.get("children_by_name", {}):
|
|
assert " " not in name, f"looks like a command line, not a name: {name!r}"
|
|
assert "://" not in name
|
|
|
|
def test_heap_census_is_skipped_on_a_large_process(self):
|
|
"""gc.get_objects() materialises every tracked object, so the census
|
|
costs most on the process that can least afford it. A bundle generated
|
|
to diagnose runaway memory must not be the allocation that tips the
|
|
host over."""
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import backend.app.api.routes.support as support_module
|
|
|
|
fake = MagicMock()
|
|
fake.memory_info.return_value = MagicMock(rss=8 * 1024**3, vms=12 * 1024**3)
|
|
fake.num_threads.return_value = 40
|
|
fake.create_time.return_value = 0.0
|
|
fake.open_files.return_value = []
|
|
fake.net_connections.return_value = []
|
|
fake.children.return_value = []
|
|
|
|
with patch("psutil.Process", return_value=fake):
|
|
info = support_module._collect_process_info()
|
|
|
|
assert "gc_top_types" not in info
|
|
assert "skipped" in info["gc_census"]
|
|
# The discriminating numbers still come through — those are the point.
|
|
assert info["rss_bytes"] == 8 * 1024**3
|
|
assert info["num_threads"] == 40
|
|
|
|
def test_heap_census_runs_on_a_normal_process(self):
|
|
from backend.app.api.routes.support import _collect_process_info
|
|
|
|
info = _collect_process_info()
|
|
|
|
assert info["gc_tracked_objects"] > 0
|
|
assert len(info["gc_top_types"]) <= 15
|
|
|
|
def test_survives_a_hostile_psutil(self):
|
|
"""psutil raises on hardened kernels and in restricted containers. A
|
|
support bundle must still be produced when it does — the bundle is how
|
|
someone reports the problem in the first place."""
|
|
from unittest.mock import patch
|
|
|
|
import backend.app.api.routes.support as support_module
|
|
|
|
with patch("psutil.Process", side_effect=RuntimeError("no /proc for you")):
|
|
info = support_module._collect_process_info()
|
|
|
|
assert info == {"available": False}
|
|
|
|
def test_partial_failures_do_not_lose_the_rest(self):
|
|
"""One inaccessible metric must not cost the others."""
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import backend.app.api.routes.support as support_module
|
|
|
|
fake = MagicMock()
|
|
fake.memory_info.return_value = MagicMock(rss=100, vms=200)
|
|
fake.num_threads.side_effect = PermissionError("denied")
|
|
fake.create_time.return_value = 0.0
|
|
fake.open_files.side_effect = PermissionError("denied")
|
|
fake.net_connections.side_effect = PermissionError("denied")
|
|
fake.children.return_value = []
|
|
|
|
with patch("psutil.Process", return_value=fake):
|
|
info = support_module._collect_process_info()
|
|
|
|
assert info["rss_bytes"] == 100
|
|
assert "num_threads" not in info
|
|
assert info["children_total"] == 0
|