| |
| |
| """ |
| (c) 2018 - Copyright Red Hat Inc |
| |
| Authors: |
| Pierre-Yves Chibon <pingou@pingoured.fr> |
| |
| """ |
| |
| from __future__ import unicode_literals, absolute_import |
| |
| import json |
| import os |
| import re |
| import shutil |
| import sys |
| import tempfile |
| import time |
| import unittest |
| |
| import pygit2 |
| import wtforms |
| from mock import patch, MagicMock |
| from bs4 import BeautifulSoup |
| |
| sys.path.insert( |
| 0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") |
| ) |
| |
| import pagure.lib.query |
| import tests |
| from pagure.lib.repo import PagureRepo |
| from pagure.lib.git import _make_signature |
| |
| |
| class PagureRemotePRtests(tests.Modeltests): |
| """ Tests for remote PRs in pagure """ |
| |
| def setUp(self): |
| """ Set up the environment. """ |
| super(PagureRemotePRtests, self).setUp() |
| |
| self.newpath = tempfile.mkdtemp(prefix="pagure-fork-test") |
| self.old_value = pagure.config.config["REMOTE_GIT_FOLDER"] |
| pagure.config.config["REMOTE_GIT_FOLDER"] = os.path.join( |
| self.path, "remotes" |
| ) |
| |
| def tearDown(self): |
| """ Clear things up. """ |
| super(PagureRemotePRtests, self).tearDown() |
| |
| pagure.config.config["REMOTE_GIT_FOLDER"] = self.old_value |
| shutil.rmtree(self.newpath) |
| |
| def set_up_git_repo(self, new_project=None, branch_from="feature"): |
| """ Set up the git repo and create the corresponding PullRequest |
| object. |
| """ |
| |
| |
| gitrepo = os.path.join(self.path, "repos", "test.git") |
| repo = pygit2.init_repository(gitrepo, bare=True) |
| |
| repopath = os.path.join(self.newpath, "test") |
| clone_repo = pygit2.clone_repository(gitrepo, repopath) |
| |
| |
| with open(os.path.join(repopath, "sources"), "w") as stream: |
| stream.write("foo\n bar") |
| clone_repo.index.add("sources") |
| clone_repo.index.write() |
| |
| try: |
| com = repo.revparse_single("HEAD") |
| prev_commit = [com.oid.hex] |
| except: |
| prev_commit = [] |
| |
| |
| tree = clone_repo.index.write_tree() |
| author = _make_signature("Alice Author", "alice@authors.tld") |
| committer = _make_signature("Cecil Committer", "cecil@committers.tld") |
| clone_repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add sources file for testing", |
| |
| tree, |
| |
| prev_commit, |
| ) |
| |
| refname = "refs/heads/master:refs/heads/master" |
| ori_remote = clone_repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| first_commit = repo.revparse_single("HEAD") |
| |
| with open(os.path.join(repopath, ".gitignore"), "w") as stream: |
| stream.write("*~") |
| clone_repo.index.add(".gitignore") |
| clone_repo.index.write() |
| |
| |
| tree = clone_repo.index.write_tree() |
| author = _make_signature("Alice Äuthòr", "alice@äuthòrs.tld") |
| committer = _make_signature("Cecil Cõmmîttër", "cecil@cõmmîttërs.tld") |
| clone_repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add .gitignore file for testing", |
| |
| tree, |
| |
| [first_commit.oid.hex], |
| ) |
| refname = "refs/heads/master:refs/heads/master" |
| ori_remote = clone_repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| |
| new_gitrepo = repopath |
| if new_project: |
| |
| new_gitrepo = os.path.join(self.newpath, new_project.fullname) |
| if not os.path.exists(new_gitrepo): |
| os.makedirs(new_gitrepo) |
| new_repo = pygit2.clone_repository(gitrepo, new_gitrepo) |
| |
| repo = pygit2.Repository(new_gitrepo) |
| |
| |
| with open(os.path.join(new_gitrepo, "sources"), "w") as stream: |
| stream.write("foo\n bar\nbaz\n boose") |
| repo.index.add("sources") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = _make_signature("Alice Author", "alice@authors.tld") |
| committer = _make_signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/%s" % branch_from, |
| author, |
| committer, |
| "A commit on branch %s" % branch_from, |
| tree, |
| [first_commit.oid.hex], |
| ) |
| refname = "refs/heads/%s" % (branch_from) |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_new_remote_pr_unauth(self): |
| """ Test creating a new remote PR un-authenticated. """ |
| |
| tests.create_projects(self.session) |
| tests.create_projects_git( |
| os.path.join(self.path, "requests"), bare=True |
| ) |
| self.set_up_git_repo() |
| |
| |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(len(project.requests), 0) |
| |
| |
| output = self.app.get("/test/diff/remote") |
| self.assertEqual(output.status_code, 302) |
| self.assertIn( |
| "You should be redirected automatically to target URL: " |
| '<a href="/login/?', |
| output.get_data(as_text=True), |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_new_remote_pr_auth(self): |
| """ Test creating a new remote PR authenticated. """ |
| |
| tests.create_projects(self.session) |
| tests.create_projects_git( |
| os.path.join(self.path, "requests"), bare=True |
| ) |
| self.set_up_git_repo() |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(len(project.requests), 0) |
| |
| |
| user = tests.FakeUser(username="foo") |
| with tests.user_set(self.app.application, user): |
| output = self.app.get("/test/diff/remote") |
| self.assertEqual(output.status_code, 200) |
| self.assertIn( |
| "<h2>New remote pull-request</h2>", |
| output.get_data(as_text=True), |
| ) |
| |
| csrf_token = self.get_csrf(output=output) |
| with patch( |
| "pagure.forms.RemoteRequestPullForm.git_repo.args", |
| MagicMock( |
| return_value=( |
| "Git Repo address", |
| [wtforms.validators.DataRequired()], |
| ) |
| ), |
| ): |
| data = { |
| "csrf_token": csrf_token, |
| "title": "Remote PR title", |
| "branch_from": "feature", |
| "branch_to": "master", |
| "git_repo": os.path.join(self.newpath, "test"), |
| } |
| output = self.app.post("/test/diff/remote", data=data) |
| self.assertEqual(output.status_code, 200) |
| output_text = output.get_data(as_text=True) |
| self.assertIn("Create Pull Request\n </div>\n", output_text) |
| self.assertIn('<div class="card mb-3" id="_1">\n', output_text) |
| self.assertIn('<div class="card mb-3" id="_2">\n', output_text) |
| self.assertNotIn( |
| '<div class="card mb-3" id="_3">\n', output_text |
| ) |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| self.assertEqual(len(project.requests), 0) |
| |
| data = { |
| "csrf_token": csrf_token, |
| "title": "Remote PR title", |
| "branch_from": "feature", |
| "branch_to": "master", |
| "git_repo": os.path.join(self.newpath, "test"), |
| "confirm": 1, |
| } |
| self.old_value = pagure.config.config["DISABLE_REMOTE_PR"] |
| pagure.config.config["DISABLE_REMOTE_PR"] = True |
| output = self.app.post( |
| "/test/diff/remote", data=data, follow_redirects=True |
| ) |
| self.assertEqual(output.status_code, 404) |
| pagure.config.config["DISABLE_REMOTE_PR"] = self.old_value |
| output = self.app.post( |
| "/test/diff/remote", data=data, follow_redirects=True |
| ) |
| self.assertEqual(output.status_code, 200) |
| output_text = output.get_data(as_text=True) |
| self.assertIn( |
| '<span class="text-success font-weight-bold">#1', |
| output_text, |
| ) |
| self.assertIn('<div class="card mb-3" id="_1">\n', output_text) |
| self.assertIn('<div class="card mb-3" id="_2">\n', output_text) |
| self.assertNotIn( |
| '<div class="card mb-3" id="_3">\n', output_text |
| ) |
| |
| |
| self.assertIn( |
| '<a href="#_1" class="list-group-item', output_text |
| ) |
| self.assertIn( |
| '<div class="ellipsis pr-changes-description">' |
| "\n <small>.gitignore</small>", |
| output_text, |
| ) |
| self.assertIn( |
| '<a href="#_2" class="list-group-item', output_text |
| ) |
| self.assertIn( |
| '<div class="ellipsis pr-changes-description">' |
| "\n <small>sources</small>", |
| output_text, |
| ) |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(len(project.requests), 1) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_new_remote_no_title(self): |
| """ Test creating a new remote PR authenticated when no title is |
| specified. """ |
| |
| tests.create_projects(self.session) |
| tests.create_projects_git( |
| os.path.join(self.path, "requests"), bare=True |
| ) |
| self.set_up_git_repo() |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(len(project.requests), 0) |
| |
| |
| user = tests.FakeUser(username="foo") |
| with tests.user_set(self.app.application, user): |
| output = self.app.get("/test/diff/remote") |
| self.assertEqual(output.status_code, 200) |
| self.assertIn( |
| "<h2>New remote pull-request</h2>", |
| output.get_data(as_text=True), |
| ) |
| |
| csrf_token = self.get_csrf(output=output) |
| with patch( |
| "pagure.forms.RemoteRequestPullForm.git_repo.args", |
| MagicMock( |
| return_value=( |
| "Git Repo address", |
| [wtforms.validators.DataRequired()], |
| ) |
| ), |
| ): |
| data = { |
| "csrf_token": csrf_token, |
| "branch_from": "master", |
| "branch_to": "feature", |
| "git_repo": os.path.join(self.newpath, "test"), |
| } |
| output = self.app.post("/test/diff/remote", data=data) |
| self.assertEqual(output.status_code, 200) |
| output_text = output.get_data(as_text=True) |
| self.assertIn("<h2>New remote pull-request</h2>", output_text) |
| self.assertIn("<option selected>feature</option>", output_text) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_new_remote_pr_empty_target(self): |
| """ Test creating a new remote PR authenticated against an empty |
| git repo. """ |
| |
| tests.create_projects(self.session) |
| tests.create_projects_git( |
| os.path.join(self.path, "requests"), bare=True |
| ) |
| |
| |
| gitrepo = os.path.join(self.path, "repos", "test.git") |
| pygit2.init_repository(gitrepo, bare=True) |
| |
| |
| gitrepo = os.path.join(self.path, "repos", "test_origin.git") |
| repo = pygit2.init_repository(gitrepo) |
| |
| |
| with open(os.path.join(gitrepo, "sources"), "w") as stream: |
| stream.write("foo\n bar") |
| repo.index.add("sources") |
| repo.index.write() |
| |
| prev_commit = [] |
| |
| |
| tree = repo.index.write_tree() |
| author = _make_signature("Alice Author", "alice@authors.tld") |
| committer = _make_signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/feature", |
| author, |
| committer, |
| "Add sources file for testing", |
| |
| tree, |
| |
| prev_commit, |
| ) |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(len(project.requests), 0) |
| |
| |
| user = tests.FakeUser(username="foo") |
| with tests.user_set(self.app.application, user): |
| output = self.app.get("/test/diff/remote") |
| self.assertEqual(output.status_code, 200) |
| self.assertIn( |
| "<h2>New remote pull-request</h2>", |
| output.get_data(as_text=True), |
| ) |
| |
| csrf_token = self.get_csrf(output=output) |
| |
| with patch( |
| "pagure.forms.RemoteRequestPullForm.git_repo.args", |
| MagicMock( |
| return_value=( |
| "Git Repo address", |
| [wtforms.validators.DataRequired()], |
| ) |
| ), |
| ): |
| data = { |
| "csrf_token": csrf_token, |
| "title": "Remote PR title", |
| "branch_from": "feature", |
| "branch_to": "master", |
| "git_repo": gitrepo, |
| } |
| output = self.app.post("/test/diff/remote", data=data) |
| self.assertEqual(output.status_code, 200) |
| output_text = output.get_data(as_text=True) |
| self.assertIn("Create Pull Request\n </div>\n", output_text) |
| self.assertIn('<div class="card mb-3" id="_1">\n', output_text) |
| self.assertNotIn( |
| '<div class="card mb-3" id="_2">\n', output_text |
| ) |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| self.assertEqual(len(project.requests), 0) |
| |
| data = { |
| "csrf_token": csrf_token, |
| "title": "Remote PR title", |
| "branch_from": "feature", |
| "branch_to": "master", |
| "git_repo": gitrepo, |
| "confirm": 1, |
| } |
| output = self.app.post( |
| "/test/diff/remote", data=data, follow_redirects=True |
| ) |
| self.assertEqual(output.status_code, 200) |
| output_text = output.get_data(as_text=True) |
| self.assertIn( |
| "<title>PR#1: Remote PR title - test\n - Pagure</title>", |
| output_text, |
| ) |
| self.assertIn('<div class="card mb-3" id="_1">\n', output_text) |
| self.assertNotIn( |
| '<div class="card mb-3" id="_2">\n', output_text |
| ) |
| |
| |
| self.assertIn( |
| '<a href="#_1" class="list-group-item', output_text |
| ) |
| self.assertIn( |
| '<div class="ellipsis pr-changes-description">' |
| "\n <small>sources</small>", |
| output_text, |
| ) |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(len(project.requests), 1) |
| |
| |
| data = {"csrf_token": csrf_token, "requestid": project.requests[0].uid} |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 200) |
| output_text = output.get_data(as_text=True) |
| data = json.loads(output_text) |
| self.assertEqual( |
| data, |
| { |
| "code": "FFORWARD", |
| "message": "The pull-request can be merged and fast-forwarded", |
| "short_code": "Ok", |
| }, |
| ) |
| |
| user = tests.FakeUser(username="pingou") |
| with tests.user_set(self.app.application, user): |
| |
| data = {"csrf_token": csrf_token} |
| output = self.app.post( |
| "/test/pull-request/1/merge", data=data, follow_redirects=True |
| ) |
| output_text = output.get_data(as_text=True) |
| self.assertEqual(output.status_code, 200) |
| self.assertIn( |
| "<title>Overview - test - Pagure</title>", output_text |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| @patch("pagure.lib.tasks_services.trigger_ci_build") |
| def test_new_remote_pr_ci_off(self, trigger_ci): |
| """ Test creating a new remote PR when CI is not configured. """ |
| |
| tests.create_projects(self.session) |
| tests.create_projects_git( |
| os.path.join(self.path, "requests"), bare=True |
| ) |
| self.set_up_git_repo() |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(len(project.requests), 0) |
| |
| |
| user = tests.FakeUser(username="foo") |
| with tests.user_set(self.app.application, user): |
| |
| csrf_token = self.get_csrf() |
| data = { |
| "csrf_token": csrf_token, |
| "title": "Remote PR title", |
| "branch_from": "feature", |
| "branch_to": "master", |
| "git_repo": os.path.join(self.newpath, "test"), |
| } |
| with patch( |
| "pagure.forms.RemoteRequestPullForm.git_repo.args", |
| MagicMock( |
| return_value=( |
| "Git Repo address", |
| [wtforms.validators.DataRequired()], |
| ) |
| ), |
| ): |
| |
| output = self.app.post( |
| "/test/diff/remote", data=data, follow_redirects=True |
| ) |
| self.assertEqual(output.status_code, 200) |
| |
| data["confirm"] = 1 |
| output = self.app.post( |
| "/test/diff/remote", data=data, follow_redirects=True |
| ) |
| self.assertEqual(output.status_code, 200) |
| output_text = output.get_data(as_text=True) |
| self.assertIn( |
| '<span class="text-success font-weight-bold">#1', |
| output_text, |
| ) |
| self.assertIn('<div class="card mb-3" id="_1">\n', output_text) |
| self.assertIn('<div class="card mb-3" id="_2">\n', output_text) |
| self.assertNotIn( |
| '<div class="card mb-3" id="_3">\n', output_text |
| ) |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(len(project.requests), 1) |
| trigger_ci.assert_not_called() |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| @patch("pagure.lib.tasks_services.trigger_ci_build") |
| def test_new_remote_pr_ci_on(self, trigger_ci): |
| """ Test creating a new remote PR when CI is configured. """ |
| |
| tests.create_projects(self.session) |
| tests.create_projects_git( |
| os.path.join(self.path, "requests"), bare=True |
| ) |
| self.set_up_git_repo() |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(len(project.requests), 0) |
| |
| |
| user = tests.FakeUser(username="pingou") |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = { |
| "active_pr": "y", |
| "ci_url": "https://jenkins.fedoraproject.org", |
| "ci_job": "test/job", |
| "ci_type": "jenkins", |
| "csrf_token": csrf_token, |
| } |
| output = self.app.post( |
| "/test/settings/Pagure CI", data=data, follow_redirects=True |
| ) |
| self.assertEqual(output.status_code, 200) |
| |
| user = tests.FakeUser(username="foo") |
| with tests.user_set(self.app.application, user): |
| data = { |
| "csrf_token": csrf_token, |
| "title": "Remote PR title", |
| "branch_from": "feature", |
| "branch_to": "master", |
| "git_repo": os.path.join(self.newpath, "test"), |
| } |
| |
| with patch( |
| "pagure.forms.RemoteRequestPullForm.git_repo.args", |
| MagicMock( |
| return_value=( |
| "Git Repo address", |
| [wtforms.validators.DataRequired()], |
| ) |
| ), |
| ): |
| |
| |
| output = self.app.post( |
| "/test/diff/remote", data=data, follow_redirects=True |
| ) |
| self.assertEqual(output.status_code, 200) |
| |
| |
| data["confirm"] = 1 |
| output = self.app.post( |
| "/test/diff/remote", data=data, follow_redirects=True |
| ) |
| self.assertEqual(output.status_code, 200) |
| output_text = output.get_data(as_text=True) |
| self.assertIn( |
| '<span class="text-success font-weight-bold">#1', |
| output_text, |
| ) |
| self.assertIn('<div class="card mb-3" id="_1">\n', output_text) |
| self.assertIn('<div class="card mb-3" id="_2">\n', output_text) |
| self.assertNotIn( |
| '<div class="card mb-3" id="_3">\n', output_text |
| ) |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(len(project.requests), 1) |
| trigger_ci.assert_not_called() |
| |
| |
| if __name__ == "__main__": |
| unittest.main(verbosity=2) |