Merge pull request #1193 from coreos-inc/keystonetest
Add basic tests for keystone auth
This commit is contained in:
commit
d5920319de
1 changed files with 112 additions and 0 deletions
112
test/test_keystone_auth.py
Normal file
112
test/test_keystone_auth.py
Normal file
|
@ -0,0 +1,112 @@
|
|||
import unittest
|
||||
import requests
|
||||
import os
|
||||
import json
|
||||
|
||||
from flask import Flask, request, abort
|
||||
from flask.ext.testing import LiveServerTestCase
|
||||
from data.users.keystone import KeystoneUsers
|
||||
|
||||
_PORT_NUMBER = 5001
|
||||
|
||||
class KeystoneAuthTests(LiveServerTestCase):
|
||||
maxDiff = None
|
||||
|
||||
def create_app(self):
|
||||
global _PORT_NUMBER
|
||||
_PORT_NUMBER = _PORT_NUMBER + 1
|
||||
|
||||
users = [
|
||||
{'username': 'adminuser', 'name': 'Admin User', 'password': 'adminpass'},
|
||||
{'username': 'cooluser', 'name': 'Cool User', 'password': 'password'},
|
||||
{'username': 'some.neat.user', 'name': 'Neat User', 'password': 'foobar'},
|
||||
]
|
||||
|
||||
ks_app = Flask('testks')
|
||||
ks_app.config['LIVESERVER_PORT'] = _PORT_NUMBER
|
||||
|
||||
if os.environ.get('DEBUG') == 'true':
|
||||
ks_app.config['DEBUG'] = True
|
||||
|
||||
@ks_app.route('/v2.0/admin/users/<userid>', methods=['GET'])
|
||||
def getuser(userid):
|
||||
for user in users:
|
||||
if user['username'] == userid:
|
||||
return json.dumps({
|
||||
'user': {
|
||||
'email': userid + '@example.com',
|
||||
}
|
||||
})
|
||||
|
||||
abort(404)
|
||||
|
||||
|
||||
@ks_app.route('/v2.0/auth/tokens', methods=['POST'])
|
||||
def tokens():
|
||||
creds = request.json['auth'][u'passwordCredentials']
|
||||
for user in users:
|
||||
if creds['username'] == user['username'] and creds['password'] == user['password']:
|
||||
return json.dumps({
|
||||
"access": {
|
||||
"token": {
|
||||
"issued_at": "2014-06-16T22:24:26.089380",
|
||||
"expires": "2020-06-16T23:24:26Z",
|
||||
"id": creds['username'],
|
||||
"tenant": {"id": "sometenant"},
|
||||
},
|
||||
"serviceCatalog":[
|
||||
{
|
||||
"endpoints": [
|
||||
{
|
||||
"adminURL": self.get_server_url() + '/v2.0/admin',
|
||||
}
|
||||
],
|
||||
"endpoints_links": [],
|
||||
"type": "identity",
|
||||
"name": "admin",
|
||||
},
|
||||
],
|
||||
"user": {
|
||||
"username": creds['username'],
|
||||
"roles_links": [],
|
||||
"id": creds['username'],
|
||||
"roles": [],
|
||||
"name": user['name'],
|
||||
},
|
||||
"metadata": {
|
||||
"is_admin": 0,
|
||||
"roles": [],
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
abort(403)
|
||||
|
||||
return ks_app
|
||||
|
||||
def setUp(self):
|
||||
self.session = requests.Session()
|
||||
self.keystone = KeystoneUsers(self.get_server_url() + '/v2.0/auth', 'adminuser', 'adminpass',
|
||||
'admintenant')
|
||||
|
||||
def test_invalid_user(self):
|
||||
(user, _) = self.keystone.verify_credentials('unknownuser', 'password')
|
||||
self.assertIsNone(user)
|
||||
|
||||
def test_invalid_password(self):
|
||||
(user, _) = self.keystone.verify_credentials('cooluser', 'notpassword')
|
||||
self.assertIsNone(user)
|
||||
|
||||
def test_cooluser(self):
|
||||
(user, _) = self.keystone.verify_credentials('cooluser', 'password')
|
||||
self.assertEquals(user.username, 'cooluser')
|
||||
self.assertEquals(user.email, 'cooluser@example.com')
|
||||
|
||||
def test_neatuser(self):
|
||||
(user, _) = self.keystone.verify_credentials('some.neat.user', 'foobar')
|
||||
self.assertEquals(user.username, 'some.neat.user')
|
||||
self.assertEquals(user.email, 'some.neat.user@example.com')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
Reference in a new issue