| |
| |
| """ |
| (c) 2015-2017 - Copyright Red Hat Inc |
| |
| Authors: |
| Pierre-Yves Chibon <pingou@pingoured.fr> |
| |
| """ |
| |
| from __future__ import unicode_literals, absolute_import |
| |
| import datetime |
| import json |
| import unittest |
| import shutil |
| import sys |
| import time |
| import os |
| |
| import pygit2 |
| from mock import patch, MagicMock, Mock |
| |
| 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 |
| |
| |
| class PagureFlaskInternaltests(tests.Modeltests): |
| """ Tests for flask Internal controller of pagure """ |
| |
| maxDiff = None |
| |
| def setUp(self): |
| """ Set up the environnment, ran before every tests. """ |
| super(PagureFlaskInternaltests, self).setUp() |
| |
| pagure.config.config["IP_ALLOWED_INTERNAL"] = list( |
| set(pagure.config.config["IP_ALLOWED_INTERNAL"] + [None]) |
| ) |
| |
| pagure.config.config["GIT_FOLDER"] = os.path.join(self.path, "repos") |
| |
| @patch.dict("pagure.config.config", {"IP_ALLOWED_INTERNAL": []}) |
| def test_internal_access_only(self): |
| output = self.app.post("/pv/ssh/lookupkey/") |
| self.assertEqual(output.status_code, 403) |
| |
| |
| output = self.app.post("/pv/ssh/lookupkey/") |
| self.assertEqual(output.status_code, 403) |
| |
| output = self.app.post( |
| "/pv/ssh/lookupkey/", |
| headers={"Authorization": "token doesntexist"}, |
| ) |
| self.assertEqual(output.status_code, 401) |
| |
| pagure.lib.query.add_token_to_user( |
| self.session, None, ["internal_access"], "pingou" |
| ) |
| token = pagure.lib.query.search_token( |
| self.session, acls=["internal_access"], user="pingou" |
| ) |
| output = self.app.post( |
| "/pv/ssh/lookupkey/", |
| headers={"Authorization": "token %s" % token[0].id}, |
| ) |
| self.assertEqual(output.status_code, 400) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_pull_request_add_comment(self, send_email): |
| """ Test the pull_request_add_comment function. """ |
| send_email.return_value = True |
| |
| tests.create_projects(self.session) |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=repo, |
| branch_from="feature", |
| repo_to=repo, |
| branch_to="master", |
| title="PR from the feature branch", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "PR from the feature branch") |
| |
| request = repo.requests[0] |
| self.assertEqual(len(request.comments), 0) |
| self.assertEqual(len(request.discussion), 0) |
| |
| data = {"objid": "foo"} |
| |
| |
| output = self.app.post("/pv/pull-request/comment/", data=data) |
| self.assertEqual(output.status_code, 405) |
| |
| |
| output = self.app.put("/pv/pull-request/comment/", data=data) |
| self.assertEqual(output.status_code, 400) |
| |
| data = {"objid": "foo", "useremail": "foo@pingou.com"} |
| |
| |
| output = self.app.put("/pv/pull-request/comment/", data=data) |
| self.assertEqual(output.status_code, 404) |
| |
| data = {"objid": request.uid, "useremail": "foo@pingou.com"} |
| |
| |
| output = self.app.put("/pv/pull-request/comment/", data=data) |
| self.assertEqual(output.status_code, 400) |
| |
| data = { |
| "objid": request.uid, |
| "useremail": "foo@pingou.com", |
| "comment": "Looks good to me!", |
| } |
| |
| |
| output = self.app.put("/pv/pull-request/comment/", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, {"message": "Comment added"}) |
| |
| self.session.commit() |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| request = repo.requests[0] |
| self.assertEqual(len(request.comments), 1) |
| self.assertEqual(len(request.discussion), 1) |
| |
| |
| before = pagure.config.config["IP_ALLOWED_INTERNAL"][:] |
| pagure.config.config["IP_ALLOWED_INTERNAL"] = [] |
| |
| output = self.app.put("/pv/pull-request/comment/", data=data) |
| self.assertEqual(output.status_code, 403) |
| |
| pagure.config.config["IP_ALLOWED_INTERNAL"] = before[:] |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_ticket_add_comment(self, send_email): |
| """ Test the ticket_add_comment function. """ |
| send_email.return_value = True |
| |
| tests.create_projects(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| msg = pagure.lib.query.new_issue( |
| session=self.session, |
| repo=repo, |
| title="Test issue", |
| content="We should work on this", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(msg.title, "Test issue") |
| |
| issue = repo.issues[0] |
| self.assertEqual(len(issue.comments), 0) |
| |
| data = {"objid": "foo"} |
| |
| |
| output = self.app.post("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 405) |
| |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 400) |
| |
| data = {"objid": "foo", "useremail": "foo@pingou.com"} |
| |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 404) |
| |
| data = {"objid": issue.uid, "useremail": "foo@pingou.com"} |
| |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 400) |
| |
| data = { |
| "objid": issue.uid, |
| "useremail": "foo@pingou.com", |
| "comment": "Looks good to me!", |
| } |
| |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, {"message": "Comment added"}) |
| |
| self.session.commit() |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| issue = repo.issues[0] |
| self.assertEqual(len(issue.comments), 1) |
| |
| |
| pagure.config.config["IP_ALLOWED_INTERNAL"].remove(None) |
| before = pagure.config.config["IP_ALLOWED_INTERNAL"][:] |
| pagure.config.config["IP_ALLOWED_INTERNAL"] = [] |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 403) |
| |
| pagure.config.config["IP_ALLOWED_INTERNAL"] = before[:] |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_private_ticket_add_comment(self, send_email): |
| """ Test the ticket_add_comment function on a private ticket. """ |
| send_email.return_value = True |
| |
| tests.create_projects(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| msg = pagure.lib.query.new_issue( |
| session=self.session, |
| repo=repo, |
| title="Test issue", |
| content="We should work on this, really", |
| user="pingou", |
| private=True, |
| ) |
| self.session.commit() |
| self.assertEqual(msg.title, "Test issue") |
| |
| issue = repo.issues[0] |
| self.assertEqual(len(issue.comments), 0) |
| |
| data = {"objid": "foo"} |
| |
| |
| output = self.app.post("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 405) |
| |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 400) |
| |
| data = {"objid": "foo", "useremail": "foo@pingou.com"} |
| |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 404) |
| |
| data = {"objid": issue.uid, "useremail": "foo@bar.com"} |
| |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 403) |
| |
| data = {"objid": issue.uid, "useremail": "foo@pingou.com"} |
| |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 400) |
| |
| data = { |
| "objid": issue.uid, |
| "useremail": "foo@pingou.com", |
| "comment": "Looks good to me!", |
| } |
| |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, {"message": "Comment added"}) |
| |
| self.session.commit() |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| issue = repo.issues[0] |
| self.assertEqual(len(issue.comments), 1) |
| |
| |
| before = pagure.config.config["IP_ALLOWED_INTERNAL"][:] |
| pagure.config.config["IP_ALLOWED_INTERNAL"] = [] |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 403) |
| |
| pagure.config.config["IP_ALLOWED_INTERNAL"] = before[:] |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_private_ticket_add_comment_acl(self, send_email): |
| """ Test the ticket_add_comment function on a private ticket. """ |
| send_email.return_value = True |
| |
| tests.create_projects(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| msg = pagure.lib.query.new_issue( |
| session=self.session, |
| repo=repo, |
| title="Test issue", |
| content="We should work on this, really", |
| user="pingou", |
| private=True, |
| ) |
| self.session.commit() |
| self.assertEqual(msg.title, "Test issue") |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| issue = repo.issues[0] |
| self.assertEqual(len(issue.comments), 0) |
| |
| |
| |
| data = { |
| "objid": issue.uid, |
| "useremail": "foo@bar.com", |
| "comment": "Looks good to me!", |
| } |
| |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 403) |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| |
| |
| msg = pagure.lib.query.add_user_to_project( |
| self.session, |
| project=repo, |
| new_user="foo", |
| user="pingou", |
| access="ticket", |
| ) |
| self.session.commit() |
| self.assertEqual(msg, "User added") |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(sorted([u.username for u in repo.users]), ["foo"]) |
| self.assertEqual(sorted([u.username for u in repo.committers]), []) |
| self.assertEqual(sorted([u.username for u in repo.admins]), []) |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 403) |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| |
| |
| msg = pagure.lib.query.add_user_to_project( |
| self.session, |
| project=repo, |
| new_user="foo", |
| user="pingou", |
| access="commit", |
| ) |
| self.session.commit() |
| self.assertEqual(msg, "User access updated") |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(sorted([u.username for u in repo.users]), ["foo"]) |
| self.assertEqual( |
| sorted([u.username for u in repo.committers]), ["foo"] |
| ) |
| self.assertEqual(sorted([u.username for u in repo.admins]), []) |
| |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, {"message": "Comment added"}) |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| issue = repo.issues[0] |
| self.assertEqual(len(issue.comments), 1) |
| |
| |
| |
| msg = pagure.lib.query.add_user_to_project( |
| self.session, |
| project=repo, |
| new_user="foo", |
| user="pingou", |
| access="admin", |
| ) |
| self.session.commit() |
| self.assertEqual(msg, "User access updated") |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(sorted([u.username for u in repo.users]), ["foo"]) |
| self.assertEqual( |
| sorted([u.username for u in repo.committers]), ["foo"] |
| ) |
| self.assertEqual(sorted([u.username for u in repo.admins]), ["foo"]) |
| |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, {"message": "Comment added"}) |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| issue = repo.issues[0] |
| self.assertEqual(len(issue.comments), 2) |
| |
| |
| before = pagure.config.config["IP_ALLOWED_INTERNAL"][:] |
| pagure.config.config["IP_ALLOWED_INTERNAL"] = [] |
| |
| output = self.app.put("/pv/ticket/comment/", data=data) |
| self.assertEqual(output.status_code, 403) |
| |
| pagure.config.config["IP_ALLOWED_INTERNAL"] = before[:] |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_mergeable_request_pull_FF(self, send_email): |
| """ Test the mergeable_request_pull endpoint with a fast-forward |
| merge. |
| """ |
| send_email.return_value = True |
| |
| |
| |
| origgitrepo = os.path.join(self.path, "repos", "test.git") |
| self.assertFalse(os.path.exists(origgitrepo)) |
| os.makedirs(origgitrepo) |
| orig_repo = pygit2.init_repository(origgitrepo, bare=True) |
| os.makedirs(os.path.join(self.path, "repos_tmp")) |
| gitrepo = os.path.join(self.path, "repos_tmp", "test.git") |
| repo = pygit2.clone_repository(origgitrepo, gitrepo) |
| |
| |
| with open(os.path.join(gitrepo, "sources"), "w") as stream: |
| stream.write("foo\n bar") |
| repo.index.add("sources") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add sources file for testing", |
| |
| tree, |
| |
| [], |
| ) |
| |
| first_commit = repo.revparse_single("HEAD") |
| refname = "refs/heads/master:refs/heads/master" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| with open(os.path.join(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 = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/feature", |
| author, |
| committer, |
| "Add baz and boose to the sources\n\n There are more objects to " |
| "consider", |
| |
| tree, |
| |
| [first_commit.oid.hex], |
| ) |
| |
| second_commit = repo.revparse_single("HEAD") |
| refname = "refs/heads/feature:refs/heads/feature" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| tests.create_projects(self.session) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=project, |
| branch_from="feature", |
| repo_to=project, |
| branch_to="master", |
| title="PR from the feature branch", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "PR from the feature branch") |
| |
| |
| data = {"objid": "blah"} |
| |
| |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 400) |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| output = self.app.get("/test/adduser") |
| csrf_token = ( |
| output.get_data(as_text=True) |
| .split('name="csrf_token" type="hidden" value="')[1] |
| .split('">')[0] |
| ) |
| |
| |
| data = {"csrf_token": csrf_token} |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 404) |
| |
| |
| project = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| 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) |
| exp = { |
| "code": "FFORWARD", |
| "message": "The pull-request can be merged and fast-forwarded", |
| "short_code": "Ok", |
| } |
| |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, exp) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_mergeable_request_pull_no_change(self, send_email): |
| """ Test the mergeable_request_pull endpoint when there are no |
| changes to merge. |
| """ |
| send_email.return_value = True |
| |
| |
| |
| gitrepo = os.path.join(self.path, "repos", "test.git") |
| self.assertFalse(os.path.exists(gitrepo)) |
| os.makedirs(gitrepo) |
| 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() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add sources file for testing", |
| |
| tree, |
| |
| [], |
| ) |
| |
| first_commit = repo.revparse_single("HEAD") |
| |
| |
| with open(os.path.join(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 = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add baz and boose to the sources\n\n There are more objects to " |
| "consider", |
| |
| tree, |
| |
| [first_commit.oid.hex], |
| ) |
| |
| second_commit = repo.revparse_single("HEAD") |
| |
| |
| tests.create_projects(self.session) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=project, |
| branch_from="master", |
| repo_to=project, |
| branch_to="master", |
| title="PR from the feature branch", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "PR from the feature branch") |
| |
| |
| data = {"objid": "blah"} |
| |
| |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 400) |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| output = self.app.get("/test/adduser") |
| csrf_token = ( |
| output.get_data(as_text=True) |
| .split('name="csrf_token" type="hidden" value="')[1] |
| .split('">')[0] |
| ) |
| |
| |
| data = {"csrf_token": csrf_token} |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 404) |
| |
| |
| project = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| 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) |
| exp = { |
| "code": "NO_CHANGE", |
| "message": "Nothing to change, git is up to date", |
| "short_code": "No changes", |
| } |
| |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, exp) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_mergeable_request_pull_merge(self, send_email): |
| """ Test the mergeable_request_pull endpoint when the changes can |
| be merged with a merge commit. |
| """ |
| send_email.return_value = True |
| |
| |
| |
| origgitrepo = os.path.join(self.path, "repos", "test.git") |
| self.assertFalse(os.path.exists(origgitrepo)) |
| os.makedirs(origgitrepo) |
| orig_repo = pygit2.init_repository(origgitrepo, bare=True) |
| os.makedirs(os.path.join(self.path, "repos_tmp")) |
| gitrepo = os.path.join(self.path, "repos_tmp", "test.git") |
| repo = pygit2.clone_repository(origgitrepo, gitrepo) |
| |
| |
| with open(os.path.join(gitrepo, "sources"), "w") as stream: |
| stream.write("foo\n bar") |
| repo.index.add("sources") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add sources file for testing", |
| |
| tree, |
| |
| [], |
| ) |
| |
| first_commit = repo.revparse_single("HEAD") |
| refname = "refs/heads/master:refs/heads/master" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| with open(os.path.join(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 = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/feature", |
| author, |
| committer, |
| "Add baz and boose to the sources\n\n There are more objects to " |
| "consider", |
| |
| tree, |
| |
| [first_commit.oid.hex], |
| ) |
| refname = "refs/heads/feature:refs/heads/feature" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| with open(os.path.join(gitrepo, ".gitignore"), "w") as stream: |
| stream.write("*~") |
| repo.index.add(".gitignore") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| 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 = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| tests.create_projects(self.session) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=project, |
| branch_from="feature", |
| repo_to=project, |
| branch_to="master", |
| title="PR from the feature branch", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "PR from the feature branch") |
| |
| |
| data = {} |
| |
| |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 400) |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| output = self.app.get("/test/adduser") |
| csrf_token = ( |
| output.get_data(as_text=True) |
| .split('name="csrf_token" type="hidden" value="')[1] |
| .split('">')[0] |
| ) |
| |
| |
| data = {"csrf_token": csrf_token} |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 404) |
| |
| |
| project = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| data = { |
| "csrf_token": csrf_token, |
| "requestid": project.requests[0].uid, |
| "force": True, |
| } |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 200) |
| exp = { |
| "code": "MERGE", |
| "message": "The pull-request can be merged with a merge commit", |
| "short_code": "With merge", |
| } |
| |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, exp) |
| |
| |
| 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) |
| exp = { |
| "code": "MERGE", |
| "message": "The pull-request can be merged with a merge commit", |
| "short_code": "With merge", |
| } |
| |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, exp) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_mergeable_request_pull_conflicts(self, send_email): |
| """ Test the mergeable_request_pull endpoint when the changes cannot |
| be merged due to conflicts. |
| """ |
| send_email.return_value = True |
| |
| |
| origgitrepo = os.path.join(self.path, "repos", "test.git") |
| self.assertFalse(os.path.exists(origgitrepo)) |
| os.makedirs(origgitrepo) |
| orig_repo = pygit2.init_repository(origgitrepo, bare=True) |
| os.makedirs(os.path.join(self.path, "repos_tmp")) |
| gitrepo = os.path.join(self.path, "repos_tmp", "test.git") |
| repo = pygit2.clone_repository(origgitrepo, gitrepo) |
| |
| |
| with open(os.path.join(gitrepo, "sources"), "w") as stream: |
| stream.write("foo\n bar") |
| repo.index.add("sources") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add sources file for testing", |
| |
| tree, |
| |
| [], |
| ) |
| |
| first_commit = repo.revparse_single("HEAD") |
| refname = "refs/heads/master:refs/heads/master" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| with open(os.path.join(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 = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/feature", |
| author, |
| committer, |
| "Add baz and boose to the sources\n\n There are more objects to " |
| "consider", |
| |
| tree, |
| |
| [first_commit.oid.hex], |
| ) |
| refname = "refs/heads/feature:refs/heads/feature" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| with open(os.path.join(gitrepo, "sources"), "w") as stream: |
| stream.write("foo\n bar\nbaz\n") |
| repo.index.add("sources") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| 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 = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| tests.create_projects(self.session) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=project, |
| branch_from="feature", |
| repo_to=project, |
| branch_to="master", |
| title="PR from the feature branch", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "PR from the feature branch") |
| |
| |
| data = {} |
| |
| |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 400) |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| output = self.app.get("/test/adduser") |
| csrf_token = ( |
| output.get_data(as_text=True) |
| .split('name="csrf_token" type="hidden" value="')[1] |
| .split('">')[0] |
| ) |
| |
| |
| data = {"csrf_token": csrf_token} |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 404) |
| |
| |
| project = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| 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) |
| exp = { |
| "code": "CONFLICTS", |
| "message": "The pull-request cannot be merged due to conflicts", |
| "short_code": "Conflicts", |
| } |
| |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, exp) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_mergeable_request_pull_merge_no_nonff_merges(self, send_email): |
| """ Test the mergeable_request_pull endpoint when the changes can |
| be merged with a merge commit, but project settings prohibit this. |
| """ |
| send_email.return_value = True |
| |
| |
| |
| origgitrepo = os.path.join(self.path, "repos", "test.git") |
| self.assertFalse(os.path.exists(origgitrepo)) |
| os.makedirs(origgitrepo) |
| orig_repo = pygit2.init_repository(origgitrepo, bare=True) |
| os.makedirs(os.path.join(self.path, "repos_tmp")) |
| gitrepo = os.path.join(self.path, "repos_tmp", "test.git") |
| repo = pygit2.clone_repository(origgitrepo, gitrepo) |
| |
| |
| with open(os.path.join(gitrepo, "sources"), "w") as stream: |
| stream.write("foo\n bar") |
| repo.index.add("sources") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add sources file for testing", |
| |
| tree, |
| |
| [], |
| ) |
| |
| first_commit = repo.revparse_single("HEAD") |
| refname = "refs/heads/master:refs/heads/master" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| with open(os.path.join(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 = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/feature", |
| author, |
| committer, |
| "Add baz and boose to the sources\n\n There are more objects to " |
| "consider", |
| |
| tree, |
| |
| [first_commit.oid.hex], |
| ) |
| refname = "refs/heads/feature:refs/heads/feature" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| with open(os.path.join(gitrepo, ".gitignore"), "w") as stream: |
| stream.write("*~") |
| repo.index.add(".gitignore") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| 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 = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| tests.create_projects(self.session) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=project, |
| branch_from="feature", |
| repo_to=project, |
| branch_to="master", |
| title="PR from the feature branch", |
| user="pingou", |
| ) |
| settings = {"disable_non_fast-forward_merges": True} |
| project.settings = settings |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "PR from the feature branch") |
| |
| |
| data = {} |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| output = self.app.get("/test/adduser") |
| csrf_token = ( |
| output.get_data(as_text=True) |
| .split('name="csrf_token" type="hidden" value="')[1] |
| .split('">')[0] |
| ) |
| |
| |
| project = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| data = { |
| "csrf_token": csrf_token, |
| "requestid": project.requests[0].uid, |
| "force": True, |
| } |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 200) |
| exp = { |
| "code": "NEEDSREBASE", |
| "message": "The pull-request must be rebased before merging", |
| "short_code": "Needs rebase", |
| } |
| |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, exp) |
| |
| |
| 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) |
| exp = { |
| "code": "NEEDSREBASE", |
| "message": "The pull-request must be rebased before merging", |
| "short_code": "Needs rebase", |
| } |
| |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, exp) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_mergeable_request_pull_minimum_score(self, send_email): |
| """ Test the mergeable_request_pull endpoint when the changes can |
| be merged with a merge FF, but project settings enforces vote on |
| the PR. |
| """ |
| send_email.return_value = True |
| |
| |
| |
| origgitrepo = os.path.join(self.path, "repos", "test.git") |
| self.assertFalse(os.path.exists(origgitrepo)) |
| os.makedirs(origgitrepo) |
| orig_repo = pygit2.init_repository(origgitrepo, bare=True) |
| os.makedirs(os.path.join(self.path, "repos_tmp")) |
| gitrepo = os.path.join(self.path, "repos_tmp", "test.git") |
| repo = pygit2.clone_repository(origgitrepo, gitrepo) |
| |
| |
| with open(os.path.join(gitrepo, "sources"), "w") as stream: |
| stream.write("foo\n bar") |
| repo.index.add("sources") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add sources file for testing", |
| |
| tree, |
| |
| [], |
| ) |
| |
| first_commit = repo.revparse_single("HEAD") |
| refname = "refs/heads/master:refs/heads/master" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| with open(os.path.join(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 = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/feature", |
| author, |
| committer, |
| "Add baz and boose to the sources\n\n There are more objects to " |
| "consider", |
| |
| tree, |
| |
| [first_commit.oid.hex], |
| ) |
| refname = "refs/heads/feature:refs/heads/feature" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| tests.create_projects(self.session) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=project, |
| branch_from="feature", |
| repo_to=project, |
| branch_to="master", |
| title="PR from the feature branch", |
| user="pingou", |
| ) |
| settings = {"Minimum_score_to_merge_pull-request": 2} |
| project.settings = settings |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "PR from the feature branch") |
| |
| |
| data = {} |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| output = self.app.get("/test/adduser") |
| csrf_token = ( |
| output.get_data(as_text=True) |
| .split('name="csrf_token" type="hidden" value="')[1] |
| .split('">')[0] |
| ) |
| |
| |
| project = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| data = { |
| "csrf_token": csrf_token, |
| "requestid": project.requests[0].uid, |
| "force": True, |
| } |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 400) |
| exp = { |
| "code": "CONFLICTS", |
| "message": "Pull-Request does not meet the minimal number " |
| "of review required: 0/2", |
| } |
| |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, exp) |
| |
| |
| pub_api_call = self.app.get("/api/0/test/pull-request/1") |
| data = json.loads(pub_api_call.get_data(as_text=True)) |
| self.assertIn(data["cached_merge_status"], ("MERGE", "FFORWARD")) |
| |
| |
| 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, 400) |
| exp = { |
| "code": "CONFLICTS", |
| "message": "Pull-Request does not meet the minimal number " |
| "of review required: 0/2", |
| } |
| |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, exp) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| @patch( |
| "pagure.lib.git.merge_pull_request", |
| MagicMock(side_effect=pagure.exceptions.PagureException("error")), |
| ) |
| def test_mergeable_request_pull_merge_pagureerror(self): |
| """ Test the mergeable_request_pull endpoint when the backend |
| raises an GitError exception. |
| """ |
| |
| |
| origgitrepo = os.path.join(self.path, "repos", "test.git") |
| self.assertFalse(os.path.exists(origgitrepo)) |
| os.makedirs(origgitrepo) |
| orig_repo = pygit2.init_repository(origgitrepo, bare=True) |
| os.makedirs(os.path.join(self.path, "repos_tmp")) |
| gitrepo = os.path.join(self.path, "repos_tmp", "test.git") |
| repo = pygit2.clone_repository(origgitrepo, gitrepo) |
| |
| |
| with open(os.path.join(gitrepo, "sources"), "w") as stream: |
| stream.write("foo\n bar") |
| repo.index.add("sources") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add sources file for testing", |
| |
| tree, |
| |
| [], |
| ) |
| |
| first_commit = repo.revparse_single("HEAD") |
| refname = "refs/heads/master:refs/heads/master" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| with open(os.path.join(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 = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/feature", |
| author, |
| committer, |
| "Add baz and boose to the sources\n\n There are more objects to " |
| "consider", |
| |
| tree, |
| |
| [first_commit.oid.hex], |
| ) |
| refname = "refs/heads/feature:refs/heads/feature" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| with open(os.path.join(gitrepo, ".gitignore"), "w") as stream: |
| stream.write("*~") |
| repo.index.add(".gitignore") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| 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 = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| tests.create_projects(self.session) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=project, |
| branch_from="feature", |
| repo_to=project, |
| branch_to="master", |
| title="PR from the feature branch", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "PR from the feature branch") |
| |
| |
| data = {} |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| output = self.app.get("/test/adduser") |
| csrf_token = self.get_csrf(output=output) |
| |
| |
| project = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| data = { |
| "csrf_token": csrf_token, |
| "requestid": project.requests[0].uid, |
| "force": True, |
| } |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 500) |
| exp = {"code": "CONFLICTS", "message": "error"} |
| |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, exp) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| @patch( |
| "pagure.lib.git.merge_pull_request", |
| MagicMock(side_effect=pygit2.GitError("git error")), |
| ) |
| def test_mergeable_request_pull_merge_giterror(self): |
| """ Test the mergeable_request_pull endpoint when the backend |
| raises an GitError exception. |
| """ |
| |
| |
| origgitrepo = os.path.join(self.path, "repos", "test.git") |
| self.assertFalse(os.path.exists(origgitrepo)) |
| os.makedirs(origgitrepo) |
| orig_repo = pygit2.init_repository(origgitrepo, bare=True) |
| os.makedirs(os.path.join(self.path, "repos_tmp")) |
| gitrepo = os.path.join(self.path, "repos_tmp", "test.git") |
| repo = pygit2.clone_repository(origgitrepo, gitrepo) |
| |
| |
| with open(os.path.join(gitrepo, "sources"), "w") as stream: |
| stream.write("foo\n bar") |
| repo.index.add("sources") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add sources file for testing", |
| |
| tree, |
| |
| [], |
| ) |
| |
| first_commit = repo.revparse_single("HEAD") |
| refname = "refs/heads/master:refs/heads/master" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| with open(os.path.join(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 = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/feature", |
| author, |
| committer, |
| "Add baz and boose to the sources\n\n There are more objects to " |
| "consider", |
| |
| tree, |
| |
| [first_commit.oid.hex], |
| ) |
| refname = "refs/heads/feature:refs/heads/feature" |
| ori_remote = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| with open(os.path.join(gitrepo, ".gitignore"), "w") as stream: |
| stream.write("*~") |
| repo.index.add(".gitignore") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| 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 = repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| tests.create_projects(self.session) |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=project, |
| branch_from="feature", |
| repo_to=project, |
| branch_to="master", |
| title="PR from the feature branch", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "PR from the feature branch") |
| |
| |
| data = {} |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| output = self.app.get("/test/adduser") |
| csrf_token = self.get_csrf(output=output) |
| |
| |
| project = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| data = { |
| "csrf_token": csrf_token, |
| "requestid": project.requests[0].uid, |
| "force": True, |
| } |
| output = self.app.post("/pv/pull-request/merge", data=data) |
| self.assertEqual(output.status_code, 409) |
| exp = {"code": "CONFLICTS", "message": "git error"} |
| |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, exp) |
| |
| def test_get_branches_of_commit(self): |
| """ Test the get_branches_of_commit from the internal API. """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos")) |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| output = self.app.get("/test/adduser") |
| self.assertEqual(output.status_code, 200) |
| csrf_token = ( |
| output.get_data(as_text=True) |
| .split('name="csrf_token" type="hidden" value="')[1] |
| .split('">')[0] |
| ) |
| |
| |
| data = {"repo": "fakerepo", "commit_id": "foo"} |
| output = self.app.post("/pv/branches/commit/", data=data) |
| self.assertEqual(output.status_code, 400) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, {"code": "ERROR", "message": "Invalid input submitted"} |
| ) |
| |
| |
| data = { |
| "repo": "fakerepo", |
| "commit_id": "foo", |
| "csrf_token": csrf_token, |
| } |
| output = self.app.post("/pv/branches/commit/", data=data) |
| self.assertEqual(output.status_code, 404) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, |
| { |
| "code": "ERROR", |
| "message": "No repo found with the information provided", |
| }, |
| ) |
| |
| |
| data = {"repo": "test", "csrf_token": csrf_token} |
| |
| output = self.app.post("/pv/branches/commit/", data=data) |
| self.assertEqual(output.status_code, 400) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, {"code": "ERROR", "message": "No commit id submitted"} |
| ) |
| |
| |
| item = pagure.lib.model.Project( |
| user_id=1, |
| name="test20", |
| description="test project #20", |
| hook_token="aaabbbhhh", |
| ) |
| self.session.add(item) |
| self.session.commit() |
| |
| data = {"repo": "test20", "commit_id": "foo", "csrf_token": csrf_token} |
| output = self.app.post("/pv/branches/commit/", data=data) |
| self.assertEqual(output.status_code, 404) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, |
| { |
| "code": "ERROR", |
| "message": "No git repo found with the information provided", |
| }, |
| ) |
| |
| |
| gitrepo = os.path.join(self.path, "repos", "test.git") |
| self.assertTrue(os.path.exists(gitrepo)) |
| repo = pygit2.Repository(gitrepo) |
| |
| |
| with open(os.path.join(gitrepo, "sources"), "w") as stream: |
| stream.write("foo\n bar") |
| repo.index.add("sources") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add sources file for testing", |
| |
| tree, |
| |
| [], |
| ) |
| |
| first_commit = repo.revparse_single("HEAD") |
| |
| |
| with open(os.path.join(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 = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/feature", |
| author, |
| committer, |
| "Add baz and boose to the sources\n\n There are more objects to " |
| "consider", |
| |
| tree, |
| |
| [first_commit.oid.hex], |
| ) |
| |
| |
| with open(os.path.join(gitrepo, ".gitignore"), "w") as stream: |
| stream.write("*~") |
| repo.index.add(".gitignore") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| commit_hash = repo.create_commit( |
| "refs/heads/feature_branch", |
| author, |
| committer, |
| "Add .gitignore file for testing", |
| |
| tree, |
| |
| [first_commit.oid.hex], |
| ) |
| |
| |
| data = {"repo": "test", "commit_id": "foo", "csrf_token": csrf_token} |
| output = self.app.post("/pv/branches/commit/", data=data) |
| self.assertEqual(output.status_code, 404) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, |
| { |
| "code": "ERROR", |
| "message": "This commit could not be found in this repo", |
| }, |
| ) |
| |
| |
| data = { |
| "repo": "test", |
| "commit_id": commit_hash, |
| "csrf_token": csrf_token, |
| } |
| output = self.app.post("/pv/branches/commit/", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, {"code": "OK", "branches": ["feature_branch"]} |
| ) |
| |
| def test_get_branches_of_commit_with_unrelated_branches(self): |
| """ Test the get_branches_of_commit from the internal API. """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos")) |
| |
| user = tests.FakeUser(username="pingou") |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| gitrepo = os.path.join(self.path, "repos", "test.git") |
| self.assertTrue(os.path.exists(gitrepo)) |
| repo = pygit2.Repository(gitrepo) |
| |
| |
| with open(os.path.join(gitrepo, "sources"), "w") as stream: |
| stream.write("foo\n bar") |
| repo.index.add("sources") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add sources file for testing", |
| |
| tree, |
| |
| [], |
| ) |
| |
| first_commit = repo.revparse_single("HEAD") |
| |
| |
| with open(os.path.join(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 = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| commit = repo.create_commit( |
| "refs/heads/feature", |
| author, |
| committer, |
| "Add baz and boose to the sources\n\n There are more objects to " |
| "consider", |
| |
| tree, |
| |
| [], |
| ) |
| commit_hash = commit.hex |
| |
| |
| data = { |
| "repo": "test", |
| "commit_id": commit_hash, |
| "csrf_token": csrf_token, |
| } |
| output = self.app.post("/pv/branches/commit/", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(js_data, {"code": "OK", "branches": ["feature"]}) |
| |
| def test_get_branches_head(self): |
| """ Test the get_branches_head from the internal API. """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos")) |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"repo": "fakerepo"} |
| output = self.app.post("/pv/branches/heads/", data=data) |
| self.assertEqual(output.status_code, 400) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, {"code": "ERROR", "message": "Invalid input submitted"} |
| ) |
| |
| |
| data = { |
| "repo": "fakerepo", |
| "commit_id": "foo", |
| "csrf_token": csrf_token, |
| } |
| output = self.app.post("/pv/branches/heads/", data=data) |
| self.assertEqual(output.status_code, 404) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, |
| { |
| "code": "ERROR", |
| "message": "No repo found with the information provided", |
| }, |
| ) |
| |
| |
| data = {"repo": "test", "csrf_token": csrf_token} |
| |
| output = self.app.post("/pv/branches/heads/", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, {"branches": {}, "code": "OK", "heads": {}} |
| ) |
| |
| |
| item = pagure.lib.model.Project( |
| user_id=1, |
| name="test20", |
| description="test project #20", |
| hook_token="aaabbbhhh", |
| ) |
| self.session.add(item) |
| self.session.commit() |
| |
| data = {"repo": "test20", "csrf_token": csrf_token} |
| output = self.app.post("/pv/branches/heads/", data=data) |
| self.assertEqual(output.status_code, 404) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, |
| { |
| "code": "ERROR", |
| "message": "No git repo found with the information provided", |
| }, |
| ) |
| |
| |
| gitrepo = os.path.join(self.path, "repos", "test.git") |
| self.assertTrue(os.path.exists(gitrepo)) |
| repo = pygit2.Repository(gitrepo) |
| |
| |
| with open(os.path.join(gitrepo, "sources"), "w") as stream: |
| stream.write("foo\n bar") |
| repo.index.add("sources") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/master", |
| author, |
| committer, |
| "Add sources file for testing", |
| |
| tree, |
| |
| [], |
| ) |
| |
| first_commit = repo.revparse_single("HEAD") |
| |
| |
| with open(os.path.join(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 = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| repo.create_commit( |
| "refs/heads/feature", |
| author, |
| committer, |
| "Add baz and boose to the sources\n\n There are more objects to " |
| "consider", |
| |
| tree, |
| |
| [first_commit.oid.hex], |
| ) |
| |
| |
| with open(os.path.join(gitrepo, ".gitignore"), "w") as stream: |
| stream.write("*~") |
| repo.index.add(".gitignore") |
| repo.index.write() |
| |
| |
| tree = repo.index.write_tree() |
| author = pygit2.Signature("Alice Author", "alice@authors.tld") |
| committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld") |
| commit_hash = repo.create_commit( |
| "refs/heads/feature_branch", |
| author, |
| committer, |
| "Add .gitignore file for testing", |
| |
| tree, |
| |
| [first_commit.oid.hex], |
| ) |
| |
| |
| data = {"repo": "test", "csrf_token": csrf_token} |
| output = self.app.post("/pv/branches/heads/", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| |
| |
| self.assertEqual(sorted(js_data.keys()), ["branches", "code", "heads"]) |
| self.assertEqual(js_data["code"], "OK") |
| self.assertEqual(len(js_data["heads"]), 3) |
| self.assertEqual(len(js_data["branches"]), 3) |
| |
| def test_get_stats_commits_no_token(self): |
| """ Test the get_stats_commits from the internal API. """ |
| |
| data = {"repo": "fakerepo"} |
| output = self.app.post("/pv/stats/commits/authors", data=data) |
| self.assertEqual(output.status_code, 400) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, {"code": "ERROR", "message": "Invalid input submitted"} |
| ) |
| |
| def test_get_stats_commits_invalid_repo(self): |
| """ Test the get_stats_commits from the internal API. """ |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"repo": "fakerepo", "csrf_token": csrf_token} |
| output = self.app.post("/pv/stats/commits/authors", data=data) |
| self.assertEqual(output.status_code, 404) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, |
| { |
| "code": "ERROR", |
| "message": "No repo found with the information provided", |
| }, |
| ) |
| |
| def test_get_stats_commits_empty_git(self): |
| """ Test the get_stats_commits from the internal API. """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos")) |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"repo": "test", "csrf_token": csrf_token} |
| output = self.app.post("/pv/stats/commits/authors", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| sorted(js_data.keys()), ["code", "message", "task_id", "url"] |
| ) |
| self.assertEqual(js_data["code"], "OK") |
| self.assertEqual(js_data["message"], "Stats asked") |
| self.assertTrue(js_data["url"].startswith("/pv/task/")) |
| |
| output = self.app.get(js_data["url"]) |
| js_data2 = json.loads(output.get_data(as_text=True)) |
| self.assertTrue( |
| js_data2 |
| in [ |
| {"results": "reference 'refs/heads/master' not found"}, |
| {"results": "Reference 'refs/heads/master' not found"}, |
| ] |
| ) |
| |
| def test_get_stats_commits_git_populated(self): |
| """ Test the get_stats_commits from the internal API. """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "test.git") |
| ) |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"repo": "test", "csrf_token": csrf_token} |
| output = self.app.post("/pv/stats/commits/authors", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| sorted(js_data.keys()), ["code", "message", "task_id", "url"] |
| ) |
| self.assertEqual(js_data["code"], "OK") |
| self.assertEqual(js_data["message"], "Stats asked") |
| self.assertTrue(js_data["url"].startswith("/pv/task/")) |
| |
| output = self.app.get(js_data["url"]) |
| while output.status_code == 418: |
| output = self.app.get(js_data["url"]) |
| js_data2 = json.loads(output.get_data(as_text=True)) |
| self.assertTrue(js_data2["results"][3] > 1509110062) |
| js_data2["results"][3] = 1509110062 |
| self.assertTrue( |
| js_data2 |
| in [ |
| { |
| "results": [ |
| 2, |
| [ |
| [ |
| 2, |
| [ |
| [ |
| "Alice Author", |
| "alice@authors.tld", |
| "https://seccdn.libravatar.org/avatar/" |
| "96c52c78570ffc4cfefcdadf5f8e77aeebcb11e07225df11bbf2fce381cdb8bd" |
| "?s=32&d=retro", |
| ] |
| ], |
| ] |
| ], |
| 1, |
| 1509110062, |
| ] |
| }, |
| { |
| "results": [ |
| 2, |
| [ |
| [ |
| 2, |
| [ |
| [ |
| "Alice Author", |
| "alice@authors.tld", |
| "https://seccdn.libravatar.org/avatar/" |
| "96c52c78570ffc4cfefcdadf5f8e77aeebcb11e07225df11bbf2fce381cdb8bd" |
| "?d=retro&s=32", |
| ] |
| ], |
| ] |
| ], |
| 1, |
| 1509110062, |
| ] |
| }, |
| ] |
| ) |
| |
| def test_get_stats_commits_trend_no_token(self): |
| """ Test the get_stats_commits_trend from the internal API. """ |
| |
| data = {"repo": "fakerepo"} |
| output = self.app.post("/pv/stats/commits/trend", data=data) |
| self.assertEqual(output.status_code, 400) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, {"code": "ERROR", "message": "Invalid input submitted"} |
| ) |
| |
| def test_get_stats_commits_trend_invalid_repo(self): |
| """ Test the get_stats_commits_trend from the internal API. """ |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"repo": "fakerepo", "csrf_token": csrf_token} |
| output = self.app.post("/pv/stats/commits/trend", data=data) |
| self.assertEqual(output.status_code, 404) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| js_data, |
| { |
| "code": "ERROR", |
| "message": "No repo found with the information provided", |
| }, |
| ) |
| |
| def test_get_stats_commits_trend_empty_git(self): |
| """ Test the get_stats_commits_trend from the internal API. """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos")) |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"repo": "test", "csrf_token": csrf_token} |
| output = self.app.post("/pv/stats/commits/trend", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| sorted(js_data.keys()), ["code", "message", "task_id", "url"] |
| ) |
| self.assertEqual(js_data["code"], "OK") |
| self.assertEqual(js_data["message"], "Stats asked") |
| self.assertTrue(js_data["url"].startswith("/pv/task/")) |
| |
| output = self.app.get(js_data["url"]) |
| js_data2 = json.loads(output.get_data(as_text=True)) |
| self.assertTrue( |
| js_data2 |
| in [ |
| {"results": "reference 'refs/heads/master' not found"}, |
| {"results": "Reference 'refs/heads/master' not found"}, |
| ] |
| ) |
| |
| def test_get_stats_commits_trend_git_populated(self): |
| """ Test the get_stats_commits_trend from the internal API. """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "test.git") |
| ) |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"repo": "test", "csrf_token": csrf_token} |
| output = self.app.post("/pv/stats/commits/trend", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| sorted(js_data.keys()), ["code", "message", "task_id", "url"] |
| ) |
| self.assertEqual(js_data["code"], "OK") |
| self.assertEqual(js_data["message"], "Stats asked") |
| self.assertTrue(js_data["url"].startswith("/pv/task/")) |
| |
| output = self.app.get(js_data["url"]) |
| js_data2 = json.loads(output.get_data(as_text=True)) |
| today = datetime.datetime.utcnow().date() |
| self.assertDictEqual(js_data2, {"results": [[str(today), 2]]}) |
| |
| def test_get_project_family_no_project(self): |
| """ Test the get_project_family from the internal API. """ |
| output = self.app.post("/pv/test/family") |
| self.assertEqual(output.status_code, 404) |
| |
| def test_get_project_family_no_csrf(self): |
| """ Test the get_project_family from the internal API. """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "test.git") |
| ) |
| |
| output = self.app.post("/pv/test/family") |
| self.assertEqual(output.status_code, 400) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(js_data.keys()), ["code", "message"]) |
| self.assertEqual(js_data["code"], "ERROR") |
| self.assertEqual(js_data["message"], "Invalid input submitted") |
| |
| def test_get_project_family(self): |
| """ Test the get_project_family from the internal API. """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "test.git") |
| ) |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| data = {"csrf_token": csrf_token} |
| output = self.app.post("/pv/test/family", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(js_data.keys()), ["code", "family"]) |
| self.assertEqual(js_data["code"], "OK") |
| self.assertEqual(js_data["family"], ["test"]) |
| |
| def test_get_project_larger_family(self): |
| """ Test the get_project_family from the internal API. """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| |
| |
| item = pagure.lib.model.User( |
| user="ralph", |
| fullname="Ralph bar", |
| password="ralph_foo", |
| default_email="ralph@bar.com", |
| ) |
| self.session.add(item) |
| item = pagure.lib.model.UserEmail(user_id=3, email="ralph@bar.com") |
| self.session.add(item) |
| self.session.commit() |
| |
| |
| item = pagure.lib.model.Project( |
| user_id=2, |
| name="test", |
| is_fork=True, |
| parent_id=1, |
| description="test project #1", |
| hook_token="aaabbbcccddd", |
| ) |
| item.close_status = [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ] |
| self.session.add(item) |
| item = pagure.lib.model.Project( |
| user_id=3, |
| name="test", |
| is_fork=True, |
| parent_id=1, |
| description="test project #1", |
| hook_token="aaabbbccceee", |
| ) |
| item.close_status = [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ] |
| self.session.add(item) |
| self.session.commit() |
| |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| data = {"csrf_token": csrf_token} |
| output = self.app.post("/pv/test/family", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(js_data.keys()), ["code", "family"]) |
| self.assertEqual(js_data["code"], "OK") |
| self.assertEqual( |
| js_data["family"], ["test", "fork/foo/test", "fork/ralph/test"] |
| ) |
| |
| def test_get_project_larger_family_pr_only(self): |
| """ Test the get_project_family from the internal API. """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| |
| |
| item = pagure.lib.model.User( |
| user="ralph", |
| fullname="Ralph bar", |
| password="ralph_foo", |
| default_email="ralph@bar.com", |
| ) |
| self.session.add(item) |
| item = pagure.lib.model.UserEmail(user_id=3, email="ralph@bar.com") |
| self.session.add(item) |
| self.session.commit() |
| |
| |
| item = pagure.lib.model.Project( |
| user_id=2, |
| name="test", |
| is_fork=True, |
| parent_id=1, |
| description="test project #1", |
| hook_token="aaabbbcccddd", |
| ) |
| item.close_status = [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ] |
| |
| default_repo_settings = item.settings |
| default_repo_settings["issue_tracker"] = False |
| item.settings = default_repo_settings |
| self.session.add(item) |
| |
| item = pagure.lib.model.Project( |
| user_id=3, |
| name="test", |
| is_fork=True, |
| parent_id=1, |
| description="test project #1", |
| hook_token="aaabbbccceee", |
| ) |
| item.close_status = [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ] |
| |
| default_repo_settings = item.settings |
| default_repo_settings["pull_requests"] = False |
| item.settings = default_repo_settings |
| self.session.add(item) |
| self.session.commit() |
| |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| data = {"csrf_token": csrf_token, "allows_pr": "1"} |
| output = self.app.post("/pv/test/family", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(js_data.keys()), ["code", "family"]) |
| self.assertEqual(js_data["code"], "OK") |
| self.assertEqual(js_data["family"], ["test", "fork/foo/test"]) |
| |
| data = {"csrf_token": csrf_token, "allows_issues": "1"} |
| output = self.app.post("/pv/test/family", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(js_data.keys()), ["code", "family"]) |
| self.assertEqual(js_data["code"], "OK") |
| self.assertEqual(js_data["family"], ["test", "fork/ralph/test"]) |
| |
| def test_get_pull_request_ready_branch_no_csrf(self): |
| """Test the get_pull_request_ready_branch from the internal API |
| on the main repository |
| """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| |
| |
| data = {"repo": "test"} |
| output = self.app.post("/pv/pull-request/ready", data=data) |
| self.assertEqual(output.status_code, 400) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(js_data.keys()), ["code", "message"]) |
| self.assertEqual(js_data["code"], "ERROR") |
| self.assertEqual(js_data["message"], "Invalid input submitted") |
| |
| def test_get_pull_request_ready_branch_no_repo(self): |
| """Test the get_pull_request_ready_branch from the internal API |
| on the main repository |
| """ |
| with tests.user_set(self.app.application, tests.FakeUser()): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"repo": "test", "namespace": "fake", "csrf_token": csrf_token} |
| output = self.app.post("/pv/pull-request/ready", data=data) |
| self.assertEqual(output.status_code, 404) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(js_data.keys()), ["code", "message"]) |
| self.assertEqual(js_data["code"], "ERROR") |
| self.assertEqual( |
| js_data["message"], "No repo found with the information provided" |
| ) |
| |
| def test_get_pull_request_ready_branch_main_repo_no_branch(self): |
| """Test the get_pull_request_ready_branch from the internal API |
| on the main repository |
| """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"csrf_token": csrf_token, "repo": "test"} |
| output = self.app.post("/pv/pull-request/ready", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(js_data.keys()), ["code", "task"]) |
| self.assertEqual(js_data["code"], "OK") |
| |
| def test_get_pull_request_ready_branch_on_fork(self): |
| """Test the get_pull_request_ready_branch from the internal API on |
| a fork |
| """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| tests.create_projects_git( |
| os.path.join(self.path, "repos", "forks", "foo"), bare=True |
| ) |
| |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "forks", "foo", "test.git"), |
| branch="feature", |
| ) |
| |
| |
| item = pagure.lib.model.Project( |
| user_id=2, |
| name="test", |
| is_fork=True, |
| parent_id=1, |
| description="test project #1", |
| hook_token="aaabbbcccddd", |
| ) |
| item.close_status = [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ] |
| self.session.add(item) |
| self.session.commit() |
| |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"csrf_token": csrf_token, "repo": "test", "repouser": "foo"} |
| output = self.app.post("/pv/pull-request/ready", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(js_data.keys()), ["code", "task"]) |
| self.assertEqual(js_data["code"], "OK") |
| output = self.app.get("/pv/task/" + js_data["task"]) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| js_data, |
| { |
| "results": { |
| "branch_w_pr": {}, |
| "new_branch": { |
| "feature": {"commits": 2, "target_branch": "master"} |
| }, |
| } |
| }, |
| ) |
| |
| def test_get_pull_request_ready_branch_on_fork_no_parent_no_pr(self): |
| """Test the get_pull_request_ready_branch from the internal API on |
| a fork that has no parent repo (deleted) and doesn't allow PR |
| """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| tests.create_projects_git( |
| os.path.join(self.path, "repos", "forks", "foo"), bare=True |
| ) |
| |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "forks", "foo", "test.git"), |
| branch="feature", |
| ) |
| |
| |
| item = pagure.lib.model.Project( |
| user_id=2, |
| name="test", |
| is_fork=True, |
| parent_id=1, |
| description="test project #1", |
| hook_token="aaabbbcccddd", |
| ) |
| item.close_status = [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ] |
| self.session.add(item) |
| self.session.commit() |
| settings = item.settings |
| settings["pull_requests"] = False |
| item.settings = settings |
| self.session.add(item) |
| self.session.commit() |
| |
| |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.session.delete(project) |
| self.session.commit() |
| |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"csrf_token": csrf_token, "repo": "test", "repouser": "foo"} |
| output = self.app.post("/pv/pull-request/ready", data=data) |
| self.assertEqual(output.status_code, 400) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(js_data.keys()), ["code", "message"]) |
| self.assertEqual(js_data["code"], "ERROR") |
| self.assertEqual( |
| js_data["message"], "Pull-request have been disabled for this repo" |
| ) |
| |
| def test_get_pull_request_ready_branch_on_fork_no_parent(self): |
| """Test the get_pull_request_ready_branch from the internal API on |
| a fork that has no parent repo (deleted). |
| """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| tests.create_projects_git( |
| os.path.join(self.path, "repos", "forks", "foo"), bare=True |
| ) |
| |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "forks", "foo", "test.git"), |
| branch="feature", |
| ) |
| |
| |
| item = pagure.lib.model.Project( |
| user_id=2, |
| name="test", |
| is_fork=True, |
| parent_id=1, |
| description="test project #1", |
| hook_token="aaabbbcccddd", |
| ) |
| item.close_status = [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ] |
| self.session.add(item) |
| self.session.commit() |
| settings = item.settings |
| settings["pull_requests"] = True |
| item.settings = settings |
| self.session.add(item) |
| self.session.commit() |
| |
| |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.session.delete(project) |
| self.session.commit() |
| |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"csrf_token": csrf_token, "repo": "test", "repouser": "foo"} |
| output = self.app.post("/pv/pull-request/ready", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(js_data.keys()), ["code", "task"]) |
| self.assertEqual(js_data["code"], "OK") |
| output = self.app.get("/pv/task/" + js_data["task"]) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| js_data, |
| { |
| "results": { |
| "branch_w_pr": {}, |
| "new_branch": { |
| "feature": {"commits": 2, "target_branch": "master"} |
| }, |
| } |
| }, |
| ) |
| |
| def test_get_pull_request_ready_branch_matching_target_off(self): |
| """Test the get_pull_request_ready_branch from the internal API on |
| a fork while PR_TARGET_MATCHING_BRANCH is False |
| """ |
| tests.create_projects(self.session) |
| |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="master" |
| ) |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="feature" |
| ) |
| |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| tests.create_projects_git( |
| os.path.join(self.path, "repos", "forks", "foo"), bare=True |
| ) |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "forks", "foo", "test.git"), |
| branch="feature", |
| ) |
| |
| |
| item = pagure.lib.model.Project( |
| user_id=2, |
| name="test", |
| is_fork=True, |
| parent_id=1, |
| description="test project #1", |
| hook_token="aaabbbcccddd", |
| ) |
| item.close_status = [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ] |
| self.session.add(item) |
| self.session.commit() |
| |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"csrf_token": csrf_token, "repo": "test", "repouser": "foo"} |
| output = self.app.post("/pv/pull-request/ready", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(js_data.keys()), ["code", "task"]) |
| self.assertEqual(js_data["code"], "OK") |
| output = self.app.get("/pv/task/" + js_data["task"]) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| js_data, |
| { |
| "results": { |
| "branch_w_pr": {}, |
| "new_branch": { |
| "feature": {"commits": 2, "target_branch": "master"} |
| }, |
| } |
| }, |
| ) |
| |
| @patch.dict("pagure.config.config", {"PR_TARGET_MATCHING_BRANCH": True}) |
| def test_get_pull_request_ready_branch_matching_target_on(self): |
| """Test the get_pull_request_ready_branch from the internal API on |
| a fork while PR_TARGET_MATCHING_BRANCH is True |
| """ |
| tests.create_projects(self.session) |
| |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="master" |
| ) |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="feature" |
| ) |
| |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| tests.create_projects_git( |
| os.path.join(self.path, "repos", "forks", "foo"), bare=True |
| ) |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "forks", "foo", "test.git"), |
| append="testing from foo's fork", |
| branch="feature", |
| ) |
| |
| |
| item = pagure.lib.model.Project( |
| user_id=2, |
| name="test", |
| is_fork=True, |
| parent_id=1, |
| description="test project #1", |
| hook_token="aaabbbcccddd", |
| ) |
| item.close_status = [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ] |
| self.session.add(item) |
| self.session.commit() |
| |
| |
| user = tests.FakeUser() |
| user.username = "pingou" |
| with tests.user_set(self.app.application, user): |
| csrf_token = self.get_csrf() |
| |
| |
| data = {"csrf_token": csrf_token, "repo": "test", "repouser": "foo"} |
| output = self.app.post("/pv/pull-request/ready", data=data) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(js_data.keys()), ["code", "task"]) |
| self.assertEqual(js_data["code"], "OK") |
| output = self.app.get("/pv/task/" + js_data["task"]) |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| js_data, |
| { |
| "results": { |
| "branch_w_pr": {}, |
| "new_branch": { |
| "feature": {"commits": 1, "target_branch": "feature"} |
| }, |
| } |
| }, |
| ) |
| |
| def test_task_info_task_running(self): |
| """ Test the task_info internal API endpoint when the task isn't |
| ready. |
| """ |
| task = MagicMock() |
| task.get = MagicMock(return_value="FAILED") |
| task.ready = MagicMock(return_value=False) |
| with patch( |
| "pagure.lib.tasks.get_result", MagicMock(return_value=task) |
| ): |
| output = self.app.get("/pv/task/2") |
| self.assertEqual(output.status_code, 418) |
| |
| def test_task_info_task_passed(self): |
| """ Test the task_info internal API endpoint when the task failed. |
| """ |
| task = MagicMock() |
| task.get = MagicMock(return_value="PASSED") |
| with patch( |
| "pagure.lib.tasks.get_result", MagicMock(return_value=task) |
| ): |
| output = self.app.get("/pv/task/2") |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(js_data, {"results": "PASSED"}) |
| |
| def test_task_info_task_failed(self): |
| """ Test the task_info internal API endpoint when the task failed. |
| """ |
| task = MagicMock() |
| task.get = MagicMock(return_value=Exception("Random error")) |
| with patch( |
| "pagure.lib.tasks.get_result", MagicMock(return_value=task) |
| ): |
| output = self.app.get("/pv/task/2") |
| self.assertEqual(output.status_code, 200) |
| js_data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(js_data, {"results": "Random error"}) |
| |
| def test_lookup_ssh_key(self): |
| """ Test the mergeable_request_pull endpoint when the backend |
| raises an GitError exception. |
| """ |
| tests.create_projects(self.session) |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| project_key = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIC4zmifEL8TLLZUZnjAuVL8495DAkpAAM2eBhwHwawBm" |
| project_key_fp = "SHA256:ZSUQAqpPDWi90Fs6Ow8Epc8F3qiKVfU+H5ssvo7jiI0" |
| |
| pingou = pagure.lib.query.get_user(self.session, "pingou") |
| user_key = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIDTsfdTcXw4rlU1aQwOTbLOXqossLwpPIk27S/G17kUz" |
| user_key_fp = "SHA256:jUJHzrq2Ct6Ubf7Y9rnB6tGnbHM9dMVsveyfPojm+i0" |
| |
| pagure.lib.query.add_sshkey_to_project_or_user( |
| self.session, |
| user_key, |
| pushaccess=True, |
| creator=pingou, |
| user=pingou, |
| ) |
| pagure.lib.query.add_sshkey_to_project_or_user( |
| self.session, |
| project_key, |
| pushaccess=True, |
| creator=pingou, |
| project=repo, |
| ) |
| self.session.commit() |
| |
| url = "/pv/ssh/lookupkey/" |
| |
| output = self.app.post(url, data={"search_key": "asdf"}) |
| self.assertEqual(output.status_code, 200) |
| result = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(result["found"], False) |
| |
| output = self.app.post(url, data={"search_key": user_key_fp}) |
| self.assertEqual(output.status_code, 200) |
| result = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(result["found"], True) |
| self.assertEqual(result["username"], "pingou") |
| self.assertEqual(result["public_key"], user_key) |
| |
| output = self.app.post( |
| url, data={"search_key": user_key_fp, "username": "pingou"} |
| ) |
| self.assertEqual(output.status_code, 200) |
| result = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(result["found"], True) |
| self.assertEqual(result["username"], "pingou") |
| self.assertEqual(result["public_key"], user_key) |
| |
| output = self.app.post( |
| url, data={"search_key": user_key_fp, "username": "foo"} |
| ) |
| self.assertEqual(output.status_code, 200) |
| result = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(result["found"], False) |
| |
| output = self.app.post(url, data={"search_key": project_key_fp}) |
| self.assertEqual(output.status_code, 200) |
| result = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(result["found"], True) |
| self.assertEqual(result["username"], "deploykey_test_2") |
| self.assertEqual(result["public_key"], project_key) |
| |
| @patch.dict( |
| "pagure.config.config", |
| {"REPOSPANNER_REGIONS": {"region": {"repo_prefix": "prefix"}}}, |
| ) |
| def test_check_ssh_access(self): |
| """ Test the SSH access check endpoint. """ |
| tests.create_projects(self.session) |
| self.session.query(pagure.lib.model.Project).filter( |
| pagure.lib.model.Project.name == "test2" |
| ).update({pagure.lib.model.Project.repospanner_region: "region"}) |
| self.session.commit() |
| |
| url = "/pv/ssh/checkaccess/" |
| |
| def runtest(project, username, access): |
| output = self.app.post( |
| url, data={"gitdir": project, "username": username} |
| ) |
| self.assertEqual(output.status_code, 200) |
| result = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(result["access"], access) |
| return result |
| |
| runtest("project.git", "pingou", False) |
| i1 = runtest("test.git", "pingou", True) |
| i2 = runtest("test.git", "foo", True) |
| i3 = runtest("tickets/test.git", "pingou", True) |
| i4 = runtest("test2.git", "pingou", True) |
| runtest("tickets/test.git", "foo", False) |
| |
| self.assertEqual(i1["reponame"], "test.git") |
| self.assertEqual(i1["repospanner_reponame"], None) |
| self.assertEqual(i1["repotype"], "main") |
| self.assertEqual(i1["region"], None) |
| self.assertEqual(i1["project_name"], "test") |
| self.assertEqual(i1["project_user"], None) |
| self.assertEqual(i1["project_namespace"], None) |
| self.assertEqual(i1, i2) |
| self.assertEqual(i3["reponame"], "tickets/test.git") |
| self.assertEqual(i3["repospanner_reponame"], None) |
| self.assertEqual(i3["repotype"], "tickets") |
| self.assertEqual(i3["region"], None) |
| self.assertEqual(i3["project_name"], "test") |
| self.assertEqual(i3["project_user"], None) |
| self.assertEqual(i3["project_namespace"], None) |
| self.assertEqual(i4["repospanner_reponame"], "prefix/main/test2") |
| self.assertEqual(i4["region"], "region") |
| |
| |
| if __name__ == "__main__": |
| unittest.main(verbosity=2) |