Integration test: WebSocket /v1/events/openstack (FastAPI TestClient)

[berendt]

Part of #2400. Follow-up to #2368. Requires the shared httpx prerequisite (see the meta issue).

What

Cover the WebSocket endpoint /v1/events/openstack (osism/api.py:631) and the in-process WebSocketManager (osism/services/websocket_manager.py): connect, set filters, broadcast an event, receive it. The manager broadcasts in-process via an asyncio queue, so this needs no extra service beyond the TestClient.

Scope

• TestClient.websocket_connect("/v1/events/openstack") accepts the connection.
• Sending a filter message updates the connection's filters and is acknowledged.
• An event pushed through the websocket manager is delivered to a connected client; an event that does not match the client's filters is not.
• Disconnect cleans up the connection (manager connection count drops).

Notes / Where

New file tests/integration/test_api_websocket.py. Strictly this path is in-process and does not require Redis, but it shares the FastAPI / httpx setup with the facts-endpoint test and belongs in the same Tier 2 batch. Mark it integration for consistency. Add httpx to [dev-packages] in Pipfile if not already done for the facts test.

[berendt]

Implementation Plan (issue #2407)

Summary

Add one new test module, tests/integration/test_api_websocket.py, that drives the in-process WebSocket endpoint /v1/events/openstack through fastapi.testclient.TestClient and the global websocket_manager singleton. The test connects, sets filters (asserting the ack), pushes events through the manager's asyncio queue and asserts matching delivery / non-matching filtering, and asserts the connection count drops on disconnect. No production code changes; httpx is already in [dev-packages]. The only non-obvious constraint is that the manager's module-global asyncio primitives bind to one event loop, so the TestClient fixture must be module-scoped and all pushes must run on the app loop via client.portal.call(...).

Ground-Truth Notes

• osism/api.py:631 — confirmed. @app.websocket("/v1/events/openstack") async def websocket_openstack_events(websocket): calls websocket_manager.connect, loops on receive_text(), on {"action": "set_filters", ...} calls update_filters and always sends ack {"type": "filter_update", "status": "success", "event_filters", "node_filters", "service_filters"} (api.py:669-676); finally: await websocket_manager.disconnect(websocket) (api.py:695).
• osism/api.py:30 — confirmed from osism.services.websocket_manager import websocket_manager. The handler uses the same global singleton a test imports, so manipulating it from the test reaches the handler's manager. api.py:147 event_bridge.set_websocket_manager(websocket_manager) runs at import → importing osism.api touches Redis at import time (hence lazy import in the fixture, mirroring the facts test).
• osism/services/websocket_manager.py — confirmed. EventMessage(event_type, source, data, node_name=None) with to_dict() → keys id, timestamp, event_type, source, node_name, data. WebSocketManager.connect accepts, registers in self.connections (a dict) under self._lock, and starts the _broadcast_events task on first connect. add_event(event) is await self.event_queue.put(event). matches_filters (websocket_manager.py:57): with only event_filters set, an event whose event_type is in the list matches; otherwise not. Global instance websocket_manager at websocket_manager.py:271.
• tests/integration/test_api_facts.py (merged PR Integration test: API facts endpoints (FastAPI TestClient + Redis) #2411) — confirmed source of truth for the shared setup: function-scoped local client fixture that lazily imports osism.api and wraps TestClient(api.app) in a with block; pytestmark = pytest.mark.integration; uuid-suffixed keys; per-test cleanup.
• Pipfile:45 — confirmed httpx = "==0.28.1" already in [dev-packages]. No Pipfile change needed (PR Integration test: API facts endpoints (FastAPI TestClient + Redis) #2411 already did it).
• setup.cfg:178-185 — confirmed [tool:pytest]: testpaths = tests/unit (so plain pytest excludes integration), addopts = -ra --strict-markers, and the integration marker is registered. No marker/config change needed. No pytest-timeout plugin is configured.
• tests/integration/conftest.py — confirmed: pytest_collection_modifyitems skips integration-marked tests when Redis is unreachable (or pytest.exit when OSISM_REQUIRE_REDIS is set). The new module needs no conftest change.
• Installed: starlette 1.0.0, fastapi 0.136.1, anyio 4.7.0, Python 3.12. WebSocketTestSession (starlette testclient.py) exposes self.portal and, while inside with TestClient(app) as client:, client.portal is a shared BlockingPortal whose loop runs the app; WebSocketTestSession.__exit__ calls fut.result(), so the handler's finally: disconnect() is guaranteed complete before the with ws: block returns (makes the disconnect-count assertion deterministic, no sleep).
• playbooks/test-integration.yml — confirmed CI runs pipenv run pytest tests/integration with REDIS_HOST=localhost and OSISM_REQUIRE_REDIS=1.
• No root README.md and no docs/ page enumerates the integration test modules; PR Integration test: API facts endpoints (FastAPI TestClient + Redis) #2411 added no docs. → Documentation Plan is "none".

Change Set

• tests/integration/test_api_websocket.py — create. New module, header # SPDX-License-Identifier: Apache-2.0, a module docstring, pytestmark = pytest.mark.integration. Black-formatted (matches the suite). Contents:
  • Top-level imports: import uuid, import pytest, and from osism.services.websocket_manager import EventMessage, websocket_manager. (These are Redis-free and safe at collection time. Do not import osism.api at module top — keep it lazy in the fixture.)
  • Module-scoped client fixture (this is the critical, non-obvious requirement — see Risks):

    @pytest.fixture(scope="module")
    def client():
        """A module-scoped TestClient bound to the FastAPI app.
    
        ``osism.api`` is imported lazily because importing it wires the event
        bridge to Redis at module load -- safe only in the integration env
        where Redis is up. The fixture is module-scoped on purpose: the global
        ``websocket_manager`` owns module-level ``asyncio`` primitives
        (``event_queue``, ``_lock``) that bind to the first event loop that
        touches them and raise "bound to a different event loop" on any other.
        Sharing one TestClient (one loop) across this module keeps them valid.
        """
        from fastapi.testclient import TestClient
        from osism import api
    
        with TestClient(api.app) as test_client:
            yield test_client

  • Helper rationale baked into the tests: push events onto the in-process queue with client.portal.call(websocket_manager.add_event, event) so add_event runs on the app loop (the queue is loop-bound; put_nowait from the test thread would be unsafe). Receive with ws.receive_json() (blocks until the broadcaster sends — no sleep). Assert connection counts relative to a captured baseline (before/before + 1/before) so the test is independent of any other connections.

Commit Sequence

1. Add integration test for OpenStack events WebSocket endpoint — adds tests/integration/test_api_websocket.py; covers connect/accept, filter-ack, matching delivery, non-matching filtering, and disconnect cleanup of the in-process websocket_manager via the FastAPI TestClient. (Single commit; no other files change. httpx and the integration marker are already in place.)

Test Plan

All five functions live in tests/integration/test_api_websocket.py and use the module-scoped client fixture. Event-type filters use distinct service prefixes so matching/non-matching is unambiguous.

• test_websocket_connect_is_accepted(client) — with client.websocket_connect("/v1/events/openstack") as ws: enters without raising WebSocketDisconnect/WebSocketDenialResponse. Locks: the endpoint accepts the upgrade. (Acceptance: connection accepted.)
• test_set_filters_is_acknowledged(client) — inside the with ws:, ws.send_json({"action": "set_filters", "event_filters": ["compute.instance.create.end"], "node_filters": ["server-01"], "service_filters": ["compute"]}); ack = ws.receive_json(); assert ack["type"] == "filter_update", ack["status"] == "success", and the three filter lists are echoed back verbatim. Locks: filter message is processed and acknowledged. (Acceptance: filters set + acknowledged.)
• test_matching_event_is_delivered(client) — set event_filters=["compute.instance.create.end"] and consume the ack; build event = EventMessage("compute.instance.create.end", "openstack", {"server": str(uuid.uuid4())}); client.portal.call(websocket_manager.add_event, event); received = ws.receive_json(); assert received == event.to_dict() (full-object equality: id/timestamp/event_type/source/node_name/data). Locks: a queued event that matches the filter is delivered. (Acceptance: matching event delivered.)
• test_non_matching_event_is_filtered_out(client) — set event_filters=["compute.instance.create.end"] and consume the ack; push a non-matching EventMessage("baremetal.node.power_set.end", "openstack", {...}) then a matching sentinel EventMessage("compute.instance.create.end", "openstack", {...}), both via client.portal.call(...); received = ws.receive_json(); assert received["id"] == sentinel.id and received["event_type"] == "compute.instance.create.end". Because the broadcaster drains the queue FIFO and skips the non-matching event, the first (and only) message received is the sentinel — a deterministic way to prove the non-matching event was not delivered without relying on a timeout/absence check. Locks: non-matching event is filtered out. (Acceptance: non-matching event not delivered.)
• test_disconnect_drops_connection_count(client) — before = len(websocket_manager.connections); with client.websocket_connect(...) as ws: send a set_filters message and consume the ack (forces the handler past connect()'s registration, defeating the accept-vs-register micro-race), then assert len(websocket_manager.connections) == before + 1; after the with block, assert len(websocket_manager.connections) == before (guaranteed by fut.result() in __exit__). Locks: disconnect runs the finally cleanup and removes the connection. (Acceptance: disconnect cleans up; count drops.)

Documentation Plan

• none — test-only change with no user-visible behavior; consistent with the merged sibling PR Integration test: API facts endpoints (FastAPI TestClient + Redis) #2411, which added no docs. The integration marker is already documented in setup.cfg:185.

Verification Commands

Per project practice these run in the python-osism-integration-tests Zuul job (playbooks/test-integration.yml); local runs need a reachable Redis.

• CI-equivalent (real run, needs Redis): REDIS_HOST=localhost OSISM_REQUIRE_REDIS=1 pipenv run pytest tests/integration/test_api_websocket.py
• Whole Tier-2 batch as CI runs it: REDIS_HOST=localhost OSISM_REQUIRE_REDIS=1 pipenv run pytest tests/integration
• Local sanity without Redis (collection + skip path; must exit 0 with tests skipped, proving imports/marker are valid): pipenv run pytest tests/integration/test_api_websocket.py
• Lint: pipenv run flake8 tests/integration/test_api_websocket.py (.flake8, max line length 200).

Risks & Open Questions

• Fixture scope is load-bearing, not cosmetic. Copying the facts test's function-scoped client would make the first websocket test pass and the second fail with RuntimeError: ... is bound to a different event loop, because websocket_manager.event_queue/_lock are module-globals that bind to the first loop. The plan specifies scope="module" to keep one loop for the whole module. Documented inline in the fixture docstring so the next maintainer doesn't "simplify" it back.
• No per-test timeout. The suite has no pytest-timeout. A real regression where a matching event is never delivered would make ws.receive_json() hang until the Zuul job timeout rather than failing fast. Mitigation kept in-design: all waits are deterministic (blocking receive returns the moment the broadcaster sends; the negative case uses the sentinel pattern, never an absence/sleep). Adding a bounded receive (e.g. wrapping via client.portal.call with anyio.fail_after) is possible but is extra machinery the rest of the suite doesn't use; not planned unless the project later adopts pytest-timeout.
• Stray events. In the isolated CI Redis nothing publishes to osism:events, so the event-bridge subscriber injects no events into the connection. The exact-id assertions would fail loudly (not false-pass) if an unexpected event ever arrived, which is the correct failure mode.
• Scope is one new test file with no production or config changes — well within the issue's blast radius.

Status

STATUS: PLAN_READY

---

Implementation plan generated by planwerk-review a24f4d9 implement with Claude

[berendt]

Implementation Report (issue #2407)

Acceptance Criteria

• TestClient.websocket_connect("/v1/events/openstack") accepts the connection.
  • Status: satisfied
  • Evidence: tests/integration/test_api_websocket.py:43 test_websocket_connect_is_accepted — entering the with client.websocket_connect(...) block without a WebSocketDisconnect/WebSocketDenialResponse proves the upgrade was accepted (starlette's __enter__ raises on a close/denial response).
• Sending a filter message updates the connection's filters and is acknowledged.
  • Status: satisfied
  • Evidence: tests/integration/test_api_websocket.py:49 test_set_filters_is_acknowledged — asserts the ack type/status and that all three filter lists are echoed back verbatim (the handler at osism/api.py:668-676 only sends this ack after update_filters).
• An event pushed through the websocket manager is delivered to a connected client; an event that does not match the client's filters is not.
  • Status: satisfied
  • Evidence: tests/integration/test_api_websocket.py:71 test_matching_event_is_delivered (full-object equality with event.to_dict()) and tests/integration/test_api_websocket.py:93 test_non_matching_event_is_filtered_out — the edge/negative path: a non-matching event is queued ahead of a matching sentinel, and the first message received is the sentinel, proving the non-matching event was dropped (no timeout/absence check).
• Disconnect cleans up the connection (manager connection count drops).
  • Status: satisfied
  • Evidence: tests/integration/test_api_websocket.py:124 test_disconnect_drops_connection_count — asserts before + 1 inside the block and before after; deterministic because starlette's WebSocketTestSession.__exit__ blocks on fut.result() until the handler's finally: disconnect() (osism/api.py:695) has run.

Commits

• a12660f Add integration test for OpenStack events WebSocket endpoint

Local verification

• pipenv run flake8 tests/integration/test_api_websocket.py — pass
• pipenv run black --check tests/integration/test_api_websocket.py — pass
• pipenv run python -m py_compile tests/integration/test_api_websocket.py — pass
• pipenv run python -c "from osism.services.websocket_manager import EventMessage, websocket_manager; ..." (top-level import + to_dict() key check) — pass
• pipenv run pytest tests/integration/test_api_websocket.py — skipped: per project memory (don't run tests locally) and because the integration suite requires a reachable Redis plus httpx (declared in Pipfile but not installed in this local venv); CI runs it via playbooks/test-integration.yml with REDIS_HOST=localhost OSISM_REQUIRE_REDIS=1.

Branch

• implement/issue-2407-websocket-integration-test (committed, not pushed — the finalize step opens the PR)

Deviations from the issue

• No Pipfile change: httpx = "==0.28.1" is already in [dev-packages] (Pipfile:45, added with the sibling facts test in Integration test: API facts endpoints (FastAPI TestClient + Redis) #2411). The issue scoped this as conditional ("if not already done"), so no change was needed.

Status

STATUS: DONE

---

Implementation report generated by planwerk-review a24f4d9 implement with Claude

[berendt]

Review Report

Resolved

• Missing receive timeout — a WebSocket regression hangs CI indefinitely — Root cause: four of the five tests block on ws.receive_json(), which bottoms out in Starlette's untimed WebSocketTestSession.receive() → queue.get(); with no pytest-timeout safety net, a regression that stops emitting the ack/matching event would hang until the outer CI wall-clock killed the worker. Fix (module-scoped, minimal-invasive per the finding's alternative): added pytest-timeout==2.4.0 to [dev-packages] (Pipfile) with a hand-locked, hash-verified entry in Pipfile.lock (recomputed _meta.hash.sha256 so pipenv install --dev --deploy still verifies), and changed pytestmark = pytest.mark.integration to pytestmark = [pytest.mark.integration, pytest.mark.timeout(30)] so every test in the module turns an indefinite hang into a 30-second failure at the exact receive_json() call. pytest-timeout registers the timeout marker via pytest_configure, so it satisfies the existing --strict-markers. (folded into 35d94ea Add integration test for OpenStack events WebSocket endpoint)

Skipped

• Tests share the process-global websocket_manager singleton [INFO] — Skipped the optional extra-isolation fixture by design. The finding itself states the cascade only triggers when a test is killed inside the websocket_connect block, and that "the only realistic way to leak a connection" is the hang from finding Add loadbalancer to kolla #1 — which is now resolved. The teardown fixture is framed as "if extra isolation is desired"; adding it would be speculative, untriggerable defensive code (YAGNI / minimal-invasive), so it is intentionally not added. Finding Add cgit playbook #2 is resolved transitively by the timeout fix.

Diff summary

• Files: Pipfile, Pipfile.lock, tests/integration/test_api_websocket.py
• Approx lines added/removed: +17 / -2

Status

Verification: python3 -m py_compile tests/integration/test_api_websocket.py → OK; Pipfile↔lock --deploy hash check → PASS; Pipfile.lock parses as valid JSON with a correct hash-pinned pytest-timeout entry. The pytest suite itself was not run locally — the integration tests need a running Redis and the pipenv venv (the system interpreter lacks fastapi/starlette/pytest), and per the project convention these run in CI. The marker/strict-markers interaction is verified by reasoning, not execution.

STATUS: DONE

---

Review report generated by planwerk-review a24f4d9 implement with Claude