Merge "Store hash of access as primary key for ec2 type."

keystone/tests/test_v3_credential.py

@@ -14,6 +14,8 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import hashlib
+import json
 import uuid
 
 import test_v3
@@ -76,3 +78,53 @@ class CredentialTestCase(test_v3.RestfulTestCase):
         self.delete(
             '/credentials/%(credential_id)s' % {
                 'credential_id': self.credential_id})
+
+    def test_create_ec2_credential(self):
+        """Call ``POST /credentials`` for creating ec2 credential."""
+        ref = self.new_credential_ref(user_id=self.user['id'])
+        blob = {"access": uuid.uuid4().hex,
+                "secret": uuid.uuid4().hex}
+        ref['blob'] = json.dumps(blob)
+        ref['type'] = 'ec2'
+        r = self.post(
+            '/credentials',
+            body={'credential': ref})
+        self.assertValidCredentialResponse(r, ref)
+        # Assert credential id is same as hash of access key id for
+        # ec2 credentials
+        self.assertEqual(r.result['credential']['id'],
+                         hashlib.sha256(blob['access']).hexdigest())
+        # Create second ec2 credential with the same access key id and check
+        # for conflict.
+        self.post(
+            '/credentials',
+            body={'credential': ref}, expected_status=409)
+
+    def test_create_non_ec2_credential(self):
+        """Call ``POST /credentials`` for creating non-ec2 credential."""
+        ref = self.new_credential_ref(user_id=self.user['id'])
+        blob = {"access": uuid.uuid4().hex,
+                "secret": uuid.uuid4().hex}
+        ref['blob'] = json.dumps(blob)
+        r = self.post(
+            '/credentials',
+            body={'credential': ref})
+        self.assertValidCredentialResponse(r, ref)
+        # Assert credential id is not same as hash of access key id for
+        # non-ec2 credentials
+        self.assertNotEqual(r.result['credential']['id'],
+                            hashlib.sha256(blob['access']).hexdigest())
+
+    def test_create_ec2_credential_with_invalid_blob(self):
+        """Call ``POST /credentials`` for creating ec2
+           credential with invalid blob.
+        """
+        ref = self.new_credential_ref(user_id=self.user['id'])
+        ref['blob'] = '{"abc":"def"d}'
+        ref['type'] = 'ec2'
+        # Assert 400 status for bad request containing invalid
+        # blob
+        response = self.post(
+            '/credentials',
+            body={'credential': ref}, expected_status=400)
+        self.assertValidErrorResponse(response)