Merge pull request #1197 from coreos-inc/webpytest

Tests for endpoints/web and some small fixes

buildtrigger/bitbuckethandler.py

@@ -4,7 +4,7 @@ import re
 from jsonschema import validate
 from buildtrigger.triggerutil import (RepositoryReadException, TriggerActivationException,
                                       TriggerDeactivationException, TriggerStartException,
-                                      InvalidPayloadException,
+                                      InvalidPayloadException, TriggerProviderException,
                                       determine_build_ref, raise_if_skipped_build,
                                       find_matching_branches)
 

endpoints/web.py

@@ -614,18 +614,19 @@ def attach_bitbucket_trigger(namespace, repository_name):
 
     try:
       oauth_info = BuildTriggerHandler.get_handler(trigger).get_oauth_url()
-
-      config = {
-        'access_token': oauth_info['access_token']
-      }
-
-      access_token_secret = oauth_info['access_token_secret']
-      model.build.update_build_trigger(trigger, config, auth_token=access_token_secret)
-
-      return redirect(oauth_info['url'])
     except TriggerProviderException:
       trigger.delete_instance()
-      abort(400, message='Could not retrieve OAuth URL from Bitbucket')
+      logger.debug('Could not retrieve Bitbucket OAuth URL')
+      abort(500)
+
+    config = {
+      'access_token': oauth_info['access_token']
+    }
+
+    access_token_secret = oauth_info['access_token_secret']
+    model.build.update_build_trigger(trigger, config, auth_token=access_token_secret)
+
+    return redirect(oauth_info['url'])
 
   abort(403)
 

test/test_endpoints.py

@@ -0,0 +1,156 @@
+# 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()