1
0

test_pagure_flask_api_fork_assign.py 8.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256
  1. # -*- coding: utf-8 -*-
  2. """
  3. (c) 2019 - Copyright Red Hat Inc
  4. Authors:
  5. Pierre-Yves Chibon <pingou@pingoured.fr>
  6. """
  7. from __future__ import unicode_literals, absolute_import
  8. import arrow
  9. import copy
  10. import datetime
  11. import unittest
  12. import shutil
  13. import sys
  14. import time
  15. import os
  16. import flask
  17. import json
  18. import munch
  19. from mock import patch, MagicMock
  20. from sqlalchemy.exc import SQLAlchemyError
  21. sys.path.insert(
  22. 0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
  23. )
  24. import pagure.lib.query
  25. import tests
  26. class PagureFlaskApiForkAssigntests(tests.SimplePagureTest):
  27. """ Tests for the flask API of pagure for assigning a PR """
  28. maxDiff = None
  29. @patch("pagure.lib.git.update_git", MagicMock(return_value=True))
  30. @patch("pagure.lib.notify.send_email", MagicMock(return_value=True))
  31. def setUp(self):
  32. """ Set up the environnment, ran before every tests. """
  33. super(PagureFlaskApiForkAssigntests, self).setUp()
  34. tests.create_projects(self.session)
  35. tests.add_content_git_repo(
  36. os.path.join(self.path, "repos", "test.git")
  37. )
  38. # Fork
  39. project = pagure.lib.query.get_authorized_project(self.session, "test")
  40. task = pagure.lib.query.fork_project(
  41. session=self.session, user="pingou", repo=project
  42. )
  43. self.session.commit()
  44. self.assertEqual(
  45. task.get(),
  46. {
  47. "endpoint": "ui_ns.view_repo",
  48. "repo": "test",
  49. "namespace": None,
  50. "username": "pingou",
  51. },
  52. )
  53. tests.add_readme_git_repo(
  54. os.path.join(self.path, "repos", "forks", "pingou", "test.git")
  55. )
  56. project = pagure.lib.query.get_authorized_project(self.session, "test")
  57. fork = pagure.lib.query.get_authorized_project(
  58. self.session, "test", user="pingou"
  59. )
  60. tests.create_tokens(self.session)
  61. tests.create_tokens_acl(self.session)
  62. req = pagure.lib.query.new_pull_request(
  63. session=self.session,
  64. repo_from=fork,
  65. branch_from="master",
  66. repo_to=project,
  67. branch_to="master",
  68. title="test pull-request",
  69. user="pingou",
  70. )
  71. self.session.commit()
  72. self.assertEqual(req.id, 1)
  73. self.assertEqual(req.title, "test pull-request")
  74. # Assert the PR is open
  75. self.session = pagure.lib.query.create_session(self.dbpath)
  76. project = pagure.lib.query.get_authorized_project(self.session, "test")
  77. self.assertEqual(len(project.requests), 1)
  78. self.assertEqual(project.requests[0].status, "Open")
  79. # Check how the PR renders in the API and the UI
  80. output = self.app.get("/api/0/test/pull-request/1")
  81. self.assertEqual(output.status_code, 200)
  82. output = self.app.get("/test/pull-request/1")
  83. self.assertEqual(output.status_code, 200)
  84. def test_api_assign_pr_invalid_project_namespace(self):
  85. """ Test api_pull_request_assign method when the project doesn't exist.
  86. """
  87. headers = {"Authorization": "token aaabbbcccddd"}
  88. # Valid token, wrong project
  89. output = self.app.post(
  90. "/api/0/somenamespace/test3/pull-request/1/assign", headers=headers
  91. )
  92. self.assertEqual(output.status_code, 401)
  93. data = json.loads(output.get_data(as_text=True))
  94. self.assertDictEqual(
  95. data,
  96. {
  97. "error": "Invalid or expired token. Please visit "
  98. "http://localhost.localdomain/settings#nav-api-tab to get or renew your "
  99. "API token.",
  100. "error_code": "EINVALIDTOK",
  101. },
  102. )
  103. def test_api_assign_pr_invalid_project(self):
  104. """ Test api_pull_request_assign method when the project doesn't exist.
  105. """
  106. headers = {"Authorization": "token aaabbbcccddd"}
  107. # Invalid project
  108. output = self.app.post(
  109. "/api/0/foo/pull-request/1/assign", headers=headers
  110. )
  111. self.assertEqual(output.status_code, 404)
  112. data = json.loads(output.get_data(as_text=True))
  113. self.assertDictEqual(
  114. data, {"error": "Project not found", "error_code": "ENOPROJECT"}
  115. )
  116. def test_api_assign_pr_invalid_project_token(self):
  117. """ Test api_pull_request_assign method when the token doesn't correspond
  118. to the project.
  119. """
  120. headers = {"Authorization": "token aaabbbcccddd"}
  121. # Valid token, wrong project
  122. output = self.app.post(
  123. "/api/0/test2/pull-request/1/assign", headers=headers
  124. )
  125. self.assertEqual(output.status_code, 401)
  126. data = json.loads(output.get_data(as_text=True))
  127. self.assertEqual(sorted(data.keys()), ["error", "error_code"])
  128. self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"])
  129. self.assertEqual(
  130. pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"]
  131. )
  132. def test_api_assign_pr_invalid_pr(self):
  133. """ Test api_pull_request_assign method when asking for an invalid PR
  134. """
  135. headers = {"Authorization": "token aaabbbcccddd"}
  136. # No input
  137. output = self.app.post(
  138. "/api/0/test/pull-request/404/assign", headers=headers
  139. )
  140. self.assertEqual(output.status_code, 404)
  141. data = json.loads(output.get_data(as_text=True))
  142. self.assertDictEqual(
  143. data, {"error": "Pull-Request not found", "error_code": "ENOREQ"}
  144. )
  145. def test_api_assign_pr_no_input(self):
  146. """ Test api_pull_request_assign method when no input is specified
  147. """
  148. headers = {"Authorization": "token aaabbbcccddd"}
  149. # No input
  150. output = self.app.post(
  151. "/api/0/test/pull-request/1/assign", headers=headers
  152. )
  153. self.assertEqual(output.status_code, 200)
  154. data = json.loads(output.get_data(as_text=True))
  155. self.assertDictEqual(data, {"message": "Nothing to change"})
  156. def test_api_assign_pr_assigned(self):
  157. """ Test api_pull_request_assign method when with valid input
  158. """
  159. headers = {"Authorization": "token aaabbbcccddd"}
  160. data = {"assignee": "pingou"}
  161. # Valid request
  162. output = self.app.post(
  163. "/api/0/test/pull-request/1/assign", data=data, headers=headers
  164. )
  165. self.assertEqual(output.status_code, 200)
  166. data = json.loads(output.get_data(as_text=True))
  167. self.assertDictEqual(data, {"message": "Request assigned"})
  168. def test_api_assign_pr_unassigned(self):
  169. """ Test api_pull_request_assign method when unassigning
  170. """
  171. self.test_api_assign_pr_assigned()
  172. headers = {"Authorization": "token aaabbbcccddd"}
  173. data = {}
  174. # Un-assign
  175. output = self.app.post(
  176. "/api/0/test/pull-request/1/assign", data=data, headers=headers
  177. )
  178. self.assertEqual(output.status_code, 200)
  179. data = json.loads(output.get_data(as_text=True))
  180. self.assertDictEqual(data, {"message": "Request assignee reset"})
  181. def test_api_assign_pr_unassigned_twice(self):
  182. """ Test api_pull_request_assign method when unassigning
  183. """
  184. self.test_api_assign_pr_unassigned()
  185. headers = {"Authorization": "token aaabbbcccddd"}
  186. data = {"assignee": None}
  187. # Un-assign
  188. output = self.app.post(
  189. "/api/0/test/pull-request/1/assign", data=data, headers=headers
  190. )
  191. self.assertEqual(output.status_code, 200)
  192. data = json.loads(output.get_data(as_text=True))
  193. self.assertDictEqual(data, {"message": "Nothing to change"})
  194. def test_api_assign_pr_unassigned_empty_string(self):
  195. """ Test api_pull_request_assign method when unassigning with an
  196. empty string
  197. """
  198. self.test_api_assign_pr_assigned()
  199. headers = {"Authorization": "token aaabbbcccddd"}
  200. # Un-assign
  201. data = {"assignee": ""}
  202. output = self.app.post(
  203. "/api/0/test/pull-request/1/assign", data=data, headers=headers
  204. )
  205. self.assertEqual(output.status_code, 200)
  206. data = json.loads(output.get_data(as_text=True))
  207. self.assertDictEqual(data, {"message": "Request assignee reset"})
  208. if __name__ == "__main__":
  209. unittest.main(verbosity=2)