add test case for password credential

Author: lepture

tests/oauth2_server.py

@@ -9,6 +9,13 @@
 db = SQLAlchemy()
 
 
+def enable_log(name='flask_oauthlib'):
+    import logging
+    logger = logging.getLogger(name)
+    logger.addHandler(logging.StreamHandler())
+    logger.setLevel(logging.DEBUG)
+
+
 class User(db.Model):
     id = db.Column(db.Integer, primary_key=True)
     username = db.Column(db.Unicode(40), unique=True, index=True,
@@ -102,17 +109,26 @@ def prepare_app(app):
     db.app = app
     db.create_all()
 
-    client = Client(
+    client1 = Client(
         name=u'dev', client_id=u'dev', client_secret=u'dev',
         _redirect_uris=u'http://localhost:8000/authorized'
     )
+
+    client2 = Client(
+        name=u'confidential', client_id=u'confidential',
+        client_secret=u'confidential', client_type=u'confidential',
+        _redirect_uris=u'http://localhost:8000/authorized'
+    )
+
     user = User(username=u'admin')
+
     try:
+        db.session.add(client1)
+        db.session.add(client2)
         db.session.add(user)
-        db.session.add(client)
         db.session.commit()
     except:
-        pass
+        db.session.rollback()
     return app
 
 
@@ -161,6 +177,12 @@ def set_token(token, request, *args, **kwargs):
         db.session.add(tok)
         db.session.commit()
 
+    @oauth.usergetter
+    def get_user(username, password, *args, **kwargs):
+        # This is optional, if you don't need password credential
+        # there is no need to implement this method
+        return User.query.get(1)
+
     @app.before_request
     def load_current_user():
         user = User.query.get(1)

tests/test_oauth2.py

@@ -2,13 +2,14 @@
 
 import os
 import tempfile
+import unittest
 from urlparse import urlparse 
 from flask import Flask
-from .oauth2_server import create_server
+from .oauth2_server import create_server, db
 from .oauth2_client import create_client
 
 
-class BaseSuite(object):
+class BaseSuite(unittest.TestCase):
     def setUp(self):
         app = Flask(__name__)
         app.debug = True
@@ -28,6 +29,9 @@ def setUp(self):
         return app
 
     def tearDown(self):
+        db.session.remove()
+        db.drop_all()
+
         os.close(self.db_fd)
         os.unlink(self.db_file)
 
@@ -37,8 +41,7 @@ def tearDown(self):
     '&redirect_uri=http%3A%2F%2Flocalhost%3A8000%2Fauthorized&scope=email'
 )
 
-
-class TestAuth(BaseSuite):
+class TestWebAuth(BaseSuite):
     def test_login(self):
         rv = self.client.get('/login')
         assert 'response_type=code' in rv.location
@@ -72,10 +75,23 @@ def test_get_access_token(self):
         assert 'access_token' in rv.data
 
     def test_full_flow(self):
-        self.test_get_access_token()
+        rv = self.client.post(authorize_url, data={'confirm': 'yes'})
+        rv = self.client.get(clean_url(rv.location))
+        assert 'access_token' in rv.data
+
         rv = self.client.get('/')
         assert 'username' in rv.data
 
+class TestPasswordAuth(BaseSuite):
+    def test_get_access_token(self):
+        auth_code = 'confidential:confidential'.encode('base64').strip()
+        url = ('/oauth/access_token?grant_type=password'
+               '&scope=email+address&username=admin&password=admin')
+        rv = self.client.get(url, headers={
+            'HTTP_AUTHORIZATION': 'Basic %s' % auth_code,
+        }, data={'confirm': 'yes'})
+        assert 'access_token' in rv.data
+
 
 def clean_url(location):
     ret = urlparse(location)