Continue on API unit tests
This commit is contained in:
Joseph Schorr
parent
900ccd4c47
commit
05b33dced4
2 changed files with 248 additions and 14 deletions
|
|
@ -261,7 +261,6 @@ def convert_user_to_organization():
|
|||
@internal_api_call
|
||||
def change_user_details():
|
||||
user = current_user.db_user()
|
||||
|
||||
user_data = request.get_json()
|
||||
|
||||
try:
|
||||
|
|
@ -422,6 +421,7 @@ def get_matching_entities(prefix):
|
|||
|
||||
team_data = [entity_team_view(team) for team in teams]
|
||||
user_data = [user_view(user) for user in users]
|
||||
|
||||
return jsonify({
|
||||
'results': team_data + user_data
|
||||
})
|
||||
|
|
@ -447,11 +447,16 @@ def create_organization():
|
|||
existing = None
|
||||
|
||||
try:
|
||||
existing = (model.get_organization(org_data['name']) or
|
||||
model.get_user(org_data['name']))
|
||||
existing = model.get_organization(org_data['name'])
|
||||
except model.InvalidOrganizationException:
|
||||
pass
|
||||
|
||||
if not existing:
|
||||
try:
|
||||
existing = model.get_user(org_data['name'])
|
||||
except model.InvalidUserException:
|
||||
pass
|
||||
|
||||
if existing:
|
||||
msg = 'A user or organization with this name already exists'
|
||||
return request_error(message=msg)
|
||||
|
|
@ -550,7 +555,7 @@ def prototype_view(proto, org_members):
|
|||
'id': proto.uuid,
|
||||
}
|
||||
|
||||
@api.route('/api/organization/<orgname>/prototypes', methods=['GET'])
|
||||
@api.route('/organization/<orgname>/prototypes', methods=['GET'])
|
||||
@api_login_required
|
||||
def get_organization_prototype_permissions(orgname):
|
||||
permission = AdministerOrganizationPermission(orgname)
|
||||
|
|
@ -588,7 +593,7 @@ def log_prototype_action(action_kind, orgname, prototype, **kwargs):
|
|||
log_action(action_kind, orgname, log_params)
|
||||
|
||||
|
||||
@api.route('/api/organization/<orgname>/prototypes', methods=['POST'])
|
||||
@api.route('/organization/<orgname>/prototypes', methods=['POST'])
|
||||
@api_login_required
|
||||
def create_organization_prototype_permission(orgname):
|
||||
permission = AdministerOrganizationPermission(orgname)
|
||||
|
|
@ -636,7 +641,7 @@ def create_organization_prototype_permission(orgname):
|
|||
abort(403)
|
||||
|
||||
|
||||
@api.route('/api/organization/<orgname>/prototypes/<prototypeid>',
|
||||
@api.route('/organization/<orgname>/prototypes/<prototypeid>',
|
||||
methods=['DELETE'])
|
||||
@api_login_required
|
||||
def delete_organization_prototype_permission(orgname, prototypeid):
|
||||
|
|
@ -658,7 +663,7 @@ def delete_organization_prototype_permission(orgname, prototypeid):
|
|||
abort(403)
|
||||
|
||||
|
||||
@api.route('/api/organization/<orgname>/prototypes/<prototypeid>',
|
||||
@api.route('/organization/<orgname>/prototypes/<prototypeid>',
|
||||
methods=['PUT'])
|
||||
@api_login_required
|
||||
def update_organization_prototype_permission(orgname, prototypeid):
|
||||
|
|
@ -1363,7 +1368,7 @@ def get_image_changes(namespace, repository, image_id):
|
|||
abort(403)
|
||||
|
||||
|
||||
@api.route('/api/repository/<path:repository>/tag/<tag>',
|
||||
@api.route('/repository/<path:repository>/tag/<tag>',
|
||||
methods=['DELETE'])
|
||||
@parse_repository_name
|
||||
def delete_full_tag(namespace, repository, tag):
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ from endpoints.api import api
|
|||
from app import app
|
||||
from initdb import setup_database_for_testing, finished_database_for_testing
|
||||
from specs import build_specs
|
||||
from data import model
|
||||
|
||||
app.register_blueprint(api, url_prefix='/api')
|
||||
|
||||
|
|
@ -14,6 +15,12 @@ READ_ACCESS_USER = 'reader'
|
|||
ADMIN_ACCESS_USER = 'devtable'
|
||||
ORGANIZATION = 'buynlarge'
|
||||
|
||||
NEW_USER_DETAILS = {
|
||||
'username': 'bobby',
|
||||
'password': 'password',
|
||||
'email': 'bobby@tables.com',
|
||||
}
|
||||
|
||||
class ApiTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
setup_database_for_testing(self)
|
||||
|
|
@ -25,17 +32,50 @@ class ApiTestCase(unittest.TestCase):
|
|||
finished_database_for_testing(self)
|
||||
self.ctx.__exit__(True, None, None)
|
||||
|
||||
def getJsonResponse(self, method_name):
|
||||
rv = self.app.get(url_for(method_name))
|
||||
assert rv.status_code == 200
|
||||
def getJsonResponse(self, method_name, params={}):
|
||||
rv = self.app.get(url_for(method_name, **params))
|
||||
self.assertEquals(200, rv.status_code)
|
||||
data = rv.data
|
||||
parsed = json.loads(data)
|
||||
return parsed
|
||||
|
||||
def login(self, username):
|
||||
self.app.post(url_for('api.signin_user'),
|
||||
data=json.dumps(dict(username=username, password='password')),
|
||||
def postResponse(self, method_name, params={}, data={}, expected_code=200):
|
||||
rv = self.app.post(url_for(method_name, **params), data=json.dumps(data),
|
||||
headers={"Content-Type": "application/json"})
|
||||
self.assertEquals(rv.status_code, expected_code)
|
||||
return rv.data
|
||||
|
||||
def getResponse(self, method_name, params={}, expected_code=200):
|
||||
rv = self.app.get(url_for(method_name, **params))
|
||||
self.assertEquals(rv.status_code, expected_code)
|
||||
return rv.data
|
||||
|
||||
def postJsonResponse(self, method_name, params={}, data={}, expected_code=200):
|
||||
rv = self.app.post(url_for(method_name, **params), data=json.dumps(data),
|
||||
headers={"Content-Type": "application/json"})
|
||||
|
||||
if rv.status_code != expected_code:
|
||||
print 'Mismatch data for method %s: %s' % (method_name, rv.data)
|
||||
|
||||
self.assertEquals(rv.status_code, expected_code)
|
||||
data = rv.data
|
||||
parsed = json.loads(data)
|
||||
return parsed
|
||||
|
||||
def putJsonResponse(self, method_name, params={}, data={}, expected_code=200):
|
||||
rv = self.app.put(url_for(method_name, **params), data=json.dumps(data),
|
||||
headers={"Content-Type": "application/json"})
|
||||
|
||||
if rv.status_code != expected_code:
|
||||
print 'Mismatch data for method %s: %s' % (method_name, rv.data)
|
||||
|
||||
self.assertEquals(rv.status_code, expected_code)
|
||||
data = rv.data
|
||||
parsed = json.loads(data)
|
||||
return parsed
|
||||
|
||||
def login(self, username, password='password'):
|
||||
return self.postJsonResponse('api.signin_user', data=dict(username=username, password=password))
|
||||
|
||||
class TestDiscovery(ApiTestCase):
|
||||
def test_discovery(self):
|
||||
|
|
@ -81,5 +121,194 @@ class TestGetUserPrivateCount(ApiTestCase):
|
|||
assert json['privateCount'] == 6
|
||||
assert json['reposAllowed'] == 125
|
||||
|
||||
class TestConvertToOrganization(ApiTestCase):
|
||||
def test_sameadminuser(self):
|
||||
self.login(READ_ACCESS_USER)
|
||||
json = self.postJsonResponse('api.convert_user_to_organization',
|
||||
data={'adminUser': READ_ACCESS_USER, 'adminPassword': 'password'},
|
||||
expected_code=400)
|
||||
|
||||
self.assertEqual('The admin user is not valid', json['message'])
|
||||
|
||||
def test_invalidadminuser(self):
|
||||
self.login(READ_ACCESS_USER)
|
||||
json = self.postJsonResponse('api.convert_user_to_organization',
|
||||
data={'adminUser': 'unknownuser', 'adminPassword': 'password'},
|
||||
expected_code=400)
|
||||
|
||||
self.assertEqual('The admin user credentials are not valid', json['message'])
|
||||
|
||||
def test_invalidadminpassword(self):
|
||||
self.login(READ_ACCESS_USER)
|
||||
json = self.postJsonResponse('api.convert_user_to_organization',
|
||||
data={'adminUser': ADMIN_ACCESS_USER, 'adminPassword': 'invalidpass'},
|
||||
expected_code=400)
|
||||
|
||||
self.assertEqual('The admin user credentials are not valid', json['message'])
|
||||
|
||||
def test_convert(self):
|
||||
self.login(READ_ACCESS_USER)
|
||||
json = self.postJsonResponse('api.convert_user_to_organization',
|
||||
data={'adminUser': ADMIN_ACCESS_USER,
|
||||
'adminPassword': 'password',
|
||||
'plan': 'free'})
|
||||
|
||||
self.assertEqual(True, json['success'])
|
||||
|
||||
# Verify the organization exists.
|
||||
organization = model.get_organization(READ_ACCESS_USER)
|
||||
assert organization is not None
|
||||
|
||||
|
||||
class TestChangeUserDetails(ApiTestCase):
|
||||
def test_changepassword(self):
|
||||
self.login(READ_ACCESS_USER)
|
||||
self.putJsonResponse('api.change_user_details', data=dict(password='newpasswordiscool'))
|
||||
self.login(READ_ACCESS_USER, password='newpasswordiscool')
|
||||
|
||||
def test_changeinvoiceemail(self):
|
||||
self.login(READ_ACCESS_USER)
|
||||
|
||||
json = self.putJsonResponse('api.change_user_details', data=dict(invoice_email=True))
|
||||
self.assertEquals(True, json['invoice_email'])
|
||||
|
||||
json = self.putJsonResponse('api.change_user_details', data=dict(invoice_email=False))
|
||||
self.assertEquals(False, json['invoice_email'])
|
||||
|
||||
|
||||
class TestCreateNewUser(ApiTestCase):
|
||||
def test_existingusername(self):
|
||||
json = self.postJsonResponse('api.create_new_user',
|
||||
data=dict(username=READ_ACCESS_USER,
|
||||
password='password',
|
||||
email='test@example.com'),
|
||||
expected_code=400)
|
||||
|
||||
self.assertEquals('The username already exists', json['message'])
|
||||
|
||||
def test_createuser(self):
|
||||
data = self.postResponse('api.create_new_user',
|
||||
data=NEW_USER_DETAILS,
|
||||
expected_code=201)
|
||||
self.assertEquals('Created', data)
|
||||
|
||||
|
||||
class TestSignout(ApiTestCase):
|
||||
def test_signout(self):
|
||||
self.login(READ_ACCESS_USER)
|
||||
|
||||
json = self.getJsonResponse('api.get_logged_in_user')
|
||||
assert json['username'] == READ_ACCESS_USER
|
||||
|
||||
self.postResponse('api.logout')
|
||||
|
||||
json = self.getJsonResponse('api.get_logged_in_user')
|
||||
assert json['anonymous'] == True
|
||||
|
||||
|
||||
class TestGetMatchingEntities(ApiTestCase):
|
||||
def test_notinorg(self):
|
||||
self.login(NO_ACCESS_USER)
|
||||
|
||||
json = self.getJsonResponse('api.get_matching_entities',
|
||||
params=dict(prefix='o', namespace=ORGANIZATION, includeTeams=True))
|
||||
|
||||
names = set([r['name'] for r in json['results']])
|
||||
assert 'outsideorg' in names
|
||||
assert not 'owners' in names
|
||||
|
||||
def test_inorg(self):
|
||||
self.login(ADMIN_ACCESS_USER)
|
||||
|
||||
json = self.getJsonResponse('api.get_matching_entities',
|
||||
params=dict(prefix='o', namespace=ORGANIZATION, includeTeams=True))
|
||||
|
||||
names = set([r['name'] for r in json['results']])
|
||||
assert 'outsideorg' in names
|
||||
assert 'owners' in names
|
||||
|
||||
|
||||
class TestCreateOrganization(ApiTestCase):
|
||||
def test_existinguser(self):
|
||||
self.login(ADMIN_ACCESS_USER)
|
||||
|
||||
json = self.postJsonResponse('api.create_organization', data=dict(name=ADMIN_ACCESS_USER),
|
||||
expected_code=400)
|
||||
|
||||
self.assertEquals('A user or organization with this name already exists', json['message'])
|
||||
|
||||
def test_existingorg(self):
|
||||
self.login(ADMIN_ACCESS_USER)
|
||||
|
||||
json = self.postJsonResponse('api.create_organization', data=dict(name=ORGANIZATION),
|
||||
expected_code=400)
|
||||
|
||||
self.assertEquals('A user or organization with this name already exists', json['message'])
|
||||
|
||||
def test_createorg(self):
|
||||
self.login(ADMIN_ACCESS_USER)
|
||||
|
||||
data = self.postResponse('api.create_organization',
|
||||
data=dict(name='neworg', email='test@example.com'),
|
||||
expected_code=201)
|
||||
|
||||
self.assertEquals('Created', data)
|
||||
|
||||
# Ensure the org was created.
|
||||
organization = model.get_organization('neworg')
|
||||
assert organization is not None
|
||||
|
||||
|
||||
class TestGetOrganization(ApiTestCase):
|
||||
def test_unknownorg(self):
|
||||
self.login(ADMIN_ACCESS_USER)
|
||||
self.getResponse('api.get_organization', params=dict(orgname='notvalid'),
|
||||
expected_code=403)
|
||||
|
||||
def test_cannotaccess(self):
|
||||
self.login(NO_ACCESS_USER)
|
||||
self.getResponse('api.get_organization', params=dict(orgname=ORGANIZATION),
|
||||
expected_code=403)
|
||||
|
||||
def test_getorganization(self):
|
||||
self.login(READ_ACCESS_USER)
|
||||
json = self.getJsonResponse('api.get_organization', params=dict(orgname=ORGANIZATION))
|
||||
|
||||
self.assertEquals(ORGANIZATION, json['name'])
|
||||
self.assertEquals(False, json['is_admin'])
|
||||
|
||||
def test_getorganization_asadmin(self):
|
||||
self.login(ADMIN_ACCESS_USER)
|
||||
json = self.getJsonResponse('api.get_organization', params=dict(orgname=ORGANIZATION))
|
||||
|
||||
self.assertEquals(ORGANIZATION, json['name'])
|
||||
self.assertEquals(True, json['is_admin'])
|
||||
|
||||
class TestChangeOrganizationDetails(ApiTestCase):
|
||||
def test_changeinvoiceemail(self):
|
||||
self.login(ADMIN_ACCESS_USER)
|
||||
|
||||
json = self.putJsonResponse('api.change_organization_details',
|
||||
params=dict(orgname=ORGANIZATION),
|
||||
data=dict(invoice_email=True))
|
||||
|
||||
self.assertEquals(True, json['invoice_email'])
|
||||
|
||||
json = self.putJsonResponse('api.change_organization_details',
|
||||
params=dict(orgname=ORGANIZATION),
|
||||
data=dict(invoice_email=False))
|
||||
self.assertEquals(False, json['invoice_email'])
|
||||
|
||||
|
||||
def test_changemail(self):
|
||||
self.login(ADMIN_ACCESS_USER)
|
||||
|
||||
json = self.putJsonResponse('api.change_organization_details',
|
||||
params=dict(orgname=ORGANIZATION),
|
||||
data=dict(email='newemail@example.com'))
|
||||
|
||||
self.assertEquals('newemail@example.com', json['email'])
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
|
|||
Reference in a new issue