From 2eba728dd2c51a437fe22e9a7ddcdaf6e443bde3 Mon Sep 17 00:00:00 2001 From: arabcoders Date: Fri, 12 Sep 2025 02:31:47 +0300 Subject: [PATCH] Refactor event handlers to be asynchronous across multiple modules --- app/library/DataStore.py | 3 +- app/library/Download.py | 8 +- app/library/DownloadQueue.py | 83 +- app/library/Events.py | 200 ++-- app/library/HttpSocket.py | 19 +- app/library/Notifications.py | 7 +- app/library/Presets.py | 2 +- app/library/Scheduler.py | 10 +- app/library/Tasks.py | 13 +- app/library/conditions.py | 2 +- app/library/dl_fields.py | 2 +- app/library/task_handlers/_base_handler.py | 10 +- app/library/task_handlers/twitch.py | 6 +- app/library/task_handlers/youtube.py | 6 +- app/main.py | 15 +- app/routes/api/conditions.py | 2 +- app/routes/api/dl_fields.py | 2 +- app/routes/api/history.py | 6 +- app/routes/api/notifications.py | 2 +- app/routes/api/presets.py | 2 +- app/routes/api/system.py | 6 +- app/routes/socket/connection.py | 4 +- app/routes/socket/history.py | 14 +- app/routes/socket/terminal.py | 14 +- app/tests/test_events.py | 1014 ++++++++++++++++++++ pyproject.toml | 8 +- uv.lock | 14 + 27 files changed, 1214 insertions(+), 260 deletions(-) create mode 100644 app/tests/test_events.py diff --git a/app/library/DataStore.py b/app/library/DataStore.py index 753ce0dc..55ff4bfd 100644 --- a/app/library/DataStore.py +++ b/app/library/DataStore.py @@ -1,4 +1,3 @@ -import asyncio import copy import json import logging @@ -151,7 +150,7 @@ class DataStore: if "error" == value.info.status and not no_notify: from app.library.Events import EventBus, Events - asyncio.create_task(EventBus.get_instance().emit(Events.ITEM_ERROR, value.info), name="emit_item_error") + EventBus.get_instance().emit(Events.ITEM_ERROR, value.info) self._dict.update({value.info._id: value}) self._update_store_item(self._type, value.info) diff --git a/app/library/Download.py b/app/library/Download.py index b42a4dca..00acd227 100644 --- a/app/library/Download.py +++ b/app/library/Download.py @@ -350,7 +350,7 @@ class Download: self.proc.start() self.info.status = "preparing" - await self._notify.emit(Events.ITEM_UPDATED, data=self.info) + self._notify.emit(Events.ITEM_UPDATED, data=self.info) asyncio.create_task(self.progress_update(), name=f"update-{self.id}") ret = await asyncio.get_running_loop().run_in_executor(None, self.proc.join) @@ -516,7 +516,7 @@ class Download: self.logger.debug(f"Status Update: {self.info._id=} {status=}") if isinstance(status, str): - await self._notify.emit(Events.ITEM_UPDATED, data=self.info) + self._notify.emit(Events.ITEM_UPDATED, data=self.info) return self.tmpfilename: str | None = status.get("tmpfilename") @@ -547,7 +547,7 @@ class Download: if "error" == self.info.status and "error" in status: self.info.error = status.get("error") - await self._notify.emit( + self._notify.emit( Events.LOG_ERROR, data=self.info, title="Download Error", @@ -584,7 +584,7 @@ class Download: self.logger.error(f"Failed to run ffprobe. {status.get}. {e}") if not self.final_update or fl: - await self._notify.emit(Events.ITEM_UPDATED, data=self.info) + self._notify.emit(Events.ITEM_UPDATED, data=self.info) async def progress_update(self): """ diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index bcfbfebd..d77fd066 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -3,6 +3,7 @@ import functools import glob import logging import time +import traceback import uuid from datetime import UTC, datetime, timedelta from email.utils import formatdate @@ -107,9 +108,11 @@ class DownloadQueue(metaclass=Singleton): _ (web.Application): The application to attach the download queue to. """ - self._notify.subscribe( - Events.STARTED, lambda _, __: self.initialize(), f"{__class__.__name__}.{__class__.initialize.__name__}" - ) + + async def event_handler(_, __): + await self.initialize() + + self._notify.subscribe(Events.STARTED, event_handler, f"{__class__.__name__}.{__class__.initialize.__name__}") Scheduler.get_instance().add( timer="* * * * *", @@ -155,7 +158,6 @@ class DownloadQueue(metaclass=Singleton): """ status: dict[str, str] = {"status": "ok"} started = False - tasks: list = [] for item_id in ids: try: @@ -173,14 +175,12 @@ class DownloadQueue(metaclass=Singleton): item.info.auto_start = True updated: Download = self.queue.put(item) - tasks.append(self._notify.emit(Events.ITEM_UPDATED, data=updated.info)) - tasks.append( - self._notify.emit( - Events.ITEM_RESUMED, - data=item.info, - title="Download Resumed", - message=f"Download '{item.info.title}' has been resumed.", - ) + self._notify.emit(Events.ITEM_UPDATED, data=updated.info) + self._notify.emit( + Events.ITEM_RESUMED, + data=item.info, + title="Download Resumed", + message=f"Download '{item.info.title}' has been resumed.", ) status[item_id] = "started" started = True @@ -189,9 +189,6 @@ class DownloadQueue(metaclass=Singleton): if started: self.event.set() - if len(tasks) > 0: - await asyncio.gather(*tasks) - return status async def pause_items(self, ids: list[str]) -> dict[str, str]: @@ -206,7 +203,6 @@ class DownloadQueue(metaclass=Singleton): """ status: dict[str, str] = {"status": "ok"} - tasks: list = [] for item_id in ids: try: @@ -229,21 +225,16 @@ class DownloadQueue(metaclass=Singleton): item.info.auto_start = False updated: Download = self.queue.put(item) - tasks.append(self._notify.emit(Events.ITEM_UPDATED, data=updated.info)) - tasks.append( - self._notify.emit( - Events.ITEM_PAUSED, - data=item.info, - title="Download Paused", - message=f"Download '{item.info.title}' has been paused.", - ) + self._notify.emit(Events.ITEM_UPDATED, data=updated.info) + self._notify.emit( + Events.ITEM_PAUSED, + data=item.info, + title="Download Paused", + message=f"Download '{item.info.title}' has been paused.", ) status[item_id] = "paused" LOG.debug(f"Item {item.info.name()} marked as paused.") - if len(tasks) > 0: - await asyncio.gather(*tasks) - return status def pause(self, shutdown: bool = False) -> bool: @@ -495,7 +486,7 @@ class DownloadQueue(metaclass=Singleton): dlInfo.info.status = "not_live" dlInfo.info.msg = nMessage.replace(f" '{dlInfo.info.title}'", "") - await self._notify.emit( + self._notify.emit( Events.LOG_INFO, data={"preset": dlInfo.info.preset, "lowPriority": True}, title=nTitle, @@ -517,7 +508,7 @@ class DownloadQueue(metaclass=Singleton): dlInfo.info.status = "error" itemDownload = self.done.put(dlInfo) - await self._notify.emit( + self._notify.emit( Events.LOG_WARNING, data={"preset": dlInfo.info.preset, "logs": text_logs}, title=nTitle, @@ -557,7 +548,7 @@ class DownloadQueue(metaclass=Singleton): nStore = "history" nEvent = Events.ITEM_MOVED nTitle = "Premiering right now" - await self._notify.emit( + self._notify.emit( Events.LOG_INFO, data={"preset": dlInfo.info.preset}, title=nTitle, message=nMessage ) else: @@ -570,7 +561,7 @@ class DownloadQueue(metaclass=Singleton): else: LOG.debug(f"Item {itemDownload.info.name()} is not set to auto-start.") - await self._notify.emit( + self._notify.emit( nEvent, data={"to": nStore, "preset": itemDownload.info.preset, "item": itemDownload.info} if Events.ITEM_MOVED == nEvent @@ -702,7 +693,7 @@ class DownloadQueue(metaclass=Singleton): self.done.put(dlInfo) - await self._notify.emit( + self._notify.emit( Events.ITEM_MOVED, data={"to": "history", "preset": dlInfo.info.preset, "item": dlInfo.info}, title="Download History Update", @@ -712,7 +703,7 @@ class DownloadQueue(metaclass=Singleton): message: str = f"The URL '{item.url}' is already downloaded and recorded in archive." LOG.error(message) - await self._notify.emit( + self._notify.emit( Events.LOG_INFO, data={"preset": item.preset}, title="Already Downloaded", message=message ) @@ -866,7 +857,7 @@ class DownloadQueue(metaclass=Singleton): await item.close() LOG.debug(f"Deleting from queue {item_ref}") self.queue.delete(id) - await self._notify.emit( + self._notify.emit( Events.ITEM_CANCELLED, data=item.info, title="Download Cancelled", @@ -874,7 +865,7 @@ class DownloadQueue(metaclass=Singleton): ) item.info.status = "cancelled" self.done.put(item) - await self._notify.emit( + self._notify.emit( Events.ITEM_MOVED, data={"to": "history", "preset": item.info.preset, "item": item.info}, title="Download Cancelled", @@ -949,7 +940,7 @@ class DownloadQueue(metaclass=Singleton): self.done.delete(id) _status: str = "Removed" if removed_files > 0 else "Cleared" - await self._notify.emit( + self._notify.emit( Events.ITEM_DELETED, data=item.info, title=f"Download {_status}", @@ -1086,7 +1077,6 @@ class DownloadQueue(metaclass=Singleton): await entry.close() if self.queue.exists(key=id): - _tasks = [] LOG.debug(f"Download Task '{id}' is completed. Removing from queue.") self.queue.delete(key=id) @@ -1096,7 +1086,7 @@ class DownloadQueue(metaclass=Singleton): if entry.is_cancelled() is True: nTitle = "Download Cancelled" nMessage = f"Cancelled '{entry.info.title}' download." - await self._notify.emit(Events.ITEM_CANCELLED, data=entry.info, title=nTitle, message=nMessage) + self._notify.emit(Events.ITEM_CANCELLED, data=entry.info, title=nTitle, message=nMessage) entry.info.status = "cancelled" if entry.info.status == "finished" and entry.info.filename: @@ -1105,19 +1095,15 @@ class DownloadQueue(metaclass=Singleton): if entry.info.is_archivable and not entry.info.is_archived: entry.info.is_archived = True - _tasks.append(self._notify.emit(Events.ITEM_COMPLETED, data=entry.info, title=nTitle, message=nMessage)) + self._notify.emit(Events.ITEM_COMPLETED, data=entry.info, title=nTitle, message=nMessage) self.done.put(entry) - _tasks.append( - self._notify.emit( - Events.ITEM_MOVED, - data={"to": "history", "preset": entry.info.preset, "item": entry.info}, - title=nTitle, - message=nMessage, - ) + self._notify.emit( + Events.ITEM_MOVED, + data={"to": "history", "preset": entry.info.preset, "item": entry.info}, + title=nTitle, + message=nMessage, ) - - await asyncio.gather(*_tasks) else: LOG.warning(f"Download '{id}' not found in queue.") @@ -1223,7 +1209,6 @@ class DownloadQueue(metaclass=Singleton): return if exc := task.exception(): - import traceback task_name: str = task.get_name() if task.get_name() else "unknown_task" LOG.error(f"Unhandled exception in background task '{task_name}': {exc!s}. {traceback.format_exc()}") diff --git a/app/library/Events.py b/app/library/Events.py index a987386d..2f2f2571 100644 --- a/app/library/Events.py +++ b/app/library/Events.py @@ -145,6 +145,9 @@ class Event: data: Any """The data that was passed to the event.""" + extras: dict = field(default_factory=dict) + """Listeners can add extra data to the event.""" + def serialize(self) -> dict: """ Serialize the event. @@ -162,9 +165,20 @@ class Event: "data": self.data, } - def __repr__(self): + def __repr__(self) -> str: return f"Event(id={self.id}, created_at={self.created_at}, event={self.event}, title={self.title}, message={self.message} data={self.data})" + def put(self, key: str, value: Any) -> None: + """ + Put extra data to the event. + + Args: + key (str): The key of the extra data. + value (Any): The value of the extra data. + + """ + self.extras[key] = value + def datatype(self) -> str: """ Get the datatype of the data. @@ -210,17 +224,12 @@ class EventBus(metaclass=Singleton): debug: bool = False """Whether to log debug messages or not.""" - _offload: BackgroundWorker + _offload: BackgroundWorker = None """The background worker to offload tasks to.""" def __init__(self): EventBus._instance = self - from .config import Config - - self.debug = Config.get_instance().debug - self._offload = BackgroundWorker.get_instance() - @staticmethod def get_instance() -> "EventBus": """ @@ -305,84 +314,11 @@ class EventBus(metaclass=Singleton): return self - def sync_emit( - self, - event: str, - data: Any | None = None, - title: str | None = None, - message: str | None = None, - loop=None, - wait: bool = True, - **kwargs, - ): - """ - Emit event and (optionally) wait for results. - - Args: - event (str): The event to emit. - data (Any|None): The data to pass to the event. - title (str | None): The title of the event, if any. - message (str | None): The message of the event, if any. - loop (asyncio.AbstractEventLoop | None): The event loop to use. If None, the current running loop is used. - wait (bool): Whether to wait for the results of the event handlers. Defaults to True. - **kwargs: Additional keyword arguments to pass to the event - - Returns: - list: The results are the return values of the coroutines. If the coroutine raises an exception, - the exception is caught and logged. If event does not exist, an empty list is returned. - If wait is False, a list of asyncio.Task objects is returned instead if we are in a running event loop, - or the result of the coroutine if we are not in a running event loop. - - """ - if event not in self._listeners: - return [] - - if not data: - data = {} - - async def emit_all(): - ev = Event(event=event, title=title, message=message, data=data) - LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data}) - - res: list = [] - - for h in self._listeners[event].values(): - try: - res.append(await h.handle(ev, **kwargs)) - except Exception as e: - LOG.exception(e) - LOG.error(f"Failed to emit event '{event}' to '{h.name}'. Error message '{e!s}'.") - - return res - - try: - loop = loop or asyncio.get_running_loop() - in_same_loop: bool = asyncio.get_running_loop() is loop - except RuntimeError: - loop = None - in_same_loop = False - - if loop is None or not loop.is_running(): - return asyncio.run(emit_all()) - - if in_same_loop: - if wait: - msg = ( - "Calling EventsBus.sync_emit(...,wait=True) from within the running event loop would cause dead-lock. " - "Use `await EventsBus.emit(...)` or `EventsBus.sync_emit(..., wait=False)`." - ) - raise RuntimeError(msg) - - return loop.create_task(emit_all()) - - fut = asyncio.run_coroutine_threadsafe(emit_all(), loop) - return fut.result() if wait else fut - - async def emit( + def emit( self, event: str, data: Any | None = None, title: str | None = None, message: str | None = None, **kwargs - ) -> Awaitable: + ) -> None: """ - Emit an event. + Emit an event to all registered listeners. Args: event (str): The event to emit. @@ -391,44 +327,6 @@ class EventBus(metaclass=Singleton): message (str | None): The message of the event, if any. **kwargs: The keyword arguments to pass to the event. - Returns: - Awaitable: The task that was created to run the event. - - """ - if event not in self._listeners: - return [] - - if not data: - data = {} - - ev = Event(event=event, title=title, message=message, data=data) - - if self.debug or event not in Events.only_debug(): - LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data}) - - tasks = [] - for handler in self._listeners[event].values(): - try: - tasks.append(handler.handle(ev, **kwargs)) - except Exception as e: - LOG.exception(e) - LOG.error(f"Failed to emit event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.") - - return asyncio.gather(*tasks) - - def offload( - self, event: str, title: str | None = None, message: str | None = None, data: Any | None = None, **kwargs - ) -> None: - """ - Offload an dispatching event to a background worker. - - Args: - event (str): The event to offload. - data (Any|None): The data to pass to the event. - title (str | None): The title of the event, if any. - message (str | None): The message of the event, if any. - **kwargs: Additional keyword arguments to pass to the event. - """ if event not in self._listeners: return @@ -439,11 +337,57 @@ class EventBus(metaclass=Singleton): ev = Event(event=event, title=title, message=message, data=data) if self.debug or event not in Events.only_debug(): - LOG.debug(f"Offloading event '{ev.id}: {ev.event}'.", extra={"data": data}) + LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data}) - for handler in self._listeners[event].values(): - try: - self._offload.submit(handler.handle, ev, **kwargs) - except Exception as e: - LOG.exception(e) - LOG.error(f"Failed to offload event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.") + try: + loop = asyncio.get_running_loop() + + for handler in self._listeners[event].values(): + try: + if handler.is_coroutine: + coro = handler.call_back(ev, handler.name, **kwargs) + if asyncio.iscoroutine(coro): + loop.create_task(coro) + else: + LOG.warning(f"Expected coroutine from async handler '{handler.name}', got {type(coro)}") + else: + loop.create_task(self._call(handler, ev, kwargs), name=f"sync-handler-{handler.name}-{ev.id}") + except Exception as e: + LOG.exception(e) + LOG.error(f"Failed to emit event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.") + except RuntimeError: + LOG.debug(f"No event loop detected - using BackgroundWorker for {len(self._listeners[event])} handlers") + for handler in self._listeners[event].values(): + try: + if not self._offload: + self._offload = BackgroundWorker.get_instance() + + self._offload.submit(handler.handle, ev, **kwargs) + except Exception as e: + LOG.exception(e) + LOG.error(f"Failed to emit event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.") + + def clear(self) -> None: + """ + Clear all listeners. Useful for testing. + """ + self._listeners.clear() + + def debug_enable(self) -> None: + """ + Enable debug logging. + """ + self.debug = True + + def debug_disable(self) -> None: + """ + Disable debug logging. + """ + self.debug = False + + @staticmethod + def _call(h, event, kw): + async def call_handler(): + return h.call_back(event, h.name, **kw) + + return call_handler() diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index 28c41297..5c497633 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -52,10 +52,10 @@ class HttpSocket: ping_timeout=5, ) encoder = encoder or Encoder() - self.rootPath = root_path + self.rootPath: Path = root_path - def emit(e: Event, _, **kwargs): - return self.sio.emit(event=e.event, data=encoder.encode(e), **kwargs) + async def event_handler(e: Event, _, **kwargs): + await self.sio.emit(event=e.event, data=encoder.encode(e), **kwargs) services = Services.get_instance() services.add_all( @@ -73,7 +73,7 @@ class HttpSocket: } ) - self._notify.subscribe("frontend", emit, f"{__class__.__name__}.emit") + self._notify.subscribe("frontend", event_handler, f"{__class__.__name__}.emit") @staticmethod def ws_event(func): # type: ignore @@ -101,11 +101,12 @@ class HttpSocket: app.on_shutdown.append(self.on_shutdown) self.sio.attach(app, socketio_path=f"{self.config.base_path.rstrip('/')}/socket.io") - self._notify.subscribe( - Events.ADD_URL, - lambda data, _, **kwargs: self.queue.add(item=Item.format(data.data)), # noqa: ARG005 - f"{__class__.__name__}.add", - ) + + async def event_handler(data: Event, _): + if data and data.data: + await self.queue.add(item=Item.format(data.data)) + + self._notify.subscribe(Events.ADD_URL, event_handler, f"{__class__.__name__}.add") load_modules(self.rootPath, self.rootPath / "routes" / "socket") diff --git a/app/library/Notifications.py b/app/library/Notifications.py index 4020f955..5ac676a1 100644 --- a/app/library/Notifications.py +++ b/app/library/Notifications.py @@ -505,13 +505,12 @@ class Notification(metaclass=Singleton): LOG.error(f"Error sending Notification event '{ev.event}: {ev.id}' to '{target.name}'. '{err_msg!s}'.") return {"url": target.request.url, "status": 500, "text": str(ev)} - def emit(self, e: Event, _, **__): + def emit(self, e: Event, _, **__) -> None: if len(self._targets) < 1 or not NotificationEvents.is_valid(e.event): - return self.noop() + return self._offload.submit(self.send, e) - - return self.noop() + return def _deep_unpack(self, data: dict) -> dict: for k, v in data.items(): diff --git a/app/library/Presets.py b/app/library/Presets.py index 425061cf..b24afd3e 100644 --- a/app/library/Presets.py +++ b/app/library/Presets.py @@ -136,7 +136,7 @@ class Presets(metaclass=Singleton): LOG.error(f"Failed to parse default preset ':{i}'. '{e!s}'.") continue - def event_handler(_, __): + async def event_handler(_, __): msg = "Not implemented" raise Exception(msg) diff --git a/app/library/Scheduler.py b/app/library/Scheduler.py index e05a3031..969eadc9 100644 --- a/app/library/Scheduler.py +++ b/app/library/Scheduler.py @@ -26,11 +26,11 @@ class Scheduler(metaclass=Singleton): self._loop = loop or asyncio.get_event_loop() - EventBus.get_instance().subscribe( - Events.SCHEDULE_ADD, - lambda data, _, **kwargs: self.add(**data.data), # noqa: ARG005 - f"{__class__.__name__}.add", - ) + async def event_handler(data, _): + if data and data.data: + self.add(**data.data) + + EventBus.get_instance().subscribe(Events.SCHEDULE_ADD, event_handler, f"{__class__.__name__}.add") @staticmethod def get_instance() -> "Scheduler": diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 59c60a9c..569befcf 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -193,11 +193,12 @@ class Tasks(metaclass=Singleton): """ self.load() - self._notify.subscribe( - Events.TASKS_ADD, - lambda data, _, **kwargs: self.save(**data.data), # noqa: ARG005 - f"{__class__.__name__}.add", - ) + + async def event_handler(data, _): + if data and data.data: + self.save(data.data) + + self._notify.subscribe(Events.TASKS_ADD, event_handler, f"{__class__.__name__}.add") self._task_handler.load() def get_all(self) -> list[Task]: @@ -449,7 +450,7 @@ class Tasks(metaclass=Singleton): await asyncio.gather(*_tasks) except Exception as e: LOG.error(f"Failed to execute '{task.name}' at '{timeNow}'. '{e!s}'.") - await self._notify.emit( + self._notify.emit( Events.LOG_ERROR, data={"preset": task.preset}, title="Task failed", diff --git a/app/library/conditions.py b/app/library/conditions.py index 1e633890..98c45de9 100644 --- a/app/library/conditions.py +++ b/app/library/conditions.py @@ -68,7 +68,7 @@ class Conditions(metaclass=Singleton): except Exception: pass - def event_handler(_, __): + async def event_handler(_, __): msg = "Not implemented" raise Exception(msg) diff --git a/app/library/dl_fields.py b/app/library/dl_fields.py index 15ee8fdc..b563581c 100644 --- a/app/library/dl_fields.py +++ b/app/library/dl_fields.py @@ -122,7 +122,7 @@ class DLFields(metaclass=Singleton): except Exception: pass - def event_handler(_, __): + async def event_handler(_, __): msg = "Not implemented" raise Exception(msg) diff --git a/app/library/task_handlers/_base_handler.py b/app/library/task_handlers/_base_handler.py index 5c79a6be..4d36e585 100644 --- a/app/library/task_handlers/_base_handler.py +++ b/app/library/task_handlers/_base_handler.py @@ -26,11 +26,11 @@ class BaseHandler: if "failure_count" not in cls.__dict__: cls.failure_count = {} - EventBus.get_instance().subscribe( - Events.ITEM_ERROR, - lambda data, _, **__: cls.on_error(data.data), - f"{cls.__name__}.on_error", - ) + async def event_handler(data, _): + if data and data.data: + await cls.on_error(data.data) + + EventBus.get_instance().subscribe(Events.ITEM_ERROR, event_handler, f"{cls.__name__}.on_error") @staticmethod def can_handle(task: Task) -> bool: diff --git a/app/library/task_handlers/twitch.py b/app/library/task_handlers/twitch.py index f764041a..be48f4ed 100644 --- a/app/library/task_handlers/twitch.py +++ b/app/library/task_handlers/twitch.py @@ -1,4 +1,3 @@ -import asyncio import logging import re from typing import TYPE_CHECKING @@ -146,9 +145,8 @@ class TwitchHandler(BaseHandler): ) try: - await asyncio.gather( - *[notify.emit(Events.ADD_URL, data=rItem.new_with(url=item["url"]).serialize()) for item in filtered] - ) + for item in filtered: + notify.emit(Events.ADD_URL, data=rItem.new_with(url=item["url"]).serialize()) except Exception as e: LOG.exception(e) LOG.error(f"Error while adding items from '{task.name}'. {e!s}") diff --git a/app/library/task_handlers/youtube.py b/app/library/task_handlers/youtube.py index e4e71967..9d652793 100644 --- a/app/library/task_handlers/youtube.py +++ b/app/library/task_handlers/youtube.py @@ -1,4 +1,3 @@ -import asyncio import logging import re from typing import TYPE_CHECKING @@ -153,9 +152,8 @@ class YoutubeHandler(BaseHandler): ) try: - await asyncio.gather( - *[notify.emit(Events.ADD_URL, data=rItem.new_with(url=item["url"]).serialize()) for item in filtered] - ) + for item in filtered: + notify.emit(Events.ADD_URL, data=rItem.new_with(url=item["url"]).serialize()) except Exception as e: LOG.exception(e) LOG.error(f"'{task.name}': Error while adding items from task feed. {e!s}") diff --git a/app/main.py b/app/main.py index 15c6ae3b..f0500960 100644 --- a/app/main.py +++ b/app/main.py @@ -38,7 +38,7 @@ ROOT_PATH: Path = Path(__file__).parent.absolute() class Main: def __init__(self, is_native: bool = False): - self._config = Config.get_instance(is_native=is_native) + self._config: Config = Config.get_instance(is_native=is_native) self._app = web.Application() self._app.on_shutdown.append(self.on_shutdown) self._background_worker = BackgroundWorker() @@ -98,7 +98,7 @@ class Main: raise async def on_shutdown(self, _: web.Application): - await EventBus.get_instance().emit( + EventBus.get_instance().emit( Events.SHUTDOWN, data={"app": self._app}, title="Application Shutdown", @@ -112,12 +112,15 @@ class Main: host = host or self._config.host port = port or self._config.port - EventBus.get_instance().sync_emit( + EventBus.get_instance().emit( Events.STARTUP, data={"app": self._app}, title="Application Startup", message="The application is starting up.", ) + if self._config.debug: + EventBus.get_instance().debug_enable() + Scheduler.get_instance().attach(self._app) self._socket.attach(self._app) @@ -131,7 +134,7 @@ class Main: DLFields.get_instance().attach(self._app) self._background_worker.attach(self._app) - EventBus.get_instance().sync_emit( + EventBus.get_instance().emit( Events.LOADED, data={"app": self._app}, title="Application Loaded", @@ -147,13 +150,11 @@ class Main: loop = asyncio.get_event_loop() - EventBus.get_instance().sync_emit( + EventBus.get_instance().emit( Events.STARTED, data={"app": self._app}, title="Application Started", message="The application has started successfully.", - loop=loop, - wait=False, ) if loop and self._config.debug: diff --git a/app/routes/api/conditions.py b/app/routes/api/conditions.py index d287df85..21796cd7 100644 --- a/app/routes/api/conditions.py +++ b/app/routes/api/conditions.py @@ -114,7 +114,7 @@ async def conditions_add(request: Request, encoder: Encoder, notify: EventBus) - status=web.HTTPInternalServerError.status_code, ) - await notify.emit(Events.CONDITIONS_UPDATE, data=items) + notify.emit(Events.CONDITIONS_UPDATE, data=items) return web.json_response(data=items, status=web.HTTPOk.status_code, dumps=encoder.encode) diff --git a/app/routes/api/dl_fields.py b/app/routes/api/dl_fields.py index 87f67dd2..c8b0fac2 100644 --- a/app/routes/api/dl_fields.py +++ b/app/routes/api/dl_fields.py @@ -96,5 +96,5 @@ async def dl_fields_add(request: Request, encoder: Encoder, notify: EventBus) -> status=web.HTTPInternalServerError.status_code, ) - await notify.emit(Events.DLFIELDS_UPDATE, data=items) + notify.emit(Events.DLFIELDS_UPDATE, data=items) return web.json_response(data=items, status=web.HTTPOk.status_code, dumps=encoder.encode) diff --git a/app/routes/api/history.py b/app/routes/api/history.py index 2bd99639..2e0929d3 100644 --- a/app/routes/api/history.py +++ b/app/routes/api/history.py @@ -169,7 +169,7 @@ async def item_update(request: Request, queue: DownloadQueue, encoder: Encoder, if updated: queue.done.put(item) - await notify.emit(Events.ITEM_UPDATED, data=item.info) + notify.emit(Events.ITEM_UPDATED, data=item.info) return web.json_response( data=item.info, @@ -309,7 +309,7 @@ async def item_archive_add(request: Request, queue: DownloadQueue, notify: Event item.info.archive_status(force=True) queue.done.put(item, no_notify=True) - await notify.emit(Events.ITEM_UPDATED, data=item.info) + notify.emit(Events.ITEM_UPDATED, data=item.info) return web.json_response( data={"message": f"item '{item.info.title}' archived."}, @@ -367,7 +367,7 @@ async def item_archive_delete(request: Request, queue: DownloadQueue, notify: Ev item.info.archive_status(force=True) queue.done.put(item, no_notify=True) - await notify.emit(Events.ITEM_UPDATED, data=item.info) + notify.emit(Events.ITEM_UPDATED, data=item.info) return web.json_response( data={"message": f"item '{item.info.title}' removed from archive."}, diff --git a/app/routes/api/notifications.py b/app/routes/api/notifications.py index fd655f8f..b1604627 100644 --- a/app/routes/api/notifications.py +++ b/app/routes/api/notifications.py @@ -104,6 +104,6 @@ async def notification_test(encoder: Encoder, notify: EventBus) -> Response: Response: The response object. """ - await notify.emit(Events.TEST, title="Test Notification", message="This is a test notification.") + notify.emit(Events.TEST, title="Test Notification", message="This is a test notification.") return web.json_response(data={}, status=web.HTTPOk.status_code, dumps=encoder.encode) diff --git a/app/routes/api/presets.py b/app/routes/api/presets.py index cd4bcdb8..69eecac7 100644 --- a/app/routes/api/presets.py +++ b/app/routes/api/presets.py @@ -96,5 +96,5 @@ async def presets_add(request: Request, encoder: Encoder, notify: EventBus) -> R status=web.HTTPInternalServerError.status_code, ) - await notify.emit(Events.PRESETS_UPDATE, data=presets) + notify.emit(Events.PRESETS_UPDATE, data=presets) return web.json_response(data=presets, status=web.HTTPOk.status_code, dumps=encoder.encode) diff --git a/app/routes/api/system.py b/app/routes/api/system.py index 123a631e..79a538b8 100644 --- a/app/routes/api/system.py +++ b/app/routes/api/system.py @@ -40,7 +40,7 @@ async def downloads_pause(queue: DownloadQueue, encoder: Encoder, notify: EventB queue.pause() msg = "Non-active downloads have been paused." - await notify.emit( + notify.emit( Events.PAUSED, data={"paused": True, "at": time.time()}, title="Downloads Paused", @@ -75,7 +75,7 @@ async def downloads_resume(queue: DownloadQueue, encoder: Encoder, notify: Event queue.resume() msg = "Resumed all downloads." - await notify.emit( + notify.emit( Events.RESUMED, data={"paused": False, "at": time.time()}, title="Downloads Resumed", @@ -109,7 +109,7 @@ async def shutdown_system(request: Request, config: Config, encoder: Encoder, no app = request.app async def do_shutdown(): - await notify.emit( + notify.emit( Events.SHUTDOWN, data={"app": app}, title="Application Shutdown", diff --git a/app/routes/socket/connection.py b/app/routes/socket/connection.py index 37c2b74e..34c685d9 100644 --- a/app/routes/socket/connection.py +++ b/app/routes/socket/connection.py @@ -37,7 +37,7 @@ async def connect(config: Config, queue: DownloadQueue, notify: EventBus, sid: s depth_limit=config.download_path_depth-1, ) - await notify.emit( + notify.emit( Events.CONNECTED, data=data, title="Client connected", @@ -78,7 +78,7 @@ async def subscribe(config: Config, notify: EventBus, sio: socketio.AsyncServer, """ if not isinstance(data, str) or not data: - await notify.emit( + notify.emit( Events.LOG_ERROR, title="Subscription Error", message="Invalid event type was expecting a string.", diff --git a/app/routes/socket/history.py b/app/routes/socket/history.py index d86e5d39..3210fa8a 100644 --- a/app/routes/socket/history.py +++ b/app/routes/socket/history.py @@ -12,13 +12,13 @@ LOG: logging.Logger = logging.getLogger(__name__) async def add_url(queue: DownloadQueue, notify: EventBus, sid: str, data: dict): data = data if isinstance(data, dict) else {} if not (url := data.get("url", None)): - await notify.emit(Events.LOG_ERROR, title="Invalid request", message="No URL provided.", to=sid) + notify.emit(Events.LOG_ERROR, title="Invalid request", message="No URL provided.", to=sid) return item: Item = Item.format(data) try: status = await queue.add(item=item) - await notify.emit( + notify.emit( event=Events.ITEM_STATUS, title="Adding URL", message=f"Adding URL '{url}' to the download queue.", @@ -27,7 +27,7 @@ async def add_url(queue: DownloadQueue, notify: EventBus, sid: str, data: dict): ) except ValueError as e: LOG.exception(e) - await notify.emit( + notify.emit( Events.LOG_ERROR, data={"preset": item.preset}, title="Error Adding URL", message=str(e), to=sid ) @@ -35,7 +35,7 @@ async def add_url(queue: DownloadQueue, notify: EventBus, sid: str, data: dict): @route(RouteType.SOCKET, "item_cancel", "item_cancel") async def item_cancel(queue: DownloadQueue, notify: EventBus, sid: str, data: str): if not (data := data if isinstance(data, str) else None): - await notify.emit(Events.LOG_ERROR, title="Invalid Request", message="No item ID provided.", to=sid) + notify.emit(Events.LOG_ERROR, title="Invalid Request", message="No item ID provided.", to=sid) return await queue.cancel([data]) @@ -46,7 +46,7 @@ async def item_delete(queue: DownloadQueue, notify: EventBus, sid: str, data: di data = data if isinstance(data, dict) else {} if not (id := data.get("id", None)): - await notify.emit(Events.LOG_ERROR, title="Invalid Request", message="No item ID provided.", to=sid) + notify.emit(Events.LOG_ERROR, title="Invalid Request", message="No item ID provided.", to=sid) return await queue.clear([id], remove_file=bool(data.get("remove_file", False))) @@ -55,7 +55,7 @@ async def item_delete(queue: DownloadQueue, notify: EventBus, sid: str, data: di @route(RouteType.SOCKET, "item_start", "item_start") async def item_start(queue: DownloadQueue, notify: EventBus, sid: str, data: list | str) -> None: if not data: - await notify.emit( + notify.emit( Events.LOG_ERROR, title="Invalid Request", message="No items provided to start.", @@ -72,7 +72,7 @@ async def item_start(queue: DownloadQueue, notify: EventBus, sid: str, data: lis @route(RouteType.SOCKET, "item_pause", "item_pause") async def item_pause(queue: DownloadQueue, notify: EventBus, sid: str, data: list | str) -> None: if not data: - await notify.emit( + notify.emit( Events.LOG_ERROR, title="Invalid Request", message="No items provided to pause.", diff --git a/app/routes/socket/terminal.py b/app/routes/socket/terminal.py index affc0cbe..401951dd 100644 --- a/app/routes/socket/terminal.py +++ b/app/routes/socket/terminal.py @@ -17,7 +17,7 @@ LOG: logging.Logger = logging.getLogger(__name__) @route(RouteType.SOCKET, "cli_post", "socket_cli_post") async def cli_post(config: Config, notify: EventBus, sid: str, data: str): if not config.console_enabled: - await notify.emit( + notify.emit( Events.LOG_ERROR, title="Feature disabled", message="Console feature is disabled.", @@ -26,7 +26,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str): return if not data: - await notify.emit(Events.CLI_CLOSE, data={"exitcode": 0}, to=sid) + notify.emit(Events.CLI_CLOSE, data={"exitcode": 0}, to=sid) return import asyncio @@ -93,7 +93,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str): assert proc.stdout is not None async for raw_line in proc.stdout: line = raw_line.rstrip(b"\n") - await notify.emit( + notify.emit( Events.CLI_OUTPUT, data={"type": "stdout", "line": line.decode("utf-8", errors="replace")}, to=sid, @@ -112,7 +112,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str): if not chunk: if buffer: - await notify.emit( + notify.emit( Events.CLI_OUTPUT, data={"type": "stdout", "line": buffer.decode("utf-8", errors="replace")}, to=sid, @@ -123,7 +123,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str): *lines, buffer = buffer.split(b"\n") for line in lines: - await notify.emit( + notify.emit( Events.CLI_OUTPUT, data={"type": "stdout", "line": line.decode("utf-8", errors="replace")}, to=sid, @@ -141,6 +141,6 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str): except Exception as e: LOG.error(f"CLI execute exception was thrown for client '{sid}'.") LOG.exception(e) - await notify.emit(Events.CLI_OUTPUT, data={"type": "stderr", "line": str(e)}, to=sid) + notify.emit(Events.CLI_OUTPUT, data={"type": "stderr", "line": str(e)}, to=sid) finally: - await notify.emit(Events.CLI_CLOSE, data={"exitcode": returncode}, to=sid) + notify.emit(Events.CLI_CLOSE, data={"exitcode": returncode}, to=sid) diff --git a/app/tests/test_events.py b/app/tests/test_events.py new file mode 100644 index 00000000..11079cef --- /dev/null +++ b/app/tests/test_events.py @@ -0,0 +1,1014 @@ +import asyncio +import datetime +from unittest.mock import MagicMock, patch + +import pytest + +from app.library.Events import Event, EventBus, EventListener, Events + + +class TestEvents: + """Test the Events constants class.""" + + def test_events_constants_exist(self): + """Test that all expected event constants exist.""" + # Basic lifecycle events + assert Events.STARTUP == "startup" + assert Events.LOADED == "loaded" + assert Events.STARTED == "started" + assert Events.SHUTDOWN == "shutdown" + + # Connection events + assert Events.CONNECTED == "connected" + + # Log events + assert Events.LOG_INFO == "log_info" + assert Events.LOG_WARNING == "log_warning" + assert Events.LOG_ERROR == "log_error" + assert Events.LOG_SUCCESS == "log_success" + + # Item events + assert Events.ITEM_ADDED == "item_added" + assert Events.ITEM_UPDATED == "item_updated" + assert Events.ITEM_COMPLETED == "item_completed" + assert Events.ITEM_CANCELLED == "item_cancelled" + assert Events.ITEM_DELETED == "item_deleted" + + def test_events_get_all(self): + """Test Events.get_all() method returns all constants.""" + all_events = Events.get_all() + + # Should be a list + assert isinstance(all_events, list) + + # Should contain expected events + expected_events = [ + "startup", + "loaded", + "started", + "shutdown", + "connected", + "log_info", + "log_warning", + "log_error", + "log_success", + "item_added", + "item_updated", + "item_completed", + "item_cancelled", + "item_deleted", + "item_paused", + "item_resumed", + "item_moved", + "item_status", + "item_error", + "test", + "add_url", + "paused", + "resumed", + ] + + for expected in expected_events: + assert expected in all_events + + # Should not contain private or callable attributes + for event in all_events: + assert not event.startswith("_") + assert isinstance(event, str) + + def test_events_frontend(self): + """Test Events.frontend() method returns frontend events.""" + frontend_events = Events.frontend() + + assert isinstance(frontend_events, list) + + # Check some expected frontend events + expected_frontend = [ + Events.CONNECTED, + Events.LOG_INFO, + Events.LOG_WARNING, + Events.LOG_ERROR, + Events.LOG_SUCCESS, + Events.ITEM_ADDED, + Events.ITEM_UPDATED, + Events.ITEM_COMPLETED, + ] + + for expected in expected_frontend: + assert expected in frontend_events + + def test_events_only_debug(self): + """Test Events.only_debug() method returns debug-only events.""" + debug_events = Events.only_debug() + + assert isinstance(debug_events, list) + assert Events.ITEM_UPDATED in debug_events + assert Events.CLI_OUTPUT in debug_events + + +class TestEvent: + """Test the Event dataclass.""" + + def test_event_creation_minimal(self): + """Test creating an Event with minimal required parameters.""" + event = Event(event="test_event", data={"key": "value"}) + + # Check required fields + assert event.event == "test_event" + assert event.data == {"key": "value"} + + # Check auto-generated fields + assert event.id is not None + assert isinstance(event.id, str) + assert event.created_at is not None + + # Check optional fields default to None + assert event.title is None + assert event.message is None + + def test_event_creation_with_all_fields(self): + """Test creating an Event with all fields specified.""" + test_data = {"test": "data"} + event = Event(event="custom_event", title="Test Title", message="Test Message", data=test_data) + + assert event.event == "custom_event" + assert event.title == "Test Title" + assert event.message == "Test Message" + assert event.data == test_data + assert event.id is not None + assert event.created_at is not None + + def test_event_serialize(self): + """Test Event serialization.""" + test_data = {"nested": {"key": "value"}} + event = Event(event="serialize_test", title="Serialize Title", message="Serialize Message", data=test_data) + + serialized = event.serialize() + + assert isinstance(serialized, dict) + assert serialized["event"] == "serialize_test" + assert serialized["title"] == "Serialize Title" + assert serialized["message"] == "Serialize Message" + assert serialized["data"] == test_data + assert "id" in serialized + assert "created_at" in serialized + + def test_event_datatype(self): + """Test Event datatype() method.""" + # Test with dictionary data + event_dict = Event(event="test", data={"key": "value"}) + assert event_dict.datatype() == "dict" + + # Test with list data + event_list = Event(event="test", data=["item1", "item2"]) + assert event_list.datatype() == "list" + + # Test with string data + event_str = Event(event="test", data="string_data") + assert event_str.datatype() == "str" + + # Test with None data + event_none = Event(event="test", data=None) + assert event_none.datatype() == "NoneType" + + def test_event_str_representation(self): + """Test Event string representations.""" + event = Event(event="repr_test", title="Repr Title", message="Repr Message", data={"test": True}) + + # Test __str__ method + str_repr = str(event) + assert "repr_test" in str_repr + assert "Repr Title" in str_repr + assert "Repr Message" in str_repr + assert event.id in str_repr + + # Test __repr__ method + repr_str = repr(event) + assert "Event(" in repr_str + assert "repr_test" in repr_str + assert event.id in repr_str + + def test_event_created_at_format(self): + """Test that created_at is in ISO format.""" + event = Event(event="time_test", data={}) + + # Should be able to parse as ISO datetime + parsed_time = datetime.datetime.fromisoformat(event.created_at) + assert isinstance(parsed_time, datetime.datetime) + + # Should be recent (within last few seconds) + now = datetime.datetime.now(tz=datetime.UTC) + time_diff = abs((now - parsed_time).total_seconds()) + assert time_diff < 5 # Within 5 seconds + + def test_event_put_method(self): + """Test Event.put() method for adding extra data.""" + event = Event(event="put_test", data={"original": "data"}) + + # Initially extras should be empty + assert event.extras == {} + + # Test putting single value + event.put("key1", "value1") + assert event.extras["key1"] == "value1" + assert len(event.extras) == 1 + + # Test putting multiple values + event.put("key2", 42) + event.put("key3", {"nested": "object"}) + assert event.extras["key2"] == 42 + assert event.extras["key3"] == {"nested": "object"} + assert len(event.extras) == 3 + + # Test overwriting existing key + event.put("key1", "new_value1") + assert event.extras["key1"] == "new_value1" + assert len(event.extras) == 3 + + # Test putting None value + event.put("key4", None) + assert event.extras["key4"] is None + assert len(event.extras) == 4 + + def test_event_put_with_complex_data(self): + """Test Event.put() with complex data types.""" + event = Event(event="complex_put_test", data={}) + + # Test with list + test_list = [1, 2, 3, "string", {"nested": True}] + event.put("list_data", test_list) + assert event.extras["list_data"] == test_list + + # Test with dictionary + test_dict = {"a": 1, "b": {"c": 2}, "d": [4, 5, 6]} + event.put("dict_data", test_dict) + assert event.extras["dict_data"] == test_dict + + # Test with callable (function reference) + def test_func(): + return "test" + + event.put("func_data", test_func) + assert event.extras["func_data"] == test_func + assert event.extras["func_data"]() == "test" + + def test_event_put_extras_persistence(self): + """Test that extras persist and don't interfere with original data.""" + original_data = {"original": "value"} + event = Event(event="persistence_test", data=original_data) + + # Add extras + event.put("extra1", "value1") + event.put("extra2", "value2") + + # Original data should be unchanged + assert event.data == original_data + assert event.data is original_data # Same object reference + + # Extras should be separate + assert event.extras == {"extra1": "value1", "extra2": "value2"} + assert "extra1" not in event.data + assert "extra2" not in event.data + + @pytest.mark.asyncio + async def test_event_mutation_between_listeners(self): + """Test that multiple listeners can mutate the same event and see each other's changes.""" + bus = EventBus() + mutations = [] + + async def listener1(event, name, **kwargs): # noqa: ARG001 + # First listener adds data + event.put("listener1", "data1") + event.put("shared_counter", 1) + mutations.append("listener1_executed") + + async def listener2(event, name, **kwargs): # noqa: ARG001 + # Second listener can see and modify first listener's data + assert event.extras.get("listener1") == "data1" + assert event.extras.get("shared_counter") == 1 + + # Add its own data + event.put("listener2", "data2") + # Increment shared counter + event.put("shared_counter", event.extras.get("shared_counter", 0) + 1) + mutations.append("listener2_executed") + + async def listener3(event, name, **kwargs): # noqa: ARG001 + # Third listener can see all previous mutations + assert event.extras.get("listener1") == "data1" + assert event.extras.get("listener2") == "data2" + assert event.extras.get("shared_counter") == 2 + + # Add final data + event.put("listener3", "data3") + event.put("final_count", len(event.extras)) + mutations.append("listener3_executed") + + # Subscribe listeners in order + bus.subscribe(Events.TEST, listener1, "listener1") + bus.subscribe(Events.TEST, listener2, "listener2") + bus.subscribe(Events.TEST, listener3, "listener3") + + # Emit event + bus.emit(Events.TEST, data={"original": "data"}) + + # Wait for execution (fire-and-forget) + await asyncio.sleep(0.01) + + # Verify all listeners executed + assert mutations == ["listener1_executed", "listener2_executed", "listener3_executed"] + + @pytest.mark.asyncio + async def test_event_mutation_with_async_listeners(self): + """Test event mutation with async listeners.""" + + bus = EventBus() + mutations = [] + + async def async_listener1(event, name, **kwargs): # noqa: ARG001 + event.put("async1", "value1") + mutations.append("async1") + + async def async_listener2(event, name, **kwargs): # noqa: ARG001 + # Can see first async listener's data + assert event.extras.get("async1") == "value1" + event.put("async2", "value2") + mutations.append("async2") + + async def sync_listener(event, name, **kwargs): # noqa: ARG001 + # Can see both async listeners' data + assert event.extras.get("async1") == "value1" + assert event.extras.get("async2") == "value2" + event.put("sync", "value3") + mutations.append("sync") + + bus.subscribe(Events.STARTUP, async_listener1, "async1") + bus.subscribe(Events.STARTUP, async_listener2, "async2") + bus.subscribe(Events.STARTUP, sync_listener, "sync") + + bus.emit(Events.STARTUP, data={"test": "data"}) + + # Wait for execution + await asyncio.sleep(0.01) + + assert mutations == ["async1", "async2", "sync"] + + @pytest.mark.asyncio + async def test_event_mutation_data_types(self): + """Test event mutation with various data types.""" + bus = EventBus() + final_extras = {} + + async def collector(event, name, **kwargs): # noqa: ARG001 + # Store reference to final extras + nonlocal final_extras + final_extras = event.extras + + async def mutator1(event, name, **kwargs): # noqa: ARG001 + event.put("string", "test") + event.put("number", 42) + event.put("list", [1, 2, 3]) + + async def mutator2(event, name, **kwargs): # noqa: ARG001 + # Modify existing list + existing_list = event.extras.get("list", []) + existing_list.append(4) + event.put("list", existing_list) + + event.put("dict", {"key": "value"}) + is_true = True + event.put("bool", is_true) + + async def mutator3(event, name, **kwargs): # noqa: ARG001 + # Modify existing dict + existing_dict = event.extras.get("dict", {}) + existing_dict["new_key"] = "new_value" + event.put("dict", existing_dict) + + event.put("nested", {"list": [1, 2], "dict": {"inner": "value"}}) + + bus.subscribe(Events.SHUTDOWN, mutator1, "mutator1") + bus.subscribe(Events.SHUTDOWN, mutator2, "mutator2") + bus.subscribe(Events.SHUTDOWN, mutator3, "mutator3") + bus.subscribe(Events.SHUTDOWN, collector, "collector") # Last to collect final state + + bus.emit(Events.SHUTDOWN, data={}) + + # Wait for execution + await asyncio.sleep(0.01) + + # Verify final state + assert final_extras["string"] == "test" + assert final_extras["number"] == 42 + assert final_extras["list"] == [1, 2, 3, 4] + assert final_extras["dict"] == {"key": "value", "new_key": "new_value"} + assert final_extras["bool"] is True + assert final_extras["nested"]["list"] == [1, 2] + assert final_extras["nested"]["dict"]["inner"] == "value" + + @pytest.mark.asyncio + async def test_sync_handler_async_wrapping(self): + """Test that sync handlers are properly wrapped in async functions to avoid blocking.""" + bus = EventBus() + execution_order = [] + + def slow_sync_handler(event, name, **kwargs): # noqa: ARG001 + """A sync handler that simulates blocking work""" + import time + + time.sleep(0.05) # Simulate some work + execution_order.append(f"sync_{name}") + + async def fast_async_handler(event, name, **kwargs): # noqa: ARG001 + """A fast async handler""" + execution_order.append(f"async_{name}") + + # Subscribe both handlers + bus.subscribe(Events.TEST, slow_sync_handler, "slow_sync") + bus.subscribe(Events.TEST, fast_async_handler, "fast_async") + + # Emit should return immediately (non-blocking) + import time + + start_time = time.time() + bus.emit(Events.TEST, data={"test": "wrapping"}) + emit_time = time.time() - start_time + + # Emit should be very fast (< 0.01s) because sync handler is wrapped + assert emit_time < 0.01, f"Emit took too long: {emit_time:.4f}s - sync handler may be blocking" + + # Wait for handlers to complete + await asyncio.sleep(0.1) + + # Both handlers should have executed + assert len(execution_order) == 2 + assert "sync_slow_sync" in execution_order + assert "async_fast_async" in execution_order + + @pytest.mark.asyncio + async def test_sync_handler_no_race_condition(self): + """Test that multiple sync handlers don't have race conditions with loop variables.""" + bus = EventBus() + results = [] + + def handler1(event, name, **kwargs): # noqa: ARG001 + results.append(f"handler1_{event.data['id']}") + + def handler2(event, name, **kwargs): # noqa: ARG001 + results.append(f"handler2_{event.data['id']}") + + def handler3(event, name, **kwargs): # noqa: ARG001 + results.append(f"handler3_{event.data['id']}") + + # Subscribe multiple sync handlers + bus.subscribe(Events.TEST, handler1, "handler1") + bus.subscribe(Events.TEST, handler2, "handler2") + bus.subscribe(Events.TEST, handler3, "handler3") + + # Emit multiple events quickly to test for race conditions + for i in range(3): + bus.emit(Events.TEST, data={"id": i}) + + # Wait for all handlers to complete + await asyncio.sleep(0.05) + + # Should have 9 results (3 handlers x 3 events) + assert len(results) == 9 + + # Each handler should have processed each event with correct data + for i in range(3): + assert f"handler1_{i}" in results + assert f"handler2_{i}" in results + assert f"handler3_{i}" in results + + @pytest.mark.asyncio + async def test_mixed_sync_async_handlers_execution_order(self): + """Test that mixed sync/async handlers execute properly without blocking.""" + bus = EventBus() + execution_times = [] + + def sync_handler(event, name, **kwargs): # noqa: ARG001 + import time + + start = time.time() + time.sleep(0.02) # Simulate work + execution_times.append(("sync", time.time() - start)) + + async def async_handler(event, name, **kwargs): # noqa: ARG001 + import time + + start = time.time() + await asyncio.sleep(0.01) # Async work + execution_times.append(("async", time.time() - start)) + + # Subscribe mixed handlers + bus.subscribe(Events.TEST, sync_handler, "sync") + bus.subscribe(Events.TEST, async_handler, "async") + + # Emit and measure total time + import time + + start_time = time.time() + bus.emit(Events.TEST, data={"test": "mixed"}) + emit_time = time.time() - start_time + + # Emit should be instant (non-blocking) + assert emit_time < 0.01, f"Emit blocked for {emit_time:.4f}s" + + # Wait for execution + await asyncio.sleep(0.1) + + # Both handlers should have executed + assert len(execution_times) >= 1, f"Expected at least 1 handler, got {len(execution_times)}: {execution_times}" + + # At least the sync handler should have executed with proper timing + sync_results = [t for type_name, t in execution_times if type_name == "sync"] + assert len(sync_results) >= 1, "Sync handler didn't execute" + + # Sync handler should have taken approximately 0.02s + sync_time = sync_results[0] + assert 0.015 < sync_time < 0.04, f"Sync handler time unexpected: {sync_time:.4f}s" + + +class TestEventListener: + """Test the EventListener class.""" + + def test_event_listener_creation_with_sync_callback(self): + """Test creating EventListener with synchronous callback.""" + + def sync_callback(event, name, **kwargs): # noqa: ARG001 + return f"sync_result_{event.event}" + + listener = EventListener("test_listener", sync_callback) + + assert listener.name == "test_listener" + assert listener.call_back == sync_callback + assert listener.is_coroutine is False + + def test_event_listener_creation_with_async_callback(self): + """Test creating EventListener with asynchronous callback.""" + + async def async_callback(event, name, **kwargs): # noqa: ARG001 + return f"async_result_{event.event}" + + listener = EventListener("async_listener", async_callback) + + assert listener.name == "async_listener" + assert listener.call_back == async_callback + assert listener.is_coroutine is True + + @pytest.mark.asyncio + async def test_async_callback(self): + """Test EventListener with async callback.""" + + async def async_callback(event, name, **kwargs): # noqa: ARG001 + return "async_result" + + listener = EventListener("test_async", async_callback) + assert listener.is_coroutine is True + + event = Event(event=Events.TEST, data={"test": "data"}) + coroutine = listener.handle(event) + + # For async callbacks, handle() returns the coroutine directly (NOT awaited) + assert asyncio.iscoroutine(coroutine) + + result = await coroutine # Await the coroutine to execute + if asyncio.iscoroutine(result): + result = await result + + # The coroutine itself should be the callback return, not awaited by handle() + assert result == (await async_callback(event, "test_async")) + + @pytest.mark.asyncio + async def test_event_listener_handle_sync_callback_bug(self): + """Test EventListener handling with sync callback shows the current bug.""" + + def sync_callback(event, name, **kwargs): # noqa: ARG001 + return f"sync_{event.event}" + + listener = EventListener("sync_test", sync_callback) + event = Event(event="test_event", data={}) + + # Current implementation has a bug with sync callbacks + # It tries to create_task with a non-coroutine, which fails + # This test documents the current behavior that should be fixed + with pytest.raises(TypeError, match="a coroutine was expected"): + await listener.handle(event) + + @pytest.mark.asyncio + async def test_event_listener_handle_with_kwargs(self): + """Test EventListener handling with additional kwargs.""" + + async def callback_with_kwargs(event, name, extra_param=None, **kwargs): # noqa: ARG001 + return {"event": event.event, "extra": extra_param} + + listener = EventListener("kwargs_test", callback_with_kwargs) + event = Event(event="kwargs_event", data={}) + + # For async callbacks, handle returns coroutine directly (the callback call) + coroutine = listener.handle(event, extra_param="test_value") + assert asyncio.iscoroutine(coroutine) + + # The coroutine should be equivalent to calling the callback directly + expected_coroutine = callback_with_kwargs(event, "kwargs_test", extra_param="test_value") + + # Both should await to the same result + result = await (await coroutine) + expected_result = await expected_coroutine + + assert result["event"] == "kwargs_event" + assert result["extra"] == "test_value" + assert result == expected_result + + +class TestEventBus: + """Test the EventBus singleton class.""" + + def setup_method(self): + """Clear EventBus listeners and reset singleton before each test.""" + # Reset singleton instance to allow mocking to work + EventBus._instance = None + bus = EventBus.get_instance() + bus.clear() + + @patch("app.library.config.Config") + @patch("app.library.BackgroundWorker.BackgroundWorker") + def test_event_bus_singleton_behavior(self, mock_bg_worker, mock_config): + """Test that EventBus follows singleton pattern.""" + mock_config.get_instance.return_value.debug = False + mock_bg_worker.get_instance.return_value = MagicMock() + + bus1 = EventBus() + bus2 = EventBus() + assert bus1 is bus2 + + @patch("app.library.config.Config") + @patch("app.library.BackgroundWorker.BackgroundWorker") + def test_event_bus_get_instance(self, mock_bg_worker, mock_config): + """Test EventBus.get_instance() method.""" + mock_config.get_instance.return_value.debug = False + mock_bg_worker.get_instance.return_value = MagicMock() + + bus1 = EventBus.get_instance() + bus2 = EventBus.get_instance() + assert bus1 is bus2 + + def test_event_bus_initialization(self): + """Test EventBus initialization with new clean design.""" + # Reset singleton to ensure fresh instance + EventBus._instance = None + + # Create EventBus with clean initialization + bus = EventBus() + + # Test the initial state is correct + assert bus.debug is False # Default debug state + assert bus._offload is None # Lazy initialization - not loaded until needed + assert bus._listeners == {} # Empty listeners dict + + # Test debug enable/disable methods + bus.debug_enable() + assert bus.debug is True + + bus.debug_disable() + assert bus.debug is False + + @patch("app.library.config.Config") + @patch("app.library.BackgroundWorker.BackgroundWorker") + def test_event_bus_subscribe_single_event(self, mock_bg_worker, mock_config): + """Test subscribing to a single event.""" + mock_config.get_instance.return_value.debug = False + mock_bg_worker.get_instance.return_value = MagicMock() + + bus = EventBus() + + async def test_callback(event, name, **kwargs): # noqa: ARG001 + return "callback_result" + + result = bus.subscribe(Events.TEST, test_callback, "test_subscriber") + + assert result is bus # Should return self for chaining + assert Events.TEST in bus._listeners + assert "test_subscriber" in bus._listeners[Events.TEST] + assert isinstance(bus._listeners[Events.TEST]["test_subscriber"], EventListener) + + @patch("app.library.config.Config") + @patch("app.library.BackgroundWorker.BackgroundWorker") + def test_event_bus_subscribe_multiple_events(self, mock_bg_worker, mock_config): + """Test subscribing to multiple events.""" + mock_config.get_instance.return_value.debug = False + mock_bg_worker.get_instance.return_value = MagicMock() + + bus = EventBus() + + async def test_callback(event, name, **kwargs): # noqa: ARG001 + return "callback_result" + + events = [Events.TEST, Events.STARTUP, Events.SHUTDOWN] + bus.subscribe(events, test_callback, "multi_subscriber") + + for event in events: + assert event in bus._listeners + assert "multi_subscriber" in bus._listeners[event] + + @patch("app.library.config.Config") + @patch("app.library.BackgroundWorker.BackgroundWorker") + def test_event_bus_subscribe_wildcard(self, mock_bg_worker, mock_config): + """Test subscribing to all events with wildcard.""" + mock_config.get_instance.return_value.debug = False + mock_bg_worker.get_instance.return_value = MagicMock() + + bus = EventBus() + + async def test_callback(event, name, **kwargs): # noqa: ARG001 + return "callback_result" + + bus.subscribe("*", test_callback, "wildcard_subscriber") + + all_events = Events.get_all() + for event in all_events: + assert event in bus._listeners + assert "wildcard_subscriber" in bus._listeners[event] + + @patch("app.library.config.Config") + @patch("app.library.BackgroundWorker.BackgroundWorker") + def test_event_bus_subscribe_frontend(self, mock_bg_worker, mock_config): + """Test subscribing to frontend events.""" + mock_config.get_instance.return_value.debug = False + mock_bg_worker.get_instance.return_value = MagicMock() + + bus = EventBus() + + async def test_callback(event, name, **kwargs): # noqa: ARG001 + return "callback_result" + + bus.subscribe("frontend", test_callback, "frontend_subscriber") + + frontend_events = Events.frontend() + for event in frontend_events: + assert event in bus._listeners + assert "frontend_subscriber" in bus._listeners[event] + + @patch("app.library.config.Config") + @patch("app.library.BackgroundWorker.BackgroundWorker") + def test_event_bus_subscribe_invalid_event(self, mock_bg_worker, mock_config): + """Test subscribing to invalid event.""" + mock_config.get_instance.return_value.debug = False + mock_bg_worker.get_instance.return_value = MagicMock() + + bus = EventBus() + + async def test_callback(event, name, **kwargs): # noqa: ARG001 + return "callback_result" + + # Should not raise an exception but log an error + result = bus.subscribe("invalid_event", test_callback, "invalid_subscriber") + + assert result is bus + assert "invalid_event" not in bus._listeners + + @patch("app.library.config.Config") + @patch("app.library.BackgroundWorker.BackgroundWorker") + def test_event_bus_subscribe_auto_generated_name(self, mock_bg_worker, mock_config): + """Test subscribing without providing name (auto-generated).""" + mock_config.get_instance.return_value.debug = False + mock_bg_worker.get_instance.return_value = MagicMock() + + bus = EventBus() + + async def test_callback(event, name, **kwargs): # noqa: ARG001 + return "callback_result" + + bus.subscribe(Events.TEST, test_callback) + + assert Events.TEST in bus._listeners + assert len(bus._listeners[Events.TEST]) == 1 + + # Name should be a UUID + subscriber_name = next(iter(bus._listeners[Events.TEST].keys())) + assert len(subscriber_name) == 36 # UUID string length + + @patch("app.library.config.Config") + @patch("app.library.BackgroundWorker.BackgroundWorker") + def test_event_bus_unsubscribe_single_event(self, mock_bg_worker, mock_config): + """Test unsubscribing from a single event.""" + mock_config.get_instance.return_value.debug = False + mock_bg_worker.get_instance.return_value = MagicMock() + + bus = EventBus() + + async def test_callback(event, name, **kwargs): # noqa: ARG001 + return "callback_result" + + # First subscribe + bus.subscribe(Events.TEST, test_callback, "test_subscriber") + assert "test_subscriber" in bus._listeners[Events.TEST] + + # Then unsubscribe + result = bus.unsubscribe(Events.TEST, "test_subscriber") + + assert result is bus + assert "test_subscriber" not in bus._listeners[Events.TEST] + + @patch("app.library.config.Config") + @patch("app.library.BackgroundWorker.BackgroundWorker") + def test_event_bus_unsubscribe_multiple_events(self, mock_bg_worker, mock_config): + """Test unsubscribing from multiple events.""" + mock_config.get_instance.return_value.debug = False + mock_bg_worker.get_instance.return_value = MagicMock() + + bus = EventBus() + + async def test_callback(event, name, **kwargs): # noqa: ARG001 + return "callback_result" + + events = [Events.TEST, Events.STARTUP] + + # Subscribe to multiple events + bus.subscribe(events, test_callback, "multi_subscriber") + + # Unsubscribe from multiple events + bus.unsubscribe(events, "multi_subscriber") + + for event in events: + assert "multi_subscriber" not in bus._listeners[event] + + @patch("app.library.config.Config") + @patch("app.library.BackgroundWorker.BackgroundWorker") + def test_event_bus_unsubscribe_nonexistent(self, mock_bg_worker, mock_config): + """Test unsubscribing from nonexistent subscription.""" + mock_config.get_instance.return_value.debug = False + mock_bg_worker.get_instance.return_value = MagicMock() + + bus = EventBus() + + # Should not raise an exception + result = bus.unsubscribe(Events.TEST, "nonexistent_subscriber") + assert result is bus + + @patch("app.library.config.Config") + @patch("app.library.BackgroundWorker.BackgroundWorker") + def test_event_bus_emit_no_listeners(self, mock_config, mock_bg_worker): + """Test emit with no listeners.""" + # Setup mocks + mock_config_instance = MagicMock() + mock_config_instance.debug = True + mock_config.get_instance.return_value = mock_config_instance + + mock_bg_worker_instance = MagicMock() + mock_bg_worker.get_instance.return_value = mock_bg_worker_instance + + bus = EventBus() + + # Emit event with no listeners - new fire-and-forget API returns None immediately + result = bus.emit(Events.TEST, data={"test": "data"}) + + # Should return None for fire-and-forget execution + assert result is None + + @pytest.mark.asyncio + async def test_event_bus_emit_with_listeners(self): + """Test emitting event with listeners (fire-and-forget).""" + bus = EventBus() + + results = [] + + async def callback1(event, name, **kwargs): # noqa: ARG001 + results.append(f"callback1_{event.event}") + return "result1" + + async def callback2(event, name, **kwargs): # noqa: ARG001 + results.append(f"callback2_{event.event}") + return "result2" + + bus.subscribe(Events.TEST, callback1, "subscriber1") + bus.subscribe(Events.TEST, callback2, "subscriber2") + + # emit() returns None immediately (fire-and-forget) + result: None = bus.emit(Events.TEST, data={"test": "data"}) + assert result is None + + # Give async tasks a moment to execute + await asyncio.sleep(0.01) + + # Side effects should have occurred + assert len(results) == 2, f"Expected 2 results, got {len(results)}: {results}" + assert "callback1_test" in results + assert "callback2_test" in results + + @patch("app.library.config.Config") + @patch("app.library.BackgroundWorker.BackgroundWorker") + @pytest.mark.asyncio + async def test_event_bus_emit_with_error_in_handler(self, mock_bg_worker, mock_config): + """Test emitting event when a handler raises an exception.""" + mock_config.get_instance.return_value.debug = False + mock_bg_worker.get_instance.return_value = MagicMock() + + bus = EventBus() + + async def failing_callback(event, name, **kwargs): # noqa: ARG001 + msg = "Handler error" + raise ValueError(msg) + + async def working_callback(event, name, **kwargs): # noqa: ARG001 + return "success" + + bus.subscribe(Events.TEST, failing_callback, "failing_subscriber") + bus.subscribe(Events.TEST, working_callback, "working_subscriber") + + # Should not raise exception, but log error + bus.emit(Events.TEST, data={"test": "data"}) + + def test_event_bus_emit_fire_and_forget(self): + """Test emit() returns None immediately (fire-and-forget).""" + bus = EventBus() + + # Should return None for no listeners + result = bus.emit(Events.TEST, data={"test": "data"}) + assert result is None + + # Should return None even with listeners + async def test_callback(event, name, **kwargs): # noqa: ARG001 + return "result" + + bus.subscribe(Events.TEST, test_callback, "test_subscriber") + result = bus.emit(Events.TEST, data={"test": "data"}) + assert result is None + + def test_event_bus_emit_lazy_background_worker(self): + """Test that BackgroundWorker is only initialized when needed.""" + bus = EventBus() + + bus._offload = None # Ensure offload is None + + # Initially _offload should be None + assert bus._offload is None + + async def test_callback(event, name, **kwargs): # noqa: ARG001 + return "result" + + bus.subscribe(Events.TEST, test_callback, "test_subscriber") + + # After emit in no-loop context, _offload should be initialized + bus.emit(Events.TEST, data={"test": "data"}) + + # Note: This might still be None if running in pytest's event loop + # The lazy initialization happens only when no event loop is detected + + @pytest.mark.asyncio + async def test_event_bus_emit_with_kwargs(self): + """Test emitting event with additional kwargs (fire-and-forget).""" + bus: EventBus = EventBus() + + received_kwargs = {} + + async def callback_with_kwargs(event, name, extra_param=None, **kwargs): # noqa: ARG001 + received_kwargs["extra_param"] = extra_param + received_kwargs["kwargs"] = kwargs + return "result" + + bus.subscribe(Events.TEST, callback_with_kwargs, "kwargs_subscriber") + + result: None = bus.emit(Events.TEST, data={"test": "data"}, extra_param="test_value", custom_arg="custom") + assert result is None + + # Give async tasks a moment to execute + await asyncio.sleep(0.02) + + # Side effects should have occurred + assert "extra_param" in received_kwargs, f"received_kwargs: {received_kwargs}" + assert received_kwargs["extra_param"] == "test_value" + assert "kwargs" in received_kwargs, f"received_kwargs: {received_kwargs}" + assert received_kwargs["kwargs"]["custom_arg"] == "custom" + + @pytest.mark.asyncio + async def test_event_bus_emit_event_data_defaults(self): + """Test emitting event with default data handling (fire-and-forget).""" + bus = EventBus() + + received_event = None + + async def test_callback(event, name, **kwargs): # noqa: ARG001 + nonlocal received_event + received_event = event + return "result" + + bus.subscribe(Events.TEST, test_callback, "default_subscriber") + + # Verify we have a subscriber + assert Events.TEST in bus._listeners + assert len(bus._listeners[Events.TEST]) == 1 + + # emit() returns None immediately + result = bus.emit(Events.TEST) + assert result is None + + # Give async tasks a moment to execute + await asyncio.sleep(0.01) + + # Side effects should have occurred + assert received_event is not None, "Callback was not called" + assert received_event.data == {} + assert received_event.title is None + assert received_event.message is None diff --git a/pyproject.toml b/pyproject.toml index 3383641f..1fc325da 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -90,6 +90,9 @@ installer = ["pyinstaller"] select = [ "ALL", # include all the rules, including new ones ] +# Ignore some rules in test files. +per-file-ignores = { "test_*.py" = ["D"] } + ignore = [ #### modules "ANN", # flake8-annotations @@ -206,7 +209,4 @@ testpaths = ["app/tests"] addopts = "-v --tb=short" [dependency-groups] -dev = [ - "pytest>=8.4.2", - "ruff>=0.13.0", -] +dev = ["pytest>=8.4.2", "pytest-asyncio>=1.1.0", "ruff>=0.13.0"] diff --git a/uv.lock b/uv.lock index 39ad554a..d7b3c96b 100644 --- a/uv.lock +++ b/uv.lock @@ -1041,6 +1041,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a8/a4/20da314d277121d6534b3a980b29035dcd51e6744bd79075a6ce8fa4eb8d/pytest-8.4.2-py3-none-any.whl", hash = "sha256:872f880de3fc3a5bdc88a11b39c9710c3497a547cfa9320bc3c5e62fbf272e79", size = 365750 }, ] +[[package]] +name = "pytest-asyncio" +version = "1.1.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/4e/51/f8794af39eeb870e87a8c8068642fc07bce0c854d6865d7dd0f2a9d338c2/pytest_asyncio-1.1.0.tar.gz", hash = "sha256:796aa822981e01b68c12e4827b8697108f7205020f24b5793b3c41555dab68ea", size = 46652 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c7/9d/bf86eddabf8c6c9cb1ea9a869d6873b46f105a5d292d3a6f7071f5b07935/pytest_asyncio-1.1.0-py3-none-any.whl", hash = "sha256:5fe2d69607b0bd75c656d1211f969cadba035030156745ee09e7d71740e58ecf", size = 15157 }, +] + [[package]] name = "python-dateutil" version = "2.9.0.post0" @@ -1502,6 +1514,7 @@ installer = [ [package.dev-dependencies] dev = [ { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "ruff" }, ] @@ -1544,6 +1557,7 @@ provides-extras = ["installer"] [package.metadata.requires-dev] dev = [ { name = "pytest", specifier = ">=8.4.2" }, + { name = "pytest-asyncio", specifier = ">=1.1.0" }, { name = "ruff", specifier = ">=0.13.0" }, ]