vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Add some tests to verify we're not leaking anything to completely public users (we're not)

Browse source
This commit is contained in:
yackob03 2013-11-06 17:09:22 -05:00
parent 7dc4c2b250
commit db59b5bf9c
6 changed files with 440 additions and 101 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
data/database.py
View file

@ -4,7 +4,6 @@ import logging
from random import SystemRandom
from datetime import datetime
from peewee import *
from peewee import create_model_tables
from app import app
@ -201,18 +200,7 @@ class QueueItem(BaseModel):
processing_expires = DateTimeField(null=True, index=True)
def initialize_db():
create_model_tables([User, Repository, Image, AccessToken, Role,
RepositoryPermission, Visibility, RepositoryTag,
EmailConfirmation, FederatedLogin, LoginService,
QueueItem, RepositoryBuild, Team, TeamMember,
TeamRole])
Role.create(name='admin')
Role.create(name='write')
Role.create(name='read')
TeamRole.create(name='admin')
TeamRole.create(name='creator')
TeamRole.create(name='member')
Visibility.create(name='public')
Visibility.create(name='private')
LoginService.create(name='github')
all_models = [User, Repository, Image, AccessToken, Role,
RepositoryPermission, Visibility, RepositoryTag,
EmailConfirmation, FederatedLogin, LoginService, QueueItem,
RepositoryBuild, Team, TeamMember, TeamRole]

1
data/model.py
View file

@ -580,6 +580,7 @@ def get_repository_images(namespace_name, repository_name):
return joined.where(Repository.name == repository_name,
Repository.namespace == namespace_name)
def list_repository_tags(namespace_name, repository_name):
select = RepositoryTag.select(RepositoryTag, Image)
with_repo = select.join(Repository)

41
endpoints/api.py
View file

@ -5,7 +5,7 @@ import requests
import urlparse
import json
from flask import request, make_response, jsonify, abort, url_for
from flask import request, make_response, jsonify, abort
from flask.ext.login import login_required, current_user, logout_user
from flask.ext.principal import identity_changed, AnonymousIdentity
from functools import wraps
@ -46,6 +46,19 @@ def api_login_required(f):
return decorated_view
def required_json_args(*required_args):
def wrap(f):
@wraps(f)
def wrapped(*args, **kwargs):
request_data = request.get_json()
for arg in required_args:
if arg not in request_data:
abort(400)
return f(*args, **kwargs)
return wrapped
return wrap
@app.errorhandler(model.DataModelException)
def handle_dme(ex):
return make_response(ex.message, 400)
@ -120,8 +133,10 @@ def change_user_details():
@app.route('/api/user/', methods=['POST'])
@required_json_args('username', 'password', 'email')
def create_user_api():
user_data = request.get_json()
existing_user = model.get_user(user_data['username'])
if existing_user:
error_resp = jsonify({
@ -145,6 +160,7 @@ def create_user_api():
@app.route('/api/signin', methods=['POST'])
@required_json_args('username', 'password')
def signin_api():
signin_data = request.get_json()
@ -184,6 +200,7 @@ def logout():
@app.route("/api/recovery", methods=['POST'])
@required_json_args('email')
def send_recovery():
email = request.get_json()['email']
code = model.create_reset_password_email_code(email)
@ -264,6 +281,7 @@ def team_view(orgname, t):
@app.route('/api/organization/<orgname>', methods=['GET'])
@api_login_required
def get_organization(orgname):
user = current_user.db_user()
@ -277,11 +295,9 @@ def get_organization(orgname):
'is_admin': is_admin
}
if current_user.is_anonymous():
abort(404)
org = model.get_organization(orgname)
if not org:
try:
org = model.get_organization(orgname)
except model.InvalidOrganizationException:
abort(404)
teams = model.get_teams_within_org(org)
@ -289,6 +305,7 @@ def get_organization(orgname):
@app.route('/api/organization/<orgname>/private', methods=['GET'])
@api_login_required
def get_organization_private_allowed(orgname):
permission = CreateRepositoryPermission(orgname)
if permission.can():
@ -318,6 +335,7 @@ def member_view(m):
@app.route('/api/organization/<orgname>/team/<teamname>',
methods=['PUT', 'POST'])
@api_login_required
def update_organization_team(orgname, teamname):
edit_permission = AdministerOrganizationPermission(orgname)
if edit_permission.can():
@ -354,6 +372,7 @@ def update_organization_team(orgname, teamname):
@app.route('/api/organization/<orgname>/team/<teamname>',
methods=['DELETE'])
@api_login_required
def delete_organization_team(orgname, teamname):
permission = AdministerOrganizationPermission(orgname)
if permission.can():
@ -365,6 +384,7 @@ def delete_organization_team(orgname, teamname):
@app.route('/api/organization/<orgname>/team/<teamname>/members',
methods=['GET'])
@api_login_required
def get_organization_team_members(orgname, teamname):
view_permission = ViewTeamPermission(orgname, teamname)
edit_permission = AdministerOrganizationPermission(orgname)
@ -389,6 +409,7 @@ def get_organization_team_members(orgname, teamname):
@app.route('/api/organization/<orgname>/team/<teamname>/members/<membername>',
methods=['PUT', 'POST'])
@api_login_required
def update_organization_team_member(orgname, teamname, membername):
permission = AdministerOrganizationPermission(orgname)
if permission.can():
@ -416,6 +437,7 @@ def update_organization_team_member(orgname, teamname, membername):
@app.route('/api/organization/<orgname>/team/<teamname>/members/<membername>',
methods=['DELETE'])
@api_login_required
def delete_organization_team_member(orgname, teamname, membername):
permission = AdministerOrganizationPermission(orgname)
if permission.can():
@ -666,6 +688,7 @@ def get_repo_builds(namespace, repository):
@app.route('/api/filedrop/', methods=['POST'])
@api_login_required
def get_filedrop_url():
mime_type = request.get_json()['mimeType']
(url, file_id) = user_files.prepare_for_drop(mime_type)
@ -774,7 +797,11 @@ def get_image_changes(namespace, repository, image_id):
def list_tag_images(namespace, repository, tag):
permission = ReadRepositoryPermission(namespace, repository)
if permission.can() or model.repository_is_public(namespace, repository):
tag_image = model.get_tag_image(namespace, repository, tag)
try:
tag_image = model.get_tag_image(namespace, repository, tag)
except model.DataModelException:
abort(404)
parent_images = model.get_parent_images(tag_image)
parents = list(parent_images)

185
initdb.py
View file

@ -6,18 +6,17 @@ import hashlib
from datetime import datetime, timedelta
from flask import url_for
from peewee import SqliteDatabase, create_model_tables, drop_model_tables
import storage
from data.database import initialize_db
from data.database import *
from data import model
from app import app
logger = logging.getLogger(__name__)
store = storage.load()
logging.basicConfig(**app.config['LOGGING_CONFIG'])
SAMPLE_DIFFS = ['test/data/sample/diffs/diffs%s.json' % i
for i in range(1, 10)]
@ -38,7 +37,7 @@ def __gen_image_id(repo, image_num):
global_image_num = [0]
def create_subtree(repo, structure, parent):
def __create_subtree(repo, structure, parent):
num_nodes, subtrees, last_node_tags = structure
# create the nodes
@ -76,7 +75,7 @@ def create_subtree(repo, structure, parent):
new_image.docker_image_id)
for subtree in subtrees:
create_subtree(repo, subtree, new_image)
__create_subtree(repo, subtree, new_image)
def __generate_repository(user, name, description, is_public, permissions,
@ -94,83 +93,113 @@ def __generate_repository(user, name, description, is_public, permissions,
model.set_user_repo_permission(delegate.username, user.username, name,
role)
create_subtree(repo, structure, None)
__create_subtree(repo, structure, None)
return repo
def initialize_database():
create_model_tables(all_models)
Role.create(name='admin')
Role.create(name='write')
Role.create(name='read')
TeamRole.create(name='admin')
TeamRole.create(name='creator')
TeamRole.create(name='member')
Visibility.create(name='public')
Visibility.create(name='private')
LoginService.create(name='github')
def wipe_database():
logger.debug('Wiping all data from the DB.')
# Sanity check to make sure we're not killing our prod db
db = model.db
if (not isinstance(model.db, SqliteDatabase) or
app.config['DB_DRIVER'] is not SqliteDatabase):
raise RuntimeError('Attempted to wipe production database!')
drop_model_tables(all_models, fail_silently=True)
def populate_database():
logger.debug('Populating the DB with test data.')
new_user_1 = model.create_user('devtable', 'password',
'jschorr@devtable.com')
new_user_1.verified = True
new_user_1.save()
new_user_2 = model.create_user('public', 'password',
'jacob.moshenko@gmail.com')
new_user_2.verified = True
new_user_2.save()
new_user_3 = model.create_user('freshuser', 'password', 'no@thanks.com')
new_user_3.verified = True
new_user_3.save()
__generate_repository(new_user_1, 'simple', 'Simple repository.', False,
[], (4, [], ['latest', 'prod']))
__generate_repository(new_user_1, 'complex',
'Complex repository with many branches and tags.',
False, [(new_user_2, 'read')],
(2, [(3, [], 'v2.0'),
(1, [(1, [(1, [], ['prod'])],
'staging'),
(1, [], None)], None)], None))
__generate_repository(new_user_1, 'gargantuan', None, False, [],
(2, [(3, [], 'v2.0'),
(1, [(1, [(1, [], ['latest', 'prod'])],
'staging'),
(1, [], None)], None),
(20, [], 'v3.0'),
(5, [], 'v4.0'),
(1, [(1, [], 'v5.0'), (1, [], 'v6.0')], None)],
None))
__generate_repository(new_user_2, 'publicrepo',
'Public repository pullable by the world.', True,
[], (10, [], 'latest'))
__generate_repository(new_user_1, 'shared',
'Shared repository, another user can write.', False,
[(new_user_2, 'write')], (5, [], 'latest'))
building = __generate_repository(new_user_1, 'building',
'Empty repository which is building.',
False, [], (0, [], None))
org = model.create_organization('devtableorg', 'quay@devtable.com',
new_user_1)
org_repo = __generate_repository(org, 'orgrepo',
'Repository owned by an org.', False,
[], (4, [], ['latest', 'prod']))
reader_team = model.create_team('readers', org, 'member',
'Readers of orgrepo.')
model.set_team_repo_permission(reader_team.name, org_repo.namespace,
org_repo.name, 'read')
model.add_user_to_team(new_user_2, reader_team)
token = model.create_access_token(building, 'write')
tag = 'ci.devtable.com:5000/%s/%s' % (building.namespace, building.name)
build = model.create_repository_build(building, token, '123-45-6789', tag)
build.build_node_id = 1
build.phase = 'building'
build.status_url = 'http://localhost:5000/test/build/status'
build.save()
if __name__ == '__main__':
initialize_db()
logging.basicConfig(**app.config['LOGGING_CONFIG'])
initialize_database()
if app.config.get('POPULATE_DB_TEST_DATA', False):
logger.debug('Populating the DB with test data.')
new_user_1 = model.create_user('devtable', 'password',
'jschorr@devtable.com')
new_user_1.verified = True
new_user_1.save()
new_user_2 = model.create_user('public', 'password',
'jacob.moshenko@gmail.com')
new_user_2.verified = True
new_user_2.save()
new_user_3 = model.create_user('freshuser', 'password', 'no@thanks.com')
new_user_3.verified = True
new_user_3.save()
__generate_repository(new_user_1, 'simple', 'Simple repository.', False,
[], (4, [], ['latest', 'prod']))
__generate_repository(new_user_1, 'complex',
'Complex repository with many branches and tags.',
False, [(new_user_2, 'read')],
(2, [(3, [], 'v2.0'),
(1, [(1, [(1, [], ['prod'])],
'staging'),
(1, [], None)], None)], None))
__generate_repository(new_user_1, 'gargantuan', None, False, [],
(2, [(3, [], 'v2.0'),
(1, [(1, [(1, [], ['latest', 'prod'])],
'staging'),
(1, [], None)], None),
(20, [], 'v3.0'),
(5, [], 'v4.0'),
(1, [(1, [], 'v5.0'), (1, [], 'v6.0')], None)],
None))
__generate_repository(new_user_2, 'publicrepo',
'Public repository pullable by the world.', True,
[], (10, [], 'latest'))
__generate_repository(new_user_1, 'shared',
'Shared repository, another user can write.', False,
[(new_user_2, 'write')], (5, [], 'latest'))
building = __generate_repository(new_user_1, 'building',
'Empty repository which is building.',
False, [], (0, [], None))
org = model.create_organization('devtableorg', 'quay@devtable.com',
new_user_1)
org_repo = __generate_repository(org, 'orgrepo',
'Repository owned by an org.', False,
[], (4, [], ['latest', 'prod']))
reader_team = model.create_team('readers', org, 'member',
'Readers of orgrepo.')
model.set_team_repo_permission(reader_team.name, org_repo.namespace,
org_repo.name, 'read')
model.add_user_to_team(new_user_2, reader_team)
token = model.create_access_token(building, 'write')
tag = 'ci.devtable.com:5000/%s/%s' % (building.namespace, building.name)
build = model.create_repository_build(building, token, '123-45-6789', tag)
build.build_node_id = 1
build.phase = 'building'
build.status_url = 'http://localhost:5000/test/build/status'
build.save()
populate_database()

0
test/__init__.py Normal file
View file

294
test/test_api_security.py Normal file
View file

@ -0,0 +1,294 @@
import unittest
import json
from flask import url_for
from uuid import uuid4
from collections import OrderedDict
import endpoints.api
from app import app
from data import model
from initdb import wipe_database, initialize_database, populate_database
PUBLIC_REPO = 'public/publicrepo'
PRIVATE_REPO = 'devtable/complex'
ORG = 'devtableorg'
ORG_REPO = ORG + '/orgrepo'
ORG_OWNERS = 'owners'
ORG_READERS = 'readers'
ORG_OWNER = 'devtable'
FAKE_IMAGE_ID = uuid4()
FAKE_TAG_NAME = uuid4()
FAKE_USERNAME = uuid4()
FAKE_TEAMNAME = uuid4()
FAKE_TOKEN = uuid4()
def open_kwargs(method='GET', json_object=None):
kwargs = {
'method': method,
}
if json_object is not None:
kwargs['data'] = json.dumps(json_object)
kwargs['content_type'] = 'application/json'
elif method == 'POST' or method == 'PUT':
kwargs['data'] = json.dumps({
'fake': 'json',
'data': 'here',
})
kwargs['content_type'] = 'application/json'
return kwargs
with app.test_request_context() as ctx:
ANON_SPEC = OrderedDict([
(url_for('welcome'), (200, open_kwargs())),
(url_for('plans_list'), (200, open_kwargs())),
(url_for('get_logged_in_user'), (200, open_kwargs())),
(url_for('change_user_details'), (401, open_kwargs('PUT'))),
(url_for('create_user_api'), (400, open_kwargs('POST'))),
(url_for('signin_api'), (400, open_kwargs('POST'))),
(url_for('logout'), (401, open_kwargs('POST'))),
(url_for('send_recovery'), (400, open_kwargs('POST'))),
(url_for('get_matching_users', prefix='dev'), (401, open_kwargs())),
(url_for('get_matching_entities', prefix='dev'), (401, open_kwargs())),
(url_for('get_organization', orgname=ORG), (401, open_kwargs())),
(url_for('get_organization_private_allowed', orgname=ORG),
(401, open_kwargs())),
(url_for('update_organization_team', orgname=ORG, teamname=ORG_OWNERS),
(401, open_kwargs('PUT'))),
(url_for('delete_organization_team', orgname=ORG, teamname=ORG_OWNERS),
(401, open_kwargs('DELETE'))),
(url_for('get_organization_team_members', orgname=ORG,
teamname=ORG_OWNERS), (401, open_kwargs())),
(url_for('update_organization_team_member', orgname=ORG,
teamname=ORG_OWNERS, membername=ORG_OWNER),
(401, open_kwargs('PUT'))),
(url_for('delete_organization_team_member', orgname=ORG,
teamname=ORG_OWNERS, membername=ORG_OWNER),
(401, open_kwargs('DELETE'))),
(url_for('create_repo_api'), (401, open_kwargs('POST'))),
(url_for('match_repos_api'), (200, open_kwargs())),
(url_for('list_repos_api'), (200, open_kwargs())),
(url_for('update_repo_api', repository=PUBLIC_REPO),
(401, open_kwargs('PUT'))),
(url_for('update_repo_api', repository=ORG_REPO),
(401, open_kwargs('PUT'))),
(url_for('update_repo_api', repository=PRIVATE_REPO),
(401, open_kwargs('PUT'))),
(url_for('change_repo_visibility_api', repository=PUBLIC_REPO),
(401, open_kwargs('POST'))),
(url_for('change_repo_visibility_api', repository=ORG_REPO),
(401, open_kwargs('POST'))),
(url_for('change_repo_visibility_api', repository=PRIVATE_REPO),
(401, open_kwargs('POST'))),
(url_for('delete_repository', repository=PUBLIC_REPO),
(401, open_kwargs('DELETE'))),
(url_for('delete_repository', repository=ORG_REPO),
(401, open_kwargs('DELETE'))),
(url_for('delete_repository', repository=PRIVATE_REPO),
(401, open_kwargs('DELETE'))),
(url_for('get_repo_api', repository=PUBLIC_REPO),(200, open_kwargs())),
(url_for('get_repo_api', repository=ORG_REPO), (403, open_kwargs())),
(url_for('get_repo_api', repository=PRIVATE_REPO), (403, open_kwargs())),
(url_for('get_repo_builds', repository=PUBLIC_REPO),
(401, open_kwargs())),
(url_for('get_repo_builds', repository=ORG_REPO), (401, open_kwargs())),
(url_for('get_repo_builds', repository=PRIVATE_REPO),
(401, open_kwargs())),
(url_for('get_filedrop_url'), (401, open_kwargs('POST'))),
(url_for('request_repo_build', repository=PUBLIC_REPO),
(401, open_kwargs('POST'))),
(url_for('request_repo_build', repository=ORG_REPO),
(401, open_kwargs('POST'))),
(url_for('request_repo_build', repository=PRIVATE_REPO),
(401, open_kwargs('POST'))),
(url_for('list_repository_images', repository=PUBLIC_REPO),
(200, open_kwargs())),
(url_for('list_repository_images', repository=ORG_REPO),
(403, open_kwargs())),
(url_for('list_repository_images', repository=PRIVATE_REPO),
(403, open_kwargs())),
(url_for('get_image', repository=PUBLIC_REPO, image_id=FAKE_IMAGE_ID),
(404, open_kwargs())),
(url_for('get_image', repository=ORG_REPO, image_id=FAKE_IMAGE_ID),
(403, open_kwargs())),
(url_for('get_image', repository=PRIVATE_REPO, image_id=FAKE_IMAGE_ID),
(403, open_kwargs())),
(url_for('get_image_changes', repository=PUBLIC_REPO,
image_id=FAKE_IMAGE_ID), (404, open_kwargs())),
(url_for('get_image_changes', repository=ORG_REPO,
image_id=FAKE_IMAGE_ID), (403, open_kwargs())),
(url_for('get_image_changes', repository=PRIVATE_REPO,
image_id=FAKE_IMAGE_ID), (403, open_kwargs())),
(url_for('list_tag_images', repository=PUBLIC_REPO, tag=FAKE_TAG_NAME),
(404, open_kwargs())),
(url_for('list_tag_images', repository=ORG_REPO, tag=FAKE_TAG_NAME),
(403, open_kwargs())),
(url_for('list_tag_images', repository=PRIVATE_REPO, tag=FAKE_TAG_NAME),
(403, open_kwargs())),
(url_for('list_repo_team_permissions', repository=PUBLIC_REPO),
(401, open_kwargs())),
(url_for('list_repo_team_permissions', repository=ORG_REPO),
(401, open_kwargs())),
(url_for('list_repo_team_permissions', repository=PRIVATE_REPO),
(401, open_kwargs())),
(url_for('list_repo_user_permissions', repository=PUBLIC_REPO),
(401, open_kwargs())),
(url_for('list_repo_user_permissions', repository=ORG_REPO),
(401, open_kwargs())),
(url_for('list_repo_user_permissions', repository=PRIVATE_REPO),
(401, open_kwargs())),
(url_for('get_user_permissions', repository=PUBLIC_REPO,
username=FAKE_USERNAME), (401, open_kwargs())),
(url_for('get_user_permissions', repository=ORG_REPO,
username=FAKE_USERNAME), (401, open_kwargs())),
(url_for('get_user_permissions', repository=PRIVATE_REPO,
username=FAKE_USERNAME), (401, open_kwargs())),
(url_for('get_team_permissions', repository=PUBLIC_REPO,
teamname=FAKE_TEAMNAME), (401, open_kwargs())),
(url_for('get_team_permissions', repository=ORG_REPO,
teamname=FAKE_TEAMNAME), (401, open_kwargs())),
(url_for('get_team_permissions', repository=PRIVATE_REPO,
teamname=FAKE_TEAMNAME), (401, open_kwargs())),
(url_for('change_user_permissions', repository=PUBLIC_REPO,
username=FAKE_USERNAME), (401, open_kwargs('PUT'))),
(url_for('change_user_permissions', repository=ORG_REPO,
username=FAKE_USERNAME), (401, open_kwargs('PUT'))),
(url_for('change_user_permissions', repository=PRIVATE_REPO,
username=FAKE_USERNAME), (401, open_kwargs('PUT'))),
(url_for('change_team_permissions', repository=PUBLIC_REPO,
teamname=FAKE_TEAMNAME), (401, open_kwargs('PUT'))),
(url_for('change_team_permissions', repository=ORG_REPO,
teamname=FAKE_TEAMNAME), (401, open_kwargs('PUT'))),
(url_for('change_team_permissions', repository=PRIVATE_REPO,
teamname=FAKE_TEAMNAME), (401, open_kwargs('PUT'))),
(url_for('delete_user_permissions', repository=PUBLIC_REPO,
username=FAKE_USERNAME), (401, open_kwargs('DELETE'))),
(url_for('delete_user_permissions', repository=ORG_REPO,
username=FAKE_USERNAME), (401, open_kwargs('DELETE'))),
(url_for('delete_user_permissions', repository=PRIVATE_REPO,
username=FAKE_USERNAME), (401, open_kwargs('DELETE'))),
(url_for('delete_team_permissions', repository=PUBLIC_REPO,
teamname=FAKE_TEAMNAME), (401, open_kwargs('DELETE'))),
(url_for('delete_team_permissions', repository=ORG_REPO,
teamname=FAKE_TEAMNAME), (401, open_kwargs('DELETE'))),
(url_for('delete_team_permissions', repository=PRIVATE_REPO,
teamname=FAKE_TEAMNAME), (401, open_kwargs('DELETE'))),
(url_for('list_repo_tokens', repository=PUBLIC_REPO),
(401, open_kwargs())),
(url_for('list_repo_tokens', repository=ORG_REPO), (401, open_kwargs())),
(url_for('list_repo_tokens', repository=PRIVATE_REPO),
(401, open_kwargs())),
(url_for('get_tokens', repository=PUBLIC_REPO, code=FAKE_TOKEN),
(401, open_kwargs())),
(url_for('get_tokens', repository=ORG_REPO, code=FAKE_TOKEN),
(401, open_kwargs())),
(url_for('get_tokens', repository=PRIVATE_REPO, code=FAKE_TOKEN),
(401, open_kwargs())),
(url_for('create_token', repository=PUBLIC_REPO),
(401, open_kwargs('POST'))),
(url_for('create_token', repository=ORG_REPO),
(401, open_kwargs('POST'))),
(url_for('create_token', repository=PRIVATE_REPO),
(401, open_kwargs('POST'))),
(url_for('change_token', repository=PUBLIC_REPO, code=FAKE_TOKEN),
(401, open_kwargs('PUT'))),
(url_for('change_token', repository=ORG_REPO, code=FAKE_TOKEN),
(401, open_kwargs('PUT'))),
(url_for('change_token', repository=PRIVATE_REPO, code=FAKE_TOKEN),
(401, open_kwargs('PUT'))),
(url_for('delete_token', repository=PUBLIC_REPO, code=FAKE_TOKEN),
(401, open_kwargs('DELETE'))),
(url_for('delete_token', repository=ORG_REPO, code=FAKE_TOKEN),
(401, open_kwargs('DELETE'))),
(url_for('delete_token', repository=PRIVATE_REPO, code=FAKE_TOKEN),
(401, open_kwargs('DELETE'))),
(url_for('subscribe_api'), (401, open_kwargs('PUT'))),
(url_for('subscribe_org_api', orgname=ORG), (401, open_kwargs('PUT'))),
(url_for('get_subscription'), (401, open_kwargs())),
(url_for('get_org_subscription', orgname=ORG), (401, open_kwargs())),
])
class ApiTestCase(unittest.TestCase):
def setUp(self):
wipe_database()
initialize_database()
populate_database()
self.client = app.test_client()
def signin(self, username, password):
args = {
'username': username,
'password': password,
}
return self.client.post('/signin', data=json.dumps(args),
follow_redirects=True)
def signout(self):
return self.client.get('/signout', follow_redirects=True)
class TestAnonymousAccess(ApiTestCase):
def test_anonymous_public_access(self):
for url, (expected_status, open_kwargs) in ANON_SPEC.items():
rv = self.client.open(url, **open_kwargs)
msg = '%s %s: %s expected: %s' % (open_kwargs['method'], url,
rv.status_code, expected_status)
self.assertEqual(rv.status_code, expected_status, msg)
if __name__ == '__main__':
unittest.main()