187 lines
6.3 KiB
Python
187 lines
6.3 KiB
Python
# coding=utf-8
|
|
|
|
import unittest
|
|
import json as py_json
|
|
|
|
from urllib import urlencode
|
|
from urlparse import urlparse, urlunparse, parse_qs
|
|
|
|
from flask import url_for
|
|
|
|
from app import app
|
|
from data import model
|
|
from endpoints.api import api, api_bp
|
|
from endpoints.api.user import Signin
|
|
from endpoints.web import web as web_bp
|
|
from initdb import setup_database_for_testing, finished_database_for_testing
|
|
|
|
|
|
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=200)
|
|
|
|
# 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_confirm_recovery_verified(self):
|
|
# Create a valid code and try.
|
|
user = model.user.get_user('devtable')
|
|
user.verified = False
|
|
user.save()
|
|
|
|
code = model.user.create_reset_password_email_code(user.email)
|
|
self.getResponse('web.confirm_recovery', code=code.code, expected_code=302)
|
|
|
|
# Ensure the current user is the expected user and that they are verified.
|
|
user = model.user.get_user('devtable')
|
|
self.assertTrue(user.verified)
|
|
|
|
self.getResponse('web.receipt', expected_code=404) # Will 401 if no user.
|
|
|
|
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')
|
|
|
|
|
|
class KeyServerTestCase(EndpointTestCase):
|
|
def test_list_service_keys(self):
|
|
pass
|
|
|
|
def test_get_service_key(self):
|
|
pass
|
|
|
|
def test_put_service_key(self):
|
|
pass
|
|
|
|
def test_delete_service_key(self):
|
|
pass
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|