| |
| |
| """ |
| (c) 2015 - Copyright Red Hat Inc |
| |
| Authors: |
| Pierre-Yves Chibon <pingou@pingoured.fr> |
| |
| """ |
| |
| from __future__ import unicode_literals, absolute_import |
| |
| import datetime |
| import unittest |
| import shutil |
| import sys |
| import os |
| |
| import json |
| from mock import patch, MagicMock |
| |
| sys.path.insert( |
| 0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") |
| ) |
| |
| import pagure.lib.query |
| import pagure.default_config |
| import tests |
| |
| |
| class PagureFlaskApiForktests(tests.Modeltests): |
| """ Tests for the flask API of pagure for issue """ |
| |
| maxDiff = None |
| |
| def setUp(self): |
| """ Set up the environnment, ran before every tests. """ |
| super(PagureFlaskApiForktests, self).setUp() |
| |
| pagure.config.config["REQUESTS_FOLDER"] = None |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_views_pr_disabled(self): |
| """ Test the api_pull_request_views method of the flask api when PR |
| are disabled. """ |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| settings = repo.settings |
| settings["pull_requests"] = False |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| output = self.app.get("/api/0/test/pull-requests") |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Pull-Request have been deactivated for this project", |
| "error_code": "EPULLREQUESTSDISABLED", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_views_pr_closed(self): |
| """ Test the api_pull_request_views method of the flask api to list |
| the closed PRs. """ |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| output = self.app.get("/api/0/test/pull-requests?status=closed") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| for k in ["first", "last"]: |
| self.assertIsNotNone(data["pagination"][k]) |
| data["pagination"][k] = "http://localhost..." |
| self.assertDictEqual( |
| data, |
| { |
| "args": { |
| "assignee": None, |
| "author": None, |
| "page": 1, |
| "per_page": 20, |
| "status": "closed", |
| }, |
| "pagination": { |
| "first": "http://localhost...", |
| "last": "http://localhost...", |
| "next": None, |
| "page": 1, |
| "pages": 0, |
| "per_page": 20, |
| "prev": None, |
| }, |
| "requests": [], |
| "total_requests": 0, |
| }, |
| ) |
| |
| |
| pagure.lib.query.close_pull_request( |
| self.session, request=req, user="pingou", merged=False |
| ) |
| |
| output = self.app.get("/api/0/test/pull-requests?status=closed") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| sorted(data.keys()), |
| ["args", "pagination", "requests", "total_requests"], |
| ) |
| self.assertDictEqual( |
| data["args"], |
| { |
| "assignee": None, |
| "author": None, |
| "page": 1, |
| "per_page": 20, |
| "status": "closed", |
| }, |
| ) |
| self.assertEqual(data["total_requests"], 1) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="closed pullrequest by user foo on repo test", |
| user="foo", |
| status="Closed", |
| ) |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="closed pullrequest by user pingou on repo test", |
| user="pingou", |
| status="Closed", |
| ) |
| self.session.commit() |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="merged pullrequest by user pingou on repo test", |
| user="pingou", |
| status="Merged", |
| ) |
| self.session.commit() |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="merged pullrequest by user foo on repo test", |
| user="foo", |
| status="Merged", |
| ) |
| self.session.commit() |
| |
| |
| output = self.app.get("/api/0/test/pull-requests?status=closed") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| |
| self.assertEqual(len(data["requests"]), 3) |
| self.assertEqual( |
| sorted(data.keys()), |
| ["args", "pagination", "requests", "total_requests"], |
| ) |
| for req in data["requests"]: |
| self.assertEqual(req["status"], "Closed") |
| self.assertEqual(data["args"]["status"], "closed") |
| self.assertEqual(data["args"]["page"], 1) |
| |
| self.assertEqual(data["total_requests"], 3) |
| |
| |
| output = self.app.get("/api/0/test/pull-requests?status=merged") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| |
| self.assertEqual(len(data["requests"]), 2) |
| self.assertEqual( |
| sorted(data.keys()), |
| ["args", "pagination", "requests", "total_requests"], |
| ) |
| for req in data["requests"]: |
| self.assertEqual(req["status"], "Merged") |
| self.assertEqual(data["args"]["status"], "merged") |
| self.assertEqual(data["args"]["page"], 1) |
| self.assertEqual(data["total_requests"], 2) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_views_all_pr(self): |
| """ Test the api_pull_request_views method of the flask api to list |
| all PRs. """ |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| output = self.app.get("/api/0/test/pull-requests?status=all") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| sorted(data.keys()), |
| ["args", "pagination", "requests", "total_requests"], |
| ) |
| self.assertDictEqual( |
| data["args"], |
| { |
| "assignee": None, |
| "author": None, |
| "page": 1, |
| "per_page": 20, |
| "status": "all", |
| }, |
| ) |
| self.assertEqual(data["total_requests"], 1) |
| |
| |
| pagure.lib.query.close_pull_request( |
| self.session, request=req, user="pingou", merged=False |
| ) |
| |
| output = self.app.get("/api/0/test/pull-requests?status=all") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| sorted(data.keys()), |
| ["args", "pagination", "requests", "total_requests"], |
| ) |
| self.assertDictEqual( |
| data["args"], |
| { |
| "assignee": None, |
| "author": None, |
| "page": 1, |
| "per_page": 20, |
| "status": "all", |
| }, |
| ) |
| self.assertEqual(data["total_requests"], 1) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_views(self, send_email): |
| """ Test the api_pull_request_views method of the flask api. """ |
| send_email.return_value = True |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| |
| output = self.app.get("/api/0/foo/pull-requests") |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Project not found", "error_code": "ENOPROJECT"} |
| ) |
| |
| |
| output = self.app.get("/api/0/test/pull-requests") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data["requests"][0]["date_created"] = "1431414800" |
| data["requests"][0]["updated_on"] = "1431414800" |
| data["requests"][0]["project"]["date_created"] = "1431414800" |
| data["requests"][0]["project"]["date_modified"] = "1431414800" |
| data["requests"][0]["repo_from"]["date_created"] = "1431414800" |
| data["requests"][0]["repo_from"]["date_modified"] = "1431414800" |
| data["requests"][0]["uid"] = "1431414800" |
| data["requests"][0]["last_updated"] = "1431414800" |
| for k in ["first", "last"]: |
| self.assertIsNotNone(data["pagination"][k]) |
| data["pagination"][k] = "http://localhost..." |
| expected_data = { |
| "args": { |
| "assignee": None, |
| "author": None, |
| "page": 1, |
| "per_page": 20, |
| "status": True, |
| }, |
| "pagination": { |
| "first": "http://localhost...", |
| "last": "http://localhost...", |
| "next": None, |
| "page": 1, |
| "pages": 1, |
| "per_page": 20, |
| "prev": None, |
| }, |
| "requests": [ |
| { |
| "assignee": None, |
| "branch": "master", |
| "branch_from": "master", |
| "cached_merge_status": "unknown", |
| "closed_at": None, |
| "closed_by": None, |
| "comments": [], |
| "commit_start": None, |
| "commit_stop": None, |
| "date_created": "1431414800", |
| "id": 1, |
| "initial_comment": None, |
| "last_updated": "1431414800", |
| "project": { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [], |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [], |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ], |
| "custom_keys": [], |
| "date_created": "1431414800", |
| "date_modified": "1431414800", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| "remote_git": None, |
| "repo_from": { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [], |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [], |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ], |
| "custom_keys": [], |
| "date_created": "1431414800", |
| "date_modified": "1431414800", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| "status": "Open", |
| "tags": [], |
| "threshold_reached": None, |
| "title": "test pull-request", |
| "uid": "1431414800", |
| "updated_on": "1431414800", |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| } |
| ], |
| "total_requests": 1, |
| } |
| self.assertDictEqual(data, expected_data) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.get("/api/0/test/pull-requests", headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data2 = json.loads(output.get_data(as_text=True)) |
| data2["requests"][0]["date_created"] = "1431414800" |
| data2["requests"][0]["updated_on"] = "1431414800" |
| data2["requests"][0]["project"]["date_created"] = "1431414800" |
| data2["requests"][0]["project"]["date_modified"] = "1431414800" |
| data2["requests"][0]["repo_from"]["date_created"] = "1431414800" |
| data2["requests"][0]["repo_from"]["date_modified"] = "1431414800" |
| data2["requests"][0]["uid"] = "1431414800" |
| data2["requests"][0]["last_updated"] = "1431414800" |
| for k in ["first", "last"]: |
| self.assertIsNotNone(data["pagination"][k]) |
| data2["pagination"][k] = "http://localhost..." |
| self.assertDictEqual(data, data2) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_view_pr_disabled(self, send_email): |
| """ Test the api_pull_request_view method of the flask api. """ |
| send_email.return_value = True |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| settings = repo.settings |
| settings["pull_requests"] = False |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| output = self.app.get("/api/0/test/pull-request/1") |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Pull-Request have been deactivated for this project", |
| "error_code": "EPULLREQUESTSDISABLED", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_view(self, send_email): |
| """ Test the api_pull_request_view method of the flask api. """ |
| send_email.return_value = True |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| |
| output = self.app.get("/api/0/foo/pull-request/1") |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Project not found", "error_code": "ENOPROJECT"} |
| ) |
| |
| |
| output = self.app.get("/api/0/test2/pull-request/1") |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Pull-Request not found", "error_code": "ENOREQ"} |
| ) |
| |
| |
| output = self.app.get("/api/0/test/pull-request/1") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data["date_created"] = "1431414800" |
| data["updated_on"] = "1431414800" |
| data["project"]["date_created"] = "1431414800" |
| data["project"]["date_modified"] = "1431414800" |
| data["repo_from"]["date_created"] = "1431414800" |
| data["repo_from"]["date_modified"] = "1431414800" |
| data["uid"] = "1431414800" |
| data["last_updated"] = "1431414800" |
| expected_data = { |
| "assignee": None, |
| "branch": "master", |
| "branch_from": "master", |
| "cached_merge_status": "unknown", |
| "closed_at": None, |
| "closed_by": None, |
| "comments": [], |
| "commit_start": None, |
| "commit_stop": None, |
| "date_created": "1431414800", |
| "id": 1, |
| "initial_comment": None, |
| "last_updated": "1431414800", |
| "project": { |
| "access_groups": {"admin": [], "commit": [], "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [], |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ], |
| "custom_keys": [], |
| "date_created": "1431414800", |
| "date_modified": "1431414800", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| "remote_git": None, |
| "repo_from": { |
| "access_groups": {"admin": [], "commit": [], "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [], |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ], |
| "custom_keys": [], |
| "date_created": "1431414800", |
| "date_modified": "1431414800", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| "status": "Open", |
| "tags": [], |
| "threshold_reached": None, |
| "title": "test pull-request", |
| "uid": "1431414800", |
| "updated_on": "1431414800", |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| } |
| self.assertDictEqual(data, expected_data) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.get("/api/0/test/pull-request/1", headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data2 = json.loads(output.get_data(as_text=True)) |
| data2["date_created"] = "1431414800" |
| data2["project"]["date_created"] = "1431414800" |
| data2["project"]["date_modified"] = "1431414800" |
| data2["repo_from"]["date_created"] = "1431414800" |
| data2["repo_from"]["date_modified"] = "1431414800" |
| data2["uid"] = "1431414800" |
| data2["date_created"] = "1431414800" |
| data2["updated_on"] = "1431414800" |
| data2["last_updated"] = "1431414800" |
| self.assertDictEqual(data, data2) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_by_uid_view(self, send_email): |
| """ Test the api_pull_request_by_uid_view method of the flask api. """ |
| send_email.return_value = True |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| uid = req.uid |
| |
| |
| output = self.app.get("/api/0/pull-requests/{}".format(uid + "aaa")) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Pull-Request not found", "error_code": "ENOREQ"} |
| ) |
| |
| |
| output = self.app.get("/api/0/pull-requests/{}".format(uid)) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data["date_created"] = "1431414800" |
| data["updated_on"] = "1431414800" |
| data["project"]["date_created"] = "1431414800" |
| data["project"]["date_modified"] = "1431414800" |
| data["repo_from"]["date_created"] = "1431414800" |
| data["repo_from"]["date_modified"] = "1431414800" |
| data["last_updated"] = "1431414800" |
| expected_data = { |
| "assignee": None, |
| "branch": "master", |
| "branch_from": "master", |
| "cached_merge_status": "unknown", |
| "closed_at": None, |
| "closed_by": None, |
| "comments": [], |
| "commit_start": None, |
| "commit_stop": None, |
| "date_created": "1431414800", |
| "id": 1, |
| "initial_comment": None, |
| "last_updated": "1431414800", |
| "project": { |
| "access_groups": {"admin": [], "commit": [], "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [], |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ], |
| "custom_keys": [], |
| "date_created": "1431414800", |
| "date_modified": "1431414800", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| "remote_git": None, |
| "repo_from": { |
| "access_groups": {"admin": [], "commit": [], "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [], |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ], |
| "custom_keys": [], |
| "date_created": "1431414800", |
| "date_modified": "1431414800", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| "status": "Open", |
| "tags": [], |
| "threshold_reached": None, |
| "title": "test pull-request", |
| "uid": uid, |
| "updated_on": "1431414800", |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| } |
| self.assertDictEqual(data, expected_data) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.get( |
| "/api/0/pull-requests/{}".format(uid), headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data2 = json.loads(output.get_data(as_text=True)) |
| data2["date_created"] = "1431414800" |
| data2["project"]["date_created"] = "1431414800" |
| data2["project"]["date_modified"] = "1431414800" |
| data2["repo_from"]["date_created"] = "1431414800" |
| data2["repo_from"]["date_modified"] = "1431414800" |
| data2["date_created"] = "1431414800" |
| data2["updated_on"] = "1431414800" |
| data2["last_updated"] = "1431414800" |
| self.assertDictEqual(data, data2) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_close_pr_disabled(self, send_email): |
| """ Test the api_pull_request_close method of the flask api. """ |
| send_email.return_value = True |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| settings = repo.settings |
| settings["pull_requests"] = False |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/close", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Pull-Request have been deactivated for this project", |
| "error_code": "EPULLREQUESTSDISABLED", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_close(self, send_email): |
| """ Test the api_pull_request_close method of the flask api. """ |
| send_email.return_value = True |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.post( |
| "/api/0/foo/pull-request/1/close", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Project not found", "error_code": "ENOPROJECT"} |
| ) |
| |
| |
| output = self.app.post( |
| "/api/0/test2/pull-request/1/close", headers=headers |
| ) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"] |
| ) |
| self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"]) |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/2/close", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Pull-Request not found", "error_code": "ENOREQ"} |
| ) |
| |
| |
| item = pagure.lib.model.Token( |
| id="foobar_token", |
| user_id=2, |
| project_id=1, |
| expiration=datetime.datetime.utcnow() |
| + datetime.timedelta(days=30), |
| ) |
| self.session.add(item) |
| self.session.commit() |
| |
| |
| acls = pagure.lib.query.get_acls(self.session) |
| for acl in acls: |
| if acl.name == "pull_request_close": |
| break |
| item = pagure.lib.model.TokenAcl( |
| token_id="foobar_token", acl_id=acl.id |
| ) |
| self.session.add(item) |
| self.session.commit() |
| |
| headers = {"Authorization": "token foobar_token"} |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/close", headers=headers |
| ) |
| self.assertEqual(output.status_code, 403) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "You are not allowed to merge/close pull-request " |
| "for this project", |
| "error_code": "ENOPRCLOSE", |
| }, |
| ) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/close", headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Pull-request closed!"}) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_merge_pr_disabled(self, send_email): |
| """ Test the api_pull_request_merge method of the flask api when PR |
| are disabled. """ |
| send_email.return_value = True |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="test", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| settings = repo.settings |
| settings["pull_requests"] = False |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Pull-Request have been deactivated for this project", |
| "error_code": "EPULLREQUESTSDISABLED", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_merge_only_assigned(self, send_email): |
| """ Test the api_pull_request_merge method of the flask api when |
| only assignee can merge the PR and the PR isn't assigned. """ |
| send_email.return_value = True |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="test", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| settings = repo.settings |
| settings["Only_assignee_can_merge_pull-request"] = True |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 403) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "This request must be assigned to be merged", |
| "error_code": "ENOTASSIGNED", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_merge_only_assigned_not_assignee( |
| self, send_email |
| ): |
| """ Test the api_pull_request_merge method of the flask api when |
| only assignee can merge the PR and the PR isn't assigned to the |
| user asking to merge. """ |
| send_email.return_value = True |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="test", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| req.assignee = pagure.lib.query.search_user(self.session, "foo") |
| self.session.add(req) |
| self.session.commit() |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| settings = repo.settings |
| settings["Only_assignee_can_merge_pull-request"] = True |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 403) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Only the assignee can merge this review", |
| "error_code": "ENOTASSIGNEE", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_merge_minimal_score(self, send_email): |
| """ Test the api_pull_request_merge method of the flask api when |
| a PR requires a certain minimal score to be merged. """ |
| send_email.return_value = True |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="test", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| settings = repo.settings |
| settings["Minimum_score_to_merge_pull-request"] = 2 |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 403) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "This request does not have the minimum review " |
| "score necessary to be merged", |
| "error_code": "EPRSCORE", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_merge(self, send_email): |
| """ Test the api_pull_request_merge method of the flask api. """ |
| send_email.return_value = True |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="test", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.post( |
| "/api/0/foo/pull-request/1/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Project not found", "error_code": "ENOPROJECT"} |
| ) |
| |
| |
| output = self.app.post( |
| "/api/0/test2/pull-request/1/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"] |
| ) |
| self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"]) |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/2/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Pull-Request not found", "error_code": "ENOREQ"} |
| ) |
| |
| |
| item = pagure.lib.model.Token( |
| id="foobar_token", |
| user_id=2, |
| project_id=1, |
| expiration=datetime.datetime.utcnow() |
| + datetime.timedelta(days=30), |
| ) |
| self.session.add(item) |
| self.session.commit() |
| |
| |
| acls = pagure.lib.query.get_acls(self.session) |
| for acl in acls: |
| if acl.name == "pull_request_merge": |
| break |
| item = pagure.lib.model.TokenAcl( |
| token_id="foobar_token", acl_id=acl.id |
| ) |
| self.session.add(item) |
| self.session.commit() |
| |
| headers = {"Authorization": "token foobar_token"} |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 403) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "You are not allowed to merge/close pull-request " |
| "for this project", |
| "error_code": "ENOPRCLOSE", |
| }, |
| ) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Changes merged!"}) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_merge_conflicting(self, send_email): |
| """ Test the api_pull_request_merge method of the flask api. """ |
| send_email.return_value = True |
| |
| tests.create_projects(self.session) |
| tests.add_content_git_repo( |
| os.path.join(self.path, "repos", "test.git") |
| ) |
| |
| |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| task = pagure.lib.query.fork_project( |
| session=self.session, user="pingou", repo=project |
| ) |
| self.session.commit() |
| self.assertEqual( |
| task.get(), |
| { |
| "endpoint": "ui_ns.view_repo", |
| "repo": "test", |
| "namespace": None, |
| "username": "pingou", |
| }, |
| ) |
| |
| |
| tests.add_content_to_git( |
| os.path.join(self.path, "repos", "forks", "pingou", "test.git"), |
| filename="foobar", |
| content="content from the fork", |
| ) |
| |
| |
| tests.add_content_to_git( |
| os.path.join(self.path, "repos", "test.git"), |
| filename="foobar", |
| content="content from the main repo", |
| ) |
| |
| project = pagure.lib.query.get_authorized_project(self.session, "test") |
| fork = pagure.lib.query.get_authorized_project( |
| self.session, "test", user="pingou" |
| ) |
| |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=fork, |
| branch_from="master", |
| repo_to=project, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 409) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "This pull-request conflicts and thus cannot be merged", |
| "error_code": "EPRCONFLICTS", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_merge_user_token(self, send_email): |
| """ Test the api_pull_request_merge method of the flask api. """ |
| send_email.return_value = True |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="test", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.post( |
| "/api/0/foo/pull-request/1/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Project not found", "error_code": "ENOPROJECT"} |
| ) |
| |
| |
| output = self.app.post( |
| "/api/0/test2/pull-request/1/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Pull-Request not found", "error_code": "ENOREQ"} |
| ) |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/2/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Pull-Request not found", "error_code": "ENOREQ"} |
| ) |
| |
| |
| item = pagure.lib.model.Token( |
| id="foobar_token", |
| user_id=2, |
| project_id=1, |
| expiration=datetime.datetime.utcnow() |
| + datetime.timedelta(days=30), |
| ) |
| self.session.add(item) |
| self.session.commit() |
| |
| |
| acls = pagure.lib.query.get_acls(self.session) |
| acl = None |
| for acl in acls: |
| if acl.name == "pull_request_merge": |
| break |
| item = pagure.lib.model.TokenAcl( |
| token_id="foobar_token", acl_id=acl.id |
| ) |
| self.session.add(item) |
| self.session.commit() |
| |
| headers = {"Authorization": "token foobar_token"} |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 403) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "You are not allowed to merge/close pull-request " |
| "for this project", |
| "error_code": "ENOPRCLOSE", |
| }, |
| ) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/merge", headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Changes merged!"}) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_add_comment(self, mockemail): |
| """ Test the api_pull_request_add_comment method of the flask api. """ |
| mockemail.return_value = True |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.post( |
| "/api/0/foo/pull-request/1/comment", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Project not found", "error_code": "ENOPROJECT"} |
| ) |
| |
| |
| output = self.app.post( |
| "/api/0/test2/pull-request/1/comment", headers=headers |
| ) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"] |
| ) |
| self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"]) |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Pull-Request not found", "error_code": "ENOREQ"} |
| ) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| |
| self.session.commit() |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual(len(request.comments), 0) |
| |
| data = {"title": "test issue"} |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": {"comment": ["This field is required."]}, |
| }, |
| ) |
| |
| |
| self.session.commit() |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual(len(request.comments), 0) |
| |
| data = {"comment": "This is a very interesting question"} |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Comment added"}) |
| |
| |
| self.session.commit() |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual(len(request.comments), 1) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_add_comment_wrong_user(self): |
| """ Test the api_pull_request_add_comment method of the flask api |
| when the user is not found in the DB. """ |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| data = {"comment": "This is a very interesting question"} |
| |
| |
| with patch( |
| "pagure.lib.query.add_pull_request_comment", |
| side_effect=pagure.exceptions.PagureException("error"), |
| ): |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", |
| data=data, |
| headers=headers, |
| ) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "error", "error_code": "ENOCODE"} |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_add_comment_pr_disabled(self): |
| """ Test the api_pull_request_add_comment method of the flask api |
| when PRs are disabled. """ |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| settings = repo.settings |
| settings["pull_requests"] = False |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| data = {"comment": "This is a very interesting question"} |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Pull-Request have been deactivated for this project", |
| "error_code": "EPULLREQUESTSDISABLED", |
| }, |
| ) |
| |
| |
| self.session.commit() |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual(len(request.comments), 0) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_pull_request_add_comment_user_token(self, mockemail): |
| """ Test the api_pull_request_add_comment method of the flask api. """ |
| mockemail.return_value = True |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.post( |
| "/api/0/foo/pull-request/1/comment", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Project not found", "error_code": "ENOPROJECT"} |
| ) |
| |
| |
| output = self.app.post( |
| "/api/0/test2/pull-request/1/comment", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Pull-Request not found", "error_code": "ENOREQ"} |
| ) |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Pull-Request not found", "error_code": "ENOREQ"} |
| ) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| forked_repo = pagure.lib.query.get_authorized_project( |
| self.session, "test" |
| ) |
| req = pagure.lib.query.new_pull_request( |
| session=self.session, |
| repo_from=forked_repo, |
| branch_from="master", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| |
| self.session.commit() |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual(len(request.comments), 0) |
| |
| data = {"title": "test issue"} |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": {"comment": ["This field is required."]}, |
| }, |
| ) |
| |
| |
| self.session.commit() |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual(len(request.comments), 0) |
| |
| data = {"comment": "This is a very interesting question"} |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Comment added"}) |
| |
| |
| self.session.commit() |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual(len(request.comments), 1) |
| |
| @patch("pagure.lib.notify.send_email") |
| def test_api_subscribe_pull_request_pr_disabled(self, p_send_email): |
| """ Test the api_subscribe_pull_request method of the flask api. """ |
| p_send_email.return_value = True |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| settings = repo.settings |
| settings["pull_requests"] = False |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/subscribe", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Pull-Request have been deactivated for this project", |
| "error_code": "EPULLREQUESTSDISABLED", |
| }, |
| ) |
| |
| @patch("pagure.lib.git.update_git") |
| @patch("pagure.lib.notify.send_email") |
| def test_api_subscribe_pull_request_invalid_token( |
| self, p_send_email, p_ugt |
| ): |
| """ Test the api_subscribe_pull_request method of the flask api. """ |
| p_send_email.return_value = True |
| p_ugt.return_value = True |
| |
| item = pagure.lib.model.User( |
| user="bar", |
| fullname="bar foo", |
| password="foo", |
| default_email="bar@bar.com", |
| ) |
| self.session.add(item) |
| item = pagure.lib.model.UserEmail(user_id=3, email="bar@bar.com") |
| self.session.add(item) |
| |
| self.session.commit() |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, user_id=3, project_id=2) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| 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="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual( |
| pagure.lib.query.get_watch_list(self.session, request), |
| set(["pingou"]), |
| ) |
| |
| data = {} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/subscribe", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or expired token. Please visit " |
| "http://localhost.localdomain/settings#api-keys to get or " |
| "renew your API token.", |
| "error_code": "EINVALIDTOK", |
| }, |
| ) |
| |
| @patch("pagure.lib.git.update_git") |
| @patch("pagure.lib.notify.send_email") |
| def test_api_subscribe_pull_request(self, p_send_email, p_ugt): |
| """ Test the api_subscribe_pull_request method of the flask api. """ |
| p_send_email.return_value = True |
| p_ugt.return_value = True |
| |
| item = pagure.lib.model.User( |
| user="bar", |
| fullname="bar foo", |
| password="foo", |
| default_email="bar@bar.com", |
| ) |
| self.session.add(item) |
| item = pagure.lib.model.UserEmail(user_id=3, email="bar@bar.com") |
| self.session.add(item) |
| |
| self.session.commit() |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, user_id=3) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| |
| output = self.app.post( |
| "/api/0/foo/pull-request/1/subscribe", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Project not found", "error_code": "ENOPROJECT"} |
| ) |
| |
| |
| output = self.app.post( |
| "/api/0/test2/pull-request/1/subscribe", headers=headers |
| ) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"] |
| ) |
| self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"]) |
| |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/1/subscribe", headers=headers |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Pull-Request not found", "error_code": "ENOREQ"} |
| ) |
| |
| |
| 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="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual( |
| pagure.lib.query.get_watch_list(self.session, request), |
| set(["pingou"]), |
| ) |
| |
| |
| data = {} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/subscribe", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." |
| self.assertDictEqual( |
| data, |
| { |
| "message": "You are no longer watching this pull-request", |
| "avatar_url": "https://seccdn.libravatar.org/avatar/...", |
| "user": "bar", |
| }, |
| ) |
| |
| data = {} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/subscribe", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." |
| self.assertDictEqual( |
| data, |
| { |
| "message": "You are no longer watching this pull-request", |
| "avatar_url": "https://seccdn.libravatar.org/avatar/...", |
| "user": "bar", |
| }, |
| ) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual( |
| pagure.lib.query.get_watch_list(self.session, request), |
| set(["pingou"]), |
| ) |
| |
| |
| data = {"status": True} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/subscribe", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." |
| self.assertDictEqual( |
| data, |
| { |
| "message": "You are now watching this pull-request", |
| "avatar_url": "https://seccdn.libravatar.org/avatar/...", |
| "user": "bar", |
| }, |
| ) |
| |
| |
| data = {"status": True} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/subscribe", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." |
| self.assertDictEqual( |
| data, |
| { |
| "message": "You are now watching this pull-request", |
| "avatar_url": "https://seccdn.libravatar.org/avatar/...", |
| "user": "bar", |
| }, |
| ) |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual( |
| pagure.lib.query.get_watch_list(self.session, request), |
| set(["pingou", "bar"]), |
| ) |
| |
| |
| data = {} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/subscribe", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." |
| self.assertDictEqual( |
| data, |
| { |
| "message": "You are no longer watching this pull-request", |
| "avatar_url": "https://seccdn.libravatar.org/avatar/...", |
| "user": "bar", |
| }, |
| ) |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual( |
| pagure.lib.query.get_watch_list(self.session, request), |
| set(["pingou"]), |
| ) |
| |
| @patch("pagure.lib.git.update_git") |
| @patch("pagure.lib.notify.send_email") |
| def test_api_subscribe_pull_request_logged_in(self, p_send_email, p_ugt): |
| """ Test the api_subscribe_pull_request method of the flask api |
| when the user is logged in via the UI. """ |
| p_send_email.return_value = True |
| p_ugt.return_value = True |
| |
| item = pagure.lib.model.User( |
| user="bar", |
| fullname="bar foo", |
| password="foo", |
| default_email="bar@bar.com", |
| ) |
| self.session.add(item) |
| item = pagure.lib.model.UserEmail(user_id=3, email="bar@bar.com") |
| self.session.add(item) |
| |
| self.session.commit() |
| |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, user_id=3) |
| tests.create_tokens_acl(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="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual( |
| pagure.lib.query.get_watch_list(self.session, request), |
| set(["pingou"]), |
| ) |
| |
| |
| user = tests.FakeUser(username="foo") |
| with tests.user_set(self.app.application, user): |
| data = {"status": True} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/subscribe", data=data |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." |
| self.assertDictEqual( |
| data, |
| { |
| "message": "You are now watching this pull-request", |
| "avatar_url": "https://seccdn.libravatar.org/avatar/...", |
| "user": "foo", |
| }, |
| ) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| request = pagure.lib.query.search_pull_requests( |
| self.session, project_id=1, requestid=1 |
| ) |
| self.assertEqual( |
| pagure.lib.query.get_watch_list(self.session, request), |
| set(["pingou", "foo"]), |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_invalid_project(self): |
| """ Test the api_pull_request_create method of the flask api when |
| not the project doesn't exist. |
| """ |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "branch_to": "master", |
| "branch_from": "test", |
| } |
| |
| output = self.app.post( |
| "/api/0/foobar/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, {"error": "Project not found", "error_code": "ENOPROJECT"} |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_missing_title(self): |
| """ Test the api_pull_request_create method of the flask api when |
| not title is submitted. |
| """ |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "branch_to": "master", |
| "branch_from": "test", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": {"title": ["This field is required."]}, |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_missing_branch_to(self): |
| """ Test the api_pull_request_create method of the flask api when |
| not branch to is submitted. |
| """ |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "title": "Test PR", |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "branch_from": "test", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": {"branch_to": ["This field is required."]}, |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_missing_branch_from(self): |
| """ Test the api_pull_request_create method of the flask api when |
| not branch from is submitted. |
| """ |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "title": "Test PR", |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "branch_to": "master", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": {"branch_from": ["This field is required."]}, |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_pr_disabled(self): |
| """ Test the api_pull_request_create method of the flask api when |
| the parent repo disabled pull-requests. |
| """ |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| settings = repo.settings |
| settings["pull_requests"] = False |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "title": "Test PR", |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "branch_to": "master", |
| "branch_from": "test", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Pull-Request have been deactivated for this project", |
| "error_code": "EPULLREQUESTSDISABLED", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_signed_pr(self): |
| """ Test the api_pull_request_create method of the flask api when |
| the parent repo enforces signed-off pull-requests. |
| """ |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| settings = repo.settings |
| settings["Enforce_signed-off_commits_in_pull-request"] = True |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "title": "Test PR", |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "branch_to": "master", |
| "branch_from": "test", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "This repo enforces that all commits are signed " |
| "off by their author.", |
| "error_code": "ENOSIGNEDOFF", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_invalid_branch_from(self): |
| """ Test the api_pull_request_create method of the flask api when |
| the branch from does not exist. |
| """ |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| settings = repo.settings |
| settings["Enforce_signed-off_commits_in_pull-request"] = True |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "title": "Test PR", |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "branch_to": "master", |
| "branch_from": "foobarbaz", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": "Branch foobarbaz does not exist", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_invalid_token(self): |
| """ Test the api_pull_request_create method of the flask api when |
| queried with an invalid token. |
| """ |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "title": "Test PR", |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "branch_to": "master", |
| "branch_from": "foobarbaz", |
| } |
| |
| output = self.app.post( |
| "/api/0/test2/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or expired token. Please visit " |
| "http://localhost.localdomain/settings#api-keys to get or " |
| "renew your API token.", |
| "error_code": "EINVALIDTOK", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_invalid_access(self): |
| """ Test the api_pull_request_create method of the flask api when |
| the user opening the PR doesn't have commit access. |
| """ |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session, user_id=2) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "title": "Test PR", |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "branch_to": "master", |
| "branch_from": "foobarbaz", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "You do not have sufficient permissions to " |
| "perform this action", |
| "error_code": "ENOTHIGHENOUGH", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_invalid_branch_to(self): |
| """ Test the api_pull_request_create method of the flask api when |
| the branch to does not exist. |
| """ |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| settings = repo.settings |
| settings["Enforce_signed-off_commits_in_pull-request"] = True |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "title": "Test PR", |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "branch_to": "foobarbaz", |
| "branch_from": "test", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": "Branch foobarbaz could not be found in the " |
| "target repo", |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_project_token_different_project(self): |
| """Test the api_pull_request_create method with the project token |
| of a different project - fails""" |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session, project_id=2) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token foo_token"} |
| data = { |
| "title": "Test of PR", |
| "inicial comment": "Some readme adjustment", |
| "branch_to": "master", |
| "branch_from": "test", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 401) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_user_token_invalid_acls(self): |
| """Test the api_pull_request_create method with the user token, but with |
| no acls for opening pull request - fails""" |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session, project_id=None) |
| for acl in ( |
| "create_project", |
| "fork_project", |
| "modify_project", |
| "update_watch_status", |
| ): |
| tests.create_tokens_acl(self.session, acl_name=acl) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "title": "Test of PR", |
| "initial_comment": "Some readme adjustment", |
| "branch_to": "master", |
| "branch_from": "test", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 401) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_from_branch_to_origin(self): |
| """Test the api_pull_request_create method from a fork to a master, |
| with project token of a origin with all the acls""" |
| |
| tests.create_projects(self.session) |
| tests.create_projects( |
| self.session, is_fork=True, hook_token_suffix="foo" |
| ) |
| project_query = self.session.query(pagure.lib.model.Project) |
| for project in project_query.filter_by(name="test").all(): |
| if project.parent_id == None: |
| parent = project |
| else: |
| child = project |
| tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) |
| tests.create_projects_git( |
| os.path.join(self.path, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo( |
| os.path.join(self.path, "repos", "forks", "pingou", "test.git"), |
| branch="branch", |
| ) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "forks", "pingou", "test.git"), |
| branch="branch", |
| ) |
| |
| |
| parent_token = pagure.lib.model.Token( |
| id="iamparenttoken", |
| user_id=parent.user_id, |
| project_id=parent.id, |
| expiration=datetime.datetime.utcnow() |
| + datetime.timedelta(days=30), |
| ) |
| self.session.add(parent_token) |
| |
| fork_token = pagure.lib.model.Token( |
| id="iamforktoken", |
| user_id=child.user_id, |
| project_id=child.id, |
| expiration=datetime.datetime.utcnow() |
| + datetime.timedelta(days=30), |
| ) |
| self.session.add(fork_token) |
| self.session.commit() |
| |
| tests.create_tokens_acl(self.session, token_id="iamparenttoken") |
| for acl in pagure.default_config.CROSS_PROJECT_ACLS: |
| tests.create_tokens_acl( |
| self.session, token_id="iamforktoken", acl_name=acl |
| ) |
| |
| headers = {"Authorization": "token iamforktoken"} |
| |
| data = { |
| "title": "war of tomatoes", |
| "initial_comment": "the manifest", |
| "branch_to": "master", |
| "branch_from": "branch", |
| "repo_from": "test", |
| "repo_from_username": "pingou", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 200) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open(self): |
| """ Test the api_pull_request_create method of the flask api. """ |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "title": "Test PR", |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "branch_to": "master", |
| "branch_from": "test", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data["project"]["date_created"] = "1516348115" |
| data["project"]["date_modified"] = "1516348115" |
| data["repo_from"]["date_created"] = "1516348115" |
| data["repo_from"]["date_modified"] = "1516348115" |
| data["uid"] = "e8b68df8711648deac67c3afed15a798" |
| data["commit_start"] = "114f1b468a5f05e635fcb6394273f3f907386eab" |
| data["commit_stop"] = "114f1b468a5f05e635fcb6394273f3f907386eab" |
| data["date_created"] = "1516348115" |
| data["last_updated"] = "1516348115" |
| data["updated_on"] = "1516348115" |
| self.assertDictEqual( |
| data, |
| { |
| "assignee": None, |
| "branch": "master", |
| "branch_from": "test", |
| "cached_merge_status": "unknown", |
| "closed_at": None, |
| "closed_by": None, |
| "comments": [], |
| "commit_start": "114f1b468a5f05e635fcb6394273f3f907386eab", |
| "commit_stop": "114f1b468a5f05e635fcb6394273f3f907386eab", |
| "date_created": "1516348115", |
| "id": 1, |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "last_updated": "1516348115", |
| "project": { |
| "access_groups": {"admin": [], "commit": [], "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [], |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ], |
| "custom_keys": [], |
| "date_created": "1516348115", |
| "date_modified": "1516348115", |
| "description": "test project #1", |
| "fullname": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "url_path": "test", |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| "remote_git": None, |
| "repo_from": { |
| "access_groups": {"admin": [], "commit": [], "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [], |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ], |
| "custom_keys": [], |
| "date_created": "1516348115", |
| "date_modified": "1516348115", |
| "description": "test project #1", |
| "fullname": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "url_path": "test", |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| "status": "Open", |
| "tags": [], |
| "threshold_reached": None, |
| "title": "Test PR", |
| "uid": "e8b68df8711648deac67c3afed15a798", |
| "updated_on": "1516348115", |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| ) |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_open_missing_initial_comment(self): |
| """ Test the api_pull_request_create method of the flask api when |
| not initial comment is submitted. |
| """ |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "title": "Test PR", |
| "branch_to": "master", |
| "branch_from": "test", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data["project"]["date_created"] = "1516348115" |
| data["project"]["date_modified"] = "1516348115" |
| data["repo_from"]["date_created"] = "1516348115" |
| data["repo_from"]["date_modified"] = "1516348115" |
| data["uid"] = "e8b68df8711648deac67c3afed15a798" |
| data["commit_start"] = "114f1b468a5f05e635fcb6394273f3f907386eab" |
| data["commit_stop"] = "114f1b468a5f05e635fcb6394273f3f907386eab" |
| data["date_created"] = "1516348115" |
| data["last_updated"] = "1516348115" |
| data["updated_on"] = "1516348115" |
| self.assertDictEqual( |
| data, |
| { |
| "assignee": None, |
| "branch": "master", |
| "branch_from": "test", |
| "cached_merge_status": "unknown", |
| "closed_at": None, |
| "closed_by": None, |
| "comments": [], |
| "commit_start": "114f1b468a5f05e635fcb6394273f3f907386eab", |
| "commit_stop": "114f1b468a5f05e635fcb6394273f3f907386eab", |
| "date_created": "1516348115", |
| "id": 1, |
| "initial_comment": None, |
| "last_updated": "1516348115", |
| "project": { |
| "access_groups": {"admin": [], "commit": [], "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [], |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ], |
| "custom_keys": [], |
| "date_created": "1516348115", |
| "date_modified": "1516348115", |
| "description": "test project #1", |
| "fullname": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "url_path": "test", |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| "remote_git": None, |
| "repo_from": { |
| "access_groups": {"admin": [], "commit": [], "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [], |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ], |
| "custom_keys": [], |
| "date_created": "1516348115", |
| "date_modified": "1516348115", |
| "description": "test project #1", |
| "fullname": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "url_path": "test", |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| "status": "Open", |
| "tags": [], |
| "threshold_reached": None, |
| "title": "Test PR", |
| "uid": "e8b68df8711648deac67c3afed15a798", |
| "updated_on": "1516348115", |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| ) |
| |
| |
| class PagureFlaskApiForkPRDiffStatstests(tests.Modeltests): |
| """ Tests for the flask API of pagure for the diff stats endpoint of PRs |
| """ |
| |
| maxDiff = None |
| |
| def setUp(self): |
| """ Set up the environnment, ran before every tests. """ |
| super(PagureFlaskApiForkPRDiffStatstests, self).setUp() |
| |
| pagure.config.config["REQUESTS_FOLDER"] = None |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), ncommits=5 |
| ) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| |
| |
| 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="test", |
| repo_to=repo, |
| branch_to="master", |
| title="test pull-request", |
| user="pingou", |
| ) |
| self.session.commit() |
| self.assertEqual(req.id, 1) |
| self.assertEqual(req.title, "test pull-request") |
| |
| @patch("pagure.lib.git.update_git", MagicMock(return_value=True)) |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_diffstats_no_repo(self): |
| """ Test the api_pull_request_merge method of the flask api. """ |
| output = self.app.get("/api/0/invalid/pull-request/404/diffstats") |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, {"error": "Project not found", "error_code": "ENOPROJECT"} |
| ) |
| |
| @patch("pagure.lib.git.update_git", MagicMock(return_value=True)) |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_diffstats_no_pr(self): |
| """ Test the api_pull_request_merge method of the flask api. """ |
| output = self.app.get("/api/0/test/pull-request/404/diffstats") |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, {"error": "Pull-Request not found", "error_code": "ENOREQ"} |
| ) |
| |
| @patch("pagure.lib.git.update_git", MagicMock(return_value=True)) |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_diffstats_file_modified(self): |
| """ Test the api_pull_request_merge method of the flask api. """ |
| output = self.app.get("/api/0/test/pull-request/1/diffstats") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| { |
| "sources": { |
| "lines_added": 10, |
| "lines_removed": 0, |
| "new_id": "540916fbd3d825d14cc0c0b2397606fda69379ce", |
| "old_id": "265f133a7c94ede4cb183dd808219c5bf9e08f87", |
| "old_path": "sources", |
| "status": "M", |
| } |
| }, |
| ) |
| |
| @patch("pagure.lib.git.update_git", MagicMock(return_value=True)) |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_diffstats_file_added_mofidied(self): |
| """ Test the api_pull_request_merge method of the flask api. """ |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), ncommits=5 |
| ) |
| tests.add_readme_git_repo( |
| os.path.join(self.path, "repos", "test.git"), |
| readme_name="README.md", |
| branch="test", |
| ) |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(len(repo.requests), 1) |
| |
| output = self.app.get("/api/0/test/pull-request/1/diffstats") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertTrue( |
| data |
| in [ |
| { |
| "README.md": { |
| "lines_added": 5, |
| "lines_removed": 0, |
| "new_id": "bd913ea153650b94f33f53e5164c36a28b761bf4", |
| "old_id": "0000000000000000000000000000000000000000", |
| "old_path": "README.md", |
| "status": "A", |
| }, |
| "sources": { |
| "lines_added": 5, |
| "lines_removed": 0, |
| "new_id": "540916fbd3d825d14cc0c0b2397606fda69379ce", |
| "old_id": "293500070b9dfc6ab66e31383f8f7fccf6a95fe2", |
| "old_path": "sources", |
| "status": "M", |
| }, |
| }, |
| { |
| "README.md": { |
| "lines_added": 5, |
| "lines_removed": 0, |
| "new_id": "bd913ea153650b94f33f53e5164c36a28b761bf4", |
| "old_id": "0000000000000000000000000000000000000000", |
| "old_path": "README.md", |
| "status": "A", |
| }, |
| "sources": { |
| "lines_added": 10, |
| "lines_removed": 0, |
| "new_id": "540916fbd3d825d14cc0c0b2397606fda69379ce", |
| "old_id": "265f133a7c94ede4cb183dd808219c5bf9e08f87", |
| "old_path": "sources", |
| "status": "M", |
| }, |
| }, |
| ] |
| ) |
| |
| @patch("pagure.lib.git.update_git", MagicMock(return_value=True)) |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def test_api_pull_request_diffstats_file_modified_deleted(self): |
| """ Test the api_pull_request_merge method of the flask api. """ |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(len(repo.requests), 1) |
| pagure.lib.tasks.update_pull_request(repo.requests[0].uid) |
| |
| tests.add_readme_git_repo( |
| os.path.join(self.path, "repos", "test.git"), |
| readme_name="README.md", |
| branch="test", |
| ) |
| tests.remove_file_git_repo( |
| os.path.join(self.path, "repos", "test.git"), |
| filename="sources", |
| branch="test", |
| ) |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, "test") |
| self.assertEqual(len(repo.requests), 1) |
| pagure.lib.tasks.update_pull_request(repo.requests[0].uid) |
| |
| output = self.app.get("/api/0/test/pull-request/1/diffstats") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| { |
| "README.md": { |
| "lines_added": 5, |
| "lines_removed": 0, |
| "new_id": "bd913ea153650b94f33f53e5164c36a28b761bf4", |
| "old_id": "0000000000000000000000000000000000000000", |
| "old_path": "README.md", |
| "status": "A", |
| }, |
| "sources": { |
| "lines_added": 0, |
| "lines_removed": 5, |
| "new_id": "0000000000000000000000000000000000000000", |
| "old_id": "265f133a7c94ede4cb183dd808219c5bf9e08f87", |
| "old_path": "sources", |
| "status": "D", |
| }, |
| }, |
| ) |
| |
| |
| class PagureApiThresholdReachedTests(tests.Modeltests): |
| """ Test the behavior of the threshold_reached value returned by the API. |
| """ |
| |
| maxDiff = None |
| |
| def _clean_data(self, data): |
| data["project"]["date_created"] = "1516348115" |
| data["project"]["date_modified"] = "1516348115" |
| data["repo_from"]["date_created"] = "1516348115" |
| data["repo_from"]["date_modified"] = "1516348115" |
| data["uid"] = "e8b68df8711648deac67c3afed15a798" |
| data["commit_start"] = "114f1b468a5f05e635fcb6394273f3f907386eab" |
| data["commit_stop"] = "114f1b468a5f05e635fcb6394273f3f907386eab" |
| data["date_created"] = "1516348115" |
| data["last_updated"] = "1516348115" |
| data["updated_on"] = "1516348115" |
| data["comments"] = [] |
| return data |
| |
| @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) |
| def setUp(self): |
| """ Set up the environment for the tests. """ |
| super(PagureApiThresholdReachedTests, self).setUp() |
| |
| 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, "requests"), bare=True |
| ) |
| tests.add_readme_git_repo(os.path.join(self.path, "repos", "test.git")) |
| tests.add_commit_git_repo( |
| os.path.join(self.path, "repos", "test.git"), branch="test" |
| ) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| |
| item = pagure.lib.model.Token( |
| id="aaabbbcccddd_foo", |
| user_id=2, |
| project_id=1, |
| expiration=datetime.datetime.utcnow() |
| + datetime.timedelta(days=30), |
| ) |
| self.session.add(item) |
| self.session.commit() |
| tests.create_tokens_acl(self.session, token_id="aaabbbcccddd_foo") |
| |
| |
| repo = pagure.lib.query._get_project(self.session, "test") |
| settings = repo.settings |
| settings["Minimum_score_to_merge_pull-request"] = 2 |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = { |
| "title": "Test PR", |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "branch_to": "master", |
| "branch_from": "test", |
| } |
| |
| output = self.app.post( |
| "/api/0/test/pull-request/new", headers=headers, data=data |
| ) |
| self.assertEqual(output.status_code, 200) |
| |
| self.expected_data = { |
| "assignee": None, |
| "branch": "master", |
| "branch_from": "test", |
| "cached_merge_status": "unknown", |
| "closed_at": None, |
| "closed_by": None, |
| "comments": [], |
| "commit_start": "114f1b468a5f05e635fcb6394273f3f907386eab", |
| "commit_stop": "114f1b468a5f05e635fcb6394273f3f907386eab", |
| "date_created": "1516348115", |
| "id": 1, |
| "initial_comment": "Nothing much, the changes speak for themselves", |
| "last_updated": "1516348115", |
| "project": { |
| "access_groups": {"admin": [], "commit": [], "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [], |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ], |
| "custom_keys": [], |
| "date_created": "1516348115", |
| "date_modified": "1516348115", |
| "description": "test project #1", |
| "fullname": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "url_path": "test", |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| "remote_git": None, |
| "repo_from": { |
| "access_groups": {"admin": [], "commit": [], "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [], |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate", |
| ], |
| "custom_keys": [], |
| "date_created": "1516348115", |
| "date_modified": "1516348115", |
| "description": "test project #1", |
| "fullname": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "url_path": "test", |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| }, |
| "status": "Open", |
| "tags": [], |
| "threshold_reached": None, |
| "title": "Test PR", |
| "uid": "e8b68df8711648deac67c3afed15a798", |
| "updated_on": "1516348115", |
| "user": {"fullname": "PY C", "name": "pingou"}, |
| } |
| |
| def test_api_pull_request_no_comments(self): |
| """ Check the value of threshold_reach when the PR has no comments. |
| """ |
| |
| |
| output = self.app.get("/api/0/test/pull-request/1") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data = self._clean_data(data) |
| self.expected_data["threshold_reached"] = False |
| self.assertDictEqual(data, self.expected_data) |
| |
| def test_api_pull_request_one_comments(self): |
| """ Check the value of threshold_reach when the PR has one comment. |
| """ |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = {"comment": "This is a very interesting solution :thumbsup:"} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Comment added"}) |
| |
| output = self.app.get("/api/0/test/pull-request/1") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data = self._clean_data(data) |
| self.expected_data["threshold_reached"] = False |
| self.assertDictEqual(data, self.expected_data) |
| |
| def test_api_pull_request_two_comments_one_person(self): |
| """ Check the value of threshold_reach when the PR has two comments |
| but from the same person. |
| """ |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = {"comment": "This is a very interesting solution :thumbsup:"} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Comment added"}) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = {"comment": "Indeed it is :thumbsup:"} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Comment added"}) |
| |
| output = self.app.get("/api/0/test/pull-request/1") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data = self._clean_data(data) |
| self.expected_data["threshold_reached"] = False |
| self.assertDictEqual(data, self.expected_data) |
| |
| def test_api_pull_request_two_comments_two_persons(self): |
| """ Check the value of threshold_reach when the PR has two comments |
| from two different persons. |
| """ |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = {"comment": "This is a very interesting solution :thumbsup:"} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Comment added"}) |
| |
| headers = {"Authorization": "token aaabbbcccddd_foo"} |
| data = {"comment": "Indeed it is :thumbsup:"} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Comment added"}) |
| |
| output = self.app.get("/api/0/test/pull-request/1") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data = self._clean_data(data) |
| data["comments"] = [] |
| self.expected_data["threshold_reached"] = True |
| self.assertDictEqual(data, self.expected_data) |
| |
| def test_api_pull_request_three_comments_two_persons_changed_to_no(self): |
| """ Check the value of threshold_reach when the PR has three |
| comments from two person among which one changed their mind from |
| +1 to -1. |
| """ |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = {"comment": "This is a very interesting solution :thumbsup:"} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Comment added"}) |
| |
| headers = {"Authorization": "token aaabbbcccddd_foo"} |
| data = {"comment": "Indeed it is :thumbsup:"} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Comment added"}) |
| |
| data = { |
| "comment": "Nevermind the bug is elsewhere in fact :thumbsdown:" |
| } |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Comment added"}) |
| |
| output = self.app.get("/api/0/test/pull-request/1") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data = self._clean_data(data) |
| data["comments"] = [] |
| self.expected_data["threshold_reached"] = False |
| self.assertDictEqual(data, self.expected_data) |
| |
| def test_api_pull_request_three_comments_two_persons_changed_to_yes(self): |
| """ Check the value of threshold_reach when the PR has three |
| comments from two person among which one changed their mind from |
| -1 to +1 |
| """ |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| data = {"comment": "This is a very interesting solution :thumbsup:"} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Comment added"}) |
| |
| headers = {"Authorization": "token aaabbbcccddd_foo"} |
| data = {"comment": "I think the bug is elsewhere :thumbsdown:"} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Comment added"}) |
| |
| data = {"comment": "Nevermind it is here :thumbsup:"} |
| output = self.app.post( |
| "/api/0/test/pull-request/1/comment", data=data, headers=headers |
| ) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, {"message": "Comment added"}) |
| |
| output = self.app.get("/api/0/test/pull-request/1") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data = self._clean_data(data) |
| data["comments"] = [] |
| self.expected_data["threshold_reached"] = True |
| self.assertDictEqual(data, self.expected_data) |
| |
| |
| if __name__ == "__main__": |
| unittest.main(verbosity=2) |