Show More
test_integration.py
216 lines
| 7.3 KiB
| text/x-python
|
PythonLexer
r427 | # -*- coding: utf-8 -*- | |||
r2487 | # Copyright (C) 2010-2018 RhodeCode GmbH | |||
r427 | # | |||
# This program is free software: you can redistribute it and/or modify | ||||
# it under the terms of the GNU Affero General Public License, version 3 | ||||
# (only), as published by the Free Software Foundation. | ||||
# | ||||
# This program is distributed in the hope that it will be useful, | ||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of | ||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | ||||
# GNU General Public License for more details. | ||||
# | ||||
# You should have received a copy of the GNU Affero General Public License | ||||
# along with this program. If not, see <http://www.gnu.org/licenses/>. | ||||
# | ||||
# This program is dual-licensed. If you wish to learn more about the | ||||
# RhodeCode Enterprise Edition, including its added features, Support services, | ||||
# and proprietary license terms, please see https://rhodecode.com/licenses/ | ||||
r731 | import time | |||
r427 | import pytest | |||
from rhodecode import events | ||||
r731 | from rhodecode.tests.fixture import Fixture | |||
r427 | from rhodecode.model.db import Session, Integration | |||
from rhodecode.model.integration import IntegrationModel | ||||
r731 | class TestDeleteScopesDeletesIntegrations(object): | |||
r1990 | def test_delete_repo_with_integration_deletes_integration( | |||
self, repo_integration_stub): | ||||
r731 | Session().delete(repo_integration_stub.repo) | |||
Session().commit() | ||||
Session().expire_all() | ||||
integration = Integration.get(repo_integration_stub.integration_id) | ||||
assert integration is None | ||||
r427 | ||||
r1990 | def test_delete_repo_group_with_integration_deletes_integration( | |||
self, repogroup_integration_stub): | ||||
r427 | ||||
r731 | Session().delete(repogroup_integration_stub.repo_group) | |||
Session().commit() | ||||
Session().expire_all() | ||||
integration = Integration.get(repogroup_integration_stub.integration_id) | ||||
assert integration is None | ||||
r427 | ||||
@pytest.fixture | ||||
r731 | def integration_repos(request, StubIntegrationType, stub_integration_settings): | |||
""" | ||||
Create repositories and integrations for testing, and destroy them after | ||||
r793 | ||||
Structure: | ||||
root_repo | ||||
parent_group/ | ||||
parent_repo | ||||
child_group/ | ||||
child_repo | ||||
other_group/ | ||||
other_repo | ||||
r731 | """ | |||
fixture = Fixture() | ||||
r793 | ||||
parent_group_id = 'int_test_parent_group_%s' % time.time() | ||||
parent_group = fixture.create_repo_group(parent_group_id) | ||||
other_group_id = 'int_test_other_group_%s' % time.time() | ||||
other_group = fixture.create_repo_group(other_group_id) | ||||
r731 | ||||
r793 | child_group_id = ( | |||
parent_group_id + '/' + 'int_test_child_group_%s' % time.time()) | ||||
child_group = fixture.create_repo_group(child_group_id) | ||||
parent_repo_id = 'int_test_parent_repo_%s' % time.time() | ||||
parent_repo = fixture.create_repo(parent_repo_id, repo_group=parent_group) | ||||
child_repo_id = 'int_test_child_repo_%s' % time.time() | ||||
child_repo = fixture.create_repo(child_repo_id, repo_group=child_group) | ||||
other_repo_id = 'int_test_other_repo_%s' % time.time() | ||||
other_repo = fixture.create_repo(other_repo_id, repo_group=other_group) | ||||
r731 | ||||
root_repo_id = 'int_test_repo_root_%s' % time.time() | ||||
root_repo = fixture.create_repo(root_repo_id) | ||||
r427 | ||||
r793 | integrations = {} | |||
for name, repo, repo_group, child_repos_only in [ | ||||
('global', None, None, None), | ||||
('root_repos', None, None, True), | ||||
('parent_repo', parent_repo, None, None), | ||||
('child_repo', child_repo, None, None), | ||||
('other_repo', other_repo, None, None), | ||||
('root_repo', root_repo, None, None), | ||||
('parent_group', None, parent_group, True), | ||||
('parent_group_recursive', None, parent_group, False), | ||||
('child_group', None, child_group, True), | ||||
('child_group_recursive', None, child_group, False), | ||||
('other_group', None, other_group, True), | ||||
('other_group_recursive', None, other_group, False), | ||||
]: | ||||
integrations[name] = IntegrationModel().create( | ||||
StubIntegrationType, settings=stub_integration_settings, | ||||
enabled=True, name='test %s integration' % name, | ||||
repo=repo, repo_group=repo_group, child_repos_only=child_repos_only) | ||||
r731 | ||||
Session().commit() | ||||
r427 | ||||
r731 | def _cleanup(): | |||
r793 | for integration in integrations.values(): | |||
Session.delete(integration) | ||||
r731 | fixture.destroy_repo(root_repo) | |||
r793 | fixture.destroy_repo(child_repo) | |||
fixture.destroy_repo(parent_repo) | ||||
fixture.destroy_repo(other_repo) | ||||
fixture.destroy_repo_group(child_group) | ||||
fixture.destroy_repo_group(parent_group) | ||||
fixture.destroy_repo_group(other_group) | ||||
r731 | ||||
request.addfinalizer(_cleanup) | ||||
return { | ||||
r793 | 'integrations': integrations, | |||
r731 | 'repos': { | |||
'root_repo': root_repo, | ||||
r793 | 'other_repo': other_repo, | |||
'parent_repo': parent_repo, | ||||
'child_repo': child_repo, | ||||
r731 | } | |||
} | ||||
r427 | ||||
r731 | def test_enabled_integration_repo_scopes(integration_repos): | |||
integrations = integration_repos['integrations'] | ||||
repos = integration_repos['repos'] | ||||
triggered_integrations = IntegrationModel().get_for_event( | ||||
events.RepoEvent(repos['root_repo'])) | ||||
assert triggered_integrations == [ | ||||
integrations['global'], | ||||
r793 | integrations['root_repos'], | |||
integrations['root_repo'], | ||||
] | ||||
triggered_integrations = IntegrationModel().get_for_event( | ||||
events.RepoEvent(repos['other_repo'])) | ||||
assert triggered_integrations == [ | ||||
integrations['global'], | ||||
integrations['other_repo'], | ||||
integrations['other_group'], | ||||
integrations['other_group_recursive'], | ||||
r731 | ] | |||
triggered_integrations = IntegrationModel().get_for_event( | ||||
r793 | events.RepoEvent(repos['parent_repo'])) | |||
r427 | ||||
r731 | assert triggered_integrations == [ | |||
integrations['global'], | ||||
r793 | integrations['parent_repo'], | |||
integrations['parent_group'], | ||||
integrations['parent_group_recursive'], | ||||
r731 | ] | |||
triggered_integrations = IntegrationModel().get_for_event( | ||||
r793 | events.RepoEvent(repos['child_repo'])) | |||
r731 | ||||
assert triggered_integrations == [ | ||||
integrations['global'], | ||||
r793 | integrations['child_repo'], | |||
integrations['parent_group_recursive'], | ||||
integrations['child_group'], | ||||
integrations['child_group_recursive'], | ||||
r731 | ] | |||
r427 | ||||
r731 | def test_disabled_integration_repo_scopes(integration_repos): | |||
integrations = integration_repos['integrations'] | ||||
repos = integration_repos['repos'] | ||||
for integration in integrations.values(): | ||||
integration.enabled = False | ||||
r427 | Session().commit() | |||
r731 | ||||
triggered_integrations = IntegrationModel().get_for_event( | ||||
events.RepoEvent(repos['root_repo'])) | ||||
assert triggered_integrations == [] | ||||
triggered_integrations = IntegrationModel().get_for_event( | ||||
r793 | events.RepoEvent(repos['parent_repo'])) | |||
r731 | ||||
assert triggered_integrations == [] | ||||
triggered_integrations = IntegrationModel().get_for_event( | ||||
r793 | events.RepoEvent(repos['child_repo'])) | |||
r731 | ||||
assert triggered_integrations == [] | ||||
r793 | triggered_integrations = IntegrationModel().get_for_event( | |||
events.RepoEvent(repos['other_repo'])) | ||||
assert triggered_integrations == [] | ||||
r731 | def test_enabled_non_repo_integrations(integration_repos): | |||
integrations = integration_repos['integrations'] | ||||
triggered_integrations = IntegrationModel().get_for_event( | ||||
events.UserPreCreate({})) | ||||
assert triggered_integrations == [integrations['global']] | ||||