From 75499350caffedf93e952ca1b6789c428ea2cc96 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Strausmann?= Date: Mon, 11 May 2026 10:01:15 +0000 Subject: [PATCH 1/3] feat(integrations): LabelData schema + Snipe-IT lookup client Introduces the app-agnostic LabelData value object (frozen Pydantic model) and the SnipeITClient that maps Snipe-IT asset payloads to LabelData. Adds 9 tests (4 schema, 5 client); total suite: 81 passed. Co-Authored-By: Claude Sonnet 4.6 --- backend/app/schemas/__init__.py | 1 + backend/app/schemas/label_data.py | 27 ++++++ backend/app/services/snipeit_client.py | 71 ++++++++++++++++ backend/tests/unit/schemas/__init__.py | 0 backend/tests/unit/schemas/test_label_data.py | 50 +++++++++++ .../unit/services/test_snipeit_client.py | 83 +++++++++++++++++++ 6 files changed, 232 insertions(+) create mode 100644 backend/app/schemas/__init__.py create mode 100644 backend/app/schemas/label_data.py create mode 100644 backend/app/services/snipeit_client.py create mode 100644 backend/tests/unit/schemas/__init__.py create mode 100644 backend/tests/unit/schemas/test_label_data.py create mode 100644 backend/tests/unit/services/test_snipeit_client.py diff --git a/backend/app/schemas/__init__.py b/backend/app/schemas/__init__.py new file mode 100644 index 0000000..d80c449 --- /dev/null +++ b/backend/app/schemas/__init__.py @@ -0,0 +1 @@ +"""Pydantic schemas — Request/Response/Domain value-objects.""" diff --git a/backend/app/schemas/label_data.py b/backend/app/schemas/label_data.py new file mode 100644 index 0000000..571194c --- /dev/null +++ b/backend/app/schemas/label_data.py @@ -0,0 +1,27 @@ +"""App-agnostic label data passed from lookup-clients to the LabelRenderer. + +LabelData is what a `*_client.lookup(id)` call produces. It is the +serialisable view of a real-world entity (Snipe-IT asset, Grocy product, +Spoolman spool) condensed into the minimal set of fields a label needs: +a title, an identifier to print, a QR-encodable URL, optional secondary +lines, and a source-app tag for downstream routing. + +Layout, font, geometry, and tape-fit decisions live on the LabelRenderer +side — they are NOT in this model. +""" + +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field + + +class LabelData(BaseModel): + """Immutable, app-agnostic label payload.""" + + model_config = ConfigDict(frozen=True) + + title: str + primary_id: str + qr_payload: str + source_app: str + secondary: list[str] = Field(default_factory=list) diff --git a/backend/app/services/snipeit_client.py b/backend/app/services/snipeit_client.py new file mode 100644 index 0000000..aa19a0d --- /dev/null +++ b/backend/app/services/snipeit_client.py @@ -0,0 +1,71 @@ +"""Snipe-IT REST API client — asset lookup by asset_tag. + +The client emits domain-level `LabelData` records so downstream consumers +(LabelRenderer, queue submitters) never see Snipe-IT's raw schema. Add new +fields by extending the mapping in `lookup()`, never by leaking the upstream +shape. +""" + +from __future__ import annotations + +from typing import Any + +import httpx + +from app.schemas.label_data import LabelData + + +class SnipeITNotFoundError(Exception): + """Raised when no Snipe-IT asset matches the given tag.""" + + +class SnipeITClient: + """Async client for Snipe-IT's REST API. + + Authenticates with a bearer token (Snipe-IT API key). Configuration — + base URL, API key, timeout — is injected so the same class can hit the + user's live instance from production and a respx-mocked endpoint from + tests, with no hidden global state. + """ + + def __init__( + self, + *, + base_url: str, + api_key: str, + timeout: float = 5.0, + ) -> None: + self._base_url = base_url.rstrip("/") + self._api_key = api_key + self._timeout = timeout + + async def lookup(self, asset_tag: str) -> LabelData: + """Return LabelData for `asset_tag`, or raise SnipeITNotFoundError.""" + url = f"{self._base_url}/api/v1/hardware/bytag/{asset_tag}" + headers = { + "Authorization": f"Bearer {self._api_key}", + "Accept": "application/json", + } + async with httpx.AsyncClient(timeout=self._timeout) as client: + response = await client.get(url, headers=headers) + + if response.status_code == 404: + raise SnipeITNotFoundError(f"Asset {asset_tag!r} not found") + response.raise_for_status() + + payload: dict[str, Any] = response.json() + return self._payload_to_label(payload, asset_tag) + + def _payload_to_label(self, payload: dict[str, Any], asset_tag: str) -> LabelData: + secondary: list[str] = [] + serial = payload.get("serial") + if serial: + secondary.append(f"S/N: {serial}") + + return LabelData( + title=str(payload.get("name") or asset_tag), + primary_id=str(payload.get("asset_tag") or asset_tag), + qr_payload=f"{self._base_url}/hardware/{payload.get('id')}", + source_app="snipeit", + secondary=secondary, + ) diff --git a/backend/tests/unit/schemas/__init__.py b/backend/tests/unit/schemas/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/tests/unit/schemas/test_label_data.py b/backend/tests/unit/schemas/test_label_data.py new file mode 100644 index 0000000..0abbb26 --- /dev/null +++ b/backend/tests/unit/schemas/test_label_data.py @@ -0,0 +1,50 @@ +import pytest +from app.schemas.label_data import LabelData + + +def test_label_data_minimal() -> None: + data = LabelData( + title="MacBook Pro 16", + primary_id="ASSET-12345", + qr_payload="https://snipe-it.example/assets/12345", + source_app="snipeit", + ) + assert data.title == "MacBook Pro 16" + assert data.primary_id == "ASSET-12345" + assert data.qr_payload == "https://snipe-it.example/assets/12345" + assert data.source_app == "snipeit" + assert data.secondary == [] + + +def test_label_data_with_secondary_fields() -> None: + data = LabelData( + title="BambuLab PLA", + primary_id="#42", + qr_payload="https://spoolman.example/spool/42", + source_app="spoolman", + secondary=["Color: Black", "Weight: 850g"], + ) + assert len(data.secondary) == 2 + assert data.secondary[0] == "Color: Black" + + +def test_label_data_is_frozen() -> None: + """LabelData is an immutable value object — mutating fields after construction must fail.""" + import pydantic + + data = LabelData( + title="t", + primary_id="p", + qr_payload="q", + source_app="snipeit", + ) + with pytest.raises(pydantic.ValidationError): + data.title = "different" # type: ignore[misc] + + +def test_label_data_default_secondary_is_distinct_per_instance() -> None: + """The default empty list must NOT be a shared mutable default.""" + a = LabelData(title="a", primary_id="a", qr_payload="a", source_app="snipeit") + b = LabelData(title="b", primary_id="b", qr_payload="b", source_app="snipeit") + # With frozen=True we cannot append, but we can still verify the lists are distinct objects. + assert a.secondary is not b.secondary diff --git a/backend/tests/unit/services/test_snipeit_client.py b/backend/tests/unit/services/test_snipeit_client.py new file mode 100644 index 0000000..d06d3af --- /dev/null +++ b/backend/tests/unit/services/test_snipeit_client.py @@ -0,0 +1,83 @@ +import httpx +import pytest +import respx +from app.services.snipeit_client import SnipeITClient, SnipeITNotFoundError + + +@pytest.mark.asyncio +@respx.mock +async def test_lookup_asset_returns_label_data() -> None: + respx.get("https://snipe-it.example/api/v1/hardware/bytag/ASSET-12345").mock( + return_value=httpx.Response( + 200, + json={ + "id": 123, + "asset_tag": "ASSET-12345", + "name": "MacBook Pro 16", + "serial": "C02XYZ", + }, + ) + ) + + client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key") + data = await client.lookup("ASSET-12345") + + assert data.title == "MacBook Pro 16" + assert data.primary_id == "ASSET-12345" + assert data.qr_payload == "https://snipe-it.example/hardware/123" + assert data.source_app == "snipeit" + assert data.secondary == ["S/N: C02XYZ"] + + +@pytest.mark.asyncio +@respx.mock +async def test_lookup_asset_404_raises_not_found() -> None: + respx.get("https://snipe-it.example/api/v1/hardware/bytag/UNKNOWN").mock( + return_value=httpx.Response(404) + ) + + client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key") + + with pytest.raises(SnipeITNotFoundError, match="UNKNOWN"): + await client.lookup("UNKNOWN") + + +@pytest.mark.asyncio +@respx.mock +async def test_lookup_asset_without_serial_has_no_secondary_line() -> None: + """Missing optional serial field must not add a 'S/N: None' line.""" + respx.get("https://snipe-it.example/api/v1/hardware/bytag/A-1").mock( + return_value=httpx.Response( + 200, + json={"id": 1, "asset_tag": "A-1", "name": "Thing"}, + ) + ) + + client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key") + data = await client.lookup("A-1") + + assert data.secondary == [] + + +@pytest.mark.asyncio +@respx.mock +async def test_lookup_strips_trailing_slash_from_base_url() -> None: + """base_url='https://snipe-it.example/' must not produce a double slash.""" + respx.get("https://snipe-it.example/api/v1/hardware/bytag/A-1").mock( + return_value=httpx.Response(200, json={"id": 1, "asset_tag": "A-1", "name": "Thing"}) + ) + client = SnipeITClient(base_url="https://snipe-it.example/", api_key="test-key") + data = await client.lookup("A-1") + assert data.qr_payload == "https://snipe-it.example/hardware/1" + + +@pytest.mark.asyncio +@respx.mock +async def test_lookup_5xx_raises_httpx_error() -> None: + """A 500 from upstream must surface as httpx.HTTPStatusError (no swallowing).""" + respx.get("https://snipe-it.example/api/v1/hardware/bytag/A-1").mock( + return_value=httpx.Response(500) + ) + client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key") + with pytest.raises(httpx.HTTPStatusError): + await client.lookup("A-1") From bae14ff7f55cc03c568db421bf8d45a3834b7ebb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Strausmann?= Date: Mon, 11 May 2026 10:06:04 +0000 Subject: [PATCH 2/3] fix(snipeit): require Snipe-IT 'id' field, tighten frozen-test exception, document HTTP error policy - Guard against missing 'id' in Snipe-IT responses: raise ValueError before constructing LabelData so callers never receive a silently broken .../hardware/None QR URL. - Add regression test asserting ValueError with match on 'missing required field 'id'' for a 200 response that omits the id field. - Tighten test_label_data_is_frozen: switch from bare pydantic.ValidationError to pydantic_core.ValidationError with match='frozen_instance' so the test fails if Pydantic ever changes the freeze behaviour. - Add one-line comment above raise_for_status() documenting the HTTP error propagation policy for future maintainers. Co-Authored-By: Claude Sonnet 4.6 --- backend/app/services/snipeit_client.py | 8 ++++++-- backend/tests/unit/schemas/test_label_data.py | 5 ++--- .../tests/unit/services/test_snipeit_client.py | 18 ++++++++++++++++++ 3 files changed, 26 insertions(+), 5 deletions(-) diff --git a/backend/app/services/snipeit_client.py b/backend/app/services/snipeit_client.py index aa19a0d..f73f9a4 100644 --- a/backend/app/services/snipeit_client.py +++ b/backend/app/services/snipeit_client.py @@ -51,21 +51,25 @@ async def lookup(self, asset_tag: str) -> LabelData: if response.status_code == 404: raise SnipeITNotFoundError(f"Asset {asset_tag!r} not found") + # 401/403/5xx surface as httpx.HTTPStatusError — callers (AppLookupService) + # decide whether to treat them as configuration errors vs transient failures. response.raise_for_status() payload: dict[str, Any] = response.json() return self._payload_to_label(payload, asset_tag) def _payload_to_label(self, payload: dict[str, Any], asset_tag: str) -> LabelData: + asset_id = payload.get("id") + if asset_id is None: + raise ValueError(f"Snipe-IT response for {asset_tag!r} is missing required field 'id'") secondary: list[str] = [] serial = payload.get("serial") if serial: secondary.append(f"S/N: {serial}") - return LabelData( title=str(payload.get("name") or asset_tag), primary_id=str(payload.get("asset_tag") or asset_tag), - qr_payload=f"{self._base_url}/hardware/{payload.get('id')}", + qr_payload=f"{self._base_url}/hardware/{asset_id}", source_app="snipeit", secondary=secondary, ) diff --git a/backend/tests/unit/schemas/test_label_data.py b/backend/tests/unit/schemas/test_label_data.py index 0abbb26..3bd2023 100644 --- a/backend/tests/unit/schemas/test_label_data.py +++ b/backend/tests/unit/schemas/test_label_data.py @@ -1,5 +1,6 @@ import pytest from app.schemas.label_data import LabelData +from pydantic_core import ValidationError def test_label_data_minimal() -> None: @@ -30,15 +31,13 @@ def test_label_data_with_secondary_fields() -> None: def test_label_data_is_frozen() -> None: """LabelData is an immutable value object — mutating fields after construction must fail.""" - import pydantic - data = LabelData( title="t", primary_id="p", qr_payload="q", source_app="snipeit", ) - with pytest.raises(pydantic.ValidationError): + with pytest.raises(ValidationError, match="frozen_instance"): data.title = "different" # type: ignore[misc] diff --git a/backend/tests/unit/services/test_snipeit_client.py b/backend/tests/unit/services/test_snipeit_client.py index d06d3af..3694167 100644 --- a/backend/tests/unit/services/test_snipeit_client.py +++ b/backend/tests/unit/services/test_snipeit_client.py @@ -71,6 +71,24 @@ async def test_lookup_strips_trailing_slash_from_base_url() -> None: assert data.qr_payload == "https://snipe-it.example/hardware/1" +@pytest.mark.asyncio +@respx.mock +async def test_lookup_missing_id_raises_value_error() -> None: + """Snipe-IT response without 'id' field must fail loudly. + + Regression guard: must not silently produce …/hardware/None. + """ + respx.get("https://snipe-it.example/api/v1/hardware/bytag/A-1").mock( + return_value=httpx.Response( + 200, + json={"asset_tag": "A-1", "name": "Broken Asset"}, # no 'id' + ) + ) + client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key") + with pytest.raises(ValueError, match="missing required field 'id'"): + await client.lookup("A-1") + + @pytest.mark.asyncio @respx.mock async def test_lookup_5xx_raises_httpx_error() -> None: From eab3ea5a468b715210764182db435c46b7742410 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Strausmann?= Date: Mon, 11 May 2026 10:13:33 +0000 Subject: [PATCH 3/3] refactor(snipeit): tuple-immutable secondary + URL-encoded asset_tag + auth-header test - LabelData.secondary changed from list[str] to tuple[str, ...] to enforce true immutability under frozen=True (list.append() was still possible before) - Pydantic v2 coerces list inputs to tuple automatically; callers unchanged - _payload_to_label now passes tuple(secondary) explicitly for mypy clarity - urllib.parse.quote(asset_tag, safe="") prevents URL breakage on tags with special characters (/, ?, space, #) - Added TODO(phase6) comment on lookup() for future httpx.AsyncClient pooling - Schema tests: replaced default-distinct identity test with immutability test (tuple.append raises AttributeError); updated secondary == () comparisons - New tests: test_lookup_url_encodes_asset_tag, test_lookup_sends_bearer_auth_header Co-Authored-By: Claude Sonnet 4.6 --- backend/app/schemas/label_data.py | 5 +-- backend/app/services/snipeit_client.py | 8 +++-- backend/tests/unit/schemas/test_label_data.py | 21 +++++++---- .../unit/services/test_snipeit_client.py | 35 +++++++++++++++++-- 4 files changed, 56 insertions(+), 13 deletions(-) diff --git a/backend/app/schemas/label_data.py b/backend/app/schemas/label_data.py index 571194c..45a7805 100644 --- a/backend/app/schemas/label_data.py +++ b/backend/app/schemas/label_data.py @@ -12,7 +12,7 @@ from __future__ import annotations -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict class LabelData(BaseModel): @@ -24,4 +24,5 @@ class LabelData(BaseModel): primary_id: str qr_payload: str source_app: str - secondary: list[str] = Field(default_factory=list) + secondary: tuple[str, ...] = () + """Additional label lines below the primary identifier.""" diff --git a/backend/app/services/snipeit_client.py b/backend/app/services/snipeit_client.py index f73f9a4..0d8bebe 100644 --- a/backend/app/services/snipeit_client.py +++ b/backend/app/services/snipeit_client.py @@ -9,6 +9,7 @@ from __future__ import annotations from typing import Any +from urllib.parse import quote import httpx @@ -41,7 +42,10 @@ def __init__( async def lookup(self, asset_tag: str) -> LabelData: """Return LabelData for `asset_tag`, or raise SnipeITNotFoundError.""" - url = f"{self._base_url}/api/v1/hardware/bytag/{asset_tag}" + # TODO(phase6): inject a shared httpx.AsyncClient for connection pooling + # when this client is consumed by the FastAPI request handler. + encoded_tag = quote(asset_tag, safe="") + url = f"{self._base_url}/api/v1/hardware/bytag/{encoded_tag}" headers = { "Authorization": f"Bearer {self._api_key}", "Accept": "application/json", @@ -71,5 +75,5 @@ def _payload_to_label(self, payload: dict[str, Any], asset_tag: str) -> LabelDat primary_id=str(payload.get("asset_tag") or asset_tag), qr_payload=f"{self._base_url}/hardware/{asset_id}", source_app="snipeit", - secondary=secondary, + secondary=tuple(secondary), ) diff --git a/backend/tests/unit/schemas/test_label_data.py b/backend/tests/unit/schemas/test_label_data.py index 3bd2023..c23d407 100644 --- a/backend/tests/unit/schemas/test_label_data.py +++ b/backend/tests/unit/schemas/test_label_data.py @@ -14,7 +14,7 @@ def test_label_data_minimal() -> None: assert data.primary_id == "ASSET-12345" assert data.qr_payload == "https://snipe-it.example/assets/12345" assert data.source_app == "snipeit" - assert data.secondary == [] + assert data.secondary == () def test_label_data_with_secondary_fields() -> None: @@ -27,6 +27,8 @@ def test_label_data_with_secondary_fields() -> None: ) assert len(data.secondary) == 2 assert data.secondary[0] == "Color: Black" + # The tuple is the actual immutability guarantee — confirm. + assert isinstance(data.secondary, tuple) def test_label_data_is_frozen() -> None: @@ -41,9 +43,14 @@ def test_label_data_is_frozen() -> None: data.title = "different" # type: ignore[misc] -def test_label_data_default_secondary_is_distinct_per_instance() -> None: - """The default empty list must NOT be a shared mutable default.""" - a = LabelData(title="a", primary_id="a", qr_payload="a", source_app="snipeit") - b = LabelData(title="b", primary_id="b", qr_payload="b", source_app="snipeit") - # With frozen=True we cannot append, but we can still verify the lists are distinct objects. - assert a.secondary is not b.secondary +def test_label_data_secondary_is_immutable() -> None: + """A tuple field cannot be mutated in-place — append must raise AttributeError.""" + data = LabelData( + title="t", + primary_id="p", + qr_payload="q", + source_app="snipeit", + secondary=["a"], + ) + with pytest.raises(AttributeError): + data.secondary.append("b") # type: ignore[attr-defined] diff --git a/backend/tests/unit/services/test_snipeit_client.py b/backend/tests/unit/services/test_snipeit_client.py index 3694167..8c0ee61 100644 --- a/backend/tests/unit/services/test_snipeit_client.py +++ b/backend/tests/unit/services/test_snipeit_client.py @@ -26,7 +26,7 @@ async def test_lookup_asset_returns_label_data() -> None: assert data.primary_id == "ASSET-12345" assert data.qr_payload == "https://snipe-it.example/hardware/123" assert data.source_app == "snipeit" - assert data.secondary == ["S/N: C02XYZ"] + assert data.secondary == ("S/N: C02XYZ",) @pytest.mark.asyncio @@ -56,7 +56,7 @@ async def test_lookup_asset_without_serial_has_no_secondary_line() -> None: client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key") data = await client.lookup("A-1") - assert data.secondary == [] + assert data.secondary == () @pytest.mark.asyncio @@ -99,3 +99,34 @@ async def test_lookup_5xx_raises_httpx_error() -> None: client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key") with pytest.raises(httpx.HTTPStatusError): await client.lookup("A-1") + + +@pytest.mark.asyncio +@respx.mock +async def test_lookup_url_encodes_asset_tag() -> None: + """Asset tags with special chars (/, ?, space) must be percent-encoded.""" + respx.get("https://snipe-it.example/api/v1/hardware/bytag/A%2F1%20test").mock( + return_value=httpx.Response( + 200, + json={"id": 1, "asset_tag": "A/1 test", "name": "Thing"}, + ) + ) + client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key") + data = await client.lookup("A/1 test") + assert data.title == "Thing" + + +@pytest.mark.asyncio +@respx.mock +async def test_lookup_sends_bearer_auth_header() -> None: + """lookup() must send Authorization: Bearer … and Accept: application/json.""" + route = respx.get("https://snipe-it.example/api/v1/hardware/bytag/A-1").mock( + return_value=httpx.Response(200, json={"id": 1, "asset_tag": "A-1", "name": "T"}) + ) + client = SnipeITClient(base_url="https://snipe-it.example", api_key="secret-key-42") + await client.lookup("A-1") + + assert route.called + sent_request = route.calls.last.request + assert sent_request.headers["Authorization"] == "Bearer secret-key-42" + assert sent_request.headers["Accept"] == "application/json"