-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: ExApps: occ-commands registration (#247)
API fore registering & unregistering OCC commands Reference: nextcloud/app_api#272 Signed-off-by: Alexander Piskun <[email protected]>
- Loading branch information
Showing
6 changed files
with
297 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,153 @@ | ||
"""Nextcloud API for registering OCC commands for ExApps.""" | ||
|
||
import dataclasses | ||
|
||
from .._exceptions import NextcloudExceptionNotFound | ||
from .._misc import clear_from_params_empty, require_capabilities | ||
from .._session import AsyncNcSessionApp, NcSessionApp | ||
|
||
_EP_SUFFIX: str = "occ_command" | ||
|
||
|
||
@dataclasses.dataclass | ||
class OccCommand: | ||
"""OccCommand description.""" | ||
|
||
def __init__(self, raw_data: dict): | ||
self._raw_data = raw_data | ||
|
||
@property | ||
def name(self) -> str: | ||
"""Unique ID for the command.""" | ||
return self._raw_data["name"] | ||
|
||
@property | ||
def description(self) -> str: | ||
"""Command description.""" | ||
return self._raw_data["description"] | ||
|
||
@property | ||
def hidden(self) -> bool: | ||
"""Flag determining ss command hidden or not.""" | ||
return self._raw_data["hidden"] | ||
|
||
@property | ||
def arguments(self) -> dict: | ||
"""Look at PHP Symfony framework for details.""" | ||
return self._raw_data["arguments"] | ||
|
||
@property | ||
def options(self) -> str: | ||
"""Look at PHP Symfony framework for details.""" | ||
return self._raw_data["options"] | ||
|
||
@property | ||
def usages(self) -> str: | ||
"""Look at PHP Symfony framework for details.""" | ||
return self._raw_data["usages"] | ||
|
||
@property | ||
def action_handler(self) -> str: | ||
"""Relative ExApp url which will be called by Nextcloud.""" | ||
return self._raw_data["execute_handler"] | ||
|
||
def __repr__(self): | ||
return f"<{self.__class__.__name__} name={self.name}, handler={self.action_handler}>" | ||
|
||
|
||
class OccCommandsAPI: | ||
"""API for registering OCC commands, avalaible as **nc.occ_command.<method>**.""" | ||
|
||
def __init__(self, session: NcSessionApp): | ||
self._session = session | ||
|
||
def register( | ||
self, | ||
name: str, | ||
callback_url: str, | ||
arguments: list | None = None, | ||
options: list | None = None, | ||
usages: list | None = None, | ||
description: str = "", | ||
hidden: bool = False, | ||
) -> None: | ||
"""Registers or edit the OCC command.""" | ||
require_capabilities("app_api", self._session.capabilities) | ||
params = { | ||
"name": name, | ||
"description": description, | ||
"arguments": arguments, | ||
"hidden": hidden, | ||
"options": options, | ||
"usages": usages, | ||
"execute_handler": callback_url, | ||
} | ||
clear_from_params_empty(["arguments", "options", "usages"], params) | ||
self._session.ocs("POST", f"{self._session.ae_url}/{_EP_SUFFIX}", json=params) | ||
|
||
def unregister(self, name: str, not_fail=True) -> None: | ||
"""Removes the OCC command.""" | ||
require_capabilities("app_api", self._session.capabilities) | ||
try: | ||
self._session.ocs("DELETE", f"{self._session.ae_url}/{_EP_SUFFIX}", params={"name": name}) | ||
except NextcloudExceptionNotFound as e: | ||
if not not_fail: | ||
raise e from None | ||
|
||
def get_entry(self, name: str) -> OccCommand | None: | ||
"""Get information of the OCC command.""" | ||
require_capabilities("app_api", self._session.capabilities) | ||
try: | ||
return OccCommand(self._session.ocs("GET", f"{self._session.ae_url}/{_EP_SUFFIX}", params={"name": name})) | ||
except NextcloudExceptionNotFound: | ||
return None | ||
|
||
|
||
class AsyncOccCommandsAPI: | ||
"""Async API for registering OCC commands, avalaible as **nc.occ_command.<method>**.""" | ||
|
||
def __init__(self, session: AsyncNcSessionApp): | ||
self._session = session | ||
|
||
async def register( | ||
self, | ||
name: str, | ||
callback_url: str, | ||
arguments: list | None = None, | ||
options: list | None = None, | ||
usages: list | None = None, | ||
description: str = "", | ||
hidden: bool = False, | ||
) -> None: | ||
"""Registers or edit the OCC command.""" | ||
require_capabilities("app_api", await self._session.capabilities) | ||
params = { | ||
"name": name, | ||
"description": description, | ||
"arguments": arguments, | ||
"hidden": hidden, | ||
"options": options, | ||
"usages": usages, | ||
"execute_handler": callback_url, | ||
} | ||
clear_from_params_empty(["arguments", "options", "usages"], params) | ||
await self._session.ocs("POST", f"{self._session.ae_url}/{_EP_SUFFIX}", json=params) | ||
|
||
async def unregister(self, name: str, not_fail=True) -> None: | ||
"""Removes the OCC command.""" | ||
require_capabilities("app_api", await self._session.capabilities) | ||
try: | ||
await self._session.ocs("DELETE", f"{self._session.ae_url}/{_EP_SUFFIX}", params={"name": name}) | ||
except NextcloudExceptionNotFound as e: | ||
if not not_fail: | ||
raise e from None | ||
|
||
async def get_entry(self, name: str) -> OccCommand | None: | ||
"""Get information of the OCC command.""" | ||
require_capabilities("app_api", await self._session.capabilities) | ||
try: | ||
return OccCommand( | ||
await self._session.ocs("GET", f"{self._session.ae_url}/{_EP_SUFFIX}", params={"name": name}) | ||
) | ||
except NextcloudExceptionNotFound: | ||
return None |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,124 @@ | ||
import pytest | ||
|
||
from nc_py_api import NextcloudExceptionNotFound | ||
|
||
|
||
def test_occ_commands_registration(nc_app): | ||
nc_app.occ_commands.register( | ||
"test_occ_name", | ||
"/some_url", | ||
) | ||
result = nc_app.occ_commands.get_entry("test_occ_name") | ||
assert result.name == "test_occ_name" | ||
assert result.description == "" | ||
assert result.action_handler == "some_url" | ||
assert result.hidden is False | ||
assert result.usages == [] | ||
assert result.arguments == [] | ||
assert result.options == [] | ||
nc_app.occ_commands.register( | ||
"test_occ_name2", | ||
"some_url2", | ||
description="desc", | ||
arguments=[ | ||
{ | ||
"name": "argument_name", | ||
"mode": "required", | ||
"description": "Description of the argument", | ||
"default": "default_value", | ||
}, | ||
], | ||
options=[], | ||
) | ||
result2 = nc_app.occ_commands.get_entry("test_occ_name2") | ||
assert result2.name == "test_occ_name2" | ||
assert result2.description == "desc" | ||
assert result2.action_handler == "some_url2" | ||
assert result2.hidden is False | ||
assert result2.usages == [] | ||
assert result2.arguments == [ | ||
{ | ||
"name": "argument_name", | ||
"mode": "required", | ||
"description": "Description of the argument", | ||
"default": "default_value", | ||
} | ||
] | ||
assert result2.options == [] | ||
nc_app.occ_commands.register( | ||
"test_occ_name", | ||
description="new desc", | ||
callback_url="/new_url", | ||
) | ||
result = nc_app.occ_commands.get_entry("test_occ_name") | ||
assert result.name == "test_occ_name" | ||
assert result.description == "new desc" | ||
assert result.action_handler == "new_url" | ||
nc_app.occ_commands.unregister(result.name) | ||
with pytest.raises(NextcloudExceptionNotFound): | ||
nc_app.occ_commands.unregister(result.name, not_fail=False) | ||
nc_app.occ_commands.unregister(result.name) | ||
nc_app.occ_commands.unregister(result2.name, not_fail=False) | ||
assert nc_app.occ_commands.get_entry(result2.name) is None | ||
assert str(result).find("name=") != -1 | ||
|
||
|
||
@pytest.mark.asyncio(scope="session") | ||
async def test_occ_commands_registration_async(anc_app): | ||
await anc_app.occ_commands.register( | ||
"test_occ_name", | ||
"/some_url", | ||
) | ||
result = await anc_app.occ_commands.get_entry("test_occ_name") | ||
assert result.name == "test_occ_name" | ||
assert result.description == "" | ||
assert result.action_handler == "some_url" | ||
assert result.hidden is False | ||
assert result.usages == [] | ||
assert result.arguments == [] | ||
assert result.options == [] | ||
await anc_app.occ_commands.register( | ||
"test_occ_name2", | ||
"some_url2", | ||
description="desc", | ||
arguments=[ | ||
{ | ||
"name": "argument_name", | ||
"mode": "required", | ||
"description": "Description of the argument", | ||
"default": "default_value", | ||
}, | ||
], | ||
options=[], | ||
) | ||
result2 = await anc_app.occ_commands.get_entry("test_occ_name2") | ||
assert result2.name == "test_occ_name2" | ||
assert result2.description == "desc" | ||
assert result2.action_handler == "some_url2" | ||
assert result2.hidden is False | ||
assert result2.usages == [] | ||
assert result2.arguments == [ | ||
{ | ||
"name": "argument_name", | ||
"mode": "required", | ||
"description": "Description of the argument", | ||
"default": "default_value", | ||
} | ||
] | ||
assert result2.options == [] | ||
await anc_app.occ_commands.register( | ||
"test_occ_name", | ||
description="new desc", | ||
callback_url="/new_url", | ||
) | ||
result = await anc_app.occ_commands.get_entry("test_occ_name") | ||
assert result.name == "test_occ_name" | ||
assert result.description == "new desc" | ||
assert result.action_handler == "new_url" | ||
await anc_app.occ_commands.unregister(result.name) | ||
with pytest.raises(NextcloudExceptionNotFound): | ||
await anc_app.occ_commands.unregister(result.name, not_fail=False) | ||
await anc_app.occ_commands.unregister(result.name) | ||
await anc_app.occ_commands.unregister(result2.name, not_fail=False) | ||
assert await anc_app.occ_commands.get_entry(result2.name) is None | ||
assert str(result).find("name=") != -1 |