Add API for token management

docs/examples.rst

@@ -46,3 +46,29 @@ Below is an example demonstrating how to update a user (requires admin privilege
         "username_to_update",
         update)
 
+
+Example 3: Creating a token
+---------------------------
+
+Below is an example illustrating how to create a token for you application::
+
+    import gogs_client
+    from getpass import getpass
+    from platform import node
+
+    api = GogsApi("https://try.gogs.io/")
+
+    try: token_str = open("tokenfile.txt","r").read()
+    except OSError: token_str = None
+    if token_str:
+        token = gogs_client.Token(token_str)
+    else:
+        username = input("username> ")
+        password = getpass("password> ")
+        login = gogs_client.UsernamePassword(username, password)
+        token = api.ensure_token(login, "my cool app on "+node(), username)
+        open("tokenfile.txt", "w".write(token.token))
+
+    username = api.authenticated_user(token)
+    print("User {} authenticated by token {}".format(username, token_str))
+

gogs_client/auth.py

@@ -1,6 +1,7 @@
 """
 Various classes for Gogs authentication
 """
+from gogs_client.entities import json_get
 
 
 class Authentication(object):
@@ -22,11 +23,27 @@ class Token(Authentication):
     """
     An immutable representation of a Gogs authentication token
     """
-    def __init__(self, token):
+    def __init__(self, token, name=None):
         """
         :param str token: contents of Gogs authentication token
         """
         self._token = token
+        self._name = name
+
+    @staticmethod
+    def from_json(parsed_json):
+        name = json_get(parsed_json, "name")
+        sha1 = json_get(parsed_json, "sha1")
+        return Token(sha1, name)
+
+    @property
+    def name(self):
+        """
+        The name of the token
+
+        :rtype: str
+        """
+        return self._name
 
     @property
     def token(self):

gogs_client/interface.py

@@ -2,6 +2,7 @@
 
 from gogs_client._implementation.http_utils import RelativeHttpRequestor, append_url
 from gogs_client.entities import GogsUser, GogsRepo
+from gogs_client.auth import Token
 
 
 class GogsApi(object):
@@ -43,6 +44,68 @@ def authenticated_user(self, auth):
         response = self._get("/user", auth=auth)
         return GogsUser.from_json(self._check_ok(response).json())
 
+    def get_tokens(self, auth, username=None):
+        """
+        Returns tokens defined for specified user.
+        If no user specified uses user authenticated by the given authentication.
+        Right now, authentication must be UsernamePassword (not Token).
+
+        :param auth.Authentication auth: authentication for user to retrieve
+        :param str username: username of owner of tokens
+
+        :return: list of token representation
+        :rtype: List[Token]
+        :raises NetworkFailure: if there is an error communicating with the server
+        :raises ApiFailure: if the request cannot be serviced
+        """
+        if username is None:
+            username = self.authenticated_user(auth).username
+        response = self._get("/users/{u}/tokens".format(u=username), auth=auth)
+        return [Token.from_json(o) for o in self._check_ok(response).json()]
+
+    def create_token(self, auth, name, username=None):
+        """
+        Creates new token with specified name for specified user.
+        If no user specified uses user authenticated by the given authentication.
+        Right now, authentication must be UsernamePassword (not Token).
+
+        :param auth.Authentication auth: authentication for user to retrieve
+        :param str name: name of new token
+        :param str username: username of owner of new token
+
+        :return: new token representation
+        :rtype: Token
+        :raises NetworkFailure: if there is an error communicating with the server
+        :raises ApiFailure: if the request cannot be serviced
+        """
+        if username is None:
+            username = self.authenticated_user(auth).username
+        data = {"name": name}
+        response = self._post("/users/{u}/tokens".format(u=username), auth=auth, data=data)
+        return Token.from_json(self._check_ok(response).json())
+
+    def ensure_token(self, auth, name, username=None):
+        """
+        Creates new token if token with specified name for specified user does not exists.
+        If no user specified uses user authenticated by the given authentication.
+        Right now, authentication must be UsernamePassword (not Token).
+
+        :param auth.Authentication auth: authentication for user to retrieve
+        :param str name: name of new token
+        :param str username: username of owner of new token
+
+        :return: token representation
+        :rtype: Token
+        :raises NetworkFailure: if there is an error communicating with the server
+        :raises ApiFailure: if the request cannot be serviced
+        """
+        if username is None:
+            username = self.authenticated_user(auth).username
+        tokens = [token for token in self.get_tokens(auth, username) if token.name == name]
+        if tokens:
+            return tokens[0]
+        return self.create_token(auth, name, username)
+
     def create_repo(self, auth, name, description=None, private=False, auto_init=False,
                     gitignore_templates=None, license_template=None, readme_template=None):
         """

tests/interface_test.py

@@ -43,11 +43,15 @@ def setUp(self):
                   "email": "[email protected]",
                   "avatar_url": "/avatars/1"
                 }"""
-        self.token = gogs_client.Token("mytoken")
+        self.token_json_str = """{
+                  "name": "new token",
+                  "sha1": "mytoken"
+                }"""
         self.username_password = gogs_client.UsernamePassword(
             "auth_username", "password")
         self.expected_repo = gogs_client.GogsRepo.from_json(json.loads(self.repo_json_str))
         self.expected_user = gogs_client.GogsUser.from_json(json.loads(self.user_json_str))
+        self.token = gogs_client.Token.from_json(json.loads(self.token_json_str))
 
     @responses.activate
     def test_create_repo1(self):
@@ -238,6 +242,31 @@ def test_authenticated_user(self):
         user = self.client.authenticated_user(self.token)
         self.assert_users_equals(user, self.expected_user)
 
+    @responses.activate
+    def test_ensure_token(self):
+        uri = self.path("/users/{}/tokens".format(self.username_password.username))
+        responses.add(responses.GET, uri, body="[]", status=200)
+        responses.add(responses.POST, uri, body=self.token_json_str, status=200)
+        responses.add(responses.GET, uri, body="["+self.token_json_str+"]", status=200)
+        token = self.client.ensure_token(self.username_password, self.token.name, self.username_password.username)
+        self.assert_tokens_equals(token, self.token)
+        token = self.client.ensure_token(self.username_password, self.token.name, self.username_password.username)
+        self.assert_tokens_equals(token, self.token)
+
+    @responses.activate
+    def test_ensure_auth_token(self):
+        uri = self.path("/user")
+        responses.add(responses.GET, uri, body=self.user_json_str, status=200)
+        uri = self.path("/users/{}/tokens".format(self.expected_user.username))
+        responses.add(responses.GET, uri, body="[]", status=200)
+        responses.add(responses.POST, uri, body=self.token_json_str, status=200)
+        tokens = self.client.get_tokens(self.username_password)
+        self.assertEqual(tokens, [])
+        tokeninfo = self.client.create_token(self.username_password, self.token.name)
+        self.assert_tokens_equals(tokeninfo, self.token)
+        token = self.client.ensure_token(self.username_password, self.token.name)
+        self.assert_tokens_equals(token, self.token)
+
     # helper methods
 
     @staticmethod
@@ -277,6 +306,10 @@ def assert_users_equals(self, user, expected):
         self.assertEqual(user.email, expected.email)
         self.assertEqual(user.avatar_url, expected.avatar_url)
 
+    def assert_tokens_equals(self, token, expected):
+        self.assertEqual(token.name, expected.name)
+        self.assertEqual(token.token, expected.token)
+
 
 if __name__ == "__main__":
     unittest.main()