diff --git a/NEWS b/NEWS index 15cefca..d5f06cc 100644 --- a/NEWS +++ b/NEWS @@ -1,3 +1,6 @@ +xxx-xx-2019: version 1.7.0 + - OAuth apps managemet API (#135) + Jun-17-2019: version 1.6.0 - Auth API (#94) - Kuviz API (#121 #124) diff --git a/carto/oauth_apps.py b/carto/oauth_apps.py new file mode 100644 index 0000000..00427d0 --- /dev/null +++ b/carto/oauth_apps.py @@ -0,0 +1,168 @@ +""" +Module for working with CARTO OAuth app management API + +https://carto.com/developers/oauth/apps/ + +.. module:: carto.oauth_apps + :platform: Unix, Windows + :synopsis: Module for working with CARTO OAuth app management API + +.. moduleauthor:: Alberto Romeu + + +""" + +from pyrestcli.fields import CharField, DateTimeField, BooleanField + +from .resources import Resource, Manager +from .exceptions import CartoException +from .paginators import CartoPaginator + + +API_VERSION = "v4" +API_ENDPOINT = "api/{api_version}/oauth_apps/" +GRANTED_API_ENDPOINT = "api/{api_version}/granted_oauth_apps/" + + +class OauthApp(Resource): + """ + Represents an OAuth app in CARTO. + + """ + id = CharField() + name = CharField() + client_id = CharField() + client_secret = CharField() + user_id = CharField() + redirect_uris = CharField(many=True) + icon_url = CharField() + website_url = CharField() + description = CharField() + restricted = BooleanField() + created_at = DateTimeField() + updated_at = DateTimeField() + + class Meta: + collection_endpoint = API_ENDPOINT.format(api_version=API_VERSION) + name_field = "id" + + def regenerate_client_secret(self): + """ + Regenerates the associated client secret + + :return: + + :raise: CartoException + """ + try: + endpoint = (self.Meta.collection_endpoint + + "{id}/regenerate_secret"). \ + format(id=self.id) + + self.send(endpoint, "POST") + except Exception as e: + raise CartoException(e) + + +class GrantedOauthApp(Resource): + """ + Represents an OAuth app granted to access a CARTO account. + + """ + id = CharField() + name = CharField() + icon_url = CharField() + website_url = CharField() + description = CharField() + scopes = CharField(many=True) + created_at = DateTimeField() + updated_at = DateTimeField() + + class Meta: + collection_endpoint = GRANTED_API_ENDPOINT.format(api_version=API_VERSION) + app_collection_endpoint = API_ENDPOINT.format(api_version=API_VERSION) + name_field = "id" + + def revoke(self): + """ + Revokes the access of the OAuth app to the CARTO account of the user + + :return: + + :raise: CartoException + """ + try: + endpoint = (self.Meta.app_collection_endpoint + + "{id}/revoke"). \ + format(id=self.id) + + self.send(endpoint, "POST") + except Exception as e: + raise CartoException(e) + + def save(self): + pass + + def refresh(self): + pass + + def delete(self): + pass + + +class OauthAppManager(Manager): + """ + Manager for the OauthApp class. + + """ + resource_class = OauthApp + json_collection_attribute = "result" + paginator_class = CartoPaginator + + def create(self, name, redirect_uris, icon_url, description, website_url): + """ + Creates an OauthApp. + + :param name: The OAuth app name + :param redirect_uris: An array of URIs for authorize callback. + :param icon_url: A URL with a squared icon for the Oauth app. + :param description: A description of the app to show in the dashboard. + :param website_url: A public URL to the app. + :type name: str + :type redirect_uris: list + :type icon_url: str + :type description: str + :type website_url: str + + :return: An OauthApp instance with a client_id and client_secret + """ + return super(OauthAppManager, self).create(name=name, + redirect_uris=redirect_uris, + icon_url=icon_url, + description=description, + website_url=website_url) + + def all_granted(self): + """ + Lists granted OAuth apps to access the user CARTO account. + + :return: A list of GrantedOauthApp + """ + raw_resources = [] + + for url, paginator_params in self.paginator.get_urls(GrantedOauthApp.Meta.collection_endpoint): + response = self.paginator.process_response(self.send(url, "get")) + raw_resources += self.client.get_response_data(response, self.Meta.parse_json)[self.json_collection_attribute] if self.json_collection_attribute is not None else self.client.get_response_data(response, self.Meta.parse_json) + + resources = [] + + for raw_resource in raw_resources: + try: + resource = GrantedOauthApp(self.client) + except (ValueError, TypeError): + continue + else: + resource.update_from_dict(raw_resource) + resources.append(resource) + + return resources diff --git a/tests/test_oauth_apps.py b/tests/test_oauth_apps.py new file mode 100644 index 0000000..86c5c97 --- /dev/null +++ b/tests/test_oauth_apps.py @@ -0,0 +1,76 @@ +import pytest +from time import time + +from pyrestcli.exceptions import NotFoundException, UnprocessableEntityError + +from carto.oauth_apps import OauthAppManager + + +@pytest.fixture(scope="module") +def oauth_app_manager(api_key_auth_client_usr): + """ + Returns an OauthAppManager instance that can be reused in tests + :param oauth_app_auth_client: Fixture that provides a valid OauthAppAuthClient + object + :return: OauthAppManager instance + """ + return OauthAppManager(api_key_auth_client_usr) + + +def test_get_oauth_app_not_found(oauth_app_manager): + with pytest.raises(NotFoundException): + oauth_app_manager.get('non-existent') + + +def random_oauth_app_name(): + return '_'.join(str(time()).split('.')) + + +def create_oauth_app(oauth_app_manager, oauth_app_name=None, redirect_uris=['https://localhost']): + if oauth_app_name is None: + oauth_app_name = random_oauth_app_name() + return oauth_app_manager.create(name=oauth_app_name, + redirect_uris=redirect_uris, + icon_url='https://localhost', + description='test from Python SDK', + website_url='https://localhost') + + +def test_create_oauth_app(oauth_app_manager): + oauth_app = create_oauth_app(oauth_app_manager) + oauth_app_get = oauth_app_manager.get(oauth_app.id) + assert oauth_app.id == oauth_app_get.id + assert oauth_app.name == oauth_app_get.name + assert oauth_app.redirect_uris == oauth_app_get.redirect_uris + assert oauth_app.icon_url == oauth_app_get.icon_url + assert oauth_app.website_url == oauth_app_get.website_url + assert oauth_app.client_id is not None + assert oauth_app.client_secret is not None + + oauth_app.delete() + + +def test_create_oauth_app_with_invalid_redirect_uris(oauth_app_manager): + with pytest.raises(UnprocessableEntityError): + create_oauth_app(oauth_app_manager, redirect_uris=['http://localhost']) + + +def test_regenerate_client_secret(oauth_app_manager): + oauth_app = create_oauth_app(oauth_app_manager) + old_client_secret = oauth_app.client_secret + oauth_app.regenerate_client_secret() + assert old_client_secret != oauth_app.client_secret + + oauth_app.delete() + + +@pytest.mark.skipif(True, + reason="Execute manually eventually") +def test_revoke_granted(oauth_app_manager): + granted_oauth_apps = oauth_app_manager.all_granted() + old_count = len(granted_oauth_apps) + if len(granted_oauth_apps) > 0: + granted_oauth_apps[0].revoke() + + granted_oauth_apps = oauth_app_manager.all_granted() + assert old_count > len(granted_oauth_apps)