Add unit tests for all the new API methods
This commit is contained in:
Joseph Schorr
parent
206049dbf8
commit
3f54022344
1 changed files with 156 additions and 6 deletions
|
|
@ -3,11 +3,14 @@ import json as py_json
|
|||
|
||||
from flask import url_for
|
||||
from endpoints.api import api
|
||||
from endpoints.webhooks import webhooks
|
||||
from endpoints.trigger import BuildTrigger
|
||||
from app import app
|
||||
from initdb import setup_database_for_testing, finished_database_for_testing
|
||||
from data import model
|
||||
from data import model, database
|
||||
|
||||
app.register_blueprint(api, url_prefix='/api')
|
||||
app.register_blueprint(webhooks, url_prefix='/webhooks')
|
||||
|
||||
NO_ACCESS_USER = 'freshuser'
|
||||
READ_ACCESS_USER = 'reader'
|
||||
|
|
@ -95,8 +98,6 @@ class ApiTestCase(unittest.TestCase):
|
|||
|
||||
class TestDiscovery(ApiTestCase):
|
||||
def test_discovery(self):
|
||||
""" Basic sanity check that discovery returns valid JSON in the expected
|
||||
format. """
|
||||
json = self.getJsonResponse('api.discovery')
|
||||
found = set([])
|
||||
for method_info in json['endpoints']:
|
||||
|
|
@ -107,8 +108,6 @@ class TestDiscovery(ApiTestCase):
|
|||
|
||||
class TestPlans(ApiTestCase):
|
||||
def test_plans(self):
|
||||
""" Basic sanity check that the plans are returned in the expectedformat.
|
||||
"""
|
||||
json = self.getJsonResponse('api.list_plans')
|
||||
found = set([])
|
||||
for method_info in json['plans']:
|
||||
|
|
@ -836,7 +835,7 @@ class TestGetRepository(ApiTestCase):
|
|||
self.assertEquals(True, json['is_organization'])
|
||||
|
||||
|
||||
class TestGetRepoBuilds(ApiTestCase):
|
||||
class TestRepoBuilds(ApiTestCase):
|
||||
def test_getrepo_nobuilds(self):
|
||||
self.login(ADMIN_ACCESS_USER)
|
||||
|
||||
|
|
@ -857,6 +856,23 @@ class TestGetRepoBuilds(ApiTestCase):
|
|||
assert 'id' in build
|
||||
assert 'status' in build
|
||||
|
||||
# Check the status endpoint.
|
||||
status_json = self.getJsonResponse('api.get_repo_build_status',
|
||||
params=dict(repository=ADMIN_ACCESS_USER + '/building', build_uuid=build['id']))
|
||||
|
||||
self.assertEquals(status_json, build)
|
||||
|
||||
# Check the archive URL.
|
||||
archive_json = self.getJsonResponse('api.get_repo_build_archive_url',
|
||||
params=dict(repository=ADMIN_ACCESS_USER + '/building', build_uuid=build['id']))
|
||||
|
||||
assert archive_json['url']
|
||||
|
||||
# Check the logs.
|
||||
logs_json = self.getJsonResponse('api.get_repo_build_logs',
|
||||
params=dict(repository=ADMIN_ACCESS_USER + '/building', build_uuid=build['id']))
|
||||
|
||||
assert 'logs' in logs_json
|
||||
|
||||
class TestRequestRepoBuild(ApiTestCase):
|
||||
def test_requestrepobuild(self):
|
||||
|
|
@ -1332,5 +1348,139 @@ class TestLogs(ApiTestCase):
|
|||
for log in json['logs']:
|
||||
self.assertEquals(READ_ACCESS_USER, log['performer']['name'])
|
||||
|
||||
|
||||
class FakeBuildTrigger(BuildTrigger):
|
||||
@classmethod
|
||||
def service_name(cls):
|
||||
return 'fakeservice'
|
||||
|
||||
def list_build_sources(self, auth_token):
|
||||
return [{'first': 'source'}, {'second': auth_token}]
|
||||
|
||||
def list_build_subdirs(self, auth_token, config):
|
||||
return [auth_token, 'foo', 'bar', config['somevalue']]
|
||||
|
||||
def handle_trigger_request(self, request, auth_token, config):
|
||||
return ('foo', ['bar'], 'build-name', 'subdir')
|
||||
|
||||
def is_active(self, config):
|
||||
return 'active' in config and config['active']
|
||||
|
||||
def activate(self, trigger_uuid, standard_webhook_url, auth_token, config):
|
||||
config['active'] = True
|
||||
return config
|
||||
|
||||
def deactivate(self, auth_token, config):
|
||||
config['active'] = False
|
||||
return config
|
||||
|
||||
def manual_start(self, auth_token, config):
|
||||
return ('foo', ['bar'], 'build-name', 'subdir')
|
||||
|
||||
|
||||
class TestBuildTriggers(ApiTestCase):
|
||||
def test_list_build_triggers(self):
|
||||
self.login(ADMIN_ACCESS_USER)
|
||||
|
||||
# Check a repo with no known triggers.
|
||||
json = self.getJsonResponse('api.list_build_triggers', params=dict(repository=ADMIN_ACCESS_USER + '/simple'))
|
||||
self.assertEquals(0, len(json['triggers']))
|
||||
|
||||
# Check a repo with one known trigger.
|
||||
json = self.getJsonResponse('api.list_build_triggers', params=dict(repository=ADMIN_ACCESS_USER + '/building'))
|
||||
self.assertEquals(1, len(json['triggers']))
|
||||
|
||||
trigger = json['triggers'][0]
|
||||
|
||||
assert 'id' in trigger
|
||||
assert 'is_active' in trigger
|
||||
assert 'config' in trigger
|
||||
assert 'service' in trigger
|
||||
|
||||
# Verify the get trigger method.
|
||||
trigger_json = self.getJsonResponse('api.get_build_trigger', params=dict(repository=ADMIN_ACCESS_USER + '/building',
|
||||
trigger_uuid=trigger['id']))
|
||||
|
||||
self.assertEquals(trigger, trigger_json)
|
||||
|
||||
# Check the recent builds for the trigger.
|
||||
builds_json = self.getJsonResponse('api.list_trigger_recent_builds', params=dict(repository=ADMIN_ACCESS_USER + '/building',
|
||||
trigger_uuid=trigger['id']))
|
||||
|
||||
assert 'builds' in builds_json
|
||||
|
||||
def test_delete_build_trigger(self):
|
||||
self.login(ADMIN_ACCESS_USER)
|
||||
|
||||
json = self.getJsonResponse('api.list_build_triggers', params=dict(repository=ADMIN_ACCESS_USER + '/building'))
|
||||
self.assertEquals(1, len(json['triggers']))
|
||||
trigger = json['triggers'][0]
|
||||
|
||||
# Delete the trigger.
|
||||
self.deleteResponse('api.delete_build_trigger', params=dict(repository=ADMIN_ACCESS_USER + '/building',
|
||||
trigger_uuid=trigger['id']))
|
||||
|
||||
# Verify it was deleted.
|
||||
json = self.getJsonResponse('api.list_build_triggers', params=dict(repository=ADMIN_ACCESS_USER + '/building'))
|
||||
self.assertEquals(0, len(json['triggers']))
|
||||
|
||||
|
||||
def test_fake_trigger(self):
|
||||
self.login(ADMIN_ACCESS_USER)
|
||||
|
||||
database.BuildTriggerService.create(name='fakeservice')
|
||||
|
||||
# Add a new fake trigger.
|
||||
repo = model.get_repository(ADMIN_ACCESS_USER, 'simple')
|
||||
user = model.get_user(ADMIN_ACCESS_USER)
|
||||
trigger = model.create_build_trigger(repo, 'fakeservice', 'sometoken', user)
|
||||
|
||||
# Verify the trigger.
|
||||
json = self.getJsonResponse('api.list_build_triggers', params=dict(repository=ADMIN_ACCESS_USER + '/simple'))
|
||||
self.assertEquals(1, len(json['triggers']))
|
||||
self.assertEquals(trigger.uuid, json['triggers'][0]['id'])
|
||||
self.assertEquals(trigger.service.name, json['triggers'][0]['service'])
|
||||
self.assertEquals(False, json['triggers'][0]['is_active'])
|
||||
|
||||
# List the trigger's sources.
|
||||
source_json = self.getJsonResponse('api.list_trigger_build_sources', params=dict(repository=ADMIN_ACCESS_USER + '/simple',
|
||||
trigger_uuid=trigger.uuid))
|
||||
self.assertEquals([{'first': 'source'}, {'second': 'sometoken'}], source_json['sources'])
|
||||
|
||||
# List the trigger's subdirs.
|
||||
subdir_json = self.postJsonResponse('api.list_build_trigger_subdirs',
|
||||
params=dict(repository=ADMIN_ACCESS_USER + '/simple', trigger_uuid=trigger.uuid),
|
||||
data={'somevalue': 'meh'})
|
||||
|
||||
self.assertEquals({'status': 'success', 'subdir': ['sometoken', 'foo', 'bar', 'meh']}, subdir_json)
|
||||
|
||||
# Activate the trigger.
|
||||
trigger_config = {}
|
||||
activate_json = self.postJsonResponse('api.activate_build_trigger',
|
||||
params=dict(repository=ADMIN_ACCESS_USER + '/simple', trigger_uuid=trigger.uuid),
|
||||
data=trigger_config)
|
||||
|
||||
self.assertEquals(True, activate_json['is_active'])
|
||||
|
||||
# Make sure the trigger has a write token.
|
||||
trigger = model.get_build_trigger(ADMIN_ACCESS_USER, 'simple', trigger.uuid)
|
||||
self.assertNotEquals(None, trigger.write_token)
|
||||
self.assertEquals(True, py_json.loads(trigger.config)['active'])
|
||||
|
||||
# Make sure we cannot activate again.
|
||||
self.postResponse('api.activate_build_trigger',
|
||||
params=dict(repository=ADMIN_ACCESS_USER + '/simple', trigger_uuid=trigger.uuid),
|
||||
data=trigger_config,
|
||||
expected_code=400)
|
||||
|
||||
# Start a manual build.
|
||||
start_json = self.postJsonResponse('api.manually_start_build_trigger',
|
||||
params=dict(repository=ADMIN_ACCESS_USER + '/simple', trigger_uuid=trigger.uuid),
|
||||
expected_code=201)
|
||||
|
||||
assert 'id' in start_json
|
||||
self.assertEquals("build-name", start_json['display_name'])
|
||||
self.assertEquals(['bar'], start_json['job_config']['docker_tags'])
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
|
|||
Reference in a new issue