removed events_listener; added basic test for webhooks_listener

Signed-off-by: bigcat88 <[email protected]>

Author: bigcat88

docs/reference/ExApp.rst

@@ -87,12 +87,6 @@ UI methods should be accessed with the help of :class:`~nc_py_api.nextcloud.Next
 .. autoclass:: nc_py_api.ex_app.providers.task_processing._TaskProcessingProviderAPI
     :members:
 
-.. autoclass:: nc_py_api.ex_app.events_listener.EventsListener
-    :members:
-
-.. autoclass:: nc_py_api.ex_app.events_listener.EventsListenerAPI
-    :members:
-
 .. autoclass:: nc_py_api.ex_app.occ_commands.OccCommand
     :members:
 

nc_py_api/ex_app/events_listener.py

@@ -31,7 +31,6 @@
 from .apps import _AppsAPI, _AsyncAppsAPI
 from .calendar_api import _CalendarAPI
 from .ex_app.defs import LogLvl
-from .ex_app.events_listener import AsyncEventsListenerAPI, EventsListenerAPI
 from .ex_app.occ_commands import AsyncOccCommandsAPI, OccCommandsAPI
 from .ex_app.providers.providers import AsyncProvidersApi, ProvidersApi
 from .ex_app.ui.ui import AsyncUiApi, UiApi
@@ -327,8 +326,6 @@ class NextcloudApp(_NextcloudBasic):
     ui: UiApi
     """Nextcloud UI API for ExApps"""
     providers: ProvidersApi
-    """API for registering providers for Nextcloud"""
-    events_listener: EventsListenerAPI
     """API for registering Events listeners for ExApps"""
     occ_commands: OccCommandsAPI
     """API for registering OCC command for ExApps"""
@@ -344,7 +341,6 @@ def __init__(self, **kwargs):
         self.preferences_ex = PreferencesExAPI(self._session)
         self.ui = UiApi(self._session)
         self.providers = ProvidersApi(self._session)
-        self.events_listener = EventsListenerAPI(self._session)
         self.occ_commands = OccCommandsAPI(self._session)
 
     @property
@@ -462,8 +458,6 @@ class AsyncNextcloudApp(_AsyncNextcloudBasic):
     ui: AsyncUiApi
     """Nextcloud UI API for ExApps"""
     providers: AsyncProvidersApi
-    """API for registering providers for Nextcloud"""
-    events_listener: AsyncEventsListenerAPI
     """API for registering Events listeners for ExApps"""
     occ_commands: AsyncOccCommandsAPI
     """API for registering OCC command for ExApps"""
@@ -479,7 +473,6 @@ def __init__(self, **kwargs):
         self.preferences_ex = AsyncPreferencesExAPI(self._session)
         self.ui = AsyncUiApi(self._session)
         self.providers = AsyncProvidersApi(self._session)
-        self.events_listener = AsyncEventsListenerAPI(self._session)
         self.occ_commands = AsyncOccCommandsAPI(self._session)
 
     @property

nc_py_api/nextcloud.py

@@ -2,6 +2,7 @@
 
 import dataclasses
 
+from ._exceptions import NextcloudExceptionNotFound
 from ._misc import clear_from_params_empty  # , require_capabilities
 from ._session import AppConfig, AsyncNcSessionBasic, NcSessionBasic
 
@@ -84,8 +85,11 @@ def get_list(self, uri_filter: str = "") -> list[WebhookInfo]:
         params = {"uri": uri_filter} if uri_filter else {}
         return [WebhookInfo(i) for i in self._session.ocs("GET", f"{self._ep_base}", params=params)]
 
-    def get_entry(self, webhook_id: int) -> WebhookInfo:
-        return WebhookInfo(self._session.ocs("GET", f"{self._ep_base}/{webhook_id}"))
+    def get_entry(self, webhook_id: int) -> WebhookInfo | None:
+        try:
+            return WebhookInfo(self._session.ocs("GET", f"{self._ep_base}/{webhook_id}"))
+        except NextcloudExceptionNotFound:
+            return None
 
     def register(
         self,
@@ -151,7 +155,7 @@ def unregister_all(self, appid: str = "") -> int:
 class _AsyncWebhooksAPI:
     """The class provides the async application management API on the Nextcloud server."""
 
-    _ep_base: str = "/ocs/v1.php/webhooks"
+    _ep_base: str = "/ocs/v1.php/apps/webhook_listeners/api/v1/webhooks"
 
     def __init__(self, session: AsyncNcSessionBasic):
         self._session = session
@@ -160,8 +164,11 @@ async def get_list(self, uri_filter: str = "") -> list[WebhookInfo]:
         params = {"uri": uri_filter} if uri_filter else {}
         return [WebhookInfo(i) for i in await self._session.ocs("GET", f"{self._ep_base}", params=params)]
 
-    async def get_entry(self, webhook_id: int) -> WebhookInfo:
-        return WebhookInfo(await self._session.ocs("GET", f"{self._ep_base}/{webhook_id}"))
+    async def get_entry(self, webhook_id: int) -> WebhookInfo | None:
+        try:
+            return WebhookInfo(await self._session.ocs("GET", f"{self._ep_base}/{webhook_id}"))
+        except NextcloudExceptionNotFound:
+            return None
 
     async def register(
         self,

nc_py_api/webhooks.py

@@ -0,0 +1,60 @@
+import pytest
+
+
+def test_events_registration(nc_app):
+    assert isinstance(nc_app.webhooks.get_list(), list)
+    assert isinstance(nc_app.webhooks.unregister_all(), int)
+    assert nc_app.webhooks.unregister_all() == 0
+    assert nc_app.webhooks.get_list() == []
+    webhook_info = nc_app.webhooks.register(
+        "POST",
+        "/some_url",
+        "OCP\\Files\\Events\\Node\\NodeCreatedEvent",
+    )
+    result = nc_app.webhooks.get_entry(webhook_info.webhook_id)
+    assert result.webhook_id == webhook_info.webhook_id
+    assert result.app_id == "nc_py_api"
+    assert result.http_method == "POST"
+    assert result.uri == "/some_url"
+    assert result.auth_method == "none"
+    assert result.auth_data == {}
+    assert result.user_id_filter == ""
+    assert result.event_filter == []
+    assert result.event == "OCP\\Files\\Events\\Node\\NodeCreatedEvent"
+    result = nc_app.webhooks.update(
+        webhook_info.webhook_id, http_method="PUT", uri="/some_url2", event="OCP\\Files\\Events\\Node\\NodeCreatedEvent"
+    )
+    assert result.webhook_id == webhook_info.webhook_id
+    nc_app.webhooks.unregister(webhook_info.webhook_id)
+    nc_app.webhooks.unregister(webhook_info.webhook_id)  # removing non-existing webhook should not fail
+    assert nc_app.webhooks.get_entry(webhook_info.webhook_id) is None
+
+
+@pytest.mark.asyncio(scope="session")
+async def test_events_registration_async(anc_app):
+    assert isinstance(await anc_app.webhooks.get_list(), list)
+    assert isinstance(await anc_app.webhooks.unregister_all(), int)
+    assert await anc_app.webhooks.unregister_all() == 0
+    assert await anc_app.webhooks.get_list() == []
+    webhook_info = await anc_app.webhooks.register(
+        "POST",
+        "/some_url",
+        "OCP\\Files\\Events\\Node\\NodeCreatedEvent",
+    )
+    result = await anc_app.webhooks.get_entry(webhook_info.webhook_id)
+    assert result.webhook_id == webhook_info.webhook_id
+    assert result.app_id == "nc_py_api"
+    assert result.http_method == "POST"
+    assert result.uri == "/some_url"
+    assert result.auth_method == "none"
+    assert result.auth_data == {}
+    assert result.user_id_filter == ""
+    assert result.event_filter == []
+    assert result.event == "OCP\\Files\\Events\\Node\\NodeCreatedEvent"
+    result = await anc_app.webhooks.update(
+        webhook_info.webhook_id, http_method="PUT", uri="/some_url2", event="OCP\\Files\\Events\\Node\\NodeCreatedEvent"
+    )
+    assert result.webhook_id == webhook_info.webhook_id
+    await anc_app.webhooks.unregister(webhook_info.webhook_id)
+    await anc_app.webhooks.unregister(webhook_info.webhook_id)  # removing non-existing webhook should not fail
+    assert await anc_app.webhooks.get_entry(webhook_info.webhook_id) is None