From b96c741f547fa39dfd801da01ca88af6b2546a89 Mon Sep 17 00:00:00 2001 From: dev-KingMaster <136489418+dev-KingMaster@users.noreply.github.com> Date: Wed, 3 Dec 2025 14:35:40 -0500 Subject: [PATCH] Nextcloud group notification support (#1440) --- apprise/plugins/nextcloud.py | 432 +++++++++++++++++++++++++-------- tests/test_plugin_nextcloud.py | 409 +++++++++++++++++++++++++++++-- 2 files changed, 719 insertions(+), 122 deletions(-) diff --git a/apprise/plugins/nextcloud.py b/apprise/plugins/nextcloud.py index 4e4e5868..be70f05a 100644 --- a/apprise/plugins/nextcloud.py +++ b/apprise/plugins/nextcloud.py @@ -25,17 +25,49 @@ # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. +from itertools import chain +from json import loads +import re + import requests -from ..common import NotifyType +from ..common import NotifyType, PersistentStoreMode +from ..exception import AppriseException from ..locale import gettext_lazy as _ from ..url import PrivacyMode from ..utils.parse import parse_list from .base import NotifyBase +# Is Group Detection +IS_GROUP = re.compile( + r"^\s*((#|%23)(?P[a-z0-9_-]+)|" + r"((#|%23)?(?Pall|everyone|\*)))\s*$", + re.I, +) + +# Is User Detection +IS_USER = re.compile( + r"^\s*(@|%40)?(?P[a-z0-9_-]+)\s*$", + re.I, +) + + +class NextcloudGroupDiscoveryException(AppriseException): + """Apprise Nextcloud Group Discovery Exception Class.""" + class NotifyNextcloud(NotifyBase): - """A wrapper for Nextcloud Notifications.""" + """A wrapper for Nextcloud Notifications. + + Targets can be individual users, groups, or everyone: + - user: specify one or more usernames as path segments + - group: prefix with a hash (e.g., ``#DevTeam``) + - everyone: use ``all`` (aliases: ``everyone``, ``*``) + + Group and everyone expansion uses Nextcloud's OCS provisioning API and + requires appropriate permissions (typically an admin account) and the + provisioning API enabled on the server. + """ # The default descriptive name associated with the Notification service_name = "Nextcloud" @@ -58,6 +90,16 @@ class NotifyNextcloud(NotifyBase): # Defines the maximum allowable characters per message. body_maxlen = 4000 + # Our default is to not use persistent storage beyond in-memory + # reference + storage_mode = PersistentStoreMode.AUTO + + # Defines how long we cache our discovery for + group_discovery_cache_length_sec = 86400 + + # unique identifier to cache the 'all' group category + all_group_id = "all" + # Define object templates templates = ( "{schema}://{host}/{targets}", @@ -94,6 +136,13 @@ class NotifyNextcloud(NotifyBase): "name": _("Target User"), "type": "string", "map_to": "targets", + "prefix": "@", + }, + "target_group": { + "name": _("Target Group"), + "type": "string", + "map_to": "targets", + "prefix": "#", }, "targets": { "name": _("Targets"), @@ -146,7 +195,30 @@ class NotifyNextcloud(NotifyBase): super().__init__(**kwargs) # Store our targets - self.targets = parse_list(targets) + self.targets = [] + self.groups = set() + + for target in parse_list(targets): + results = IS_GROUP.match(target) + if results: + group_id = ( + self.all_group_id + if results.group("all") else results.group("group")) + + self.groups.add(group_id) + self.logger.debug("Added Nextcloud group '%s'", group_id) + continue + + results = IS_USER.match(target) + if results: + # Store our target + self.targets.append(results.group("user")) + self.logger.debug( + "Added Nextcloud user '%s'", self.targets[-1]) + continue + + self.logger.warning( + "Ignored invalid Nextcloud user/group '%s'", target) self.version = self.template_args["version"]["default"] if version is not None: @@ -164,7 +236,8 @@ class NotifyNextcloud(NotifyBase): raise TypeError(msg) from None # Support URL Prefix - self.url_prefix = "" if not url_prefix else url_prefix.strip("/") + self.url_prefix = "" if not url_prefix else ( + "/" + url_prefix.strip("/")) self.headers = {} if headers: @@ -173,31 +246,175 @@ class NotifyNextcloud(NotifyBase): return - def send(self, body, title="", notify_type=NotifyType.INFO, **kwargs): - """Perform Nextcloud Notification.""" + def _fetch(self, payload=None, target=None, group=None): + """Wrapper to NextCloud API requests object.""" - if len(self.targets) == 0: - # There were no services to notify - self.logger.warning("There were no Nextcloud targets to notify.") - return False + # our method + method = "POST" if target else "GET" # Prepare our Header headers = { "User-Agent": self.app_id, "OCS-APIREQUEST": "true", + "Accept": "application/json", } # Apply any/all header over-rides defined headers.update(self.headers) - # error tracking (used for function return) + # Prepare base URL fragments + scheme = "https" if self.secure else "http" + host_port = ( + self.host + if not isinstance(self.port, int) + else f"{self.host}:{self.port}" + ) + base = f"{scheme}://{host_port}" + if self.url_prefix: + base = f"{base}{self.url_prefix}" + + # Auth + auth = (self.user, self.password) if self.user else None + + # our URL Parameters + params = {} + + # our response + content = None + + if target: + # Nextcloud URL based on version used + query = f'v{self.version} Notify "{target}"' + esc_target = NotifyNextcloud.quote(target) + url = ( + f"{base}/ocs/v2.php/" + "apps/admin_notifications/" + f"api/v1/notifications/{esc_target}" + if self.version < 21 + else ( + f"{base}/ocs/v2.php/" + "apps/notifications/" + f"api/v2/admin_notifications/{esc_target}" + ) + ) + + elif group: + query = f'Group "{group}"' + params = { + "format": "json", + } + esc_group = NotifyNextcloud.quote(group) + url = f"{base}/ocs/v1.php/cloud/groups/{esc_group}" + + else: # Users + query = "Users" + params = { + "format": "json", + } + url = f"{base}/ocs/v1.php/cloud/users" + + self.throttle() + self.logger.debug( + "Nextcloud %s %s URL: %s (cert_verify=%r)", + query, + method, + url, + self.verify_certificate, + ) + + if payload: + self.logger.debug( + "Nextcloud v%d Payload: %s", self.version, str(payload) + ) + + try: + # Prepare our request object + request = requests.post if target else requests.get + + r = request( + url, + headers=headers, + data=payload, + params=params, + auth=auth, + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + + try: + content = loads(r.content) + + except (AttributeError, TypeError, ValueError): + # ValueError = r.content is Unparsable + # TypeError = r.content is None + # AttributeError = r is None + content = {} + + if r.status_code != requests.codes.ok: + status_str = NotifyNextcloud.http_response_code_lookup( + r.status_code + ) + + self.logger.warning( + "Failed to send Nextcloud %s: %s%serror=%d.", + query, + status_str, + ", " if status_str else "", + r.status_code, + ) + + self.logger.debug(f"Response Details:\r\n{r.content}") + if target: + return (False, content) + + raise NextcloudGroupDiscoveryException( + f"{query} non-200 response") + + except requests.RequestException as e: + self.logger.warning( + "A Connection error occurred with Nextcloud %s", + query, + ) + self.logger.debug(f"Socket Exception: {e!s}") + + if target: + return (False, None) + raise NextcloudGroupDiscoveryException( + f"{query} socket exception") from None + + self.logger.info("Sent Nextcloud %s", query) + return (True, content) + + def send(self, body, title="", notify_type=NotifyType.INFO, **kwargs): + """Perform Nextcloud Notification.""" + + # Create a copy of our targets + targets = set(self.targets) + + # Initialize our has_error flag has_error = False - # Create a copy of the targets list - targets = list(self.targets) - while len(targets): - target = targets.pop(0) + if self.groups: + # Append our group lookup + try: + for group in self.groups: + if group == self.all_group_id: + targets |= self.all_users() + else: + # specific group + targets |= self.users_by_group(group) + + except NextcloudGroupDiscoveryException: + # logging already handled within all_users and user_by_group() + return False + + if not targets: + # There were no services to notify + self.logger.warning("There were no Nextcloud targets to notify.") + return False + + for target in targets: # Prepare our Payload payload = { "shortMessage": title if title else self.app_desc, @@ -207,96 +424,88 @@ class NotifyNextcloud(NotifyBase): # doesn't take kindly to empty longMessage entries. payload["longMessage"] = body - auth = None - if self.user: - auth = (self.user, self.password) - - # Nextcloud URL based on version used - notify_url = ( - "{schema}://{host}/{url_prefix}/ocs/v2.php/" - "apps/admin_notifications/" - "api/v1/notifications/{target}" - if self.version < 21 - else ( - "{schema}://{host}/{url_prefix}/ocs/v2.php/" - "apps/notifications/" - "api/v2/admin_notifications/{target}" - ) - ) - - notify_url = notify_url.format( - schema="https" if self.secure else "http", - host=( - self.host - if not isinstance(self.port, int) - else f"{self.host}:{self.port}" - ), - url_prefix=self.url_prefix, - target=target, - ) - - self.logger.debug( - "Nextcloud v%d POST URL: %s (cert_verify=%r)", - self.version, - notify_url, - self.verify_certificate, - ) - self.logger.debug( - "Nextcloud v%d Payload: %s", self.version, str(payload) - ) - - # Always call throttle before any remote server i/o is made - self.throttle() - - try: - r = requests.post( - notify_url, - data=payload, - headers=headers, - auth=auth, - verify=self.verify_certificate, - timeout=self.request_timeout, - ) - if r.status_code != requests.codes.ok: - # We had a problem - status_str = NotifyNextcloud.http_response_code_lookup( - r.status_code - ) - - self.logger.warning( - "Failed to send Nextcloud v{} notification:" - "{}{}error={}.".format( - self.version, - status_str, - ", " if status_str else "", - r.status_code, - ) - ) - - self.logger.debug(f"Response Details:\r\n{r.content}") - # track our failure - has_error = True - continue - - else: - self.logger.info( - "Sent Nextcloud %d notification.", self.version - ) - - except requests.RequestException as e: - self.logger.warning( - "A Connection error occurred sending Nextcloud v%d" - "notification.", - self.version, - ) - self.logger.debug(f"Socket Exception: {e!s}") - - # track our failure + is_okay, _ = self._fetch(payload, target) + if not is_okay: + # Toggle our status has_error = True - continue return not has_error + def users_by_group(self, group): + """ + Lists users associated with a provided group + """ + + # Check our cache + targets = self.store.get(group) + if targets is not None: + # Returned cached value + self.logger.trace( + f"Using Nextcloud cached response for group '{group}' " + "query") + return set(targets) + + # _fetch throws an exception if it fails, so we can + # go ahead and ignore checking for it. + _, response = self._fetch(group=group) + + # Initialize our targets + targets = set() + + # If we get here, our fetch was successful; look up our users + users = response.get("ocs", {}).get("data", {}).get("users") + if isinstance(users, list): + targets = { + s for u in users if (s := str(u).strip()) + } + + if not targets: + self.logger.warning( + "No users associated with Nextcloud group '%s'", group + ) + + self.store.set( + group, list(targets), + expires=self.group_discovery_cache_length_sec) + return targets + + def all_users(self): + """ + Lists users associated with Nextcloud instance + """ + # Check our cache + targets = self.store.get(self.all_group_id) + if targets is not None: + self.logger.trace( + "Using Nextcloud cached response for all-user query") + return set(targets) + + # _fetch throws an exception if it fails, so we can + # go ahead and ignore checking for it. + _, response = self._fetch() + + # Initialize our targets + targets = set() + + # If we get here, our fetch was successful; look up our users + users = response.get("ocs", {}).get("data", {}).get("users") + if isinstance(users, list): + targets = { + s for u in users if (s := str(u).strip()) + } + + if not targets: + self.logger.warning( + "Failed to retrieve all users from Nextcloud", + ) + # early exit; no cache + return targets + + self.store.set( + self.all_group_id, + list(targets), expires=self.group_discovery_cache_length_sec) + return targets + @property def url_identifier(self): """Returns all of the identifiers that make this URL unique from @@ -310,6 +519,7 @@ class NotifyNextcloud(NotifyBase): self.password, self.host, self.port, + self.url_prefix, ) def url(self, privacy=False, *args, **kwargs): @@ -321,7 +531,7 @@ class NotifyNextcloud(NotifyBase): # Set our version params["version"] = str(self.version) - if self.url_prefix: + if self.url_prefix.rstrip("/"): params["url_prefix"] = self.url_prefix # Extend our parameters @@ -341,6 +551,9 @@ class NotifyNextcloud(NotifyBase): user=NotifyNextcloud.quote(self.user, safe=""), ) + group_prefix = self.template_tokens["target_group"]["prefix"] + user_prefix = self.template_tokens["target_user"]["prefix"] + default_port = 443 if self.secure else 80 return "{schema}://{auth}{hostname}{port}/{targets}?{params}".format( schema=self.secure_protocol if self.secure else self.protocol, @@ -353,14 +566,23 @@ class NotifyNextcloud(NotifyBase): if self.port is None or self.port == default_port else f":{self.port}" ), - targets="/".join([NotifyNextcloud.quote(x) for x in self.targets]), + targets="/".join([ + NotifyNextcloud.quote(x, safe=(group_prefix + user_prefix)) + for x in chain( + # Groups are prefixed with a pound/hashtag symbol + [f"{group_prefix}{x}" for x in self.groups], + # Users + [f"{user_prefix}{x}" for x in self.targets], + ) + ]), + params=NotifyNextcloud.urlencode(params), ) def __len__(self): """Returns the number of targets associated with this notification.""" - targets = len(self.targets) - return targets if targets else 1 + targets = len(self.targets) + len(self.groups) + return max(1, targets) @staticmethod def parse_url(url): diff --git a/tests/test_plugin_nextcloud.py b/tests/test_plugin_nextcloud.py index 2f25e07a..4e267a87 100644 --- a/tests/test_plugin_nextcloud.py +++ b/tests/test_plugin_nextcloud.py @@ -26,15 +26,22 @@ # POSSIBILITY OF SUCH DAMAGE. # Disable logging for a cleaner testing output +from json import dumps import logging from unittest import mock from helpers import AppriseURLTester import requests -from apprise import Apprise, NotifyType +from apprise import Apprise, AppriseAsset, NotifyType, PersistentStoreMode from apprise.plugins.nextcloud import NotifyNextcloud +NEXTCLOUD_GOOD_RESPONSE = dumps({ + "ocs": { + "meta": {"status": "ok", "statuscode": 100}, + "data": {"users": ["user1", "user2"]}, + }}) + logging.disable(logging.CRITICAL) apprise_url_tests = ( @@ -45,12 +52,15 @@ apprise_url_tests = ( "ncloud://:@/", { "instance": None, + # Our response expected server response + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( "ncloud://", { "instance": None, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( @@ -58,6 +68,7 @@ apprise_url_tests = ( { # No hostname "instance": None, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( @@ -68,6 +79,7 @@ apprise_url_tests = ( # Since there are no targets specified we expect a False return on # send() "notify_response": False, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( @@ -75,6 +87,7 @@ apprise_url_tests = ( { # An invalid version was specified "instance": TypeError, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( @@ -82,6 +95,7 @@ apprise_url_tests = ( { # An invalid version was specified "instance": TypeError, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( @@ -89,88 +103,122 @@ apprise_url_tests = ( { # An invalid version was specified "instance": TypeError, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( "ncloud://localhost/admin", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( "ncloud://user@localhost/admin", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( "ncloud://user@localhost?to=user1,user2", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( "ncloud://user@localhost?to=user1,user2&version=20", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( "ncloud://user@localhost?to=user1,user2&version=21", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( "ncloud://user@localhost?to=user1&version=20&url_prefix=/abcd", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( "ncloud://user@localhost?to=user1&version=21&url_prefix=/abcd", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( "ncloud://user:pass@localhost/user1/user2", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, # Our expected url(privacy=True) startswith() response: - "privacy_url": "ncloud://user:****@localhost/user1/user2", + "privacy_url": "ncloud://user:****@localhost/@user1/@user2", + }, + ), + ( + "ncloud://user:pass@localhost/#group1/#group2/#group1", + { + # Test groups, but also note a duplicate group provided + "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, + "privacy_url": "ncloud://user:****@localhost/#group", }, ), ( "ncloud://user:pass@localhost:8080/admin", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( "nclouds://user:pass@localhost/admin", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, # Our expected url(privacy=True) startswith() response: - "privacy_url": "nclouds://user:****@localhost/admin", + "privacy_url": "nclouds://user:****@localhost/@admin", }, ), ( "nclouds://user:pass@localhost:8080/admin/", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), + ( + "nclouds://user:pass@localhost:8080/#group/", + { + "instance": NotifyNextcloud, + # Invalid JSON Response + "requests_response_text": "{", + # We will fail to make the notify() call due to our bad response + "notify_response": False, + }, + ), + ( "ncloud://localhost:8080/admin?+HeaderKey=HeaderValue", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, }, ), ( "ncloud://user:pass@localhost:8081/admin", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, # force a failure "response": False, "requests_response_code": requests.codes.internal_server_error, @@ -180,6 +228,7 @@ apprise_url_tests = ( "ncloud://user:pass@localhost:8082/admin", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, # throw a bizzare code forcing us to fail to look it up "response": False, "requests_response_code": 999, @@ -189,6 +238,7 @@ apprise_url_tests = ( "ncloud://user:pass@localhost:8083/user1/user2/user3", { "instance": NotifyNextcloud, + "requests_response_text": NEXTCLOUD_GOOD_RESPONSE, # Throws a series of i/o exceptions with this flag # is set and tests that we gracfully handle them "test_requests_exceptions": True, @@ -257,25 +307,350 @@ def test_plugin_nextcloud_url_prefix(mock_post): assert ( mock_post.call_args_list[0][0][0] == "http://localhost/abcd/ocs/v2.php/apps/" - "admin_notifications/api/v1/notifications/admin" + "admin_notifications/api/v1/notifications/admin") + + +@mock.patch("requests.post") +@mock.patch("requests.get") +def test_plugin_nextcloud_groups_and_all(mock_get, mock_post): + """NotifyNextcloud() Group and All user expansion.""" + + # Mock POST success + post_resp = mock.Mock() + post_resp.content = "" + post_resp.status_code = requests.codes.ok + mock_post.return_value = post_resp + + # Mock GET responses for group and users listing + def get_side_effect(url, *args, **kwargs): + resp = mock.Mock() + if "/ocs/v1.php/cloud/groups/" in url: + # Return JSON for group + j = { + "ocs": { + "meta": {"status": "ok", "statuscode": 100}, + "data": {"users": ["user1", "user2"]}, + } + } + resp.status_code = requests.codes.ok + resp.json = lambda: j + resp.content = dumps(j).encode() + return resp + + elif "/ocs/v1.php/cloud/users" in url: + j = { + "ocs": { + "meta": {"status": "ok", "statuscode": 100}, + "data": {"users": ["user1", "user3"]}, + } + } + resp.status_code = requests.codes.ok + resp.json = lambda: j + resp.content = dumps(j).encode() + return resp + # default + resp.status_code = requests.codes.ok + resp.content = b"" + resp.json = lambda: {} + return resp + + mock_get.side_effect = get_side_effect + + # Instantiate with a mix of targets: group, all, and direct user + obj = NotifyNextcloud( + host="localhost", + user="admin", + password="pass", + targets=["#devs", "all", "user4"], ) - mock_post.reset_mock() - - # instantiate our object (without a batch mode) - obj = Apprise.instantiate( - "ncloud://localhost/admin/?version=21&url_prefix=a/longer/path/abcd/" - ) + assert isinstance(obj, NotifyNextcloud) + # Send notification assert ( - obj.notify(body="body", title="title", notify_type=NotifyType.INFO) + obj.send(body="body", title="title", notify_type=NotifyType.INFO) is True ) - # Not set to batch, so we send 2 different messages - assert mock_post.call_count == 1 - assert ( - mock_post.call_args_list[0][0][0] - == "http://localhost/a/longer/path/abcd/" - "ocs/v2.php/apps/notifications/api/v2/admin_notifications/admin" + # Expected resolved users (deduplicated): user1, user2, user3, user4 + # Hence 4 POST calls + assert mock_post.call_count == 4 + + # Validate calls were made to expected endpoints + called_urls = [c[0][0] for c in mock_post.call_args_list] + for u in ("user1", "user2", "user3", "user4"): + assert any(u in url for url in called_urls) + + +@mock.patch("requests.post") +@mock.patch("requests.get") +def test_plugin_nextcloud_persistent_storage(mock_get, mock_post, tmpdir): + """Testing persistent storage""" + + post_resp = mock.Mock() + post_resp.content = "" + post_resp.status_code = requests.codes.ok + mock_post.return_value = post_resp + + # Set up persistent storage + asset = AppriseAsset( + storage_mode=PersistentStoreMode.FLUSH, + storage_path=str(tmpdir), ) + + def get_side_effect(url, *args, **kwargs): + resp = mock.Mock() + # Default json() to empty + resp.json = lambda: {} + + if "/ocs/v1.php/cloud/groups" in url: + resp.status_code = requests.codes.ok + payload = {"ocs": {"data": {"users": ["u1"]}}} + resp.json = lambda: payload + resp.content = dumps(payload).encode() + return resp + + elif "/ocs/v1.php/cloud/users" in url: + resp.status_code = requests.codes.ok + payload = {"ocs": {"data": {"users": ["u2"]}}} + resp.json = lambda: payload + resp.content = dumps(payload).encode() + return resp + + resp.status_code = 500 + resp.content = b"" + return resp + + mock_get.side_effect = get_side_effect + + obj = NotifyNextcloud( + host="localhost", + user="admin", + password="pass", + targets=["#devs", "all"], + asset=asset, + ) + # We failed to get our list + assert obj.send( + body="body", title="title", notify_type=NotifyType.INFO) is True + + # User and Group looked up + assert mock_get.call_count == 2 + + # Expect users u1 (group) and u2 (all) + assert mock_post.call_count == 2 + called_urls = [c[0][0] for c in mock_post.call_args_list] + for u in ("u1", "u2"): + assert any(u in url for url in called_urls) + + mock_get.reset_mock() + mock_post.reset_mock() + + obj = NotifyNextcloud( + host="localhost", + user="admin", + password="pass", + targets=["#devs"], + asset=asset, + ) + # We succeeded this time + assert obj.send( + body="body", title="title", notify_type=NotifyType.INFO) is True + + # Expect users u1 (group) only and pulled from cache + assert mock_get.call_count == 0 + assert mock_post.call_count == 1 + assert mock_post.call_args_list[0][0][0].endswith("/u1") + + +@mock.patch("requests.post") +@mock.patch("requests.get") +def test_plugin_nextcloud_groups_errors_and_dedup(mock_get, mock_post): + """Non-200/exception paths return empty lists and dedup still applies.""" + + post_resp = mock.Mock() + post_resp.content = "" + post_resp.status_code = requests.codes.ok + mock_post.return_value = post_resp + + # Non-200 for group and users; JSON invalid + def get_side_effect(url, *args, **kwargs): + resp = mock.Mock() + resp.status_code = 401 + resp.content = b"" + # Return empty/invalid JSON to drive empty path + resp.json = lambda: {} + return resp + + mock_get.side_effect = get_side_effect + + # Provide duplicates alongside failing expansions + obj = NotifyNextcloud( + host="localhost", + user="admin", + password="pass", + targets=["#devs", "all", "user1", "user1", "user2"], + ) + + # We failed to hit the server for data + assert obj.send(body="x", title="y", notify_type=NotifyType.INFO) is False + + # we have no control over the order, but we know that on the first + # GET call, we'd have gotten a 401 response; so we'd have stopped from + # that point further + assert mock_get.call_count == 1 + + # Nothing notified + assert mock_post.call_count == 0 + + +@mock.patch("requests.post") +@mock.patch("requests.get") +def test_plugin_nextcloud_req_exception_and_empty_targets(mock_get, mock_post): + """RequestException returns empty expansion; direct users send.""" + + post_resp = mock.Mock() + post_resp.content = "" + post_resp.status_code = requests.codes.ok + mock_post.return_value = post_resp + + def get_side_effect(url, *args, **kwargs): + raise requests.RequestException("boom") + + mock_get.side_effect = get_side_effect + + obj = NotifyNextcloud( + host="localhost", + user="admin", + password="pass", + targets=["", " ", "#DevTeam", "#", "userX"], + ) + + # Our Group inquiry failed to respond + assert obj.send(body="x", title="y", notify_type=NotifyType.INFO) is False + assert mock_post.call_count == 0 + assert mock_get.call_count == 1 + assert mock_get.call_args_list[0][0][0] \ + == "http://localhost/ocs/v1.php/cloud/groups/DevTeam" + assert mock_get.call_args_list[0][1]["params"].get("format") == "json" + + +@mock.patch("requests.post") +@mock.patch("requests.get") +def test_plugin_nextcloud_json_empty_returns_empty(mock_get, mock_post): + """Invalid/empty JSON returns empty; direct users still send.""" + + post_resp = mock.Mock() + post_resp.content = "" + post_resp.status_code = requests.codes.ok + mock_post.return_value = post_resp + + def get_side_effect(url, *args, **kwargs): + resp = mock.Mock() + resp.json = lambda: {} + resp.status_code = requests.codes.ok + resp.content = b"{}" + return resp + + mock_get.side_effect = get_side_effect + + obj = NotifyNextcloud( + host="localhost", + user="admin", + password="pass", + targets=["#broken", "all", "userZ"], + ) + + # Our notification + assert obj.send(body="x", title="y", notify_type=NotifyType.INFO) is True + # Only direct userZ posts because both expansions return empty + assert mock_get.call_count == 2 + assert any("/cloud/users" in call[0][0] + for call in mock_get.call_args_list) + assert any("/cloud/groups/broken" in call[0][0] + for call in mock_get.call_args_list) + + # userZ would get a notification + assert mock_post.call_count == 1 + assert mock_post.call_args_list[0][0][0].endswith("/userZ") + + +@mock.patch("requests.post") +@mock.patch("requests.get") +def test_plugin_nextcloud_caching_group_and_all(mock_get, mock_post): + """Cache hits avoid repeat OCS lookups.""" + + # First round of GETs return users for group and all + def get_side_effect(url, *args, **kwargs): + resp = mock.Mock() + resp.status_code = requests.codes.ok + if "/ocs/v1.php/cloud/groups" in url: + j = { + "ocs": { + "meta": {"status": "ok", "statuscode": 100}, + "data": {"users": ["g1", "g2"]}, + } + } + elif "/ocs/v1.php/cloud/users" in url: + j = { + "ocs": { + "meta": {"status": "ok", "statuscode": 100}, + "data": {"users": ["a1", "a2"]}, + } + } + else: + j = {"ocs": {"data": {"users": []}}} + resp.json = lambda: j + resp.content = dumps(j).encode() + return resp + + mock_get.side_effect = get_side_effect + + post_resp = mock.Mock() + post_resp.content = "" + post_resp.status_code = requests.codes.ok + mock_post.return_value = post_resp + + obj = NotifyNextcloud( + host="localhost", + user="admin", + password="pass", + targets=["#devs", "all", "@joe"], + ) + + # First send: resolves via OCS; expect 2 GETs (group + all) + assert obj.send(body="b", title="t", notify_type=NotifyType.INFO) is True + assert mock_get.call_count == 2 + called = "".join(c[0][0] for c in mock_get.call_args_list) + assert "/cloud/groups/" in called and "/cloud/users" in called + + # we sent 5 notifications + assert mock_post.call_count == 5 + expected_users = {"a1", "a2", "g1", "g2", "joe"} + + # Extract the user segment from the URL of each call + actual_users = { + call[0][0].split("/")[-1] + for call in mock_post.call_args_list + } + + # Assert that the set of actual users matches the set of expected users + assert actual_users == expected_users + + # Reset our mock object + mock_get.reset_mock() + mock_post.reset_mock() + + assert obj.send(body="b2", title="t2", notify_type=NotifyType.INFO) is True + # Cached responses were used to get our user information + assert mock_get.call_count == 0 + assert mock_post.call_count == 5 + + # We can re-verify our notifications went as expected: + actual_users = { + call[0][0].split("/")[-1] + for call in mock_post.call_args_list + } + + # Assert that the set of actual users matches the set of expected users + assert actual_users == expected_users