Commit 2e681c41 authored by Tom Pillot's avatar Tom Pillot
Browse files

Test new branches, pipeline and pull requests

parent 733df700
Pipeline #10457 passed with stage
in 2 minutes and 9 seconds
......@@ -63,7 +63,7 @@ class GhTests(TestCase):
self.assertSync(branch)
def gh_webhook_event(self, event, last_commit=''):
def gh_webhook_event(self, event, last_commit='', branch='master', pr_action='', pr_number='', pr_login=''):
"""Simulate receiving an event from a github webhook."""
data = {
'repository': {
......@@ -72,8 +72,21 @@ class GhTests(TestCase):
'name': self.namespace.slug_github
}
},
'ref': 'refs/heads/master',
'after': last_commit
'ref': f'refs/heads/{branch}',
'after': last_commit,
'action': pr_action,
'number': pr_number,
'pull_request': {
'head': {
'repo': {
'owner': {
'login': pr_login
},
'clone_url': self.project.remote_url_github()
},
'sha': last_commit
}
}
}
encoded_data = self.client._encode_json({} if data is None else data, content_type='application/json')
......@@ -87,17 +100,24 @@ class GhTests(TestCase):
HTTP_X_HUB_SIGNATURE=signature,
HTTP_X_GITHUB_EVENT=event)
def gl_webhook_event(self, event, last_commit=''):
def gl_webhook_event(self, event, last_commit='', branch='master', status=''):
"""Simulate receiving an event from a gitlab webhook."""
data = {
'repository': {
'name': self.project.slug,
},
'project': {
'name': self.project.slug,
'namespace': self.namespace.slug_gitlab
},
'ref': 'refs/heads/master',
'after': last_commit
'ref': f'refs/heads/{branch}',
'after': last_commit,
'object_attributes': {
'ref': branch,
'sha': last_commit,
'status': status,
'id': 1
}
}
return self.client.post(reverse('gl-webhook'),
data,
......@@ -164,50 +184,178 @@ class GhTests(TestCase):
self.assertEqual(response.status_code, 200)
self.assertEqual(response.content.decode(), 'already synced')
def test_push_github(self):
"""Test sync after pushing to github."""
def push_github(self, branch):
"""Test sync after pushing to the given branch on github."""
self.sync()
last_commit = self.github.get_branch('master').commit.sha
file = self.github.get_contents('README.md')
last_commit = self.github.get_branch(branch).commit.sha
file = self.github.get_contents('README.md', branch)
self.github.update_file('README.md',
message='Test push on github',
message=f'Test push on github {branch}',
content=last_commit[:8],
sha=file.sha,
branch='master')
branch=branch)
last_commit_github = self.github.get_branch('master').commit.sha
last_commit_github = self.github.get_branch(branch).commit.sha
self.assertNotEqual(last_commit, last_commit_github)
response = self.gh_webhook_event('push', last_commit_github)
response = self.gh_webhook_event('push', last_commit_github, branch)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.content.decode(), 'push event detected')
last_commit_gitlab = self.gitlab.commits.list(ref_name='master')[0].id
last_commit_gitlab = self.gitlab.commits.list(ref_name=branch)[0].id
self.assertEqual(last_commit_github, last_commit_gitlab)
def test_push_gitlab(self):
"""Test sync after pushing to gitlab."""
def test_push_github_master(self):
"""Test sync after pushing to github on master."""
self.push_github('master')
def test_push_github_devel(self):
"""Test sync after pushing to github on devel."""
self.push_github('devel')
def push_gitlab(self, branch):
"""Test sync after pushing to the given branch on gitlab."""
self.sync()
git_repo = self.project.git()
last_commit = self.gitlab.commits.list(ref_name='master')[0].id
last_commit = self.gitlab.commits.list(ref_name=branch)[0].id
# Push a new commit to gitlab
gl_remote_name = f'gitlab/{self.namespace.slug}'
if gl_remote_name not in git_repo.remotes:
git_repo.create_remote(gl_remote_name, url=self.project.remote_url_gitlab())
with open(str(self.project.git_path()) + '/README.md', 'w') as f:
f.write(last_commit[:8])
git_repo.index.add(['README.md'])
git_repo.index.commit('Test push on gitlab')
git_repo.git.push(gl_remote_name, 'master')
last_commit_gitlab = self.gitlab.commits.list(ref_name='master')[0].id
file = self.gitlab.files.get(file_path='README.md', ref=branch)
file.content = last_commit[:8]
file.save(branch=branch, commit_message=f'Test push on gitlab {branch}')
last_commit_gitlab = self.gitlab.commits.list(ref_name=branch)[0].id
self.assertNotEqual(last_commit, last_commit_gitlab)
response = self.gl_webhook_event('Push Hook', last_commit_gitlab)
response = self.gl_webhook_event('Push Hook', last_commit_gitlab, branch)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.content.decode(), 'push event detected')
last_commit_github = self.github.get_branch(branch).commit.sha
self.assertEqual(last_commit_github, last_commit_gitlab)
def test_push_gitlab_master(self):
"""Test sync after pushing to gitlab on master."""
self.push_gitlab('master')
def test_push_gitlab_devel(self):
"""Test sync after pushing to gitlab on devel."""
self.push_gitlab('devel')
def test_branch_github(self):
"""Test sync after creating or deleting a branch on github."""
self.sync()
source_branch_name = 'master'
target_branch_name = 'test-branch-github'
last_commit = self.github.get_branch(source_branch_name).commit.sha
# Test sync after creating a new branch
source_branch = self.github.get_branch(source_branch_name)
if target_branch_name in [b.name for b in self.github.get_branches()]:
branch_ref = self.github.get_git_ref(ref=f'heads/{target_branch_name}')
else:
branch_ref = self.github.create_git_ref(ref=f'refs/heads/{target_branch_name}',
sha=source_branch.commit.sha)
file = self.github.get_contents('README.md', target_branch_name)
self.github.update_file('README.md',
message='Test new branch on github',
content=last_commit[:8],
sha=file.sha,
branch=target_branch_name)
target_branch = self.github.get_branch(target_branch_name)
last_commit_github = target_branch.commit.sha
response = self.gh_webhook_event('push', last_commit_github, target_branch_name)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.content.decode(), 'push event detected')
last_commit_gitlab = self.gitlab.commits.list(ref_name=target_branch_name)[0].id
self.assertEqual(last_commit_github, last_commit_gitlab)
# Test sync after deleting a branch
branch_ref.delete()
self.assertNotIn(target_branch_name, [b.name for b in self.github.get_branches()])
response = self.gh_webhook_event('push', '0000000000000000000000000000000000000000', target_branch_name)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.content.decode(), 'push event detected')
self.assertNotIn(target_branch_name, [b.name for b in self.gitlab.branches.list()])
def test_branch_gitlab(self):
"""Test sync after creating or deleting a branch on gitlab."""
self.sync()
source_branch_name = 'master'
target_branch_name = 'test-branch-gitlab'
last_commit = self.gitlab.commits.list(ref_name=source_branch_name)[0].id
# Test sync after creating a new branch
if target_branch_name in [b.name for b in self.gitlab.branches.list()]:
target_branch = self.gitlab.branches.get(target_branch_name)
else:
target_branch = self.gitlab.branches.create({'branch': target_branch_name, 'ref': source_branch_name})
file = self.gitlab.files.get(file_path='README.md', ref=target_branch_name)
file.content = last_commit[:8]
file.save(branch=target_branch_name, commit_message='Test new branch on gitlab')
last_commit_gitlab = self.gitlab.commits.list(ref_name=target_branch_name)[0].id
response = self.gl_webhook_event('Push Hook', last_commit_gitlab, target_branch_name)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.content.decode(), 'push event detected')
last_commit_github = self.github.get_branch('master').commit.sha
last_commit_github = self.github.get_branch(target_branch_name).commit.sha
self.assertEqual(last_commit_github, last_commit_gitlab)
# Test sync after deleting a branch
target_branch.delete()
self.assertNotIn(target_branch_name, [b.name for b in self.gitlab.branches.list()])
response = self.gl_webhook_event('Push Hook', '0000000000000000000000000000000000000000', target_branch_name)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.content.decode(), 'push event detected')
self.assertNotIn(target_branch_name, [b.name for b in self.github.get_branches()])
def test_pipeline(self):
"""Test reporting the gitlab pipeline status to github."""
self.sync()
last_commit = self.gitlab.commits.list(ref_name='master')[0].id
response = self.gl_webhook_event('Pipeline Hook', last_commit, status='success')
self.assertEqual(response.status_code, 200)
self.assertIn('success', [status.state for status in self.github.get_branch('master').commit.get_statuses()])
def test_pr(self):
"""Test github's pull requests."""
self.sync()
not_accepted_string = "doesn't accept pull requests on master"
# Test pr on master
last_commit = self.github.get_branch("devel").commit.sha
pr_master = self.project.github().create_pull(title="Test pr on master", body='', head="devel", base="master")
response = self.gh_webhook_event('pull_request',
last_commit=last_commit,
pr_action="opened",
pr_login="foo",
pr_number=pr_master.number)
self.assertEqual(response.status_code, 200)
self.assertTrue([c.body for c in pr_master.get_issue_comments() if not_accepted_string in c.body])
self.assertIn(f'pr/{pr_master.number}', [b.name for b in self.gitlab.branches.list()])
# Test pr on devel
last_commit = self.github.get_branch("master").commit.sha
pr_devel = self.project.github().create_pull(title="Test pr on devel", body='', head="master", base="devel")
response = self.gh_webhook_event('pull_request',
last_commit=last_commit,
pr_action="opened",
pr_login="foo",
pr_number=pr_devel.number)
self.assertEqual(response.status_code, 200)
self.assertFalse([c.body for c in pr_devel.get_issue_comments() if not_accepted_string in c.body])
self.assertIn(f'pr/{pr_devel.number}', [b.name for b in self.gitlab.branches.list()])
# Close the pr
for pr in [pr_master, pr_devel]:
pr.edit(state="closed")
response = self.gh_webhook_event('pull_request', pr_action="closed", pr_login="foo", pr_number=pr.number)
self.assertEqual(response.status_code, 200)
self.assertNotIn(f'pr/{pr.number}', [b.name for b in self.gitlab.branches.list()])
......@@ -39,6 +39,7 @@ def pull_request(request: HttpRequest, rep: str) -> HttpResponse:
data = loads(request.body.decode())
event = data['action']
branch = f'pr/{data["number"]}'
login = slugify(data["pull_request"]["head"]["repo"]["owner"]["login"])
namespace = get_object_or_404(Namespace, slug_github=slugify(data['repository']['owner']['name']))
project = get_object_or_404(Project, main_namespace=namespace, slug=slugify(data['repository']['name']))
......@@ -47,14 +48,16 @@ def pull_request(request: HttpRequest, rep: str) -> HttpResponse:
# Prevent pull requests on master when necessary
if event in ['opened', 'reopened']:
pr = project.github().get_pull(data["number"])
gh = project.github()
pr = gh.get_pull(data["number"])
pr_branch = pr.base.ref
if not project.accept_pr_to_master and pr_branch == 'master':
if not project.accept_pr_to_master and pr_branch == 'master' \
and 'devel' in [b.name for b in gh.get_branches()] and login != namespace.slug_github:
logger.info(f"{namespace.slug}/{project.slug}: New pr {data['number']} to master")
pr.create_issue_comment("This project doesn't accept pull requests on master, please change the base "
"branch of your pull request to devel.")
gh_remote_name = f'github/{data["pull_request"]["head"]["repo"]["owner"]["login"]}'
gh_remote_name = f'github/{login}'
if gh_remote_name not in git_repo.remotes:
remote = git_repo.create_remote(gh_remote_name, data["pull_request"]["head"]["repo"]["clone_url"])
else:
......@@ -244,7 +247,7 @@ def webhook(request: HttpRequest) -> HttpResponse:
if event == 'check_suite':
return check_suite(request, 'check_suite event detected')
if event == 'pull_request':
return pull_request(request, 'check_suite event detected')
return pull_request(request, 'pull_request event detected')
return HttpResponseForbidden('event not found')
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment