diff --git a/buildtrigger/bitbuckethandler.py b/buildtrigger/bitbuckethandler.py index 089271355..529b635cc 100644 --- a/buildtrigger/bitbuckethandler.py +++ b/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) diff --git a/endpoints/web.py b/endpoints/web.py index 426a9f041..9e3a69e52 100644 --- a/endpoints/web.py +++ b/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) diff --git a/test/test_endpoints.py b/test/test_endpoints.py new file mode 100644 index 000000000..872585975 --- /dev/null +++ b/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()