mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
MartinNYHC
b30a283184
feat(spoolman-inventory): squashed feature work for rebase onto dev Squashed all commits from feature/spoolman-inventory-ui onto a single commit to enable a clean rebase onto dev. Original per-commit history preserved at backup tag backup/spoolman-inventory-ui-prerebase-20260507-105721.
208 lines
8.1 KiB
Python
208 lines
8.1 KiB
Python
"""Unit tests for SpoolBuddy schema validation (security fixes H2, M2, M4).
|
|
|
|
Tests Pydantic model validation without requiring a running server or DB.
|
|
"""
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from backend.app.schemas.spoolbuddy import (
|
|
DeviceRegisterRequest,
|
|
HeartbeatRequest,
|
|
ScaleReadingRequest,
|
|
UpdateStatusRequest,
|
|
WriteTagResultRequest,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# H2 — UpdateStatusRequest: only valid Literal values accepted
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUpdateStatusRequestValidation:
|
|
def test_valid_status_updating(self):
|
|
req = UpdateStatusRequest(status="updating")
|
|
assert req.status == "updating"
|
|
|
|
def test_valid_status_complete(self):
|
|
req = UpdateStatusRequest(status="complete")
|
|
assert req.status == "complete"
|
|
|
|
def test_valid_status_error(self):
|
|
req = UpdateStatusRequest(status="error")
|
|
assert req.status == "error"
|
|
|
|
def test_invalid_status_rejected(self):
|
|
"""Arbitrary status strings must be rejected (H2: prevents unbounded WS injection)."""
|
|
with pytest.raises(ValidationError):
|
|
UpdateStatusRequest(status="hacked")
|
|
|
|
def test_empty_status_rejected(self):
|
|
with pytest.raises(ValidationError):
|
|
UpdateStatusRequest(status="")
|
|
|
|
def test_message_max_length_enforced(self):
|
|
"""message field must not exceed 255 chars."""
|
|
with pytest.raises(ValidationError):
|
|
UpdateStatusRequest(status="updating", message="x" * 256)
|
|
|
|
def test_message_at_max_length_accepted(self):
|
|
req = UpdateStatusRequest(status="complete", message="x" * 255)
|
|
assert len(req.message) == 255
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M2 — HeartbeatRequest: system_stats size limit (4096 bytes)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHeartbeatSystemStatsValidation:
|
|
def test_none_accepted(self):
|
|
req = HeartbeatRequest(system_stats=None)
|
|
assert req.system_stats is None
|
|
|
|
def test_small_dict_accepted(self):
|
|
req = HeartbeatRequest(system_stats={"cpu": 12.5, "mem": 60.0})
|
|
assert req.system_stats["cpu"] == 12.5
|
|
|
|
def test_oversized_dict_rejected(self):
|
|
"""system_stats exceeding 4096 bytes JSON-encoded must be rejected (M2)."""
|
|
huge = {"data": "x" * 5000}
|
|
with pytest.raises(ValidationError, match="4096"):
|
|
HeartbeatRequest(system_stats=huge)
|
|
|
|
def test_exactly_4096_bytes_accepted(self):
|
|
"""A dict whose JSON is exactly 4096 bytes must pass."""
|
|
import json
|
|
|
|
# Build a dict whose JSON is exactly 4096 bytes
|
|
filler = "x" * (4096 - len('{"k": ""}'))
|
|
d = {"k": filler}
|
|
assert len(json.dumps(d)) == 4096
|
|
req = HeartbeatRequest(system_stats=d)
|
|
assert req.system_stats is not None
|
|
|
|
def test_one_byte_over_limit_rejected(self):
|
|
import json
|
|
|
|
filler = "x" * (4097 - len('{"k": ""}'))
|
|
d = {"k": filler}
|
|
assert len(json.dumps(d)) == 4097
|
|
with pytest.raises(ValidationError):
|
|
HeartbeatRequest(system_stats=d)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M4 — DeviceRegisterRequest: max_length on device-sourced string fields
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDeviceRegisterRequestValidation:
|
|
VALID_BASE = {"device_id": "dev1", "hostname": "spoolbuddy.local", "ip_address": "192.168.1.50"}
|
|
|
|
def test_valid_minimal_accepted(self):
|
|
req = DeviceRegisterRequest(**self.VALID_BASE)
|
|
assert req.device_id == "dev1"
|
|
|
|
def test_firmware_version_too_long_rejected(self):
|
|
with pytest.raises(ValidationError):
|
|
DeviceRegisterRequest(**self.VALID_BASE, firmware_version="x" * 21)
|
|
|
|
def test_firmware_version_at_max_accepted(self):
|
|
req = DeviceRegisterRequest(**self.VALID_BASE, firmware_version="x" * 20)
|
|
assert req.firmware_version == "x" * 20
|
|
|
|
def test_nfc_reader_type_too_long_rejected(self):
|
|
with pytest.raises(ValidationError):
|
|
DeviceRegisterRequest(**self.VALID_BASE, nfc_reader_type="x" * 21)
|
|
|
|
def test_nfc_connection_too_long_rejected(self):
|
|
with pytest.raises(ValidationError):
|
|
DeviceRegisterRequest(**self.VALID_BASE, nfc_connection="x" * 21)
|
|
|
|
def test_backend_url_too_long_rejected(self):
|
|
with pytest.raises(ValidationError):
|
|
DeviceRegisterRequest(**self.VALID_BASE, backend_url="http://" + "x" * 249)
|
|
|
|
def test_backend_url_at_max_accepted(self):
|
|
url = "http://" + "x" * (255 - len("http://"))
|
|
req = DeviceRegisterRequest(**self.VALID_BASE, backend_url=url)
|
|
assert req.backend_url == url
|
|
|
|
def test_device_id_too_long_rejected(self):
|
|
with pytest.raises(ValidationError):
|
|
DeviceRegisterRequest(device_id="x" * 51, hostname="h", ip_address="1.2.3.4")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M4 — WriteTagResultRequest: device_id max_length
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestWriteTagResultRequestValidation:
|
|
def test_device_id_too_long_rejected(self):
|
|
with pytest.raises(ValidationError):
|
|
WriteTagResultRequest(device_id="x" * 51, spool_id=1, tag_uid="AABBCCDD", success=True)
|
|
|
|
def test_device_id_at_max_accepted(self):
|
|
req = WriteTagResultRequest(device_id="x" * 50, spool_id=1, tag_uid="AABBCCDD", success=True)
|
|
assert len(req.device_id) == 50
|
|
|
|
def test_tag_uid_hex_pattern_accepted(self):
|
|
req = WriteTagResultRequest(device_id="dev1", spool_id=1, tag_uid="AABBCCDD", success=True)
|
|
assert req.tag_uid == "AABBCCDD"
|
|
|
|
def test_tag_uid_non_hex_rejected(self):
|
|
"""Non-hex characters in tag_uid must be rejected (prevents injection via NFC write-back)."""
|
|
with pytest.raises(ValidationError):
|
|
WriteTagResultRequest(device_id="dev1", spool_id=1, tag_uid="AABB; DROP", success=True)
|
|
|
|
def test_tag_uid_too_short_rejected(self):
|
|
with pytest.raises(ValidationError):
|
|
WriteTagResultRequest(device_id="dev1", spool_id=1, tag_uid="AABB", success=True)
|
|
|
|
def test_tag_uid_max_length_accepted(self):
|
|
req = WriteTagResultRequest(device_id="dev1", spool_id=1, tag_uid="A" * 30, success=True)
|
|
assert len(req.tag_uid) == 30
|
|
|
|
def test_tag_uid_over_max_length_rejected(self):
|
|
with pytest.raises(ValidationError):
|
|
WriteTagResultRequest(device_id="dev1", spool_id=1, tag_uid="A" * 31, success=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M4 — ScaleReadingRequest: weight_grams accepts any float (raw uncalibrated ADC)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestScaleReadingRequestValidation:
|
|
def test_valid_weight_accepted(self):
|
|
req = ScaleReadingRequest(device_id="sb1", weight_grams=250.0)
|
|
assert req.weight_grams == 250.0
|
|
|
|
def test_zero_weight_accepted(self):
|
|
req = ScaleReadingRequest(device_id="sb1", weight_grams=0.0)
|
|
assert req.weight_grams == 0.0
|
|
|
|
def test_large_raw_adc_weight_accepted(self):
|
|
# Uncalibrated scale with factor=1.0 produces raw ADC values in the millions
|
|
req = ScaleReadingRequest(device_id="sb1", weight_grams=5_000_000.0)
|
|
assert req.weight_grams == 5_000_000.0
|
|
|
|
def test_negative_weight_accepted(self):
|
|
# Scale can legitimately read negative values when tare is not calibrated
|
|
req = ScaleReadingRequest(device_id="sb1", weight_grams=-50_000.0)
|
|
assert req.weight_grams == -50_000.0
|
|
|
|
def test_nan_weight_rejected(self):
|
|
import math
|
|
|
|
with pytest.raises(ValidationError):
|
|
ScaleReadingRequest(device_id="sb1", weight_grams=math.nan)
|
|
|
|
def test_inf_weight_rejected(self):
|
|
import math
|
|
|
|
with pytest.raises(ValidationError):
|
|
ScaleReadingRequest(device_id="sb1", weight_grams=math.inf)
|