OAuth app management API

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)

carto/oauth_apps.py

@@ -0,0 +1,157 @@
+"""
+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 <[email protected]>
+
+
+"""
+
+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()
+    user_name = CharField()
+    redirect_uris = CharField(many=True)
+    icon_url = 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()
+    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):
+        """
+        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.
+        :type name: str
+        :type redirect_uris: list
+        :type icon_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)
+
+    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

tests/test_oauth_apps.py

@@ -0,0 +1,72 @@
+import pytest
+from time import time
+
+from pyrestcli.exceptions import NotFoundException, UnprocessableEntityError
+
+from carto.oauth_apps import OauthAppManager
+from carto.exceptions import CartoException
+
+
+@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')
+
+
+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.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)