Unit tests for osism/api.py — helper functions, Pydantic models, secret masking

[berendt]

Background

Follow-up to #2192 (foundation) and PR #2193 (pytest + Zuul infrastructure). Part of Tier 7 (#2199). osism/api.py (1161 LOC) implements the OSISM API server (FastAPI). Because of its size the module is split into two sub-issues: this issue covers the module-level helper functions, the Pydantic request/response models, and the inventory-secret masking; the companion issue "Unit tests for osism/api.py — HTTP & WebSocket endpoints (FastAPI TestClient)" (Tier 7, split 2/2) covers the route handlers.

Scope

Add tests/unit/test_api_helpers.py covering _mask_inventory_secrets(), find_device_by_identifier(), process_netbox_webhook() and the Pydantic models in osism/api.py. There is currently no test coverage for osism/api.py at all (only tests/unit/test_smoke.py imports the osism package). Endpoint behavior (HTTP status codes, masking applied in responses) is out of scope here — it belongs to the companion endpoint issue.

Note: importing osism.api has import-time side effects — dictConfig(LogConfig().model_dump()) is executed and osism/services/event_bridge.py:304 instantiates EventBridge(), which attempts a Redis connection (redis.Redis(...).ping()); the failure is caught and only logged, so the import works without Redis.

Test targets

_mask_inventory_secrets() — api.py:35

Pure function, no patching needed. It delegates key classification to osism.tasks.conductor.utils._is_secret_key (matches password/secret substrings case-insensitively and the ironic_osism_ prefix) — use the real implementation.

• Key containing password (also mixed case, e.g. ADMIN_Password) → value replaced by "***"
• Key containing secret → "***"
• Key starting with ironic_osism_ → "***"
• Non-secret key whose string value starts with $ANSIBLE_VAULT; → "***"; also with leading whitespace (" $ANSIBLE_VAULT;...", the code calls .strip() before the prefix check)
• String value containing $ANSIBLE_VAULT; but not at the start → unchanged
• Nested dict under a non-secret key → recursed; secret keys masked at any depth
• Secret key whose value is a dict → masked to "***" as a whole (the secret-key check wins over recursion)
• Non-dict, non-string values (int, list, None, bool) under non-secret keys → passed through unchanged
• List of dicts under a non-secret key → not traversed (lists are not recursed) — pin down the current behavior
• Empty dict → {}
• Non-string key (e.g. 42) → _is_secret_key returns False, value passed through
• Input dict is not mutated (function builds a new dict)

find_device_by_identifier() — api.py:265

Patch utils.nb — the lazy attribute on osism.utils — via monkeypatch.setattr(osism.utils, "nb", fake_nb, raising=False) (raising=False because the attribute only materializes via the module __getattr__ on first access).

• utils.nb is None/falsy → returns None, no NetBox queries made
• nb.dcim.devices.filter(name=identifier) returns devices → first device returned; the cf_inventory_hostname and serial filters are not called
• Name filter empty, filter(cf_inventory_hostname=identifier) hits → that device returned
• Name + custom-field filters empty, filter(serial=identifier) hits → that device returned
• All three filters return empty → returns None
• Filter returns multiple devices → first element of list(devices) returned

process_netbox_webhook() — api.py:588

Patch osism.api.reconciler.run (assert on .delay) and osism.utils.nb (only needed for the interfaces branch). Build real WebhookNetboxData instances (Celery is not involved beyond the patched .delay, no broker needed).

• data["url"] contains "devices", tags include "Managed by OSISM", custom_fields["device_type"] == "server" → reconciler.run.delay() called exactly once
• Managed device with device_type == "switch" → no .delay() call (TODO branch, log only)
• Managed device with custom_fields["device_type"] missing or None → falls back to "node" → no branch matches, no .delay() call
• url contains "interfaces" → device fetched via utils.nb.dcim.devices.get(id=data["device"]["id"]); tags/custom_fields taken from the fetched device; with "Managed by OSISM" in tags and custom_fields["device_type"] set → log-only branch, no .delay(); assert nb.dcim.devices.get called with the right id
• Unknown url (neither devices nor interfaces) → warning logged, early return, no NetBox access, no .delay()
• Tags without "Managed by OSISM" → "Ignoring change for unmanaged device" path, no .delay()
• data without url/name keys → KeyError propagates (the endpoint wraps this into a 500 — covered in the companion issue)

LogConfig — api.py:77

• Defaults: LOGGER_NAME == "osism", LOG_LEVEL == "INFO", version == 1, disable_existing_loggers is False
• model_dump() yields a dict usable by logging.config.dictConfig — contains formatters["default"], handlers["default"] (class logging.StreamHandler, stream ext://sys.stderr) and the loggers osism, api, uvicorn, uvicorn.error, uvicorn.access
• dictConfig(LogConfig().model_dump()) does not raise (this already runs at import time, but make it explicit)

Pydantic models — api.py:54–374

NotificationBaremetal (api.py:54), WebhookNetboxResponse (api.py:63), WebhookNetboxData (api.py:67), DeviceSearchResult (api.py:150), BaremetalNode (api.py:155), BaremetalNodesResponse (api.py:210), BaremetalNodeNetboxInfo (api.py:215), BaremetalNodesNetboxRequest (api.py:228), BaremetalNodesNetboxResponse (api.py:232), BaremetalPort (api.py:238), BaremetalPortsResponse (api.py:247), BaremetalNodeParameters (api.py:252), SinkResponse (api.py:314), HostsResponse (api.py:318), HostvarEntry (api.py:325), HostvarsResponse (api.py:330), HostvarSingleResponse (api.py:336), FactEntry (api.py:342), FactsResponse (api.py:347), FactSingleResponse (api.py:354), SearchResultEntry (api.py:361), SearchResponse (api.py:368). Use parametrized tests; no patching needed.

• NotificationBaremetal: valid payload accepted; message_id coerced to UUID; non-UUID message_id → ValidationError; each missing required field → ValidationError
• WebhookNetboxData: ISO-8601 string for timestamp coerced to datetime.datetime; request_id coerced to UUID; invalid values → ValidationError
• BaremetalNode: instantiable with no arguments (all fields optional); defaults traits == [], properties == {}, extra == {}; default_factory produces independent objects per instance (mutating one instance's traits does not leak into another)
• BaremetalNodeParameters, BaremetalNodeNetboxInfo, BaremetalPort, DeviceSearchResult: instantiable with no/partial arguments, optional fields default to None
• Required-field models (SinkResponse, HostsResponse, HostvarEntry, HostvarsResponse, HostvarSingleResponse, FactEntry, FactsResponse, FactSingleResponse, SearchResultEntry, SearchResponse, BaremetalNodesResponse, BaremetalPortsResponse, BaremetalNodesNetboxRequest, BaremetalNodesNetboxResponse, WebhookNetboxResponse): missing required field → ValidationError; happy-path round trip via model_dump()
• HostvarEntry.value / FactEntry.value / SearchResultEntry.value are Any → accept str, dict, list, None
• BaremetalNodesNetboxResponse: plain dicts as values are coerced to BaremetalNodeNetboxInfo

Mocking hints

• fastapi/pydantic/starlette are runtime dependencies (requirements.txt; CI installs the package itself via pipenv run pip install ., see playbooks/test-unit.yml) — nothing extra to install for this issue. The httpx-based TestClient is only needed in the companion endpoint issue.
• Importing osism.api triggers the EventBridge() Redis-connect attempt described above; the exception is caught internally, tests need no Redis. No patching required, just be aware of the logged error.
• For utils.nb use monkeypatch.setattr(osism.utils, "nb", fake_nb, raising=False); plain mocker.patch("osism.utils.nb", ...) would first call the module-level __getattr__ and try to build a real NetBox connection.
• A minimal valid WebhookNetboxData for process_netbox_webhook tests:

  WebhookNetboxData(
      username="admin",
      data={
          "url": "/api/dcim/devices/1/",
          "name": "sw1",
          "tags": [{"name": "Managed by OSISM"}],
          "custom_fields": {"device_type": "server"},
      },
      snapshots={},
      event="updated",
      timestamp="2026-01-01T00:00:00Z",
      model="device",
      request_id="f8a6e0a4-3c1b-4f53-9c5f-2f4f3b2a1e00",
  )

• For the interfaces branch set data={"url": "/api/dcim/interfaces/5/", "name": "eth0", "device": {"id": 1}} and give the fake device tags=["Managed by OSISM"] plus custom_fields={"device_type": "server"}.
• reconciler.run is a Celery task object; patch the whole attribute (mocker.patch("osism.api.reconciler.run")) and assert on .delay — no broker involved.

Definition of Done

• tests/unit/test_api_helpers.py created
• All listed cases covered
• pytest --cov=osism.api shows all lines of _mask_inventory_secrets, find_device_by_identifier, process_netbox_webhook and the model definitions covered (module-wide ≥ 90 % is reached together with the companion endpoint issue)
• pipenv run pytest tests/unit/test_api_helpers.py passes locally
• flake8, mypy, python-black remain green
• Zuul job python-osism-unit-tests passes

Dependencies

• Part of Meta: Unit test coverage for osism/ #2199 (meta). Blocked by Introduce foundation for unit tests (pytest + Zuul CI integration) #2192 / PR Introduce foundation for unit tests (pytest + Zuul CI) #2193.
• Companion issue: Unit tests for osism/api.py — HTTP & WebSocket endpoints (FastAPI TestClient) (Tier 7, split 2/2).