diff --git a/src/ahriman/web/schemas/__init__.py b/src/ahriman/web/schemas/__init__.py
index e5e5f2c5..e9de4369 100644
--- a/src/ahriman/web/schemas/__init__.py
+++ b/src/ahriman/web/schemas/__init__.py
@@ -28,6 +28,7 @@ from ahriman.web.schemas.configuration_schema import ConfigurationSchema
from ahriman.web.schemas.counters_schema import CountersSchema
from ahriman.web.schemas.dependencies_schema import DependenciesSchema
from ahriman.web.schemas.error_schema import ErrorSchema
+from ahriman.web.schemas.event_bus_filter_schema import EventBusFilterSchema
from ahriman.web.schemas.event_schema import EventSchema
from ahriman.web.schemas.event_search_schema import EventSearchSchema
from ahriman.web.schemas.file_schema import FileSchema
@@ -61,6 +62,7 @@ from ahriman.web.schemas.repository_id_schema import RepositoryIdSchema
from ahriman.web.schemas.repository_stats_schema import RepositoryStatsSchema
from ahriman.web.schemas.rollback_schema import RollbackSchema
from ahriman.web.schemas.search_schema import SearchSchema
+from ahriman.web.schemas.sse_schema import SSESchema
from ahriman.web.schemas.status_schema import StatusSchema
from ahriman.web.schemas.update_flags_schema import UpdateFlagsSchema
from ahriman.web.schemas.worker_schema import WorkerSchema
diff --git a/src/ahriman/web/schemas/event_bus_filter_schema.py b/src/ahriman/web/schemas/event_bus_filter_schema.py
new file mode 100644
index 00000000..b7eb840e
--- /dev/null
+++ b/src/ahriman/web/schemas/event_bus_filter_schema.py
@@ -0,0 +1,33 @@
+#
+# Copyright (c) 2021-2026 ahriman team.
+#
+# This file is part of ahriman
+# (see https://github.com/arcan1s/ahriman).
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see .
+#
+from ahriman.models.event import EventType
+from ahriman.web.apispec import fields
+from ahriman.web.schemas.repository_id_schema import RepositoryIdSchema
+
+
+class EventBusFilterSchema(RepositoryIdSchema):
+ """
+ request event bus filter schema
+ """
+
+ event = fields.List(fields.String(), metadata={
+ "description": "Event type filter",
+ "example": [EventType.PackageUpdated],
+ })
diff --git a/src/ahriman/web/schemas/sse_schema.py b/src/ahriman/web/schemas/sse_schema.py
new file mode 100644
index 00000000..7211d294
--- /dev/null
+++ b/src/ahriman/web/schemas/sse_schema.py
@@ -0,0 +1,35 @@
+#
+# Copyright (c) 2021-2026 ahriman team.
+#
+# This file is part of ahriman
+# (see https://github.com/arcan1s/ahriman).
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see .
+#
+from ahriman.models.event import EventType
+from ahriman.web.apispec import Schema, fields
+
+
+class SSESchema(Schema):
+ """
+ response SSE schema
+ """
+
+ event = fields.String(required=True, metadata={
+ "description": "Event type",
+ "example": EventType.PackageUpdated,
+ })
+ data = fields.Dict(keys=fields.String(), values=fields.Raw(), metadata={
+ "description": "Event data",
+ })
diff --git a/src/ahriman/web/views/v1/auditlog/event_bus.py b/src/ahriman/web/views/v1/auditlog/event_bus.py
index 757090ce..c3a019a4 100644
--- a/src/ahriman/web/views/v1/auditlog/event_bus.py
+++ b/src/ahriman/web/views/v1/auditlog/event_bus.py
@@ -28,7 +28,7 @@ from ahriman.core.status.event_bus import SSEvent
from ahriman.models.event import EventType
from ahriman.models.user_access import UserAccess
from ahriman.web.apispec.decorators import apidocs
-from ahriman.web.schemas import EventSchema, RepositoryIdSchema
+from ahriman.web.schemas import EventBusFilterSchema, SSESchema
from ahriman.web.views.base import BaseView
@@ -70,8 +70,8 @@ class EventBusView(BaseView):
description="Stream live updates via SSE",
permission=GET_PERMISSION,
error_404_description="Repository is unknown",
- schema=EventSchema(many=True),
- query_schema=RepositoryIdSchema,
+ schema=SSESchema(many=True),
+ query_schema=EventBusFilterSchema,
)
async def get(self) -> StreamResponse:
"""
@@ -81,15 +81,16 @@ class EventBusView(BaseView):
StreamResponse: 200 with streaming updates
"""
topics = [EventType(event) for event in self.request.query.getall("event", [])] or None
+ event_bus = self.service().event_bus
async with sse_response(self.request) as response:
- subscription_id, queue = await self.service().event_bus.subscribe(topics)
+ subscription_id, queue = await event_bus.subscribe(topics)
try:
await self._run(response, queue)
except (ConnectionResetError, QueueShutDown):
pass
finally:
- await self.service().event_bus.unsubscribe(subscription_id)
+ await event_bus.unsubscribe(subscription_id)
return response
diff --git a/tests/ahriman/core/status/conftest.py b/tests/ahriman/core/status/conftest.py
index f51af705..9a7cf214 100644
--- a/tests/ahriman/core/status/conftest.py
+++ b/tests/ahriman/core/status/conftest.py
@@ -2,6 +2,7 @@ import pytest
from ahriman.core.configuration import Configuration
from ahriman.core.status import Client
+from ahriman.core.status.event_bus import EventBus
from ahriman.core.status.web_client import WebClient
@@ -16,6 +17,17 @@ def client() -> Client:
return Client()
+@pytest.fixture
+def event_bus() -> EventBus:
+ """
+ fixture for event bus
+
+ Returns:
+ EventBus: even bus test instance
+ """
+ return EventBus(0)
+
+
@pytest.fixture
def web_client(configuration: Configuration) -> WebClient:
"""
diff --git a/tests/ahriman/core/status/test_event_bus.py b/tests/ahriman/core/status/test_event_bus.py
new file mode 100644
index 00000000..814307d5
--- /dev/null
+++ b/tests/ahriman/core/status/test_event_bus.py
@@ -0,0 +1,107 @@
+import pytest
+
+from ahriman.core.status.event_bus import EventBus
+from ahriman.models.event import EventType
+from ahriman.models.package import Package
+
+
+async def test_broadcast(event_bus: EventBus, package_ahriman: Package) -> None:
+ """
+ must broadcast event to all subscribers
+ """
+ _, queue = await event_bus.subscribe()
+ await event_bus.broadcast(EventType.PackageUpdated, package_ahriman.base, version=package_ahriman.version)
+
+ message = queue.get_nowait()
+ assert message == (
+ EventType.PackageUpdated,
+ {"object_id": package_ahriman.base, "version": package_ahriman.version},
+ )
+
+
+async def test_broadcast_with_topics(event_bus: EventBus, package_ahriman: Package) -> None:
+ """
+ must broadcast event to subscribers with matching topics
+ """
+ _, queue = await event_bus.subscribe([EventType.PackageUpdated])
+ await event_bus.broadcast(EventType.PackageUpdated, package_ahriman.base)
+ assert not queue.empty()
+
+
+async def test_broadcast_topic_isolation(event_bus: EventBus, package_ahriman: Package) -> None:
+ """
+ must not broadcast event to subscribers with non-matching topics
+ """
+ _, queue = await event_bus.subscribe([EventType.BuildLog])
+ await event_bus.broadcast(EventType.PackageUpdated, package_ahriman.base)
+ assert queue.empty()
+
+
+async def test_broadcast_queue_full(event_bus: EventBus, package_ahriman: Package) -> None:
+ """
+ must discard message to slow subscriber
+ """
+ event_bus.max_size = 1
+ _, queue = await event_bus.subscribe()
+
+ await event_bus.broadcast(EventType.PackageUpdated, package_ahriman.base)
+ await event_bus.broadcast(EventType.PackageRemoved, package_ahriman.base)
+ assert queue.qsize() == 1
+
+
+async def test_shutdown(event_bus: EventBus) -> None:
+ """
+ must send sentinel to all subscribers on shutdown
+ """
+ _, queue = await event_bus.subscribe()
+
+ await event_bus.shutdown()
+ message = queue.get_nowait()
+ assert message is None
+
+
+async def test_shutdown_queue_full(event_bus: EventBus, package_ahriman: Package) -> None:
+ """
+ must handle shutdown when queue is full
+ """
+ event_bus.max_size = 1
+ _, queue = await event_bus.subscribe()
+
+ await event_bus.broadcast(EventType.PackageUpdated, package_ahriman.base)
+ await event_bus.shutdown()
+
+
+async def test_subscribe(event_bus: EventBus) -> None:
+ """
+ must register new subscriber
+ """
+ subscriber_id, queue = await event_bus.subscribe()
+
+ assert subscriber_id
+ assert queue.empty()
+ assert subscriber_id in event_bus._subscribers
+
+
+async def test_subscribe_with_topics(event_bus: EventBus) -> None:
+ """
+ must register subscriber with topic filter
+ """
+ subscriber_id, _ = await event_bus.subscribe([EventType.BuildLog])
+ topics, _ = event_bus._subscribers[subscriber_id]
+ assert topics == [EventType.BuildLog]
+
+
+async def test_unsubscribe(event_bus: EventBus) -> None:
+ """
+ must remove subscriber
+ """
+ subscriber_id, _ = await event_bus.subscribe()
+ await event_bus.unsubscribe(subscriber_id)
+ assert subscriber_id not in event_bus._subscribers
+
+
+async def test_unsubscribe_unknown(event_bus: EventBus) -> None:
+ """
+ must not fail on unknown subscriber removal
+ """
+ await event_bus.unsubscribe("unknown")
diff --git a/tests/ahriman/core/status/test_watcher.py b/tests/ahriman/core/status/test_watcher.py
index f353a722..b7e5b6fb 100644
--- a/tests/ahriman/core/status/test_watcher.py
+++ b/tests/ahriman/core/status/test_watcher.py
@@ -5,17 +5,36 @@ from pytest_mock import MockerFixture
from ahriman.core.exceptions import UnknownPackageError
from ahriman.core.status.watcher import Watcher
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
+from ahriman.models.changes import Changes
+from ahriman.models.dependencies import Dependencies
+from ahriman.models.event import Event, EventType
+from ahriman.models.log_record import LogRecord
+from ahriman.models.log_record_id import LogRecordId
from ahriman.models.package import Package
+from ahriman.models.pkgbuild_patch import PkgbuildPatch
-def test_packages(watcher: Watcher, package_ahriman: Package) -> None:
+async def test_event_add(watcher: Watcher, mocker: MockerFixture) -> None:
"""
- must return list of available packages
+ must create new event
"""
- assert not watcher.packages
+ event = Event("event", "object")
+ cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.event_add")
- watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus())}
- assert watcher.packages
+ await watcher.event_add(event)
+ cache_mock.assert_called_once_with(event)
+
+
+async def test_event_get(watcher: Watcher, mocker: MockerFixture) -> None:
+ """
+ must retrieve events
+ """
+ event = Event("event", "object")
+ cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.event_get", return_value=[event])
+
+ result = await watcher.event_get(None, None)
+ assert result == [event]
+ cache_mock.assert_called_once_with(None, None, None, None, -1, 0)
async def test_load(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
@@ -45,6 +64,15 @@ async def test_load_known(watcher: Watcher, package_ahriman: Package, mocker: Mo
assert status.status == BuildStatusEnum.Success
+async def test_logs_rotate(watcher: Watcher, mocker: MockerFixture) -> None:
+ """
+ must rotate logs
+ """
+ cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.logs_rotate")
+ await watcher.logs_rotate(42)
+ cache_mock.assert_called_once_with(42)
+
+
async def test_package_archives(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must return package archives from package info
@@ -75,17 +103,65 @@ async def test_package_get_failed(watcher: Watcher, package_ahriman: Package) ->
await watcher.package_get(package_ahriman.base)
+async def test_package_changes_get(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+ """
+ must return package changes
+ """
+ changes = Changes("sha")
+ cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_changes_get",
+ return_value=changes)
+
+ assert await watcher.package_changes_get(package_ahriman.base) == changes
+ cache_mock.assert_called_once_with(package_ahriman.base)
+
+
+async def test_package_changes_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+ """
+ must update package changes
+ """
+ changes = Changes("sha")
+ cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_changes_update")
+
+ await watcher.package_changes_update(package_ahriman.base, changes)
+ cache_mock.assert_called_once_with(package_ahriman.base, changes)
+
+
+async def test_package_dependencies_get(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+ """
+ must return package dependencies
+ """
+ dependencies = Dependencies({"path": [package_ahriman.base]})
+ cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_dependencies_get",
+ return_value=dependencies)
+
+ assert await watcher.package_dependencies_get(package_ahriman.base) == dependencies
+ cache_mock.assert_called_once_with(package_ahriman.base)
+
+
+async def test_package_dependencies_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+ """
+ must update package dependencies
+ """
+ dependencies = Dependencies({"path": [package_ahriman.base]})
+ cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_dependencies_update")
+
+ await watcher.package_dependencies_update(package_ahriman.base, dependencies)
+ cache_mock.assert_called_once_with(package_ahriman.base, dependencies)
+
+
async def test_package_hold_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must update package hold status
"""
cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_hold_update")
+ broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus())}
await watcher.package_hold_update(package_ahriman.base, enabled=True)
cache_mock.assert_called_once_with(package_ahriman.base, enabled=True)
_, status = watcher._known[package_ahriman.base]
assert status.is_held is True
+ broadcast_mock.assert_called_once_with(EventType.PackageHeld, package_ahriman.base, is_held=True)
async def test_package_hold_update_unknown(watcher: Watcher, package_ahriman: Package) -> None:
@@ -96,16 +172,83 @@ async def test_package_hold_update_unknown(watcher: Watcher, package_ahriman: Pa
await watcher.package_hold_update(package_ahriman.base, enabled=True)
+async def test_package_logs_add(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+ """
+ must post log record
+ """
+ log_record = LogRecord(LogRecordId(package_ahriman.base, "1.0.0"), 42.0, "message")
+ cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_logs_add")
+ broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
+
+ await watcher.package_logs_add(log_record)
+ cache_mock.assert_called_once_with(log_record)
+ broadcast_mock.assert_called_once_with(EventType.BuildLog, package_ahriman.base, **log_record.view())
+
+
+async def test_package_logs_get(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+ """
+ must return package logs
+ """
+ log_record = LogRecord(LogRecordId(package_ahriman.base, "1.0.0"), 42.0, "message")
+ cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_logs_get",
+ return_value=[log_record])
+
+ assert await watcher.package_logs_get(package_ahriman.base) == [log_record]
+ cache_mock.assert_called_once_with(package_ahriman.base, None, None, -1, 0)
+
+
+async def test_package_logs_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+ """
+ must remove package logs
+ """
+ cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_logs_remove")
+ await watcher.package_logs_remove(package_ahriman.base, None)
+ cache_mock.assert_called_once_with(package_ahriman.base, None)
+
+
+async def test_package_patches_get(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+ """
+ must return package patches
+ """
+ patch = PkgbuildPatch("key", "value")
+ cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_patches_get", return_value=[patch])
+
+ assert await watcher.package_patches_get(package_ahriman.base, None) == [patch]
+ cache_mock.assert_called_once_with(package_ahriman.base, None)
+
+
+async def test_package_patches_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+ """
+ must remove package patches
+ """
+ cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_patches_remove")
+ await watcher.package_patches_remove(package_ahriman.base, None)
+ cache_mock.assert_called_once_with(package_ahriman.base, None)
+
+
+async def test_package_patches_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+ """
+ must update package patches
+ """
+ patch = PkgbuildPatch("key", "value")
+ cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_patches_update")
+
+ await watcher.package_patches_update(package_ahriman.base, patch)
+ cache_mock.assert_called_once_with(package_ahriman.base, patch)
+
+
async def test_package_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must remove package base
"""
cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_remove")
+ broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus())}
await watcher.package_remove(package_ahriman.base)
assert not watcher._known
cache_mock.assert_called_once_with(package_ahriman.base)
+ broadcast_mock.assert_called_once_with(EventType.PackageRemoved, package_ahriman.base)
async def test_package_remove_unknown(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
@@ -113,8 +256,11 @@ async def test_package_remove_unknown(watcher: Watcher, package_ahriman: Package
must not fail on unknown base removal
"""
cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_remove")
+ broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
+
await watcher.package_remove(package_ahriman.base)
cache_mock.assert_called_once_with(package_ahriman.base)
+ broadcast_mock.assert_called_once_with(EventType.PackageRemoved, package_ahriman.base)
async def test_package_status_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
@@ -122,6 +268,7 @@ async def test_package_status_update(watcher: Watcher, package_ahriman: Package,
must update package status only for known package
"""
cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_status_update")
+ broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus())}
await watcher.package_status_update(package_ahriman.base, BuildStatusEnum.Success)
@@ -129,6 +276,9 @@ async def test_package_status_update(watcher: Watcher, package_ahriman: Package,
package, status = watcher._known[package_ahriman.base]
assert package == package_ahriman
assert status.status == BuildStatusEnum.Success
+ broadcast_mock.assert_called_once_with(
+ EventType.PackageStatusChanged, package_ahriman.base, status=BuildStatusEnum.Success.value,
+ )
async def test_package_status_update_preserves_hold(watcher: Watcher, package_ahriman: Package,
@@ -137,6 +287,7 @@ async def test_package_status_update_preserves_hold(watcher: Watcher, package_ah
must preserve hold status on package status update
"""
mocker.patch("ahriman.core.status.local_client.LocalClient.package_status_update")
+ mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus(is_held=True))}
await watcher.package_status_update(package_ahriman.base, BuildStatusEnum.Success)
@@ -157,10 +308,15 @@ async def test_package_update(watcher: Watcher, package_ahriman: Package, mocker
must add package to cache
"""
cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_update")
+ broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
await watcher.package_update(package_ahriman, BuildStatusEnum.Unknown)
- assert watcher.packages
+ assert await watcher.packages()
cache_mock.assert_called_once_with(package_ahriman, pytest.helpers.anyvar(int))
+ broadcast_mock.assert_called_once_with(
+ EventType.PackageUpdated, package_ahriman.base,
+ status=BuildStatusEnum.Unknown.value, version=package_ahriman.version,
+ )
async def test_package_update_preserves_hold(watcher: Watcher, package_ahriman: Package,
@@ -176,12 +332,34 @@ async def test_package_update_preserves_hold(watcher: Watcher, package_ahriman:
assert status.is_held is True
-async def test_status_update(watcher: Watcher) -> None:
+async def test_packages(watcher: Watcher, package_ahriman: Package) -> None:
+ """
+ must return list of available packages
+ """
+ assert not await watcher.packages()
+
+ watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus())}
+ assert await watcher.packages()
+
+
+async def test_shutdown(watcher: Watcher, mocker: MockerFixture) -> None:
+ """
+ must gracefully shutdown watcher
+ """
+ shutdown_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.shutdown")
+ await watcher.shutdown()
+ shutdown_mock.assert_called_once_with()
+
+
+async def test_status_update(watcher: Watcher, mocker: MockerFixture) -> None:
"""
must update service status
"""
+ broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
+
await watcher.status_update(BuildStatusEnum.Success)
assert watcher.status.status == BuildStatusEnum.Success
+ broadcast_mock.assert_called_once_with(EventType.ServiceStatusChanged, None, status=BuildStatusEnum.Success.value)
def test_call(watcher: Watcher, package_ahriman: Package) -> None:
diff --git a/tests/ahriman/web/schemas/test_event_bus_filter_schema.py b/tests/ahriman/web/schemas/test_event_bus_filter_schema.py
new file mode 100644
index 00000000..1982fb6b
--- /dev/null
+++ b/tests/ahriman/web/schemas/test_event_bus_filter_schema.py
@@ -0,0 +1 @@
+# schema testing goes in view class tests
diff --git a/tests/ahriman/web/schemas/test_sse_schema.py b/tests/ahriman/web/schemas/test_sse_schema.py
new file mode 100644
index 00000000..1982fb6b
--- /dev/null
+++ b/tests/ahriman/web/schemas/test_sse_schema.py
@@ -0,0 +1 @@
+# schema testing goes in view class tests
diff --git a/tests/ahriman/web/views/v1/auditlog/test_view_v1_auditlog_event_bus.py b/tests/ahriman/web/views/v1/auditlog/test_view_v1_auditlog_event_bus.py
new file mode 100644
index 00000000..5e7bad0c
--- /dev/null
+++ b/tests/ahriman/web/views/v1/auditlog/test_view_v1_auditlog_event_bus.py
@@ -0,0 +1,119 @@
+import asyncio
+import pytest
+
+from aiohttp.test_utils import TestClient
+from asyncio import Queue
+from pytest_mock import MockerFixture
+from unittest.mock import AsyncMock
+
+from ahriman.core.status.watcher import Watcher
+from ahriman.models.event import EventType
+from ahriman.models.package import Package
+from ahriman.models.user_access import UserAccess
+from ahriman.web.keys import WatcherKey
+from ahriman.web.views.v1.auditlog.event_bus import EventBusView
+
+
+async def _producer(watcher: Watcher, package_ahriman: Package) -> None:
+ """
+ create producer
+
+ Args:
+ watcher(Watcher): watcher test instance
+ package_ahriman(Package): package test instance
+ """
+ await asyncio.sleep(0.1)
+ await watcher.event_bus.broadcast(EventType.PackageRemoved, package_ahriman.base)
+ await watcher.event_bus.broadcast(EventType.PackageUpdated, package_ahriman.base, status="success")
+ await asyncio.sleep(0.1)
+ await watcher.event_bus.shutdown()
+
+
+async def test_get_permission() -> None:
+ """
+ must return correct permission for the request
+ """
+ request = pytest.helpers.request("", "", "GET")
+ assert await EventBusView.get_permission(request) == UserAccess.Full
+
+
+def test_routes() -> None:
+ """
+ must return correct routes
+ """
+ assert EventBusView.ROUTES == ["/api/v1/events/stream"]
+
+
+async def test_run_timeout() -> None:
+ """
+ must handle timeout and continue loop
+ """
+ queue = Queue()
+
+ async def _shutdown() -> None:
+ await asyncio.sleep(0.05)
+ await queue.put(None)
+
+ response = AsyncMock()
+ response.is_connected = lambda: True
+ response.ping_interval = 0.01
+
+ asyncio.create_task(_shutdown())
+ await EventBusView._run(response, queue)
+
+
+async def test_get(client: TestClient, package_ahriman: Package) -> None:
+ """
+ must stream events via SSE
+ """
+ watcher = next(iter(client.app[WatcherKey].values()))
+ asyncio.create_task(_producer(watcher, package_ahriman))
+ request_schema = pytest.helpers.schema_request(EventBusView.get, location="querystring")
+ # no content validation here because it is a streaming response
+
+ assert not request_schema.validate({})
+ response = await client.get("/api/v1/events/stream")
+ assert response.status == 200
+
+ body = await response.text()
+ assert EventType.PackageUpdated in body
+ assert "ahriman" in body
+
+
+async def test_get_with_topic_filter(client: TestClient, package_ahriman: Package) -> None:
+ """
+ must filter events by topic
+ """
+ watcher = next(iter(client.app[WatcherKey].values()))
+ asyncio.create_task(_producer(watcher, package_ahriman))
+ request_schema = pytest.helpers.schema_request(EventBusView.get, location="querystring")
+
+ payload = {"event": [EventType.PackageUpdated]}
+ assert not request_schema.validate(payload)
+ response = await client.get("/api/v1/events/stream", params=payload)
+ assert response.status == 200
+
+ body = await response.text()
+ assert EventType.PackageUpdated in body
+ assert EventType.PackageRemoved not in body
+
+
+async def test_get_not_found(client: TestClient) -> None:
+ """
+ must return not found for unknown repository
+ """
+ response_schema = pytest.helpers.schema_response(EventBusView.get, code=404)
+
+ response = await client.get("/api/v1/events/stream", params={"architecture": "unknown", "repository": "unknown"})
+ assert response.status == 404
+ assert not response_schema.validate(await response.json())
+
+
+async def test_get_connection_reset(client: TestClient, mocker: MockerFixture) -> None:
+ """
+ must handle connection reset
+ """
+ mocker.patch.object(EventBusView, "_run", side_effect=ConnectionResetError)
+
+ response = await client.get("/api/v1/events/stream")
+ assert response.status == 200