fix: override head method for sse endpoint to make it actually working

This commit is contained in:
2026-07-06 14:44:29 +03:00
parent 78288befb8
commit a3b6372c11
2 changed files with 101 additions and 17 deletions
+41 -17
View File
@@ -19,7 +19,7 @@
#
import json
from aiohttp.web import HTTPBadRequest, Request, StreamResponse
from aiohttp.web import HTTPBadRequest, Request, Response, StreamResponse
from aiohttp_sse import EventSourceResponse, sse_response
from asyncio import Queue, QueueShutDown, wait_for
from typing import ClassVar
@@ -66,16 +66,13 @@ class EventBusView(BaseView):
return await BaseView.get_permission(request)
permission = UserAccess.Full
event_filter = request.query.getall("event", []) if request.query is not None else []
try:
topics = cls(request)._topics()
except HTTPBadRequest:
topics = None
if event_filter:
try:
topics = {EventType(event) for event in event_filter}
except ValueError:
pass
else:
if topics.issubset(cls.READ_EVENTS):
permission = UserAccess.Read
if topics is not None and set(topics).issubset(cls.READ_EVENTS):
permission = UserAccess.Read
return permission
@@ -98,6 +95,24 @@ class EventBusView(BaseView):
await response.send(json.dumps(data), event=event_type)
def _topics(self) -> list[EventType] | None:
"""
parse event filter from request query
Returns:
list[EventType] | None: event filter if any
Raises:
HTTPBadRequest: if invalid event type is supplied
"""
if self.request.query is None:
return None
try:
return [EventType(event) for event in self.request.query.getall("event", [])] or None
except ValueError as ex:
raise HTTPBadRequest(reason=str(ex))
@apidocs(
tags=["Audit log"],
summary="Live updates",
@@ -116,14 +131,8 @@ class EventBusView(BaseView):
Returns:
StreamResponse: 200 with streaming updates
Raises:
HTTPBadRequest: if invalid event type is supplied
"""
try:
topics = [EventType(event) for event in self.request.query.getall("event", [])] or None
except ValueError as ex:
raise HTTPBadRequest(reason=str(ex))
topics = self._topics()
object_id = self.request.query.get("object_id")
event_bus = self.service().event_bus
@@ -138,3 +147,18 @@ class EventBusView(BaseView):
await event_bus.unsubscribe(subscription_id)
return response
async def head(self) -> StreamResponse:
"""
HEAD method implementation based on the result of GET method
Returns:
StreamResponse: generated response for the request
"""
self._topics()
self.service()
return Response(headers={
"Cache-Control": "no-cache",
"Content-Type": "text/event-stream",
})
@@ -2,6 +2,7 @@ import asyncio
import pytest
from aiohttp.test_utils import TestClient
from aiohttp.web import HTTPBadRequest
from asyncio import Queue
from multidict import MultiDict
from pytest_mock import MockerFixture
@@ -110,6 +111,37 @@ async def test_run_timeout() -> None:
await EventBusView._run(response, queue)
def test_topics() -> None:
"""
must parse event filters
"""
request = pytest.helpers.request("", "", "GET", params=MultiDict([
("event", EventType.PackageUpdated),
("event", EventType.PackageRemoved),
]))
assert EventBusView(request)._topics() == [EventType.PackageUpdated, EventType.PackageRemoved]
def test_topics_empty() -> None:
"""
must return None for missing event filters
"""
request = pytest.helpers.request("", "", "GET", params=MultiDict())
assert EventBusView(request)._topics() is None
def test_topics_invalid() -> None:
"""
must raise bad request for invalid event filters
"""
request = pytest.helpers.request("", "", "GET", params=MultiDict(event="invalid"))
with pytest.raises(HTTPBadRequest):
EventBusView(request)._topics()
async def test_get(client: TestClient, package_ahriman: Package) -> None:
"""
must stream events via SSE
@@ -192,3 +224,31 @@ async def test_get_connection_reset(client: TestClient, mocker: MockerFixture) -
mocker.patch.object(EventBusView, "_run", side_effect=ConnectionResetError)
response = await client.get("/api/v1/events/stream")
assert response.status == 200
async def test_head(client: TestClient) -> None:
"""
must check stream availability without opening SSE stream
"""
response = await client.head("/api/v1/events/stream", params={"event": EventType.PackageUpdated})
assert response.status == 200
assert response.headers["Content-Type"] == "text/event-stream"
assert not await response.text()
async def test_head_bad_request(client: TestClient) -> None:
"""
must return bad request for invalid event type
"""
response = await client.head("/api/v1/events/stream", params={"event": "invalid"})
assert response.status == 400
assert not await response.text()
async def test_head_not_found(client: TestClient) -> None:
"""
must return not found for unknown repository
"""
response = await client.head("/api/v1/events/stream", params={"architecture": "unknown", "repository": "unknown"})
assert response.status == 404
assert not await response.text()