##// END OF EJS Templates
vcs: Use key lookup to explode if key is not available....
vcs: Use key lookup to explode if key is not available. The key has to be in the dictionary, otherwise something changed in the code and expoding here is better than hiding the error.

File last commit:

r793:fc8d2069 default
r890:ed7b68ec default
Show More
test_integration.py
222 lines | 7.3 KiB | text/x-python | PythonLexer
dan
integrations: fix bug where deleting a repo did not delete integrations
r427 # -*- coding: utf-8 -*-
# Copyright (C) 2010-2016 RhodeCode GmbH
#
# 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/
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 import time
dan
integrations: fix bug where deleting a repo did not delete integrations
r427 import pytest
from rhodecode import events
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 from rhodecode.tests.fixture import Fixture
dan
integrations: fix bug where deleting a repo did not delete integrations
r427 from rhodecode.model.db import Session, Integration
from rhodecode.model.integration import IntegrationModel
from rhodecode.integrations.types.base import IntegrationTypeBase
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 class TestDeleteScopesDeletesIntegrations(object):
def test_delete_repo_with_integration_deletes_integration(self,
repo_integration_stub):
Session().delete(repo_integration_stub.repo)
Session().commit()
Session().expire_all()
integration = Integration.get(repo_integration_stub.integration_id)
assert integration is None
dan
integrations: fix bug where deleting a repo did not delete integrations
r427
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 def test_delete_repo_group_with_integration_deletes_integration(self,
repogroup_integration_stub):
Session().delete(repogroup_integration_stub.repo_group)
Session().commit()
Session().expire_all()
integration = Integration.get(repogroup_integration_stub.integration_id)
assert integration is None
dan
integrations: fix bug where deleting a repo did not delete integrations
r427
@pytest.fixture
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 def integration_repos(request, StubIntegrationType, stub_integration_settings):
"""
Create repositories and integrations for testing, and destroy them after
dan
integrations: add recursive repo group scope to allow integrations...
r793
Structure:
root_repo
parent_group/
parent_repo
child_group/
child_repo
other_group/
other_repo
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 """
fixture = Fixture()
dan
integrations: add recursive repo group scope to allow integrations...
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)
dan
integrations: refactor/cleanup + features, fixes #4181...
r731
dan
integrations: add recursive repo group scope to allow integrations...
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)
dan
integrations: refactor/cleanup + features, fixes #4181...
r731
root_repo_id = 'int_test_repo_root_%s' % time.time()
root_repo = fixture.create_repo(root_repo_id)
dan
integrations: fix bug where deleting a repo did not delete integrations
r427
dan
integrations: add recursive repo group scope to allow integrations...
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)
dan
integrations: refactor/cleanup + features, fixes #4181...
r731
Session().commit()
dan
integrations: fix bug where deleting a repo did not delete integrations
r427
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 def _cleanup():
dan
integrations: add recursive repo group scope to allow integrations...
r793 for integration in integrations.values():
Session.delete(integration)
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 fixture.destroy_repo(root_repo)
dan
integrations: add recursive repo group scope to allow integrations...
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)
dan
integrations: refactor/cleanup + features, fixes #4181...
r731
request.addfinalizer(_cleanup)
return {
dan
integrations: add recursive repo group scope to allow integrations...
r793 'integrations': integrations,
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 'repos': {
'root_repo': root_repo,
dan
integrations: add recursive repo group scope to allow integrations...
r793 'other_repo': other_repo,
'parent_repo': parent_repo,
'child_repo': child_repo,
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 }
}
dan
integrations: fix bug where deleting a repo did not delete integrations
r427
dan
integrations: refactor/cleanup + features, fixes #4181...
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'],
dan
integrations: add recursive repo group scope to allow integrations...
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'],
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 ]
triggered_integrations = IntegrationModel().get_for_event(
dan
integrations: add recursive repo group scope to allow integrations...
r793 events.RepoEvent(repos['parent_repo']))
dan
integrations: fix bug where deleting a repo did not delete integrations
r427
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 assert triggered_integrations == [
integrations['global'],
dan
integrations: add recursive repo group scope to allow integrations...
r793 integrations['parent_repo'],
integrations['parent_group'],
integrations['parent_group_recursive'],
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 ]
triggered_integrations = IntegrationModel().get_for_event(
dan
integrations: add recursive repo group scope to allow integrations...
r793 events.RepoEvent(repos['child_repo']))
dan
integrations: refactor/cleanup + features, fixes #4181...
r731
assert triggered_integrations == [
integrations['global'],
dan
integrations: add recursive repo group scope to allow integrations...
r793 integrations['child_repo'],
integrations['parent_group_recursive'],
integrations['child_group'],
integrations['child_group_recursive'],
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 ]
dan
integrations: fix bug where deleting a repo did not delete integrations
r427
dan
integrations: refactor/cleanup + features, fixes #4181...
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
dan
integrations: fix bug where deleting a repo did not delete integrations
r427 Session().commit()
dan
integrations: refactor/cleanup + features, fixes #4181...
r731
triggered_integrations = IntegrationModel().get_for_event(
events.RepoEvent(repos['root_repo']))
assert triggered_integrations == []
triggered_integrations = IntegrationModel().get_for_event(
dan
integrations: add recursive repo group scope to allow integrations...
r793 events.RepoEvent(repos['parent_repo']))
dan
integrations: refactor/cleanup + features, fixes #4181...
r731
assert triggered_integrations == []
dan
integrations: fix bug where deleting a repo did not delete integrations
r427
dan
integrations: refactor/cleanup + features, fixes #4181...
r731 triggered_integrations = IntegrationModel().get_for_event(
dan
integrations: add recursive repo group scope to allow integrations...
r793 events.RepoEvent(repos['child_repo']))
dan
integrations: refactor/cleanup + features, fixes #4181...
r731
assert triggered_integrations == []
dan
integrations: add recursive repo group scope to allow integrations...
r793 triggered_integrations = IntegrationModel().get_for_event(
events.RepoEvent(repos['other_repo']))
assert triggered_integrations == []
dan
integrations: refactor/cleanup + features, fixes #4181...
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']]