diff --git a/test/test_api_usage.py b/test/test_api_usage.py index f6654fdab..04f7257c3 100644 --- a/test/test_api_usage.py +++ b/test/test_api_usage.py @@ -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()