Move to using an ACL model

Author: gmr

consulate/api/acl.py

@@ -4,6 +4,7 @@
 """
 import logging
 
+from consulate.models import acl
 from consulate.api import base
 from consulate import exceptions
 
@@ -37,10 +38,7 @@ def bootstrap(self):
         :raises: :exc:`~consulate.exceptions.Forbidden`
 
         """
-        response = self._adapter.put(self._build_uri(['bootstrap']))
-        if response.status_code == 403:
-            raise exceptions.Forbidden(response.body)
-        return response.body.get('ID')
+        return self._put_response_body(['bootstrap'])['ID']
 
     def create(self, name, acl_type='client', rules=None):
         """The create endpoint is used to make a new token. A token has a name,
@@ -71,13 +69,9 @@ def create(self, name, acl_type='client', rules=None):
         :raises: consulate.exceptions.Forbidden
 
         """
-        payload = {'Name': name, 'Type': acl_type}
-        if rules:
-            payload['Rules'] = rules
-        response = self._adapter.put(self._build_uri(['create']), payload)
-        if response.status_code == 403:
-            raise exceptions.Forbidden(response.body)
-        return response.body.get('ID') or None
+        return self._put_response_body(
+            ['create'], {}, dict(acl.ACL(
+                name=name, type=acl_type, rules=rules)))['ID']
 
     def clone(self, acl_id):
         """Clone an existing ACL returning the new ACL ID
@@ -87,10 +81,7 @@ def clone(self, acl_id):
         :raises: consulate.exceptions.Forbidden
 
         """
-        response = self._adapter.put(self._build_uri(['clone', acl_id]))
-        if response.status_code == 403:
-            raise exceptions.Forbidden(response.body)
-        return response.body.get('ID')
+        return self._put_response_body(['clone', acl_id])['ID']
 
     def destroy(self, acl_id):
         """Delete the specified ACL
@@ -114,15 +105,15 @@ def info(self, acl_id):
         :raises: consulate.exceptions.NotFound
 
         """
-        result = self._get(['info', acl_id])
-        if not result:
-            raise exceptions.NotFound()
-        return result
+        response = self._get(['info', acl_id], raise_on_404=True)
+        if not response:
+            raise exceptions.NotFound('ACL not found')
+        return response
 
     def list(self):
         """Return a list of all ACLs
 
-        :rtype: list
+        :rtype: list([dict])
         :raises: consulate.exceptions.Forbidden
 
         """
@@ -140,26 +131,22 @@ def replication(self):
         :raises: consulate.exceptions.Forbidden
 
         """
-        result = self._get(['replication'])
-        LOGGER.info('Result: %r', result)
-        return result
+        return self._get(['replication'])
 
     def update(self, acl_id, name, acl_type='client', rules=None):
         """Update an existing ACL, updating its values or add a new ACL if
-        the ACL Id specified is not found.
+        the ACL ID specified is not found.
+
+        The call will return the ID of the ACL.
 
         :param str acl_id: The ACL id
         :param str name: The name of the ACL
         :param str acl_type: The ACL type
         :param str rules: The ACL rules document
-        :rtype: bool
+        :rtype: str
         :raises: consulate.exceptions.Forbidden
 
         """
-        payload = {'ID': acl_id, 'Name': name, 'Type': acl_type}
-        if rules:
-            payload['Rules'] = rules
-        response = self._adapter.put(self._build_uri(['update']), payload)
-        if response.status_code == 403:
-            raise exceptions.Forbidden(response.body)
-        return response.status_code == 200
+        return self._put_response_body(
+            ['update'], {}, dict(acl.ACL(
+                id=acl_id, name=name, type=acl_type, rules=rules)))['ID']

tests/acl_tests.py

@@ -20,16 +20,13 @@
 
 
 class TestCase(base.TestCase):
-    def setUp(self):
-        super(TestCase, self).setUp()
-        self.acl_list = list()
 
-    def tearDown(self):
-        for acl_id in self.acl_list:
-            self.consul.acl.destroy(acl_id)
+    @staticmethod
+    def uuidv4():
+        return str(uuid.uuid4())
 
     def test_bootstrap_success(self):
-        expectation = str(uuid.uuid4())
+        expectation = self.uuidv4()
 
         @httmock.all_requests
         def response_content(_url_unused, request):
@@ -46,97 +43,75 @@ def test_bootstrap_raises(self):
             self.consul.acl.bootstrap()
 
     def test_clone_bad_acl_id(self):
-        acl_id = str(uuid.uuid4())
         with self.assertRaises(consulate.Forbidden):
-            self.forbidden_consul.acl.clone(acl_id)
+            self.consul.acl.clone(self.uuidv4())
 
-    @base.generate_key
-    def test_clone_forbidden(self, acl_id):
+    def test_clone_forbidden(self):
         with self.assertRaises(consulate.Forbidden):
-            self.forbidden_consul.acl.clone(acl_id)
+            self.forbidden_consul.acl.clone(self.uuidv4())
 
-    @base.generate_key
-    def test_create_and_destroy(self, acl_id):
-        acl_id = self.consul.acl.create(acl_id)
-        self.acl_list.append(acl_id)
+    def test_create_and_destroy(self):
+        acl_id = self.consul.acl.create(self.uuidv4())
         self.assertTrue(self.consul.acl.destroy(acl_id))
 
-    @base.generate_key
-    def test_create_with_rules(self, acl_id):
-        acl_id = self.consul.acl.create(acl_id, rules=ACL_RULES)
-        self.acl_list.append(acl_id)
+    def test_create_with_rules(self):
+        acl_id = self.consul.acl.create(self.uuidv4(), rules=ACL_RULES)
         value = self.consul.acl.info(acl_id)
         self.assertEqual(value['Rules'], ACL_RULES)
 
-    @base.generate_key
-    def test_create_and_info(self, acl_id):
-        acl_id = self.consul.acl.create(acl_id)
-        self.acl_list.append(acl_id)
+    def test_create_and_info(self):
+        acl_id = self.consul.acl.create(self.uuidv4())
         self.assertIsNotNone(acl_id)
         data = self.consul.acl.info(acl_id)
         self.assertIsNotNone(data)
         self.assertEqual(acl_id, data.get('ID'))
 
-    @base.generate_key
-    def test_create_and_list(self, acl_id):
-        acl_id = self.consul.acl.create(acl_id)
-        self.acl_list.append(acl_id)
+    def test_create_and_list(self):
+        acl_id = self.consul.acl.create(self.uuidv4())
         data = self.consul.acl.list()
         self.assertIn(acl_id, [r.get('ID') for r in data])
 
-    @base.generate_key
-    def test_create_and_clone(self, acl_id):
-        acl_id = self.consul.acl.create(acl_id)
-        self.acl_list.append(acl_id)
+    def test_create_and_clone(self):
+        acl_id = self.consul.acl.create(self.uuidv4())
         clone_id = self.consul.acl.clone(acl_id)
-        self.acl_list.append(clone_id)
         data = self.consul.acl.list()
         self.assertIn(clone_id, [r.get('ID') for r in data])
 
-    @base.generate_key
-    def test_create_and_update(self, acl_id):
-        acl_id = self.consul.acl.create(acl_id)
-        self.acl_list.append(acl_id)
+    def test_create_and_update(self):
+        acl_id = self.consul.acl.create(self.uuidv4())
         self.consul.acl.update(acl_id, 'Foo')
         data = self.consul.acl.list()
         self.assertIn('Foo', [r.get('Name') for r in data])
         self.assertIn(acl_id, [r.get('ID') for r in data])
 
-    @base.generate_key
-    def test_create_forbidden(self, acl_id):
+    def test_create_forbidden(self):
         with self.assertRaises(consulate.Forbidden):
-            self.forbidden_consul.acl.create(acl_id)
+            self.forbidden_consul.acl.create(self.uuidv4())
 
-    @base.generate_key
-    def test_destroy_forbidden(self, acl_id):
+    def test_destroy_forbidden(self):
         with self.assertRaises(consulate.Forbidden):
-            self.forbidden_consul.acl.destroy(acl_id)
+            self.forbidden_consul.acl.destroy(self.uuidv4())
 
     def test_info_acl_id_not_found(self):
-        acl_id = str(uuid.uuid4())
         with self.assertRaises(consulate.NotFound):
-            self.forbidden_consul.acl.info(acl_id)
+            self.forbidden_consul.acl.info(self.uuidv4())
 
     def test_replication(self):
         result = self.forbidden_consul.acl.replication()
         self.assertFalse(result['Enabled'])
         self.assertFalse(result['Running'])
 
-    @base.generate_key
-    def test_update_not_found_adds_new_key(self, acl_id):
-        self.assertTrue(self.consul.acl.update(acl_id, 'Foo2'))
+    def test_update_not_found_adds_new_key(self):
+        acl_id = self.consul.acl.update(self.uuidv4(), 'Foo2')
         data = self.consul.acl.list()
         self.assertIn('Foo2', [r.get('Name') for r in data])
         self.assertIn(acl_id, [r.get('ID') for r in data])
 
-    @base.generate_key
-    def test_update_with_rules(self, acl_id):
-        self.consul.acl.update(acl_id, name='test', rules=ACL_RULES)
-        self.acl_list.append(acl_id)
+    def test_update_with_rules(self):
+        acl_id = self.consul.acl.update(self.uuidv4(), name='test', rules=ACL_RULES)
         value = self.consul.acl.info(acl_id)
         self.assertEqual(value['Rules'], ACL_RULES)
 
-    @base.generate_key
-    def test_update_forbidden(self, acl_id):
+    def test_update_forbidden(self):
         with self.assertRaises(consulate.Forbidden):
-            self.forbidden_consul.acl.update(acl_id, name='test')
+            self.forbidden_consul.acl.update(self.uuidv4(), name='test')