add read and list token

Author: abaez

consulate/api/acl.py

@@ -24,13 +24,6 @@ class ACL(base.Endpoint):
     tokens.
 
     """
-    def read_self_token(self):
-        """Retrieve the currently used token.
-        :param rtype: dict
-
-        """
-        return self._get(["token", "self"])
-
     def list_policies(self):
         """List all ACL policies available in cluster.
         :param rtype: list
@@ -172,6 +165,28 @@ def delete_role(self, id):
         """
         return self._delete(["policy", id])
 
+    def list_tokens(self):
+        """List all ACL tokens available in cluster.
+        :param rtype: list
+
+        """
+        return self._get(["tokens"])
+
+    def read_token(self, accessor_id):
+        """Read an existing token with the given ID.
+        :param str id: The ID of the role.
+        :param rtype: dict
+
+        """
+        return self._get(["token", accessor_id])
+
+    def read_self_token(self):
+        """Retrieve the currently used token.
+        :param rtype: dict
+
+        """
+        return self._get(["token", "self"])
+
     def create_token(self,
                      accessor_id=None,
                      description=None,
@@ -198,7 +213,17 @@ def create_token(self,
         :param rtype: dict
 
         """
-        pass
+        return self._put_response_body(
+            ["token"], {},
+            dict(
+                model.ACLToken(accessor_id=accessor_id,
+                               description=description,
+                               expiration_time=expiration_time,
+                               expiration_ttl=expiration_ttl,
+                               local=local,
+                               policies=policies,
+                               roles=roles,
+                               service_identities=service_identities)))
 
     # NOTE: Everything below here is deprecated post consul-1.4.0.
 

tests/acl_tests.py

@@ -206,11 +206,35 @@ def test_list_roles_exception(self):
     def test_create_token(self):
         secret_id = self.uuidv4()
         accessor_id = self.uuidv4()
-        result = self.consul.acl.create_create_token(
+        result = self.consul.acl.create_token(
             accessor_id=accessor_id,
             secret_id=secret_id,
             roles=ROLELINKS_SAMPLE,
             policies=POLICYLINKS_SAMPLE,
             service_identities=SERVICE_IDENTITIES_SAMPLE)
         self.assertEqual(result['AccessorID'], accessor_id)
         self.assertEqual(result['SecretID'], secret_id)
+
+    def test_create_and_read_token(self):
+        secret_id = self.uuidv4()
+        accessor_id = self.uuidv4()
+        value = self.consul.acl.create_token(
+            accessor_id=accessor_id,
+            secret_id=secret_id,
+            roles=ROLELINKS_SAMPLE,
+            policies=POLICYLINKS_SAMPLE,
+            service_identities=SERVICE_IDENTITIES_SAMPLE)
+        result = self.consul.acl.read_token(value["AccessorID"])
+        self.assertEqual(result['AccessorID'], accessor_id)
+
+    def test_create_and_delete_token(self):
+        secret_id = self.uuidv4()
+        accessor_id = self.uuidv4()
+        value = self.consul.acl.create_token(
+            accessor_id=accessor_id,
+            secret_id=secret_id,
+            roles=ROLELINKS_SAMPLE,
+            policies=POLICYLINKS_SAMPLE,
+            service_identities=SERVICE_IDENTITIES_SAMPLE)
+        result = self.consul.acl.delete_token(value["AccessorID"])
+        self.assertTrue(result)