Unit tests for osism/api.py — HTTP & WebSocket endpoints (FastAPI TestClient)

[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 route handlers (HTTP + WebSocket) exercised through fastapi.testclient.TestClient; the companion issue "Unit tests for osism/api.py — helper functions, Pydantic models, secret masking" (Tier 7, split 1/2) covers _mask_inventory_secrets(), find_device_by_identifier(), process_netbox_webhook() and the models directly.

Dev dependency required: fastapi itself is a runtime dependency (requirements.txt:13; CI installs the package via pipenv run pip install ., see playbooks/test-unit.yml), but TestClient is httpx-based since Starlette 0.21 and httpx is not declared anywhere (not in Pipfile, Pipfile.lock or requirements.txt; import httpx fails in the current pipenv environment). Add httpx to [dev-packages] in the Pipfile (pinned like the other dev packages, e.g. httpx = "==0.28.1") and relock as part of this issue.

Scope

Add tests/unit/test_api_endpoints.py covering all HTTP endpoints and the WebSocket endpoint of osism/api.py via TestClient(app). There is currently no test coverage for osism/api.py. Direct unit tests for the helper functions and models are out of scope (companion issue) — but response-level effects such as secret masking in hostvars responses belong here. This issue is large (20 routes); it may be split further during implementation, as was done for Tier 3 issues.

Test targets

Health endpoints — root() api.py:293, v1() api.py:299, events_info() api.py:305

No patching needed.

• GET / → 200, {"result": "ok"}
• GET /v1 → 200, {"result": "ok"}
• GET /v1/events → 200, body contains websocket_endpoint == "/v1/events/openstack"

Telemetry sinks — write_sink_meters() api.py:378, write_sink_events() api.py:396

No patching needed (handlers only parse JSON and log).

• POST /v1/meters/sink with a JSON list → 200, {"result": "ok"}
• POST /v1/meters/sink with a JSON object → 200
• POST /v1/meters/sink with a malformed JSON body (content=b"{not json") → 500, detail "Failed to process meters data"
• Same three cases for POST /v1/events/sink (detail "Failed to process events data")

get_baremetal_nodes_list() — api.py:416

Patch osism.api.openstack.get_baremetal_nodes.

• Returns two node dicts → 200, count == 2, fields mapped into BaremetalNode entries
• Returns [] → 200, count == 0
• Raises → 500, detail starts with "Failed to retrieve baremetal nodes:" and includes the exception text

NetBox info — get_baremetal_node_netbox_info() api.py:443, get_baremetal_nodes_netbox_info() api.py:464

Patch osism.api.openstack.get_baremetal_node_netbox_info / osism.api.openstack.get_baremetal_nodes_netbox_info.

• GET /v1/baremetal/nodes/{node_name}/netbox happy path → 200, device_role/primary_ip4/primary_ip6/netbox_url mapped; task function called with node_name
• Single-node lookup raises → 500, detail includes the node name
• POST /v1/baremetal/nodes/netbox with {"node_names": ["n1", "n2"]} → 200, nodes keyed by name; task function called with the list
• POST body missing node_names → 422 (FastAPI validation, handler never called)
• Bulk lookup raises → 500, detail "Failed to retrieve NetBox info: ..."

get_baremetal_node_ports() api.py:488, get_baremetal_node_parameters() api.py:507

Patch osism.api.openstack.get_baremetal_node_ports / osism.api.openstack.get_baremetal_node_parameters.

• Ports happy path → 200, count matches, port fields mapped; raises → 500 with node UUID in detail
• Parameters happy path (kernel_append_params, netplan_parameters, frr_parameters) → 200; raises → 500 with node UUID in detail

notifications_baremetal() — api.py:526

Patch osism.api.baremetal_events.get_handler (method on the module-level BaremetalEvents instance) via mocker.patch.object.

• Valid notification body → 204 with empty body; get_handler called with event_type, returned handler called with payload
• Handler raises → 500, detail "Failed to process baremetal notification"
• message_id not a UUID → 422
• Optional: without patching, an unknown event_type (e.g. "foo.bar.baz.qux") with payload={"ironic_object.data": {"name": "n1"}} exercises the real default handler → 204

sonic_ztp_complete() — api.py:549

Patch osism.utils.nb (see hints) and osism.api.find_device_by_identifier (already unit-tested in the companion issue).

• utils.nb is None → 503, detail "NetBox is not enabled"
• Device found → 200, {"result": "ok", "device": <name>}; device.custom_fields["provision_state"] set to "active" and device.save() called
• No device found → 404, detail contains the identifier (the inner HTTPException is re-raised unchanged through the generic handler)
• device.save() raises → 500, detail "Failed to complete ZTP process"

webhook() — api.py:704

Patch osism.utils.nb and osism.api.process_netbox_webhook (its branch logic is unit-tested in the companion issue).

• utils.nb is None → 503, detail "NetBox webhook processing is not enabled"
• Valid body → 200, {"result": "ok"}; process_netbox_webhook called with the parsed WebhookNetboxData
• process_netbox_webhook raises → 500, detail "Failed to process NetBox webhook"
• Body missing a required field (e.g. request_id) → 422
• Note: content_length: int = Header(...) is required, but httpx/TestClient sets Content-Length automatically for JSON bodies; x_hook_signature is optional and currently unvalidated

websocket_openstack_events() — api.py:632

Patch osism.api.websocket_manager with a stub whose async connect accepts the socket (see hints) — this avoids starting the real broadcaster background task.

• client.websocket_connect("/v1/events/openstack") → connect awaited with the WebSocket
• Send {"action": "set_filters", "event_filters": [...], "node_filters": [...], "service_filters": [...]} → update_filters awaited with exactly these values; acknowledgment frame received: {"type": "filter_update", "status": "success", ...} echoing the filters
• set_filters with only some keys → missing ones passed as None and echoed as null in the ack
• Send invalid JSON text → no ack, no update_filters call, connection stays usable (a subsequent valid set_filters still gets an ack)
• Send valid JSON with action != "set_filters" → no ack, no update_filters call
• update_filters raising → inner except Exception logs and the loop continues (connection still open)
• Closing the client context → disconnect awaited (the finally block)

get_inventory_hosts() — api.py:736

Patch osism.api.get_inventory_path, osism.api.os.path.exists, osism.api.subprocess.run, osism.api.get_hosts_from_inventory (the inventory helpers themselves are covered by #2194).

• Inventory file missing (os.path.exists → False) → 503, detail contains the path
• subprocess.run result with returncode != 0 → 500, "Failed to load Ansible inventory"
• Happy path → 200, hosts + count; assert the command is ["ansible-inventory", "-i", <path>, "--list"]
• ?limit=compute* → command extended with ["--limit", "compute*"]
• subprocess.TimeoutExpired → 504
• Non-JSON stdout → 500, "Failed to parse Ansible inventory"
• get_hosts_from_inventory raises → 500, generic "Failed to retrieve hosts: ..."

Hostvars — get_host_hostvars() api.py:803, get_host_hostvar() api.py:866

Patch osism.api.get_inventory_path and osism.api.subprocess.run.

• Happy path → 200; variables sorted by name, count correct; assert get_inventory_path called with prefer_minified=False and the command uses --host <host>
• Secret masking applied in the response: stdout containing {"ansible_password": "x", "vaulted": "$ANSIBLE_VAULT;1.1;AES256\n..."} → both values are "***"
• returncode != 0 with "Could not match" in stderr → 404, "Host '<host>' not found in inventory"
• returncode != 0 with "Unable to parse" in stderr → 404
• returncode != 0 with other stderr → 500
• subprocess.TimeoutExpired → 504
• Non-JSON stdout → 500
• Single-variable endpoint: variable present → 200 with host/name/value; secret variable (e.g. database_password) → value "***" (masking happens after the membership check)
• Variable not in hostvars → 404, "Variable '<variable>' not found for host '<host>'"
• Host not found / timeout / bad JSON → 404 / 504 / 500 as above

Facts — get_host_facts() api.py:932, get_host_fact() api.py:971

Patch utils.redis — the lazy attribute on osism.utils — via monkeypatch.setattr(osism.utils, "redis", fake_redis, raising=False).

• redis.get returns None → 404, "No facts found in cache for host '<host>'"; assert the (separator-less) cache key f"ansible_facts{host}" is used
• Returns JSON bytes → 200; facts sorted by name, count correct, from_cache is True
• Returns invalid JSON → 500, "Failed to parse facts for <host>"
• redis.get raises → 500, generic "Failed to retrieve facts: ..."
• Single-fact endpoint: fact present → 200 with value and from_cache is True; fact missing → 404 "Fact '<fact>' not found..."; no cache entry → 404; invalid JSON → 500

search_inventory() — api.py:1010

Patch osism.api.get_inventory_path, osism.api.os.path.exists, osism.api.subprocess.run, osism.api.get_hosts_from_inventory, and builtins.open (facts are read from /cache/facts/<host>).

• Missing name_pattern query parameter → 422
• Invalid name_pattern regex ("[") → 400, "Invalid name_pattern regex: ..."
• Invalid host_pattern regex → 400
• source=bogus → 400, "source must be 'hostvars', 'facts', or omitted for both"
• Inventory file missing → 503
• Inventory --list call with returncode != 0 → 500
• Hostvars search happy path: --list then one --host call per host (subprocess.run side_effect list); only names matching name_pattern (case-insensitive) returned with source == "hostvars"; secret values masked ("***") before matching
• host_pattern filters the host list; hosts_searched reflects the filtered count
• source=facts: no per-host ansible-inventory --host calls; facts file read via open, matching fact names returned with source == "facts"; facts file absent → host contributes nothing
• limit honored: more matches than limit → exactly limit results, host loop breaks early
• Per-host subprocess.TimeoutExpired or bad hostvars JSON → warning, host skipped, search continues
• Facts file with invalid JSON / IOError on open → warning, search continues
• Response query dict echoes name_pattern, host_pattern, source, limit
• Two get_inventory_path calls: prefer_minified=False for hostvars, default (minified) for the host list

Mocking hints

• Add httpx to [dev-packages] in the Pipfile first — from fastapi.testclient import TestClient raises without it (Starlette ≥ 0.21 TestClient is httpx-based).
• Module-scoped fixture:

  from fastapi.testclient import TestClient
  from osism.api import app
  
  @pytest.fixture(scope="module")
  def client():
      return TestClient(app)

• Importing osism.api instantiates EventBridge() (osism/services/event_bridge.py:304), which attempts redis.Redis(...).ping() against host redis; the exception is caught and logged, so no Redis is needed — just expect the log line. The import also runs dictConfig.
• osism.api does from osism.tasks import reconciler, openstack — attribute lookups happen on the module objects, so mocker.patch("osism.api.openstack.get_baremetal_nodes") patches the same object as osism.tasks.openstack.get_baremetal_nodes. Tests call the patched task functions synchronously via the endpoint; no Celery broker is involved.
• utils.nb and utils.redis are lazy attributes materialized by osism.utils.__getattr__; use monkeypatch.setattr(osism.utils, "nb", fake, raising=False) (a plain mocker.patch("osism.utils.nb") would trigger a real connection attempt during the initial getattr).
• Fake subprocess.run results: subprocess.CompletedProcess(args=[...], returncode=0, stdout=json.dumps(payload), stderr="") (the code passes text=True, so stdout is a str). Timeouts: side_effect=subprocess.TimeoutExpired(cmd="ansible-inventory", timeout=30).
• WebSocket stub — the real websocket_manager.connect calls websocket.accept(); a bare AsyncMock would never accept and client.websocket_connect would fail. Use:

  fake_manager = mocker.MagicMock()
  async def fake_connect(ws):
      await ws.accept()
  fake_manager.connect = mocker.AsyncMock(side_effect=fake_connect)
  fake_manager.update_filters = mocker.AsyncMock()
  fake_manager.disconnect = mocker.AsyncMock()
  mocker.patch("osism.api.websocket_manager", fake_manager)

  Then with client.websocket_connect("/v1/events/openstack") as ws: ws.send_text(...); ws.receive_json().
• For the facts branch of search_inventory, patch osism.api.os.path.exists with a side_effect keyed on the path (inventory path → True, /cache/facts/<host> → as needed) and builtins.open with mock_open(read_data=json.dumps(facts)).
• A minimal valid webhook body (JSON) for POST /v1/webhook/netbox:

  {
    "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"
  }

Definition of Done

• httpx added to [dev-packages] in Pipfile (and Pipfile.lock updated)
• tests/unit/test_api_endpoints.py created
• All listed cases covered
• pytest --cov=osism.api shows ≥ 90 % (together with the companion helper issue)
• pipenv run pytest tests/unit/test_api_endpoints.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 — helper functions, Pydantic models, secret masking (Tier 7, split 1/2).