# -*- coding: utf-8 -*- """ (c) 2015 - Copyright Red Hat Inc Authors: Pierre-Yves Chibon """ 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) # Create a pull-request 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) # Create a pull-request 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, "tags": [], "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, }, ) # Close the PR and try again 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, "tags": [], "page": 1, "per_page": 20, "status": "closed", }, ) self.assertEqual(data["total_requests"], 1) # Create two closed pull-requests 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() # Test the API view of closed pull-requests 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) # Test the API view of merged pull-requests 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) # Create a pull-request 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, "tags": [], "page": 1, "per_page": 20, "status": "all", }, ) self.assertEqual(data["total_requests"], 1) # Close the PR and try again 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, "tags": [], "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) # Create a pull-request 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") # Invalid repo 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"} ) # List pull-requests 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, "tags": [], "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"} # Access Pull-Request authenticated 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_tag_filtered(self, send_email): """ Test the api_pull_request_view method of the flask api to list tag filtered open PRs. """ 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") # Add a tag pagure.lib.query.new_tag( self.session, "tag-1", "tag-1 description", "#ff0000", repo.id ) # Create a pull-request 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?tags=tag-1") 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, "tags": ["tag-1"], "page": 1, "per_page": 20, "status": True, }, "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 0, "per_page": 20, "prev": None, }, "requests": [], "total_requests": 0, }, ) # Tag the PR and try again pagure.lib.query.update_tags( self.session, obj=req, tags=["tag-1"], username="pingou" ) self.session.commit() output = self.app.get("/api/0/test/pull-requests?tags=tag-1") 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, "tags": ["tag-1"], "page": 1, "per_page": 20, "status": True, }, ) self.assertEqual(data["total_requests"], 1) # Try negative filtering output = self.app.get("/api/0/test/pull-requests?tags=!tag-1") 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, "tags": ["!tag-1"], "page": 1, "per_page": 20, "status": True, }, "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 0, "per_page": 20, "prev": None, }, "requests": [], "total_requests": 0, }, ) @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) # Create a pull-request 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) # Create a pull-request 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") # Invalid repo 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"} ) # Invalid issue for this repo 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"} ) # Valid issue 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"} # Access Pull-Request authenticated 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) # Create a pull-request 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 # Invalid request 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"} ) # Valid issue 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"} # Access Pull-Request authenticated 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) # Create the pull-request to close 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) # Create the pull-request to close 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"} # Invalid project 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"} ) # Valid token, wrong project 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"]) # Invalid PR 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"} ) # Create a token for foo for this project 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() # Allow the token to close PR 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"} # User not admin 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"} # Close PR 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) # Create the pull-request to close 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) # Create the pull-request to close 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) # Create the pull-request to close 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 request", "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) # Create the pull-request to close 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) # Create the pull-request to close 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"} # Invalid project 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"} ) # Valid token, wrong project 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"]) # Invalid PR 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"} ) # Create a token for foo for this project 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() # Allow the token to merge PR 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"} # User not admin 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"} # Merge PR 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") ) # Fork 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", }, ) # Add content to the fork tests.add_content_to_git( os.path.join(self.path, "repos", "forks", "pingou", "test.git"), filename="foobar", content="content from the fork", ) # Add content to the main repo, so they conflict 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) # Create the pull-request to close 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"} # Merge PR 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) # Create the pull-request to close 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"} # Invalid project 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"} ) # Valid token, invalid PR 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"} ) # Valid token, invalid PR - other project 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"} ) # Create a token for foo for this project 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() # Allow the token to merge PR 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"} # User not admin 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"} # Merge PR 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"} # Invalid project 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"} ) # Valid token, wrong project 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"]) # No input 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"} ) # Create a pull-request 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") # Check comments before 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"} # Incomplete request 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."]}, }, ) # No change 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"} # Valid request 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"}) # One 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"} # Create a pull-request 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"} # Valid request 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"} # Create a pull-request 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"} # Valid request 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", }, ) # no comment added 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"} # Invalid project 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"} ) # Valid token, invalid request 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"} ) # Valid token, invalid request in another project 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"} ) # Create a pull-request 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") # Check comments before 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"} # Incomplete request 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."]}, }, ) # No change 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"} # Valid request 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"}) # One 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"} # Invalid project 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"} # Create pull-request 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") # Check subscribtion before 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#nav-api-tab 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"} # Invalid project 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"} ) # Valid token, wrong project 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"]) # No input 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"} ) # Create pull-request 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") # Check subscribtion before 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"]), ) # Unsubscribe - no changes 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", }, ) # No change 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"]), ) # Subscribe 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", }, ) # Subscribe - no changes 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"]), ) # Unsubscribe 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) # Create pull-request 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") # Check subscribtion before 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"]), ) # Subscribe 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", }, ) # Check subscribtions after 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) # Check the behavior if the project disabled the issue tracker 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) # Check the behavior if the project disabled the issue tracker 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) # Check the behavior if the project disabled the issue tracker 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#nav-api-tab 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) # Check the behavior if the project disabled the issue tracker 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", ) # Create tokens 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" ) # Create the pull-request to close 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"] = [] # Let's not check the 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) # Add a token for user `foo` 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") # Add a minimal required score: 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. """ # Check the PR with 0 comment: 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. """ # Check the PR with 1 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. """ # Add two comments from the same user: 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. """ # Add two comments from two users: 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"] = [] # Let's not check the 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. """ # Add three comments from two users: 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"] = [] # Let's not check the 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 """ # Add three comments from two users: 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"] = [] # Let's not check the comments self.expected_data["threshold_reached"] = True self.assertDictEqual(data, self.expected_data) if __name__ == "__main__": unittest.main(verbosity=2)