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..45a7805 --- /dev/null +++ b/backend/app/schemas/label_data.py @@ -0,0 +1,28 @@ +"""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 + + +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: 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 new file mode 100644 index 0000000..0d8bebe --- /dev/null +++ b/backend/app/services/snipeit_client.py @@ -0,0 +1,79 @@ +"""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 +from urllib.parse import quote + +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.""" + # 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", + } + 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") + # 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/{asset_id}", + source_app="snipeit", + secondary=tuple(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..c23d407 --- /dev/null +++ b/backend/tests/unit/schemas/test_label_data.py @@ -0,0 +1,56 @@ +import pytest +from app.schemas.label_data import LabelData +from pydantic_core import ValidationError + + +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" + # The tuple is the actual immutability guarantee — confirm. + assert isinstance(data.secondary, tuple) + + +def test_label_data_is_frozen() -> None: + """LabelData is an immutable value object — mutating fields after construction must fail.""" + data = LabelData( + title="t", + primary_id="p", + qr_payload="q", + source_app="snipeit", + ) + with pytest.raises(ValidationError, match="frozen_instance"): + data.title = "different" # type: ignore[misc] + + +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 new file mode 100644 index 0000000..8c0ee61 --- /dev/null +++ b/backend/tests/unit/services/test_snipeit_client.py @@ -0,0 +1,132 @@ +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_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: + """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") + + +@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"