quay/test/test_endpoints.py

# coding=utf-8

import unittest
import json as py_json

from data import model
from flask import url_for
from app import app
from endpoints.web import web as web_bp
from endpoints.api import api, api_bp
from endpoints.api.user import Signin
from initdb import setup_database_for_testing, finished_database_for_testing

from urllib import urlencode
from urlparse import urlparse, urlunparse, parse_qs

try:
  app.register_blueprint(web_bp, url_prefix='')
except ValueError:
  # This blueprint was already registered
  pass

try:
  app.register_blueprint(api_bp, url_prefix='/api')
except ValueError:
  # This blueprint was already registered
  pass


CSRF_TOKEN_KEY = '_csrf_token'
CSRF_TOKEN = '123csrfforme'

class EndpointTestCase(unittest.TestCase):
  maxDiff = None

  @staticmethod
  def _add_csrf(without_csrf):
    parts = urlparse(without_csrf)
    query = parse_qs(parts[4])
    query[CSRF_TOKEN_KEY] = CSRF_TOKEN
    return urlunparse(list(parts[0:4]) + [urlencode(query)] + list(parts[5:]))

  def setUp(self):
    setup_database_for_testing(self)
    self.app = app.test_client()
    self.ctx = app.test_request_context()
    self.ctx.__enter__()
    self.setCsrfToken(CSRF_TOKEN)

  def tearDown(self):
    finished_database_for_testing(self)
    self.ctx.__exit__(True, None, None)

  def setCsrfToken(self, token):
    with self.app.session_transaction() as sess:
      sess[CSRF_TOKEN_KEY] = token

  def getResponse(self, resource_name, expected_code=200, **kwargs):
    rv = self.app.get(url_for(resource_name, **kwargs))
    self.assertEquals(rv.status_code, expected_code)
    return rv.data

  def login(self, username, password):
    rv = self.app.post(EndpointTestCase._add_csrf(api.url_for(Signin)),
                       data=py_json.dumps(dict(username=username, password=password)),
                       headers={"Content-Type": "application/json"})
    self.assertEquals(rv.status_code, 200)


class WebEndpointTestCase(EndpointTestCase):
  def test_index(self):
    self.getResponse('web.index')

  def test_repo_view(self):
    self.getResponse('web.repository', path='devtable/simple')

  def test_org_view(self):
    self.getResponse('web.org_view', path='buynlarge')

  def test_user_view(self):
    self.getResponse('web.user_view', path='devtable')

  def test_confirm_repo_email(self):
    code = model.repository.create_email_authorization_for_repo('devtable', 'simple', 'foo@bar.com')
    self.getResponse('web.confirm_repo_email', code=code.code)

    found = model.repository.get_email_authorized_for_repo('devtable', 'simple', 'foo@bar.com')
    self.assertTrue(found.confirmed)

  def test_confirm_email(self):
    user = model.user.get_user('devtable')
    self.assertNotEquals(user.email, 'foo@bar.com')

    code = model.user.create_confirm_email_code(user, 'foo@bar.com')
    self.getResponse('web.confirm_email', code=code.code, expected_code=302)

    user = model.user.get_user('devtable')
    self.assertEquals(user.email, 'foo@bar.com')

  def test_confirm_recovery(self):
    # Try for an invalid code.
    self.getResponse('web.confirm_recovery', code='someinvalidcode', expected_code=403)

    # Create a valid code and try.
    user = model.user.get_user('devtable')
    code = model.user.create_reset_password_email_code(user.email)
    self.getResponse('web.confirm_recovery', code=code.code, expected_code=302)

  def test_build_status_badge(self):
    # Try for an invalid repository.
    self.getResponse('web.build_status_badge', repository='foo/bar', expected_code=404)

    # Try for a public repository.
    self.getResponse('web.build_status_badge', repository='public/publicrepo')

    # Try for an private repository.
    self.getResponse('web.build_status_badge', repository='devtable/simple',
                     expected_code=404)

    # Try for an private repository with an invalid token.
    self.getResponse('web.build_status_badge', repository='devtable/simple',
                     token='sometoken', expected_code=404)

    # Try for an private repository with a valid token.
    repository = model.repository.get_repository('devtable', 'simple')
    self.getResponse('web.build_status_badge', repository='devtable/simple',
                     token=repository.badge_token)

  def test_attach_custom_build_trigger(self):
    self.getResponse('web.attach_custom_build_trigger', repository='foo/bar', expected_code=401)
    self.getResponse('web.attach_custom_build_trigger', repository='devtable/simple', expected_code=401)

    self.login('freshuser', 'password')
    self.getResponse('web.attach_custom_build_trigger', repository='devtable/simple', expected_code=403)

    self.login('devtable', 'password')
    self.getResponse('web.attach_custom_build_trigger', repository='devtable/simple', expected_code=302)

  def test_redirect_to_repository(self):
    self.getResponse('web.redirect_to_repository', repository='foo/bar', expected_code=404)
    self.getResponse('web.redirect_to_repository', repository='public/publicrepo', expected_code=302)
    self.getResponse('web.redirect_to_repository', repository='devtable/simple', expected_code=404)

    self.login('devtable', 'password')
    self.getResponse('web.redirect_to_repository', repository='devtable/simple', expected_code=302)

  def test_redirect_to_namespace(self):
    self.getResponse('web.redirect_to_namespace', namespace='unknown', expected_code=404)
    self.getResponse('web.redirect_to_namespace', namespace='devtable', expected_code=302)
    self.getResponse('web.redirect_to_namespace', namespace='buynlarge', expected_code=302)

  def test_jwk_set_uri(self):
    self.getResponse('web.jwk_set_uri')

if __name__ == '__main__':
  unittest.main()