Skip to content

Commit

Permalink
+tests, increased coverage (#7)
Browse files Browse the repository at this point in the history
* provisioning_api: edit group tests
* exposed `check_error` to use
* polished UsersGroupsAPI
---------
Signed-off-by: Alexander Piskun <bigcat88@icloud.com>
  • Loading branch information
bigcat88 authored Jun 30, 2023
1 parent b1bf6b0 commit 91ba5d7
Show file tree
Hide file tree
Showing 8 changed files with 167 additions and 50 deletions.
13 changes: 13 additions & 0 deletions .readthedocs.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
version: 2

build:
os: ubuntu-22.04
tools:
python: "3.10"

python:
install:
- method: pip
path: .
extra_requirements:
- docs
2 changes: 1 addition & 1 deletion nc_py_api/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
"""

from .constants import ApiScope, LogLvl
from .exceptions import NextcloudException
from .exceptions import NextcloudException, check_error
from .files import FsNodeInfo
from .integration_fastapi import nc_app, set_enabled_handler, set_scopes
from .nextcloud import Nextcloud, NextcloudApp
Expand Down
12 changes: 11 additions & 1 deletion nc_py_api/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,17 @@ def __str__(self):
return f"[{self.status_code}]{reason}{info}"


def check_error(code: int, info: str):
def check_error(code: int, info: str = ""):
if 996 <= code <= 999:
if code == 996:
phrase = "Server error"
elif code == 997:
phrase = "Unauthorised"
elif code == 998:
phrase = "Not found"
else:
phrase = "Unknown error"
raise NextcloudException(code, reason=phrase, info=info)
if not codes.is_error(code):
return
raise NextcloudException(code, reason=codes(code).phrase, info=info)
49 changes: 40 additions & 9 deletions nc_py_api/users_groups.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,35 +2,66 @@
Nextcloud API for working with user groups.
"""

from typing import Optional
from typing import Optional, TypedDict

from ._session import NcSessionBasic
from .misc import kwargs_to_dict

ENDPOINT = "/ocs/v1.php/cloud/groups"


class GroupDetails(TypedDict):
id: str
display_name: str
user_count: int
disabled: bool
can_add: bool
can_remove: bool


class UsersGroupsAPI:
def __init__(self, session: NcSessionBasic):
self._session = session

def list(self, mask: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None) -> dict:
def get(self, mask: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None) -> list[str]:
data = kwargs_to_dict(["search", "limit", "offset"], search=mask, limit=limit, offset=offset)
response_data = self._session.ocs(method="GET", path=ENDPOINT, params=data)
return response_data["groups"] if response_data else {}
return response_data["groups"] if response_data else []

def create(self, group_id: str) -> dict:
return self._session.ocs(method="POST", path=f"{ENDPOINT}", params={"groupid": group_id})
def get_details(
self, mask: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None
) -> list[GroupDetails]:
data = kwargs_to_dict(["search", "limit", "offset"], search=mask, limit=limit, offset=offset)
response_data = self._session.ocs(method="GET", path=f"{ENDPOINT}/details", params=data)
return [self._to_group_details(i) for i in response_data["groups"]] if response_data else []

def edit(self, group_id: str, **kwargs) -> dict:
return self._session.ocs(method="PUT", path=f"{ENDPOINT}/{group_id}", params={**kwargs})
def create(self, group_id: str, display_name: Optional[str] = None) -> None:
params = {"groupid": group_id}
if display_name is not None:
params["displayname"] = display_name
self._session.ocs(method="POST", path=f"{ENDPOINT}", params=params)

def delete(self, group_id: str) -> dict:
return self._session.ocs(method="DELETE", path=f"{ENDPOINT}/{group_id}")
def edit(self, group_id: str, display_name: str) -> None:
params = {"key": "displayname", "value": display_name}
self._session.ocs(method="PUT", path=f"{ENDPOINT}/{group_id}", params=params)

def delete(self, group_id: str) -> None:
self._session.ocs(method="DELETE", path=f"{ENDPOINT}/{group_id}")

def get_members(self, group_id: str) -> dict:
response_data = self._session.ocs(method="GET", path=f"{ENDPOINT}/{group_id}")
return response_data["users"] if response_data else {}

def get_subadmins(self, group_id: str) -> dict:
return self._session.ocs(method="GET", path=f"{ENDPOINT}/{group_id}/subadmins")

@staticmethod
def _to_group_details(reply: dict) -> GroupDetails:
return {
"id": reply["id"],
"display_name": reply["displayname"],
"user_count": reply["usercount"],
"disabled": bool(reply["disabled"]),
"can_add": reply["canAdd"],
"can_remove": reply["canRemove"],
}
4 changes: 2 additions & 2 deletions nc_py_api/users_statuses.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,15 +64,15 @@ def get(self, user_id: str) -> Optional[UserStatus]:
raise e from None

def get_predefined(self) -> list[PredefinedStatus]:
require_capabilities("user_status", self._session.capabilities)
if self._session.nc_version["major"] < 27:
return []
require_capabilities("user_status", self._session.capabilities)
return self._session.ocs(method="GET", path=f"{ENDPOINT}/predefined_statuses")

def set_predefined(self, message_id: str, clear_at: int = 0) -> None:
require_capabilities("user_status", self._session.capabilities)
if self._session.nc_version["major"] < 27:
return
require_capabilities("user_status", self._session.capabilities)
params: dict[str, Union[int, str]] = {"messageId": message_id}
if clear_at:
params["clearAt"] = clear_at
Expand Down
85 changes: 66 additions & 19 deletions tests/groups_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,46 +4,93 @@

from gfixture import NC_TO_TEST

TEST_GROUP_NAME = "test_coverage_group"
TEST_GROUP_NAME = "test_coverage_group1"
TEST_GROUP_NAME2 = "test_coverage_group2"


@pytest.mark.skipif(not isinstance(NC_TO_TEST[:1][0], Nextcloud), reason="Not available for NextcloudApp.")
@pytest.mark.parametrize("nc", NC_TO_TEST[:1])
def test_create_delete_group(nc):
@pytest.mark.parametrize("params", ((TEST_GROUP_NAME, ), (TEST_GROUP_NAME, "display name")))
def test_create_delete_group(nc, params):
test_group_name = params[0]
try:
nc.users_groups.delete(TEST_GROUP_NAME)
nc.users_groups.delete(test_group_name)
except NextcloudException:
pass
nc.users_groups.create(TEST_GROUP_NAME)
nc.users_groups.create(*params)
with pytest.raises(NextcloudException):
nc.users_groups.create(TEST_GROUP_NAME)
nc.users_groups.delete(TEST_GROUP_NAME)
nc.users_groups.create(*params)
nc.users_groups.delete(test_group_name)
with pytest.raises(NextcloudException):
nc.users_groups.delete(TEST_GROUP_NAME)
nc.users_groups.delete(test_group_name)


@pytest.mark.skipif(not isinstance(NC_TO_TEST[:1][0], Nextcloud), reason="Not available for NextcloudApp.")
@pytest.mark.parametrize("nc", NC_TO_TEST[:1])
def test_get_group(nc):
for i in (TEST_GROUP_NAME, TEST_GROUP_NAME2):
try:
nc.users_groups.create(i)
except NextcloudException:
pass
groups = nc.users_groups.get()
assert isinstance(groups, list)
assert len(groups) >= 2
assert TEST_GROUP_NAME in groups
assert TEST_GROUP_NAME2 in groups
groups = nc.users_groups.get(mask=TEST_GROUP_NAME)
assert len(groups) == 1
groups = nc.users_groups.get(limit=1)
assert len(groups) == 1
assert groups[0] != nc.users_groups.get(limit=1, offset=1)[0]
nc.users_groups.delete(TEST_GROUP_NAME)
nc.users_groups.delete(TEST_GROUP_NAME2)


@pytest.mark.skipif(not isinstance(NC_TO_TEST[:1][0], Nextcloud), reason="Not available for NextcloudApp.")
@pytest.mark.parametrize("nc", NC_TO_TEST[:1])
def test_get_non_existing_group(nc):
groups = nc.users_groups.get(mask="Such group should not be present")
assert isinstance(groups, list)
assert not groups


@pytest.mark.skipif(not isinstance(NC_TO_TEST[:1][0], Nextcloud), reason="Not available for NextcloudApp.")
@pytest.mark.parametrize("nc", NC_TO_TEST[:1])
def test_list_group(nc):
def test_get_group_details(nc):
try:
nc.users_groups.create(TEST_GROUP_NAME)
nc.users_groups.delete(TEST_GROUP_NAME)
except NextcloudException:
pass
try:
nc.users_groups.create(TEST_GROUP_NAME + "2")
nc.users_groups.create(TEST_GROUP_NAME)
except NextcloudException:
pass
groups = nc.users_groups.list()
assert len(groups) >= 2
assert TEST_GROUP_NAME in groups
assert TEST_GROUP_NAME + "2" in groups
groups = nc.users_groups.list(mask=TEST_GROUP_NAME)
assert len(groups) == 2
groups = nc.users_groups.list(limit=1)
groups = nc.users_groups.get_details(mask=TEST_GROUP_NAME)
assert len(groups) == 1
assert groups[0] != nc.users_groups.list(limit=1, offset=1)[0]
group = groups[0]
assert group["id"] == TEST_GROUP_NAME
assert group["display_name"] == TEST_GROUP_NAME
assert not group["disabled"]
assert isinstance(group["user_count"], int)
assert isinstance(group["can_add"], bool)
assert isinstance(group["can_remove"], bool)
nc.users_groups.delete(TEST_GROUP_NAME)


@pytest.mark.skipif(not isinstance(NC_TO_TEST[:1][0], Nextcloud), reason="Not available for NextcloudApp.")
@pytest.mark.parametrize("nc", NC_TO_TEST[:1])
def test_group_edit(nc):
try:
nc.users_groups.create(TEST_GROUP_NAME)
except NextcloudException:
pass
nc.users_groups.edit(TEST_GROUP_NAME, display_name="earth people")
assert nc.users_groups.get_details(mask=TEST_GROUP_NAME)[0]["display_name"] == "earth people"
nc.users_groups.delete(TEST_GROUP_NAME)
nc.users_groups.delete(TEST_GROUP_NAME + "2")
with pytest.raises(NextcloudException) as exc_info:
nc.users_groups.edit(TEST_GROUP_NAME, display_name="earth people")
assert exc_info.value.status_code == 996


@pytest.mark.skipif(not isinstance(NC_TO_TEST[:1][0], Nextcloud), reason="Not available for NextcloudApp.")
Expand Down
12 changes: 12 additions & 0 deletions tests/misc_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
import pytest

from nc_py_api import check_error, NextcloudException


@pytest.mark.parametrize("code", (995, 996, 997, 998, 999, 1000))
def test_check_error(code):
if 996 <= code <= 999:
with pytest.raises(NextcloudException):
check_error(code)
else:
check_error(code)
40 changes: 22 additions & 18 deletions tests/users_statuses_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from time import time


from gfixture import NC_TO_TEST, NC_VERSION
from gfixture import NC_TO_TEST


@pytest.mark.parametrize("nc", NC_TO_TEST)
Expand Down Expand Up @@ -36,17 +36,19 @@ def test_get_status(nc, message):
assert not r1["messageIsPredefined"]


@pytest.mark.skipif(NC_VERSION.get("major", 0) < 27, reason="NC27 required.")
@pytest.mark.parametrize("nc", NC_TO_TEST)
def test_get_predefined(nc):
r = nc.users_statuses.get_predefined()
assert isinstance(r, list)
assert r
for i in r:
assert isinstance(i["id"], str)
assert isinstance(i["message"], str)
assert isinstance(i["icon"], str)
assert isinstance(i["clearAt"], dict) or i["clearAt"] is None
if nc.srv_version["major"] < 27:
assert r == []
else:
assert isinstance(r, list)
assert r
for i in r:
assert isinstance(i["id"], str)
assert isinstance(i["message"], str)
assert isinstance(i["icon"], str)
assert isinstance(i["clearAt"], dict) or i["clearAt"] is None


@pytest.mark.parametrize("nc", NC_TO_TEST)
Expand Down Expand Up @@ -89,15 +91,17 @@ def test_set_status_type(nc, value):
assert r["statusIsUserDefined"]


@pytest.mark.skipif(NC_VERSION.get("major", 0) < 27, reason="NC27 required.")
@pytest.mark.parametrize("nc", NC_TO_TEST)
@pytest.mark.parametrize("clear_at", (None, int(time()) + 360))
def test_set_predefined(nc, clear_at):
predefined_statuses = nc.users_statuses.get_predefined()
for i in predefined_statuses:
nc.users_statuses.set_predefined(i["id"], clear_at)
r = nc.users_statuses.get_current()
assert r["message"] == i["message"]
assert r["messageId"] == i["id"]
assert r["messageIsPredefined"]
assert r["clearAt"] == clear_at
if nc.srv_version["major"] < 27:
nc.users_statuses.set_predefined("meeting")
else:
predefined_statuses = nc.users_statuses.get_predefined()
for i in predefined_statuses:
nc.users_statuses.set_predefined(i["id"], clear_at)
r = nc.users_statuses.get_current()
assert r["message"] == i["message"]
assert r["messageId"] == i["id"]
assert r["messageIsPredefined"]
assert r["clearAt"] == clear_at

0 comments on commit 91ba5d7

Please sign in to comment.