quay/test/test_keystone_auth.py

import json
import os
import unittest

import requests

from flask import Flask, request, abort
from flask_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()