From de8c170092945d97ad9ac20c743ca9dc96ea8777 Mon Sep 17 00:00:00 2001 From: bigcat88 Date: Thu, 19 Feb 2026 15:27:22 +0200 Subject: [PATCH 1/3] refactor: drop sync API for Activity, Notes, User Status, Weather Status Remove sync wrapper classes (_ActivityAPI, _NotesAPI, _UserStatusAPI, _WeatherStatusAPI) and their registrations in _NextcloudBasic. Async implementations remain unchanged. - Remove sync classes and NcSessionBasic imports from 4 module files - Remove sync type annotations, __init__ lines, and state resets from nextcloud.py - Remove all sync tests; async tests fully cover the same functionality - Fix talk_test async test to use AsyncNextcloud for user_status setup - Bump version to 0.30.0 - Add async-first deprecation notice to README --- README.md | 12 +- nc_py_api/_version.py | 2 +- nc_py_api/activity.py | 54 +-------- nc_py_api/nextcloud.py | 22 +--- nc_py_api/notes.py | 140 +--------------------- nc_py_api/user_status.py | 98 +-------------- nc_py_api/weather_status.py | 68 +---------- tests/actual_tests/activity_test.py | 31 ----- tests/actual_tests/misc_test.py | 10 -- tests/actual_tests/nc_app_test.py | 13 -- tests/actual_tests/notes_test.py | 68 ----------- tests/actual_tests/talk_test.py | 38 +----- tests/actual_tests/user_status_test.py | 102 +--------------- tests/actual_tests/weather_status_test.py | 77 ------------ 14 files changed, 26 insertions(+), 709 deletions(-) diff --git a/README.md b/README.md index ec7c252d..694f0e36 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,17 @@ Python library that provides a robust and well-documented API that allows develo * **Reliable**: Minimum number of incompatible changes. * **Robust**: All code is covered with tests as much as possible. * **Easy**: Designed to be easy to use. - * **Sync + Async**: Provides both sync and async APIs. + * **Async-first**: Full async API with sync wrappers available for most modules. + +### Deprecation notice: sync API + +Starting with version **0.30.0**, we are gradually removing sync wrappers in favour of +the async API. The following modules have already lost their sync counterparts: +**Activity**, **Notes**, **User Status**, and **Weather Status**. + +All remaining sync methods will be phased out in future releases. If you are still +using the sync `Nextcloud` / `NextcloudApp` classes, we recommend migrating to +`AsyncNextcloud` / `AsyncNextcloudApp` as soon as possible. ### Differences between the Nextcloud and NextcloudApp classes diff --git a/nc_py_api/_version.py b/nc_py_api/_version.py index 8bc79437..7cc79376 100644 --- a/nc_py_api/_version.py +++ b/nc_py_api/_version.py @@ -1,3 +1,3 @@ """Version of nc_py_api.""" -__version__ = "0.24.0" +__version__ = "0.30.0" diff --git a/nc_py_api/activity.py b/nc_py_api/activity.py index c445f5cb..01271506 100644 --- a/nc_py_api/activity.py +++ b/nc_py_api/activity.py @@ -6,7 +6,7 @@ from ._exceptions import NextcloudExceptionNotModified from ._misc import check_capabilities, nc_iso_time_to_datetime -from ._session import AsyncNcSessionBasic, NcSessionBasic +from ._session import AsyncNcSessionBasic @dataclasses.dataclass @@ -136,58 +136,6 @@ def __repr__(self): ) -class _ActivityAPI: - """The class provides the Activity Application API.""" - - _ep_base: str = "/ocs/v1.php/apps/activity" - last_given: int - """Used by ``get_activities``, when **since** param is ``True``.""" - - def __init__(self, session: NcSessionBasic): - self._session = session - self.last_given = 0 - - @property - def available(self) -> bool: - """Returns True if the Nextcloud instance supports this feature, False otherwise.""" - return not check_capabilities("activity.apiv2", self._session.capabilities) - - def get_activities( - self, - filter_id: ActivityFilter | str = "", - since: int | bool = 0, - limit: int = 50, - object_type: str = "", - object_id: int = 0, - sort: str = "desc", - ) -> list[Activity]: - """Returns activities for the current user. - - :param filter_id: Filter to apply, if needed. - :param since: Last activity ID you have seen. When specified, only activities after provided are returned. - Can be set to ``True`` to automatically use last ``last_given`` from previous calls. Default = **0**. - :param limit: Max number of activities to be returned. - :param object_type: Filter the activities to a given object. - :param object_id: Filter the activities to a given object. - :param sort: Sort activities ascending or descending. Default is ``desc``. - - .. note:: ``object_type`` and ``object_id`` should only appear together with ``filter_id`` unset. - """ - if since is True: - since = self.last_given - url, params = _get_activities(filter_id, since, limit, object_type, object_id, sort) - try: - result = self._session.ocs("GET", self._ep_base + url, params=params) - except NextcloudExceptionNotModified: - return [] - self.last_given = int(self._session.response_headers["X-Activity-Last-Given"]) - return [Activity(i) for i in result] - - def get_filters(self) -> list[ActivityFilter]: - """Returns avalaible activity filters.""" - return [ActivityFilter(i) for i in self._session.ocs("GET", self._ep_base + "/api/v2/activity/filters")] - - class _AsyncActivityAPI: """The class provides the async Activity Application API.""" diff --git a/nc_py_api/nextcloud.py b/nc_py_api/nextcloud.py index 645b8f9c..5d2736b9 100644 --- a/nc_py_api/nextcloud.py +++ b/nc_py_api/nextcloud.py @@ -27,7 +27,7 @@ ) from ._talk_api import _AsyncTalkAPI, _TalkAPI from ._theming import ThemingInfo, get_parsed_theme -from .activity import _ActivityAPI, _AsyncActivityAPI +from .activity import _AsyncActivityAPI from .apps import _AppsAPI, _AsyncAppsAPI from .calendar_api import _CalendarAPI from .ex_app.defs import LogLvl @@ -37,28 +37,24 @@ from .files.files import FilesAPI from .files.files_async import AsyncFilesAPI from .loginflow_v2 import _AsyncLoginFlowV2API, _LoginFlowV2API -from .notes import _AsyncNotesAPI, _NotesAPI +from .notes import _AsyncNotesAPI from .notifications import _AsyncNotificationsAPI, _NotificationsAPI -from .user_status import _AsyncUserStatusAPI, _UserStatusAPI +from .user_status import _AsyncUserStatusAPI from .users import _AsyncUsersAPI, _UsersAPI from .users_groups import _AsyncUsersGroupsAPI, _UsersGroupsAPI -from .weather_status import _AsyncWeatherStatusAPI, _WeatherStatusAPI +from .weather_status import _AsyncWeatherStatusAPI from .webhooks import _AsyncWebhooksAPI, _WebhooksAPI class _NextcloudBasic(ABC): # pylint: disable=too-many-instance-attributes apps: _AppsAPI """Nextcloud API for App management""" - activity: _ActivityAPI - """Activity Application API""" cal: _CalendarAPI """Nextcloud Calendar API""" files: FilesAPI """Nextcloud API for File System and Files Sharing""" preferences: PreferencesAPI """Nextcloud User Preferences API""" - notes: _NotesAPI - """Nextcloud Notes API""" notifications: _NotificationsAPI """Nextcloud API for managing user notifications""" talk: _TalkAPI @@ -67,27 +63,19 @@ class _NextcloudBasic(ABC): # pylint: disable=too-many-instance-attributes """Nextcloud API for managing users.""" users_groups: _UsersGroupsAPI """Nextcloud API for managing user groups.""" - user_status: _UserStatusAPI - """Nextcloud API for managing users statuses""" - weather_status: _WeatherStatusAPI - """Nextcloud API for managing user weather statuses""" webhooks: _WebhooksAPI """Nextcloud API for managing webhooks""" _session: NcSessionBasic def __init__(self, session: NcSessionBasic): self.apps = _AppsAPI(session) - self.activity = _ActivityAPI(session) self.cal = _CalendarAPI(session) self.files = FilesAPI(session) self.preferences = PreferencesAPI(session) - self.notes = _NotesAPI(session) self.notifications = _NotificationsAPI(session) self.talk = _TalkAPI(session) self.users = _UsersAPI(session) self.users_groups = _UsersGroupsAPI(session) - self.user_status = _UserStatusAPI(session) - self.weather_status = _WeatherStatusAPI(session) self.webhooks = _WebhooksAPI(session) @property @@ -381,8 +369,6 @@ def set_user(self, user_id: str): self._session.set_user(user_id) self.talk.config_sha = "" self.talk.modified_since = 0 - self.activity.last_given = 0 - self.notes.last_etag = "" self._session.update_server_info() @property diff --git a/nc_py_api/notes.py b/nc_py_api/notes.py index b47908f3..8e94a85c 100644 --- a/nc_py_api/notes.py +++ b/nc_py_api/notes.py @@ -9,7 +9,7 @@ from ._exceptions import check_error from ._misc import check_capabilities, clear_from_params_empty, require_capabilities -from ._session import AsyncNcSessionBasic, NcSessionBasic +from ._session import AsyncNcSessionBasic @dataclasses.dataclass @@ -91,144 +91,6 @@ class NotesSettings(typing.TypedDict): """Newly created note's files will have this file suffix. Default is **.txt**.""" -class _NotesAPI: - """Class implementing Nextcloud Notes API.""" - - _ep_base: str = "/index.php/apps/notes/api/v1" # without `index.php` we will get 405 error. - last_etag: str - """Used by ``get_list``, when **etag** param is ``True``.""" - - def __init__(self, session: NcSessionBasic): - self._session = session - self.last_etag = "" - - @property - def available(self) -> bool: - """Returns True if the Nextcloud instance supports this feature, False otherwise.""" - return not check_capabilities("notes", self._session.capabilities) - - def get_list( - self, - category: str | None = None, - modified_since: int | None = None, - limit: int | None = None, - cursor: str | None = None, - no_content: bool = False, - etag: bool = False, - ) -> list[Note]: - """Get information of all Notes. - - :param category: Filter the result by category name. Notes with another category are not included in the result. - :param modified_since: When provided only results newer than given Unix timestamp are returned. - :param limit: Limit response to contain no more than the given number of notes. - If there are more notes, then the result is chunked and the HTTP response header - **X-Notes-Chunk-Cursor** is sent with a string value. - - .. note:: Use :py:attr:`~nc_py_api.nextcloud.Nextcloud.response_headers` property to achieve that. - :param cursor: You should use the string value from the last request's HTTP response header - ``X-Notes-Chunk-Cursor`` in order to get the next chunk of notes. - :param no_content: Flag indicating should ``content`` field be excluded from response. - :param etag: Flag indicating should ``ETag`` from last call be used. Default = **False**. - """ - require_capabilities("notes", self._session.capabilities) - params = { - "category": category, - "pruneBefore": modified_since, - "exclude": "content" if no_content else None, - "chunkSize": limit, - "chunkCursor": cursor, - } - clear_from_params_empty(list(params.keys()), params) - headers = {"If-None-Match": self.last_etag} if self.last_etag and etag else {} - r = _res_to_json(self._session.adapter.get(self._ep_base + "/notes", params=params, headers=headers)) - self.last_etag = self._session.response_headers["ETag"] - return [Note(i) for i in r] - - def by_id(self, note: Note) -> Note: - """Get updated information about :py:class:`~nc_py_api.notes.Note`.""" - require_capabilities("notes", self._session.capabilities) - r = _res_to_json( - self._session.adapter.get( - self._ep_base + f"/notes/{note.note_id}", headers={"If-None-Match": f'"{note.etag}"'} - ) - ) - return Note(r) if r else note - - def create( - self, - title: str, - content: str | None = None, - category: str | None = None, - favorite: bool | None = None, - last_modified: int | str | datetime.datetime | None = None, - ) -> Note: - """Create new Note.""" - require_capabilities("notes", self._session.capabilities) - params = { - "title": title, - "content": content, - "category": category, - "favorite": favorite, - "modified": last_modified, - } - clear_from_params_empty(list(params.keys()), params) - return Note(_res_to_json(self._session.adapter.post(self._ep_base + "/notes", json=params))) - - def update( - self, - note: Note, - title: str | None = None, - content: str | None = None, - category: str | None = None, - favorite: bool | None = None, - overwrite: bool = False, - ) -> Note: - """Updates Note. - - ``overwrite`` specifies should be or not the Note updated even if it was changed on server(has different ETag). - """ - require_capabilities("notes", self._session.capabilities) - headers = {"If-Match": f'"{note.etag}"'} if not overwrite else {} - params = { - "title": title, - "content": content, - "category": category, - "favorite": favorite, - } - clear_from_params_empty(list(params.keys()), params) - if not params: - raise ValueError("Nothing to update.") - return Note( - _res_to_json( - self._session.adapter.put(self._ep_base + f"/notes/{note.note_id}", json=params, headers=headers) - ) - ) - - def delete(self, note: int | Note) -> None: - """Deletes a Note.""" - require_capabilities("notes", self._session.capabilities) - note_id = note.note_id if isinstance(note, Note) else note - check_error(self._session.adapter.delete(self._ep_base + f"/notes/{note_id}")) - - def get_settings(self) -> NotesSettings: - """Returns Notes App settings.""" - require_capabilities("notes", self._session.capabilities) - r = _res_to_json(self._session.adapter.get(self._ep_base + "/settings")) - return {"notes_path": r["notesPath"], "file_suffix": r["fileSuffix"]} - - def set_settings(self, notes_path: str | None = None, file_suffix: str | None = None) -> None: - """Change specified setting(s).""" - if notes_path is None and file_suffix is None: - raise ValueError("No setting to change.") - require_capabilities("notes", self._session.capabilities) - params = { - "notesPath": notes_path, - "fileSuffix": file_suffix, - } - clear_from_params_empty(list(params.keys()), params) - check_error(self._session.adapter.put(self._ep_base + "/settings", json=params)) - - class _AsyncNotesAPI: """Class implements Async Nextcloud Notes API.""" diff --git a/nc_py_api/user_status.py b/nc_py_api/user_status.py index 01ba2f96..a6ef65e7 100644 --- a/nc_py_api/user_status.py +++ b/nc_py_api/user_status.py @@ -5,7 +5,7 @@ from ._exceptions import NextcloudExceptionNotFound from ._misc import check_capabilities, kwargs_to_params, require_capabilities -from ._session import AsyncNcSessionBasic, NcSessionBasic +from ._session import AsyncNcSessionBasic @dataclasses.dataclass @@ -107,102 +107,6 @@ def __repr__(self): ) -class _UserStatusAPI: - """Class providing the user status management API on the Nextcloud server.""" - - _ep_base: str = "/ocs/v1.php/apps/user_status/api/v1" - - def __init__(self, session: NcSessionBasic): - self._session = session - - @property - def available(self) -> bool: - """Returns True if the Nextcloud instance supports this feature, False otherwise.""" - return not check_capabilities("user_status.enabled", self._session.capabilities) - - def get_list(self, limit: int | None = None, offset: int | None = None) -> list[UserStatus]: - """Returns statuses for all users.""" - require_capabilities("user_status.enabled", self._session.capabilities) - data = kwargs_to_params(["limit", "offset"], limit=limit, offset=offset) - result = self._session.ocs("GET", f"{self._ep_base}/statuses", params=data) - return [UserStatus(i) for i in result] - - def get_current(self) -> CurrentUserStatus: - """Returns the current user status.""" - require_capabilities("user_status.enabled", self._session.capabilities) - return CurrentUserStatus(self._session.ocs("GET", f"{self._ep_base}/user_status")) - - def get(self, user_id: str) -> UserStatus | None: - """Returns the user status for the specified user.""" - require_capabilities("user_status.enabled", self._session.capabilities) - try: - return UserStatus(self._session.ocs("GET", f"{self._ep_base}/statuses/{user_id}")) - except NextcloudExceptionNotFound: - return None - - def get_predefined(self) -> list[PredefinedStatus]: - """Returns a list of predefined statuses available for installation on this Nextcloud instance.""" - if self._session.nc_version["major"] < 27: - return [] - require_capabilities("user_status.enabled", self._session.capabilities) - result = self._session.ocs("GET", f"{self._ep_base}/predefined_statuses") - return [PredefinedStatus(i) for i in result] - - def set_predefined(self, status_id: str, clear_at: int = 0) -> None: - """Set predefined status for the current user. - - :param status_id: ``predefined`` status ID. - :param clear_at: *optional* time in seconds before the status is cleared. - """ - if self._session.nc_version["major"] < 27: - return - require_capabilities("user_status.enabled", self._session.capabilities) - params: dict[str, int | str] = {"messageId": status_id} - if clear_at: - params["clearAt"] = clear_at - self._session.ocs("PUT", f"{self._ep_base}/user_status/message/predefined", params=params) - - def set_status_type(self, value: typing.Literal["online", "away", "dnd", "invisible", "offline"]) -> None: - """Sets the status type for the current user.""" - require_capabilities("user_status.enabled", self._session.capabilities) - self._session.ocs("PUT", f"{self._ep_base}/user_status/status", params={"statusType": value}) - - def set_status(self, message: str | None = None, clear_at: int = 0, status_icon: str = "") -> None: - """Sets current user status. - - :param message: Message text to set in the status. - :param clear_at: Unix Timestamp, representing the time to clear the status. - :param status_icon: The icon picked by the user (must be one emoji) - """ - require_capabilities("user_status.enabled", self._session.capabilities) - if message is None: - self._session.ocs("DELETE", f"{self._ep_base}/user_status/message") - return - if status_icon: - require_capabilities("user_status.supports_emoji", self._session.capabilities) - params: dict[str, int | str] = {"message": message} - if clear_at: - params["clearAt"] = clear_at - if status_icon: - params["statusIcon"] = status_icon - self._session.ocs("PUT", f"{self._ep_base}/user_status/message/custom", params=params) - - def get_backup_status(self, user_id: str = "") -> UserStatus | None: - """Get the backup status of the user if any.""" - require_capabilities("user_status.enabled", self._session.capabilities) - user_id = user_id if user_id else self._session.user - if not user_id: - raise ValueError("user_id can not be empty.") - return self.get(f"_{user_id}") - - def restore_backup_status(self, status_id: str) -> CurrentUserStatus | None: - """Restores the backup state as current for the current user.""" - require_capabilities("user_status.enabled", self._session.capabilities) - require_capabilities("user_status.restore", self._session.capabilities) - result = self._session.ocs("DELETE", f"{self._ep_base}/user_status/revert/{status_id}") - return result if result else None - - class _AsyncUserStatusAPI: """Class provides async user status management API on the Nextcloud server.""" diff --git a/nc_py_api/weather_status.py b/nc_py_api/weather_status.py index 833cce68..68f3d67e 100644 --- a/nc_py_api/weather_status.py +++ b/nc_py_api/weather_status.py @@ -4,7 +4,7 @@ import enum from ._misc import check_capabilities, require_capabilities -from ._session import AsyncNcSessionBasic, NcSessionBasic +from ._session import AsyncNcSessionBasic class WeatherLocationMode(enum.IntEnum): @@ -40,72 +40,6 @@ def __init__(self, raw_location: dict): self.mode = WeatherLocationMode(int(raw_location.get("mode", 0))) -class _WeatherStatusAPI: - """Class providing the weather status management API on the Nextcloud server.""" - - _ep_base: str = "/ocs/v1.php/apps/weather_status/api/v1" - - def __init__(self, session: NcSessionBasic): - self._session = session - - @property - def available(self) -> bool: - """Returns True if the Nextcloud instance supports this feature, False otherwise.""" - return not check_capabilities("weather_status.enabled", self._session.capabilities) - - def get_location(self) -> WeatherLocation: - """Returns the current location set on the Nextcloud server for the user.""" - require_capabilities("weather_status.enabled", self._session.capabilities) - return WeatherLocation(self._session.ocs("GET", f"{self._ep_base}/location")) - - def set_location( - self, - latitude: float | None = None, - longitude: float | None = None, - address: str | None = None, - ) -> bool: - """Sets the user's location on the Nextcloud server. - - :param latitude: north-south position of a point on the surface of the Earth. - :param longitude: east-west position of a point on the surface of the Earth. - :param address: city, index(*optional*) and country, e.g. "Paris, 75007, France" - """ - require_capabilities("weather_status.enabled", self._session.capabilities) - params: dict[str, str | float] = {} - if latitude is not None and longitude is not None: - params.update({"lat": latitude, "lon": longitude}) - elif address: - params["address"] = address - else: - raise ValueError("latitude & longitude or address should be present") - result = self._session.ocs("PUT", f"{self._ep_base}/location", params=params) - return result.get("success", False) - - def get_forecast(self) -> list[dict]: - """Get forecast for the current location.""" - require_capabilities("weather_status.enabled", self._session.capabilities) - return self._session.ocs("GET", f"{self._ep_base}/forecast") - - def get_favorites(self) -> list[str]: - """Returns favorites addresses list.""" - require_capabilities("weather_status.enabled", self._session.capabilities) - return self._session.ocs("GET", f"{self._ep_base}/favorites") - - def set_favorites(self, favorites: list[str]) -> bool: - """Sets favorites addresses list.""" - require_capabilities("weather_status.enabled", self._session.capabilities) - result = self._session.ocs("PUT", f"{self._ep_base}/favorites", json={"favorites": favorites}) - return result.get("success", False) - - def set_mode(self, mode: WeatherLocationMode) -> bool: - """Change the weather status mode.""" - if int(mode) == WeatherLocationMode.UNKNOWN.value: - raise ValueError("This mode can not be set") - require_capabilities("weather_status.enabled", self._session.capabilities) - result = self._session.ocs("PUT", f"{self._ep_base}/mode", params={"mode": int(mode)}) - return result.get("success", False) - - class _AsyncWeatherStatusAPI: """Class provides async weather status management API on the Nextcloud server.""" diff --git a/tests/actual_tests/activity_test.py b/tests/actual_tests/activity_test.py index 8fb0f91b..c3ebab0a 100644 --- a/tests/actual_tests/activity_test.py +++ b/tests/actual_tests/activity_test.py @@ -5,19 +5,6 @@ from nc_py_api.activity import Activity -def test_get_filters(nc_any): - if nc_any.activity.available is False: - pytest.skip("Activity App is not installed") - r = nc_any.activity.get_filters() - assert r - for i in r: - assert i.filter_id - assert isinstance(i.icon, str) - assert i.name - assert isinstance(i.priority, int) - assert str(i).find("name=") != -1 - - @pytest.mark.asyncio(scope="session") async def test_get_filters_async(anc_any): if await anc_any.activity.available is False: @@ -53,24 +40,6 @@ def _test_get_activities(r: list[Activity]): assert str(i).find("app=") != -1 -def test_get_activities(nc_any): - if nc_any.activity.available is False: - pytest.skip("Activity App is not installed") - with pytest.raises(ValueError): - nc_any.activity.get_activities(object_id=4) - r = nc_any.activity.get_activities(since=True) - _test_get_activities(r) - r2 = nc_any.activity.get_activities(since=True) - if r2: - old_activities_id = [i.activity_id for i in r] - assert r2[0].activity_id not in old_activities_id - assert r2[-1].activity_id not in old_activities_id - assert len(nc_any.activity.get_activities(since=0, limit=1)) == 1 - while True: - if not nc_any.activity.get_activities(since=True): - break - - @pytest.mark.asyncio(scope="session") async def test_get_activities_async(anc_any): if await anc_any.activity.available is False: diff --git a/tests/actual_tests/misc_test.py b/tests/actual_tests/misc_test.py index 3f07d08d..1039104c 100644 --- a/tests/actual_tests/misc_test.py +++ b/tests/actual_tests/misc_test.py @@ -154,16 +154,6 @@ async def test_no_initial_connection_async(anc_any): assert new_nc._session._capabilities -def test_ocs_timeout(nc_any): - new_nc = Nextcloud(npa_timeout=0.01) if isinstance(nc_any, Nextcloud) else NextcloudApp(npa_timeout=0.01) - with pytest.raises(NextcloudException) as e: - if new_nc.weather_status.set_location(latitude=41.896655, longitude=12.488776): - new_nc.weather_status.get_forecast() - if e.value.status_code in (500, 996): - pytest.skip("Some network problem on the host") - assert e.value.status_code == 408 - - @pytest.mark.asyncio(scope="session") async def test_ocs_timeout_async(anc_any): new_nc = ( diff --git a/tests/actual_tests/nc_app_test.py b/tests/actual_tests/nc_app_test.py index e3f45dbe..90396764 100644 --- a/tests/actual_tests/nc_app_test.py +++ b/tests/actual_tests/nc_app_test.py @@ -34,19 +34,6 @@ async def test_app_cfg_async(anc_app): assert app_cfg.app_secret == environ["APP_SECRET"] -def test_change_user(nc_app): - orig_user = nc_app.user - try: - orig_capabilities = nc_app.capabilities - assert nc_app.user_status.available - nc_app.set_user("") - assert not nc_app.user_status.available - assert orig_capabilities != nc_app.capabilities - finally: - nc_app.set_user(orig_user) - assert orig_capabilities == nc_app.capabilities - - @pytest.mark.asyncio(scope="session") async def test_change_user_async(anc_app): orig_user = await anc_app.user diff --git a/tests/actual_tests/notes_test.py b/tests/actual_tests/notes_test.py index 275c5bf9..1ab62ed4 100644 --- a/tests/actual_tests/notes_test.py +++ b/tests/actual_tests/notes_test.py @@ -5,24 +5,6 @@ from nc_py_api import NextcloudException, notes -def test_settings(nc_any): - if nc_any.notes.available is False: - pytest.skip("Notes is not installed") - - original_settings = nc_any.notes.get_settings() - assert isinstance(original_settings["file_suffix"], str) - assert isinstance(original_settings["notes_path"], str) - nc_any.notes.set_settings(file_suffix=".ncpa") - modified_settings = nc_any.notes.get_settings() - assert modified_settings["file_suffix"] == ".ncpa" - assert modified_settings["notes_path"] == original_settings["notes_path"] - nc_any.notes.set_settings(file_suffix=original_settings["file_suffix"]) - modified_settings = nc_any.notes.get_settings() - assert modified_settings["file_suffix"] == original_settings["file_suffix"] - with pytest.raises(ValueError): - nc_any.notes.set_settings() - - @pytest.mark.asyncio(scope="session") async def test_settings_async(anc_any): if await anc_any.notes.available is False: @@ -42,15 +24,6 @@ async def test_settings_async(anc_any): await anc_any.notes.set_settings() -def test_create_delete(nc_any): - if nc_any.notes.available is False: - pytest.skip("Notes is not installed") - unix_timestamp = (datetime.utcnow() - datetime(1970, 1, 1)).total_seconds() - new_note = nc_any.notes.create(str(unix_timestamp)) - nc_any.notes.delete(new_note) - _test_create_delete(new_note) - - @pytest.mark.asyncio(scope="session") async def test_create_delete_async(anc_any): if await anc_any.notes.available is False: @@ -73,40 +46,6 @@ def _test_create_delete(new_note: notes.Note): assert str(new_note).find("title=") != -1 -def test_get_update_note(nc_any): - if nc_any.notes.available is False: - pytest.skip("Notes is not installed") - - for i in nc_any.notes.get_list(): - nc_any.notes.delete(i) - - assert not nc_any.notes.get_list() - unix_timestamp = (datetime.utcnow() - datetime(1970, 1, 1)).total_seconds() - new_note = nc_any.notes.create(str(unix_timestamp)) - try: - all_notes = nc_any.notes.get_list() - assert all_notes[0] == new_note - assert not nc_any.notes.get_list(etag=True) - assert nc_any.notes.get_list()[0] == new_note - assert nc_any.notes.by_id(new_note) == new_note - updated_note = nc_any.notes.update(new_note, content="content") - assert updated_note.content == "content" - all_notes = nc_any.notes.get_list() - assert all_notes[0].content == "content" - all_notes_no_content = nc_any.notes.get_list(no_content=True) - assert all_notes_no_content[0].content == "" - assert nc_any.notes.by_id(new_note).content == "content" - with pytest.raises(NextcloudException): - assert nc_any.notes.update(new_note, content="should be rejected") - new_note = nc_any.notes.update(new_note, content="should not be rejected", overwrite=True) - nc_any.notes.update(new_note, category="test_category", favorite=True) - new_note = nc_any.notes.by_id(new_note) - assert new_note.favorite is True - assert new_note.category == "test_category" - finally: - nc_any.notes.delete(new_note) - - @pytest.mark.asyncio(scope="session") async def test_get_update_note_async(anc_any): if await anc_any.notes.available is False: @@ -142,13 +81,6 @@ async def test_get_update_note_async(anc_any): await anc_any.notes.delete(new_note) -def test_update_note_invalid_param(nc_any): - if nc_any.notes.available is False: - pytest.skip("Notes is not installed") - with pytest.raises(ValueError): - nc_any.notes.update(notes.Note({"id": 0, "etag": "42242"})) - - @pytest.mark.asyncio(scope="session") async def test_update_note_invalid_param_async(anc_any): if await anc_any.notes.available is False: diff --git a/tests/actual_tests/talk_test.py b/tests/actual_tests/talk_test.py index 32e8ad7a..61dc23aa 100644 --- a/tests/actual_tests/talk_test.py +++ b/tests/actual_tests/talk_test.py @@ -132,45 +132,13 @@ def _test_get_conversations_include_status(participants: list[talk.Participant]) assert str(second_participant).find("last_ping=") != -1 -def test_get_conversations_include_status(nc, nc_client): - if nc.talk.available is False: - pytest.skip("Nextcloud Talk is not installed") - nc_second_user = Nextcloud(nc_auth_user=environ["TEST_USER_ID"], nc_auth_pass=environ["TEST_USER_PASS"]) - nc_second_user.user_status.set_status_type("away") - nc_second_user.user_status.set_status("my status message", status_icon="😇") - conversation = nc.talk.create_conversation(talk.ConversationType.ONE_TO_ONE, environ["TEST_USER_ID"]) - try: - conversations = nc.talk.get_user_conversations(include_status=False) - assert conversations - first_conv = next(i for i in conversations if i.conversation_id == conversation.conversation_id) - assert not first_conv.status_type - conversations = nc.talk.get_user_conversations(include_status=True) - assert conversations - first_conv = next(i for i in conversations if i.conversation_id == conversation.conversation_id) - assert first_conv.status_type == "away" - assert first_conv.status_message == "my status message" - assert first_conv.status_icon == "😇" - participants = nc.talk.list_participants(first_conv) - # 10 april 2025: something changed in Nextcloud 31+, and now here is "1" as result instead of 2 - if len(participants) == 1: - return - _test_get_conversations_include_status(participants) - participants = nc.talk.list_participants(first_conv, include_status=True) - assert len(participants) == 2 - second_participant = next(i for i in participants if i.actor_id == environ["TEST_USER_ID"]) - assert second_participant.status_message == "my status message" - assert str(conversation).find("type=") != -1 - finally: - nc.talk.leave_conversation(conversation.token) - - @pytest.mark.asyncio(scope="session") async def test_get_conversations_include_status_async(anc, anc_client): if await anc.talk.available is False: pytest.skip("Nextcloud Talk is not installed") - nc_second_user = Nextcloud(nc_auth_user=environ["TEST_USER_ID"], nc_auth_pass=environ["TEST_USER_PASS"]) - nc_second_user.user_status.set_status_type("away") - nc_second_user.user_status.set_status("my status message-async", status_icon="😇") + nc_second_user = AsyncNextcloud(nc_auth_user=environ["TEST_USER_ID"], nc_auth_pass=environ["TEST_USER_PASS"]) + await nc_second_user.user_status.set_status_type("away") + await nc_second_user.user_status.set_status("my status message-async", status_icon="😇") conversation = await anc.talk.create_conversation(talk.ConversationType.ONE_TO_ONE, environ["TEST_USER_ID"]) try: conversations = await anc.talk.get_user_conversations(include_status=False) diff --git a/tests/actual_tests/user_status_test.py b/tests/actual_tests/user_status_test.py index 4a2dab97..7f506816 100644 --- a/tests/actual_tests/user_status_test.py +++ b/tests/actual_tests/user_status_test.py @@ -10,15 +10,6 @@ ) -def test_available(nc): - assert nc.user_status.available - - -@pytest.mark.asyncio(scope="session") -async def test_available_async(anc): - assert await anc.user_status.available - - def compare_user_statuses(p1: UserStatus, p2: UserStatus): assert p1.user_id == p2.user_id assert p1.status_message == p2.status_message @@ -39,13 +30,9 @@ def _test_get_status(r1: CurrentUserStatus, message): assert str(r1).find("status_id=") != -1 -@pytest.mark.parametrize("message", ("1 2 3", None, "")) -def test_get_status(nc, message): - nc.user_status.set_status(message) - r1 = nc.user_status.get_current() - r2 = nc.user_status.get(nc.user) - compare_user_statuses(r1, r2) - _test_get_status(r1, message) +@pytest.mark.asyncio(scope="session") +async def test_available_async(anc): + assert await anc.user_status.available @pytest.mark.asyncio(scope="session") @@ -58,10 +45,6 @@ async def test_get_status_async(anc, message): _test_get_status(r1, message) -def test_get_status_non_existent_user(nc): - assert nc.user_status.get("no such user") is None - - @pytest.mark.asyncio(scope="session") async def test_get_status_non_existent_user_async(anc): assert await anc.user_status.get("no such user") is None @@ -77,14 +60,6 @@ def _test_get_predefined(r: list[PredefinedStatus]): assert isinstance(i.clear_at, ClearAt) or i.clear_at is None -def test_get_predefined(nc): - r = nc.user_status.get_predefined() - if nc.srv_version["major"] < 27: - assert r == [] - else: - _test_get_predefined(r) - - @pytest.mark.asyncio(scope="session") async def test_get_predefined_async(anc): r = await anc.user_status.get_predefined() @@ -94,17 +69,6 @@ async def test_get_predefined_async(anc): _test_get_predefined(r) -def test_get_list(nc): - r_all = nc.user_status.get_list() - assert r_all - assert isinstance(r_all, list) - r_current = nc.user_status.get_current() - for i in r_all: - if i.user_id == nc.user: - compare_user_statuses(i, r_current) - assert str(i).find("status_type=") != -1 - - @pytest.mark.asyncio(scope="session") async def test_get_list_async(anc): r_all = await anc.user_status.get_list() @@ -117,25 +81,6 @@ async def test_get_list_async(anc): assert str(i).find("status_type=") != -1 -def test_set_status(nc): - time_clear = int(time()) + 60 - nc.user_status.set_status("cool status", time_clear) - r = nc.user_status.get_current() - assert r.status_message == "cool status" - assert r.status_clear_at == time_clear - assert r.status_icon is None - nc.user_status.set_status("Sick!", status_icon="🤒") - r = nc.user_status.get_current() - assert r.status_message == "Sick!" - assert r.status_clear_at is None - assert r.status_icon == "🤒" - nc.user_status.set_status(None) - r = nc.user_status.get_current() - assert r.status_message is None - assert r.status_clear_at is None - assert r.status_icon is None - - @pytest.mark.asyncio(scope="session") async def test_set_status_async(anc): time_clear = int(time()) + 60 @@ -156,14 +101,6 @@ async def test_set_status_async(anc): assert r.status_icon is None -@pytest.mark.parametrize("value", ("online", "away", "dnd", "invisible", "offline")) -def test_set_status_type(nc, value): - nc.user_status.set_status_type(value) - r = nc.user_status.get_current() - assert r.status_type == value - assert r.status_type_defined - - @pytest.mark.asyncio(scope="session") @pytest.mark.parametrize("value", ("online", "away", "dnd", "invisible", "offline")) async def test_set_status_type_async(anc, value): @@ -173,21 +110,6 @@ async def test_set_status_type_async(anc, value): assert r.status_type_defined -@pytest.mark.parametrize("clear_at", (None, int(time()) + 60 * 60 * 9)) -def test_set_predefined(nc, clear_at): - if nc.srv_version["major"] < 27: - nc.user_status.set_predefined("meeting") - else: - predefined_statuses = nc.user_status.get_predefined() - for i in predefined_statuses: - nc.user_status.set_predefined(i.status_id, clear_at) - r = nc.user_status.get_current() - assert r.status_message == i.message - assert r.status_id == i.status_id - assert r.message_predefined - assert r.status_clear_at == clear_at - - @pytest.mark.asyncio(scope="session") @pytest.mark.parametrize("clear_at", (None, int(time()) + 60 * 60 * 9)) async def test_set_predefined_async(anc, clear_at): @@ -204,16 +126,6 @@ async def test_set_predefined_async(anc, clear_at): assert r.status_clear_at == clear_at -def test_get_back_status_from_from_empty_user(nc_app): - orig_user = nc_app._session.user - nc_app._session.set_user("") - try: - with pytest.raises(ValueError): - nc_app.user_status.get_backup_status("") - finally: - nc_app._session.set_user(orig_user) - - @pytest.mark.asyncio(scope="session") async def test_get_back_status_from_from_empty_user_async(anc_app): orig_user = await anc_app._session.user @@ -225,19 +137,11 @@ async def test_get_back_status_from_from_empty_user_async(anc_app): anc_app._session.set_user(orig_user) -def test_get_back_status_from_from_non_exist_user(nc): - assert nc.user_status.get_backup_status("mёm_m-m.l") is None - - @pytest.mark.asyncio(scope="session") async def test_get_back_status_from_from_non_exist_user_async(anc): assert await anc.user_status.get_backup_status("mёm_m-m.l") is None -def test_restore_from_non_existing_back_status(nc): - assert nc.user_status.restore_backup_status("no such backup status") is None - - @pytest.mark.asyncio(scope="session") async def test_restore_from_non_existing_back_status_async(anc): assert await anc.user_status.restore_backup_status("no such backup status") is None diff --git a/tests/actual_tests/weather_status_test.py b/tests/actual_tests/weather_status_test.py index fe8dcf4e..bf343640 100644 --- a/tests/actual_tests/weather_status_test.py +++ b/tests/actual_tests/weather_status_test.py @@ -3,48 +3,11 @@ from nc_py_api import NextcloudException, weather_status -def test_available(nc): - assert nc.weather_status.available - - @pytest.mark.asyncio(scope="session") async def test_available_async(anc): assert await anc.weather_status.available -def test_get_set_location(nc_any): - try: - nc_any.weather_status.set_location(longitude=0.0, latitude=0.0) - except NextcloudException as e: - if e.status_code in (408, 500, 996): - pytest.skip("Some network problem on the host") - raise e from None - loc = nc_any.weather_status.get_location() - assert loc.latitude == 0.0 - assert loc.longitude == 0.0 - assert isinstance(loc.address, str) - assert isinstance(loc.mode, int) - try: - assert nc_any.weather_status.set_location(address="Paris, 75007, France") - except NextcloudException as e: - if e.status_code in (500, 996): - pytest.skip("Some network problem on the host") - raise e from None - loc = nc_any.weather_status.get_location() - assert loc.latitude - assert loc.longitude - if loc.address.find("Unknown") != -1: - pytest.skip("Some network problem on the host") - assert loc.address.find("Paris") != -1 - assert nc_any.weather_status.set_location(latitude=41.896655, longitude=12.488776) - loc = nc_any.weather_status.get_location() - assert loc.latitude == 41.896655 - assert loc.longitude == 12.488776 - if loc.address.find("Unknown") != -1: - pytest.skip("Some network problem on the host") - assert loc.address.find("Rom") != -1 - - @pytest.mark.asyncio(scope="session") async def test_get_set_location_async(anc_any): try: @@ -79,27 +42,12 @@ async def test_get_set_location_async(anc_any): assert loc.address.find("Rom") != -1 -def test_get_set_location_no_lat_lon_address(nc): - with pytest.raises(ValueError): - nc.weather_status.set_location() - - @pytest.mark.asyncio(scope="session") async def test_get_set_location_no_lat_lon_address_async(anc): with pytest.raises(ValueError): await anc.weather_status.set_location() -def test_get_forecast(nc_any): - nc_any.weather_status.set_location(latitude=41.896655, longitude=12.488776) - if nc_any.weather_status.get_location().address.find("Unknown") != -1: - pytest.skip("Some network problem on the host") - forecast = nc_any.weather_status.get_forecast() - assert isinstance(forecast, list) - assert forecast - assert isinstance(forecast[0], dict) - - @pytest.mark.asyncio(scope="session") async def test_get_forecast_async(anc_any): await anc_any.weather_status.set_location(latitude=41.896655, longitude=12.488776) @@ -111,17 +59,6 @@ async def test_get_forecast_async(anc_any): assert isinstance(forecast[0], dict) -def test_get_set_favorites(nc): - nc.weather_status.set_favorites([]) - r = nc.weather_status.get_favorites() - assert isinstance(r, list) - assert not r - nc.weather_status.set_favorites(["Paris, France", "Madrid, Spain"]) - r = nc.weather_status.get_favorites() - assert any("Paris" in x for x in r) - assert any("Madrid" in x for x in r) - - @pytest.mark.asyncio(scope="session") async def test_get_set_favorites_async(anc): await anc.weather_status.set_favorites([]) @@ -134,13 +71,6 @@ async def test_get_set_favorites_async(anc): assert any("Madrid" in x for x in r) -def test_set_mode(nc): - nc.weather_status.set_mode(weather_status.WeatherLocationMode.MODE_BROWSER_LOCATION) - assert nc.weather_status.get_location().mode == weather_status.WeatherLocationMode.MODE_BROWSER_LOCATION.value - nc.weather_status.set_mode(weather_status.WeatherLocationMode.MODE_MANUAL_LOCATION) - assert nc.weather_status.get_location().mode == weather_status.WeatherLocationMode.MODE_MANUAL_LOCATION.value - - @pytest.mark.asyncio(scope="session") async def test_set_mode_async(anc): await anc.weather_status.set_mode(weather_status.WeatherLocationMode.MODE_BROWSER_LOCATION) @@ -153,13 +83,6 @@ async def test_set_mode_async(anc): ).mode == weather_status.WeatherLocationMode.MODE_MANUAL_LOCATION.value -def test_set_mode_invalid(nc): - with pytest.raises(ValueError): - nc.weather_status.set_mode(weather_status.WeatherLocationMode.UNKNOWN) - with pytest.raises(ValueError): - nc.weather_status.set_mode(0) - - @pytest.mark.asyncio(scope="session") async def test_set_mode_invalid_async(anc): with pytest.raises(ValueError): From 19a8e27e8d0f9674856887442e1bfbfd6fd75004 Mon Sep 17 00:00:00 2001 From: bigcat88 Date: Thu, 19 Feb 2026 16:45:48 +0200 Subject: [PATCH 2/3] fix: handle missing content/mediaType in TalkBotMessage for system messages The bot callback fires for all message types including system messages (conversation created, bot joined, etc.) which lack object.content and object.mediaType keys. This caused intermittent KeyError crashes in CI. - Make object_content return {} when content key is missing - Make object_media_type return "" when mediaType key is missing - Skip bot handler assertions for non-user messages in test helpers --- nc_py_api/talk_bot.py | 17 +++++++++++++---- tests/_talk_bot.py | 2 ++ tests/_talk_bot_async.py | 2 ++ 3 files changed, 17 insertions(+), 4 deletions(-) diff --git a/nc_py_api/talk_bot.py b/nc_py_api/talk_bot.py index fc2c539d..47220321 100644 --- a/nc_py_api/talk_bot.py +++ b/nc_py_api/talk_bot.py @@ -62,13 +62,22 @@ def object_name(self) -> str: @property def object_content(self) -> ObjectContent: - """Dictionary with a ``message`` and ``parameters`` keys.""" - return json.loads(self._raw_data["object"]["content"]) + """Dictionary with a ``message`` and ``parameters`` keys. + + .. note:: May return an empty dict for system messages that have no content. + """ + content = self._raw_data["object"].get("content") + if content is None: + return {} + return json.loads(content) @property def object_media_type(self) -> str: - """``text/markdown`` when the message should be interpreted as **Markdown**, otherwise ``text/plain``.""" - return self._raw_data["object"]["mediaType"] + """``text/markdown`` when the message should be interpreted as **Markdown**, otherwise ``text/plain``. + + .. note:: May return an empty string for system messages. + """ + return self._raw_data["object"].get("mediaType", "") @property def conversation_token(self) -> str: diff --git a/tests/_talk_bot.py b/tests/_talk_bot.py index 2e888c29..2e8357cf 100644 --- a/tests/_talk_bot.py +++ b/tests/_talk_bot.py @@ -14,6 +14,8 @@ def coverage_talk_bot_process_request(message: talk_bot.TalkBotMessage, request: Request): + if message.object_name != "message": + return COVERAGE_BOT.react_to_message(message, "🥳") COVERAGE_BOT.react_to_message(message, "🫡") COVERAGE_BOT.delete_reaction(message, "🫡") diff --git a/tests/_talk_bot_async.py b/tests/_talk_bot_async.py index 04afca79..f7966b22 100644 --- a/tests/_talk_bot_async.py +++ b/tests/_talk_bot_async.py @@ -14,6 +14,8 @@ async def coverage_talk_bot_process_request(message: talk_bot.TalkBotMessage, request: Request): + if message.object_name != "message": + return await COVERAGE_BOT.react_to_message(message, "🥳") await COVERAGE_BOT.react_to_message(message, "🫡") await COVERAGE_BOT.delete_reaction(message, "🫡") From 074771cfcee057175e57ec64f426c2105307be4b Mon Sep 17 00:00:00 2001 From: bigcat88 Date: Thu, 19 Feb 2026 17:46:03 +0200 Subject: [PATCH 3/3] Fix ObjectContent TypedDict to allow empty dict for system messages --- nc_py_api/talk_bot.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nc_py_api/talk_bot.py b/nc_py_api/talk_bot.py index 47220321..3b3546c0 100644 --- a/nc_py_api/talk_bot.py +++ b/nc_py_api/talk_bot.py @@ -15,7 +15,7 @@ from .nextcloud import AsyncNextcloudApp, NextcloudApp -class ObjectContent(typing.TypedDict): +class ObjectContent(typing.TypedDict, total=False): """Object content of :py:class:`~nc_py_api.talk_bot.TalkBotMessage`.""" message: str