Add tests for Resource Owner Password Credentials Flow

oidc_provider/tests/test_token_endpoint.py

@@ -51,6 +51,14 @@ class TokenTestCase(TestCase):
         self.user = create_fake_user()
         self.client = create_fake_client(response_type='code')
 
+    def _password_grant_post_data(self):
+        return {
+            'username': 'johndoe',
+            'password': '1234',
+            'grant_type': 'password',
+            'scope': 'openid email',
+        }
+
     def _auth_code_post_data(self, code):
         """
         All the data that will be POSTed to the Token Endpoint.
@@ -128,6 +136,77 @@ class TokenTestCase(TestCase):
 
         return userinfo(request)
 
+    def _auth_header(self):
+        user_pass = self.client.client_id + ':' + self.client.client_secret
+        auth = b'Basic ' + b64encode(user_pass.encode('utf-8'))
+        auth_header = {'HTTP_AUTHORIZATION': auth.decode('utf-8')}
+        return auth_header
+
+    # Resource Owner Password Credentials Grant
+    # requirements to satisfy in all test_password_grant methods
+    # https://tools.ietf.org/html/rfc6749#section-4.3.2
+    #
+    # grant_type
+    #       REQUIRED.  Value MUST be set to "password".
+    # username
+    #       REQUIRED.  The resource owner username.
+    # password
+    #       REQUIRED.  The resource owner password.
+    # scope
+    #       OPTIONAL.  The scope of the access request as described by
+    #       Section 3.3.
+    #
+    # The authorization server MUST:
+    # o  require client authentication for confidential clients or for any
+    #    client that was issued client credentials (or with other
+    #    authentication requirements),
+    # o  authenticate the client if client authentication is included, and
+    # o  validate the resource owner password credentials using its
+    #    existing password validation algorithm.
+
+    def test_password_grant_get_access_token_without_scope(self):
+        post_data = self._password_grant_post_data()
+        del (post_data['scope'])
+
+        response = self._post_request(
+            post_data=post_data,
+            extras=self._auth_header()
+        )
+
+        response_dict = json.loads(response.content.decode('utf-8'))
+        self.assertIn('access_token', response_dict)
+
+    def test_password_grant_get_access_token_with_scope(self):
+        response = self._post_request(
+            post_data=self._password_grant_post_data(),
+            extras=self._auth_header()
+        )
+
+        response_dict = json.loads(response.content.decode('utf-8'))
+        self.assertIn('access_token', response_dict)
+
+    def test_password_grant_get_access_token_invalid_user_credentials(self):
+        invalid_post = self._password_grant_post_data()
+        invalid_post['password'] = 'wrong!'
+
+        response = self._post_request(
+            post_data=invalid_post,
+            extras=self._auth_header()
+        )
+
+        self.assertEqual(400, response.status_code)
+
+    def test_password_grant_get_access_token_invalid_client_credentials(self):
+        self.client.client_id = 'foo'
+        self.client.client_secret = 'bar'
+
+        response = self._post_request(
+            post_data=self._password_grant_post_data(),
+            extras=self._auth_header()
+        )
+
+        self.assertEqual(400, response.status_code)
+
     @override_settings(OIDC_TOKEN_EXPIRE=720)
     def test_authorization_code(self):
         """
@@ -307,12 +386,7 @@ class TokenTestCase(TestCase):
         del basicauth_data['client_id']
         del basicauth_data['client_secret']
 
-        # Generate HTTP Basic Auth header with id and secret.
-        user_pass = self.client.client_id + ':' + self.client.client_secret
-        auth_header = b'Basic ' + b64encode(user_pass.encode('utf-8'))
-        response = self._post_request(basicauth_data, {
-            'HTTP_AUTHORIZATION': auth_header.decode('utf-8'),
-        })
+        response = self._post_request(basicauth_data, self._auth_header())
         response.content.decode('utf-8')
 
         self.assertEqual('invalid_client' in response.content.decode('utf-8'),