179 lines
7.3 KiB
Python
179 lines
7.3 KiB
Python
#!/usr/bin/env python
|
|
|
|
# Copyright 2012 Hewlett-Packard Development Company, L.P.
|
|
# Copyright 2014 Wikimedia Foundation Inc.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
import logging
|
|
import os
|
|
|
|
import git
|
|
import testtools
|
|
|
|
from zuul.merger.merger import Repo
|
|
from tests.base import ZuulTestCase, FIXTURE_DIR, simple_layout
|
|
|
|
|
|
class TestMergerRepo(ZuulTestCase):
|
|
|
|
log = logging.getLogger("zuul.test.merger.repo")
|
|
tenant_config_file = 'config/single-tenant/main.yaml'
|
|
workspace_root = None
|
|
|
|
def setUp(self):
|
|
super(TestMergerRepo, self).setUp()
|
|
self.workspace_root = os.path.join(self.test_root, 'workspace')
|
|
|
|
def test_ensure_cloned(self):
|
|
parent_path = os.path.join(self.upstream_root, 'org/project1')
|
|
|
|
# Forge a repo having a submodule
|
|
parent_repo = git.Repo(parent_path)
|
|
parent_repo.git.submodule('add', os.path.join(
|
|
self.upstream_root, 'org/project2'), 'subdir')
|
|
parent_repo.index.commit('Adding project2 as a submodule in subdir')
|
|
# git 1.7.8 changed .git from being a directory to a file pointing
|
|
# to the parent repository /.git/modules/*
|
|
self.assertTrue(os.path.exists(
|
|
os.path.join(parent_path, 'subdir', '.git')),
|
|
msg='.git file in submodule should be a file')
|
|
|
|
work_repo = Repo(parent_path, self.workspace_root,
|
|
'none@example.org', 'User Name', '0', '0')
|
|
self.assertTrue(
|
|
os.path.isdir(os.path.join(self.workspace_root, 'subdir')),
|
|
msg='Cloned repository has a submodule placeholder directory')
|
|
self.assertFalse(os.path.exists(
|
|
os.path.join(self.workspace_root, 'subdir', '.git')),
|
|
msg='Submodule is not initialized')
|
|
|
|
sub_repo = Repo(
|
|
os.path.join(self.upstream_root, 'org/project2'),
|
|
os.path.join(self.workspace_root, 'subdir'),
|
|
'none@example.org', 'User Name', '0', '0')
|
|
self.assertTrue(os.path.exists(
|
|
os.path.join(self.workspace_root, 'subdir', '.git')),
|
|
msg='Cloned over the submodule placeholder')
|
|
|
|
self.assertEqual(
|
|
os.path.join(self.upstream_root, 'org/project1'),
|
|
work_repo.createRepoObject().remotes[0].url,
|
|
message="Parent clone still point to upstream project1")
|
|
|
|
self.assertEqual(
|
|
os.path.join(self.upstream_root, 'org/project2'),
|
|
sub_repo.createRepoObject().remotes[0].url,
|
|
message="Sub repository points to upstream project2")
|
|
|
|
def test_clone_timeout(self):
|
|
parent_path = os.path.join(self.upstream_root, 'org/project1')
|
|
self.patch(git.Git, 'GIT_PYTHON_GIT_EXECUTABLE',
|
|
os.path.join(FIXTURE_DIR, 'fake_git.sh'))
|
|
work_repo = Repo(parent_path, self.workspace_root,
|
|
'none@example.org', 'User Name', '0', '0',
|
|
git_timeout=0.001, retry_attempts=1)
|
|
# TODO: have the merger and repo classes catch fewer
|
|
# exceptions, including this one on initialization. For the
|
|
# test, we try cloning again.
|
|
with testtools.ExpectedException(git.exc.GitCommandError,
|
|
'.*exit code\(-9\)'):
|
|
work_repo._ensure_cloned()
|
|
|
|
def test_fetch_timeout(self):
|
|
parent_path = os.path.join(self.upstream_root, 'org/project1')
|
|
work_repo = Repo(parent_path, self.workspace_root,
|
|
'none@example.org', 'User Name', '0', '0',
|
|
retry_attempts=1)
|
|
work_repo.git_timeout = 0.001
|
|
self.patch(git.Git, 'GIT_PYTHON_GIT_EXECUTABLE',
|
|
os.path.join(FIXTURE_DIR, 'fake_git.sh'))
|
|
with testtools.ExpectedException(git.exc.GitCommandError,
|
|
'.*exit code\(-9\)'):
|
|
work_repo.update()
|
|
|
|
def test_fetch_retry(self):
|
|
parent_path = os.path.join(self.upstream_root, 'org/project1')
|
|
work_repo = Repo(parent_path, self.workspace_root,
|
|
'none@example.org', 'User Name', '0', '0',
|
|
retry_interval=1)
|
|
self.patch(git.Git, 'GIT_PYTHON_GIT_EXECUTABLE',
|
|
os.path.join(FIXTURE_DIR, 'git_fetch_error.sh'))
|
|
work_repo.update()
|
|
# This is created on the first fetch
|
|
self.assertTrue(os.path.exists(os.path.join(
|
|
self.workspace_root, 'stamp1')))
|
|
# This is created on the second fetch
|
|
self.assertTrue(os.path.exists(os.path.join(
|
|
self.workspace_root, 'stamp2')))
|
|
|
|
|
|
class TestMergerWithAuthUrl(ZuulTestCase):
|
|
config_file = 'zuul-github-driver.conf'
|
|
|
|
git_url_with_auth = True
|
|
|
|
@simple_layout('layouts/merging-github.yaml', driver='github')
|
|
def test_changing_url(self):
|
|
"""
|
|
This test checks that if getGitUrl returns different urls for the same
|
|
repo (which happens if an access token is part of the url) then the
|
|
remote urls are changed in the merger accordingly. This tests directly
|
|
the merger.
|
|
"""
|
|
|
|
merger = self.executor_server.merger
|
|
repo = merger.getRepo('github', 'org/project')
|
|
first_url = repo.remote_url
|
|
|
|
repo = merger.getRepo('github', 'org/project')
|
|
second_url = repo.remote_url
|
|
|
|
# the urls should differ
|
|
self.assertNotEqual(first_url, second_url)
|
|
|
|
@simple_layout('layouts/merging-github.yaml', driver='github')
|
|
def test_changing_url_end_to_end(self):
|
|
"""
|
|
This test checks that if getGitUrl returns different urls for the same
|
|
repo (which happens if an access token is part of the url) then the
|
|
remote urls are changed in the merger accordingly. This is an end to
|
|
end test.
|
|
"""
|
|
|
|
A = self.fake_github.openFakePullRequest('org/project', 'master',
|
|
'PR title')
|
|
self.fake_github.emitEvent(A.getCommentAddedEvent('merge me'))
|
|
self.waitUntilSettled()
|
|
self.assertTrue(A.is_merged)
|
|
|
|
# get remote url of org/project in merger
|
|
repo = self.executor_server.merger.repos.get('github.com/org/project')
|
|
self.assertIsNotNone(repo)
|
|
git_repo = git.Repo(repo.local_path)
|
|
first_url = list(git_repo.remotes[0].urls)[0]
|
|
|
|
B = self.fake_github.openFakePullRequest('org/project', 'master',
|
|
'PR title')
|
|
self.fake_github.emitEvent(B.getCommentAddedEvent('merge me again'))
|
|
self.waitUntilSettled()
|
|
self.assertTrue(B.is_merged)
|
|
|
|
repo = self.executor_server.merger.repos.get('github.com/org/project')
|
|
self.assertIsNotNone(repo)
|
|
git_repo = git.Repo(repo.local_path)
|
|
second_url = list(git_repo.remotes[0].urls)[0]
|
|
|
|
# the urls should differ
|
|
self.assertNotEqual(first_url, second_url)
|