mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
maziggy
a36c0009a5
The support bundle gives each printer's connection diagnostic 15 s and discarded the whole result on overrun, recording only "timed_out". A 14-printer farm's bundle carried that marker for every printer and nothing else, so it could not say which check was slow. run_connection_diagnostic now keeps an optional progress dict current (finished checks + the step in flight). On timeout the snapshot records stalled_in, elapsed_s and the checks that completed.
369 lines
15 KiB
Python
369 lines
15 KiB
Python
"""Tests for the diagnostic snapshot helper that aggregates connection,
|
|
virtual-printer, and log-health diagnostics for the support bundle and
|
|
bug-report submission paths (#1506 follow-up).
|
|
|
|
The helper has three hard requirements:
|
|
|
|
- Always returns the three top-level keys, even when sections are empty.
|
|
- Fail-soft per probe — a single crash doesn't break the snapshot.
|
|
- Bounded total runtime — concurrent gather caps wall-clock to the slowest probe.
|
|
"""
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from backend.app.services.diagnostic_snapshot import collect_diagnostic_snapshot
|
|
|
|
|
|
def _make_db_with_printers_and_vps(printers: list, vps: list):
|
|
"""Stub AsyncSession whose two .execute() calls return printers then VPs."""
|
|
printers_result = MagicMock()
|
|
printers_result.scalars.return_value.all.return_value = printers
|
|
vps_result = MagicMock()
|
|
vps_result.scalars.return_value.all.return_value = vps
|
|
db = MagicMock()
|
|
# Two execute calls — printer query, then VP query (order matches the
|
|
# helper). side_effect cycles through the queue.
|
|
db.execute = AsyncMock(side_effect=[printers_result, vps_result])
|
|
return db
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_always_returns_three_top_level_keys_when_empty():
|
|
"""No printers, no VPs — still get the three keys (empty lists, empty
|
|
log-health). Callers downstream rely on the shape being stable."""
|
|
db = _make_db_with_printers_and_vps([], [])
|
|
with patch(
|
|
"backend.app.services.diagnostic_snapshot._run_log_health",
|
|
new=AsyncMock(return_value={"findings": []}),
|
|
):
|
|
out = await collect_diagnostic_snapshot(db)
|
|
assert set(out.keys()) == {"connection_diagnostics", "vp_diagnostics", "log_health"}
|
|
assert out["connection_diagnostics"] == []
|
|
assert out["vp_diagnostics"] == []
|
|
assert out["log_health"] == {"findings": []}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_runs_diagnostic_per_active_printer():
|
|
"""Each active printer gets a connection check; each enabled VP gets a
|
|
setup check. Result list length matches the input lists."""
|
|
printers = [
|
|
SimpleNamespace(id=1, name="P1S", ip_address="192.168.1.10", serial_number="01S00A", access_code="abc123"),
|
|
SimpleNamespace(id=2, name="X1C", ip_address="192.168.1.11", serial_number="C11Y00", access_code="xyz456"),
|
|
]
|
|
vps = [SimpleNamespace(id=10, name="VP-1")]
|
|
db = _make_db_with_printers_and_vps(printers, vps)
|
|
|
|
fake_conn = SimpleNamespace(model_dump=lambda: {"checks": []})
|
|
fake_vp = SimpleNamespace(model_dump=lambda: {"checks": []})
|
|
|
|
with (
|
|
patch(
|
|
"backend.app.services.printer_diagnostic.run_connection_diagnostic",
|
|
new=AsyncMock(return_value=fake_conn),
|
|
),
|
|
patch(
|
|
"backend.app.services.virtual_printer.virtual_printer_manager.get_instance",
|
|
return_value=None,
|
|
),
|
|
patch(
|
|
"backend.app.services.virtual_printer.diagnostic.run_vp_diagnostic",
|
|
new=AsyncMock(return_value=fake_vp),
|
|
),
|
|
patch(
|
|
"backend.app.services.diagnostic_snapshot._run_log_health",
|
|
new=AsyncMock(return_value={"findings": []}),
|
|
),
|
|
):
|
|
out = await collect_diagnostic_snapshot(db)
|
|
|
|
assert len(out["connection_diagnostics"]) == 2
|
|
assert out["connection_diagnostics"][0]["printer_id"] == 1
|
|
assert out["connection_diagnostics"][1]["printer_id"] == 2
|
|
assert all("result" in entry for entry in out["connection_diagnostics"])
|
|
assert len(out["vp_diagnostics"]) == 1
|
|
assert out["vp_diagnostics"][0]["vp_id"] == 10
|
|
assert "result" in out["vp_diagnostics"][0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_fails_soft_when_single_printer_diagnostic_raises():
|
|
"""A crash inside one printer's diagnostic emits an error marker for that
|
|
printer, but the snapshot's other sections still complete. This is the
|
|
whole point of including diagnostics in the bundle — a partial result
|
|
beats a 500."""
|
|
printers = [
|
|
SimpleNamespace(id=1, name="ok", ip_address="1.1.1.1", serial_number="s1", access_code="a"),
|
|
SimpleNamespace(id=2, name="bad", ip_address="2.2.2.2", serial_number="s2", access_code="b"),
|
|
]
|
|
db = _make_db_with_printers_and_vps(printers, [])
|
|
|
|
fake_ok = SimpleNamespace(model_dump=lambda: {"status": "ok"})
|
|
|
|
async def diag(ip_address, **_):
|
|
if ip_address == "2.2.2.2":
|
|
raise RuntimeError("simulated crash")
|
|
return fake_ok
|
|
|
|
with (
|
|
patch("backend.app.services.printer_diagnostic.run_connection_diagnostic", new=AsyncMock(side_effect=diag)),
|
|
patch(
|
|
"backend.app.services.diagnostic_snapshot._run_log_health",
|
|
new=AsyncMock(return_value={"findings": []}),
|
|
),
|
|
):
|
|
out = await collect_diagnostic_snapshot(db)
|
|
|
|
# Both printers represented; the crashing one carries an `error` field.
|
|
assert len(out["connection_diagnostics"]) == 2
|
|
ok_entry = next(e for e in out["connection_diagnostics"] if e["printer_id"] == 1)
|
|
bad_entry = next(e for e in out["connection_diagnostics"] if e["printer_id"] == 2)
|
|
assert "result" in ok_entry
|
|
assert "error" not in ok_entry
|
|
assert "error" in bad_entry
|
|
assert "simulated crash" in bad_entry["error"]
|
|
# Log-health still completes despite the per-printer crash.
|
|
assert out["log_health"] == {"findings": []}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_emits_timed_out_marker_when_probe_exceeds_cap():
|
|
"""If a single probe stalls past the per-diagnostic timeout, the entry
|
|
is marked `timed_out` rather than blocking the whole snapshot. Patch
|
|
the timeout small so the test runs fast."""
|
|
printers = [SimpleNamespace(id=1, name="slow", ip_address="1.1.1.1", serial_number="s", access_code="a")]
|
|
db = _make_db_with_printers_and_vps(printers, [])
|
|
|
|
async def slow_diag(*a, **k):
|
|
import asyncio
|
|
|
|
await asyncio.sleep(5) # well past the patched cap below
|
|
return SimpleNamespace(model_dump=lambda: {})
|
|
|
|
with (
|
|
patch(
|
|
"backend.app.services.printer_diagnostic.run_connection_diagnostic", new=AsyncMock(side_effect=slow_diag)
|
|
),
|
|
patch("backend.app.services.diagnostic_snapshot._PER_DIAGNOSTIC_TIMEOUT_SECONDS", 0.05),
|
|
patch(
|
|
"backend.app.services.diagnostic_snapshot._run_log_health",
|
|
new=AsyncMock(return_value={"findings": []}),
|
|
),
|
|
):
|
|
out = await collect_diagnostic_snapshot(db)
|
|
|
|
assert len(out["connection_diagnostics"]) == 1
|
|
assert out["connection_diagnostics"][0]["error"] == "timed_out"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timed_out_entry_names_the_stalled_step_and_keeps_finished_checks():
|
|
"""#3164: every printer of a 14-printer farm came back as a bare
|
|
`timed_out`, so the bundle could not say which step hung. The real
|
|
diagnostic runs here with the ports answering and the subnet lookup
|
|
hanging; the entry must name `subnet` and carry the port checks that
|
|
finished before it."""
|
|
import time as _time
|
|
|
|
printers = [
|
|
SimpleNamespace(
|
|
id=1,
|
|
name="A1",
|
|
model="A1",
|
|
ip_address="10.0.0.5",
|
|
serial_number="s",
|
|
access_code="a",
|
|
)
|
|
]
|
|
db = _make_db_with_printers_and_vps(printers, [])
|
|
|
|
def hanging_subnet(*_a, **_k):
|
|
_time.sleep(1.0) # blocks its worker thread well past the cap below
|
|
return True
|
|
|
|
with (
|
|
patch("backend.app.services.printer_diagnostic._check_port", new=AsyncMock(return_value=True)),
|
|
patch("backend.app.services.printer_diagnostic._check_ftps_tls", new=AsyncMock(return_value="ok")),
|
|
patch("backend.app.services.printer_diagnostic.detect_container_runtime", return_value=None),
|
|
patch("backend.app.services.printer_diagnostic._host_source_ip", return_value="10.0.0.2"),
|
|
patch("backend.app.services.printer_diagnostic._same_subnet", side_effect=hanging_subnet),
|
|
patch("backend.app.services.diagnostic_snapshot._PER_DIAGNOSTIC_TIMEOUT_SECONDS", 0.2),
|
|
patch(
|
|
"backend.app.services.diagnostic_snapshot._run_log_health",
|
|
new=AsyncMock(return_value={"findings": []}),
|
|
),
|
|
):
|
|
out = await collect_diagnostic_snapshot(db)
|
|
|
|
entry = out["connection_diagnostics"][0]
|
|
assert entry["error"] == "timed_out"
|
|
assert entry["stalled_in"] == "subnet"
|
|
assert 0.1 <= entry["elapsed_s"] < 1.0
|
|
finished = {c["id"]: c["status"] for c in entry["checks"]}
|
|
assert finished["port_mqtt"] == "pass"
|
|
assert finished["port_ftps"] == "pass"
|
|
assert finished["network_mode"] == "skip"
|
|
assert "subnet" not in finished
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timed_out_entry_without_progress_still_has_the_marker():
|
|
"""A diagnostic that hangs before recording anything still yields the
|
|
marker with an empty check list rather than a KeyError."""
|
|
printers = [SimpleNamespace(id=1, name="slow", ip_address="1.1.1.1", serial_number="s", access_code="a")]
|
|
db = _make_db_with_printers_and_vps(printers, [])
|
|
|
|
async def hang(*a, **k):
|
|
import asyncio
|
|
|
|
await asyncio.sleep(5)
|
|
|
|
with (
|
|
patch("backend.app.services.printer_diagnostic.run_connection_diagnostic", new=AsyncMock(side_effect=hang)),
|
|
patch("backend.app.services.diagnostic_snapshot._PER_DIAGNOSTIC_TIMEOUT_SECONDS", 0.05),
|
|
patch(
|
|
"backend.app.services.diagnostic_snapshot._run_log_health",
|
|
new=AsyncMock(return_value={"findings": []}),
|
|
),
|
|
):
|
|
out = await collect_diagnostic_snapshot(db)
|
|
|
|
entry = out["connection_diagnostics"][0]
|
|
assert entry["stalled_in"] is None
|
|
assert entry["checks"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_masks_ip_addresses_in_all_diagnostic_fields():
|
|
"""The diagnostic schemas embed raw IPv4 in three places — the top-level
|
|
``PrinterDiagnosticResult.ip_address``, the network-mode check's
|
|
``params.{printer_ip, host_ip}``, and the VP diagnostic's
|
|
``params.bind_ip``. None of those should leak into the submitted
|
|
snapshot. Sanitization runs after the per-probe gather; both DB-known
|
|
IPs (covered by sensitive_strings → "[IP]") and host / VP-bind IPs
|
|
(caught by the IPv4 regex fallback) end up redacted.
|
|
"""
|
|
printers = [
|
|
SimpleNamespace(
|
|
id=1, name="Workshop", ip_address="192.168.255.131", serial_number="01S00ABC123", access_code="abcd1234"
|
|
)
|
|
]
|
|
vps = [SimpleNamespace(id=10, name="VP-Workshop")]
|
|
db = _make_db_with_printers_and_vps(printers, vps)
|
|
|
|
fake_conn = SimpleNamespace(
|
|
model_dump=lambda: {
|
|
"ip_address": "192.168.255.131",
|
|
"overall": "ok",
|
|
"checks": [
|
|
{
|
|
"id": "network_mode",
|
|
"status": "warn",
|
|
"params": {"printer_ip": "192.168.255.131", "host_ip": "192.168.255.16"},
|
|
}
|
|
],
|
|
}
|
|
)
|
|
fake_vp = SimpleNamespace(
|
|
model_dump=lambda: {
|
|
"overall": "ok",
|
|
"checks": [{"id": "bind_interface", "status": "pass", "params": {"bind_ip": "192.168.254.2"}}],
|
|
}
|
|
)
|
|
|
|
with (
|
|
patch(
|
|
"backend.app.services.log_reader.collect_sensitive_strings",
|
|
new=AsyncMock(
|
|
return_value={
|
|
"Workshop": "[PRINTER]",
|
|
"192.168.255.131": "[IP]",
|
|
"01S00ABC123": "[SERIAL]",
|
|
"abcd1234": "[ACCESS_CODE]",
|
|
}
|
|
),
|
|
),
|
|
patch(
|
|
"backend.app.services.printer_diagnostic.run_connection_diagnostic", new=AsyncMock(return_value=fake_conn)
|
|
),
|
|
patch("backend.app.services.virtual_printer.virtual_printer_manager.get_instance", return_value=None),
|
|
patch("backend.app.services.virtual_printer.diagnostic.run_vp_diagnostic", new=AsyncMock(return_value=fake_vp)),
|
|
patch(
|
|
"backend.app.services.diagnostic_snapshot._run_log_health",
|
|
new=AsyncMock(return_value={"findings": [{"sample": "Connecting to 10.0.0.5..."}]}),
|
|
),
|
|
):
|
|
out = await collect_diagnostic_snapshot(db)
|
|
|
|
# Conection diagnostic — top-level ip_address and check params both masked.
|
|
conn_entry = out["connection_diagnostics"][0]
|
|
assert conn_entry["printer_name"] == "[PRINTER]"
|
|
assert conn_entry["result"]["ip_address"] == "[IP]"
|
|
check_params = conn_entry["result"]["checks"][0]["params"]
|
|
assert check_params["printer_ip"] == "[IP]"
|
|
assert check_params["host_ip"] == "[IP]" # not in DB; caught by regex fallback
|
|
|
|
# VP diagnostic — bind_ip masked (regex fallback; never in DB).
|
|
vp_entry = out["vp_diagnostics"][0]
|
|
assert vp_entry["result"]["checks"][0]["params"]["bind_ip"] == "[IP]"
|
|
|
|
# Log-health findings — IPs in log samples also masked (regex applies
|
|
# recursively through the dict, not just to known fields).
|
|
assert "10.0.0.5" not in str(out["log_health"])
|
|
assert "[IP]" in out["log_health"]["findings"][0]["sample"]
|
|
|
|
# Sanity: no raw IPv4 anywhere in the serialized snapshot.
|
|
import json
|
|
import re as _re
|
|
|
|
serialized = json.dumps(out)
|
|
raw_ipv4 = _re.search(
|
|
r"\b(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]\d|\d)(?:\.(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]\d|\d)){3}\b", serialized
|
|
)
|
|
assert raw_ipv4 is None, f"raw IPv4 leaked into snapshot: {raw_ipv4.group()}"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_runs_probes_concurrently_not_sequentially():
|
|
"""Total wall-clock for N printers should be O(slowest), not O(sum) —
|
|
this is what makes the feature usable on a fleet. Set each probe to
|
|
take 0.2 s; with 4 printers, sequential is 0.8 s, concurrent is 0.2 s.
|
|
Allow margin for scheduling and the test still catches a regression
|
|
to sequential execution.
|
|
"""
|
|
import time
|
|
|
|
printers = [
|
|
SimpleNamespace(id=i, name=f"P{i}", ip_address=f"1.1.1.{i}", serial_number=f"s{i}", access_code="a")
|
|
for i in range(4)
|
|
]
|
|
db = _make_db_with_printers_and_vps(printers, [])
|
|
|
|
async def slow_diag(*a, **k):
|
|
import asyncio
|
|
|
|
await asyncio.sleep(0.2)
|
|
return SimpleNamespace(model_dump=lambda: {"ok": True})
|
|
|
|
with (
|
|
patch(
|
|
"backend.app.services.printer_diagnostic.run_connection_diagnostic", new=AsyncMock(side_effect=slow_diag)
|
|
),
|
|
patch(
|
|
"backend.app.services.diagnostic_snapshot._run_log_health",
|
|
new=AsyncMock(return_value={"findings": []}),
|
|
),
|
|
):
|
|
start = time.monotonic()
|
|
out = await collect_diagnostic_snapshot(db)
|
|
elapsed = time.monotonic() - start
|
|
|
|
assert len(out["connection_diagnostics"]) == 4
|
|
# Concurrent should be ~0.2 s; sequential would be ~0.8 s. Use 0.5 s
|
|
# as the threshold — slack enough for slow CI, tight enough to catch
|
|
# a regression to sequential execution.
|
|
assert elapsed < 0.5, f"snapshot ran sequentially: {elapsed:.2f}s for 4 x 0.2s probes"
|