test_pagure_flask_api.py 9.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260
  1. # -*- coding: utf-8 -*-
  2. """
  3. (c) 2015 - Copyright Red Hat Inc
  4. Authors:
  5. Pierre-Yves Chibon <pingou@pingoured.fr>
  6. """
  7. from __future__ import unicode_literals, absolute_import
  8. import unittest
  9. import shutil
  10. import sys
  11. import os
  12. import json
  13. from mock import patch, MagicMock
  14. sys.path.insert(
  15. 0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
  16. )
  17. import pagure.api
  18. import pagure.flask_app
  19. import pagure.lib.query
  20. import tests
  21. class PagureFlaskApitests(tests.SimplePagureTest):
  22. """Tests for flask API controller of pagure"""
  23. maxDiff = None
  24. def test_api_doc(self):
  25. """Test the API documentation page."""
  26. print(dir(self.app))
  27. output = self.app.get("/api/0/")
  28. output_text = output.get_data(as_text=True)
  29. self.assertIn("<title> API | pagure - Pagure</title>\n", output_text)
  30. self.assertIn(">Pagure API Reference</h5>\n", output_text)
  31. def test_api_doc_authenticated(self):
  32. """Test the API documentation page."""
  33. user = tests.FakeUser(username="foo")
  34. with tests.user_set(self.app.application, user):
  35. output = self.app.get("/api/0/")
  36. output_text = output.get_data(as_text=True)
  37. self.assertIn(
  38. "<title> API | pagure - Pagure</title>\n", output_text
  39. )
  40. self.assertIn(">Pagure API Reference</h5>\n", output_text)
  41. def test_api_get_request_data(self):
  42. data = {"foo": "bar"}
  43. # test_request_context doesn't set flask.g, but some teardown
  44. # functions try to use that, so let's exclude them
  45. self._app.teardown_request_funcs = {}
  46. with self._app.test_request_context(
  47. "/api/0/version", method="POST", data=data
  48. ):
  49. self.assertEqual(pagure.api.get_request_data()["foo"], "bar")
  50. data = json.dumps(data)
  51. with self._app.test_request_context(
  52. "/api/0/version", data=data, content_type="application/json"
  53. ):
  54. self.assertEqual(pagure.api.get_request_data()["foo"], "bar")
  55. def test_api_version_old_url(self):
  56. """Test the api_version function."""
  57. output = self.app.get("/api/0/version")
  58. self.assertEqual(output.status_code, 200)
  59. data = json.loads(output.get_data(as_text=True))
  60. self.assertEqual(data["version"], pagure.__api_version__)
  61. self.assertEqual(sorted(data.keys()), ["version"])
  62. def test_api_version_new_url(self):
  63. """Test the api_version function at its new url."""
  64. output = self.app.get("/api/0/-/version")
  65. self.assertEqual(output.status_code, 200)
  66. data = json.loads(output.get_data(as_text=True))
  67. self.assertEqual(data["version"], pagure.__api_version__)
  68. self.assertEqual(sorted(data.keys()), ["version"])
  69. def test_api_groups(self):
  70. """Test the api_groups function."""
  71. # Add a couple of groups so that we can list them
  72. item = pagure.lib.model.PagureGroup(
  73. group_name="group1",
  74. group_type="user",
  75. display_name="User group",
  76. user_id=1, # pingou
  77. )
  78. self.session.add(item)
  79. item = pagure.lib.model.PagureGroup(
  80. group_name="rel-eng",
  81. group_type="user",
  82. display_name="Release engineering group",
  83. user_id=1, # pingou
  84. )
  85. self.session.add(item)
  86. self.session.commit()
  87. output = self.app.get("/api/0/groups")
  88. self.assertEqual(output.status_code, 200)
  89. data = json.loads(output.get_data(as_text=True))
  90. self.assertEqual(data["groups"], ["group1", "rel-eng"])
  91. self.assertEqual(
  92. sorted(data.keys()), ["groups", "pagination", "total_groups"]
  93. )
  94. self.assertEqual(data["total_groups"], 2)
  95. output = self.app.get("/api/0/groups?pattern=re")
  96. self.assertEqual(output.status_code, 200)
  97. data = json.loads(output.get_data(as_text=True))
  98. self.assertEqual(data["groups"], ["rel-eng"])
  99. self.assertEqual(
  100. sorted(data.keys()), ["groups", "pagination", "total_groups"]
  101. )
  102. self.assertEqual(data["total_groups"], 1)
  103. def test_api_whoami_unauth(self):
  104. """Test the api_whoami function."""
  105. output = self.app.post("/api/0/-/whoami")
  106. self.assertEqual(output.status_code, 401)
  107. data = json.loads(output.get_data(as_text=True))
  108. self.assertEqual(
  109. data,
  110. {
  111. "error": "Invalid or expired token. Please visit "
  112. "http://localhost.localdomain/settings#nav-api-tab to get or "
  113. "renew your API token.",
  114. "error_code": "EINVALIDTOK",
  115. },
  116. )
  117. def test_api_whoami_invalid_auth(self):
  118. """Test the api_whoami function with an invalid token."""
  119. tests.create_projects(self.session)
  120. tests.create_tokens(self.session)
  121. headers = {"Authorization": "token invalid"}
  122. output = self.app.post("/api/0/-/whoami", headers=headers)
  123. self.assertEqual(output.status_code, 401)
  124. data = json.loads(output.get_data(as_text=True))
  125. self.assertEqual(
  126. data,
  127. {
  128. "error": "Invalid or expired token. Please visit "
  129. "http://localhost.localdomain/settings#nav-api-tab to get or "
  130. "renew your API token.",
  131. "error_code": "EINVALIDTOK",
  132. "errors": "Invalid token",
  133. },
  134. )
  135. def test_api_whoami_auth(self):
  136. """Test the api_whoami function with a valid token."""
  137. tests.create_projects(self.session)
  138. tests.create_tokens(self.session)
  139. headers = {"Authorization": "token aaabbbcccddd"}
  140. output = self.app.post("/api/0/-/whoami", headers=headers)
  141. self.assertEqual(output.status_code, 200)
  142. data = json.loads(output.get_data(as_text=True))
  143. self.assertEqual(data, {"username": "pingou"})
  144. def test_api_error_codes(self):
  145. """Test the api_error_codes endpoint."""
  146. output = self.app.get("/api/0/-/error_codes")
  147. self.assertEqual(output.status_code, 200)
  148. data = json.loads(output.get_data(as_text=True))
  149. self.assertEqual(len(data), 47)
  150. self.assertEqual(
  151. sorted(data.keys()),
  152. sorted(
  153. [
  154. "EBRANCHNOTFOUND",
  155. "EDATETIME",
  156. "EDBERROR",
  157. "EEMPTYGIT",
  158. "EFILENOTFOUND",
  159. "EGITERROR",
  160. "EINVALIDISSUEFIELD",
  161. "EINVALIDISSUEFIELD_LINK",
  162. "EINVALIDPERPAGEVALUE",
  163. "EINVALIDPRIORITY",
  164. "EINVALIDREQ",
  165. "EINVALIDTOK",
  166. "EISSUENOTALLOWED",
  167. "EMODIFYPROJECTNOTALLOWED",
  168. "ENEWPROJECTDISABLED",
  169. "ENEWPROJECTFORBIDDEN",
  170. "ENOCODE",
  171. "ENOCOMMENT",
  172. "ENOCOMMIT",
  173. "ENOGROUP",
  174. "ENOISSUE",
  175. "ENOPRCLOSE",
  176. "ENOPROJECT",
  177. "ENOPROJECTS",
  178. "ENOPRSTATS",
  179. "ENOREQ",
  180. "ENOSIGNEDOFF",
  181. "ENOTASSIGNED",
  182. "ENOTASSIGNEE",
  183. "ENOTHIGHENOUGH",
  184. "ENOTMAINADMIN",
  185. "ENOUSER",
  186. "EPRCONFLICTS",
  187. "EPRNOTALLOWED",
  188. "EPRSCORE",
  189. "EPULLREQUESTSDISABLED",
  190. "ETIMESTAMP",
  191. "ETRACKERDISABLED",
  192. "ETRACKERREADONLY",
  193. "EUBLOCKED",
  194. "EREBASENOTALLOWED",
  195. "ENOPLUGIN",
  196. "EPLUGINDISABLED",
  197. "EPLUGINCHANGENOTALLOWED",
  198. "EPLUGINNOTINSTALLED",
  199. "ENOTAG",
  200. "EMIRRORINGDISABLED",
  201. ]
  202. ),
  203. )
  204. @patch("pagure.lib.tasks.get_result")
  205. def test_api_task_status(self, get_result):
  206. async_result = MagicMock()
  207. async_result.status = "running"
  208. async_result.ready.return_value = False
  209. get_result.return_value = async_result
  210. output = self.app.get("/api/0/task/123abc/status/")
  211. self.assertEqual(output.status_code, 200)
  212. data = json.loads(output.get_data(as_text=True))
  213. self.assertEqual(data, {"ready": False, "status": "running"})
  214. async_result = MagicMock()
  215. async_result.status = "finished"
  216. async_result.ready.return_value = True
  217. async_result.successful.return_value = True
  218. get_result.return_value = async_result
  219. output = self.app.get("/api/0/task/123abc/status/")
  220. self.assertEqual(output.status_code, 200)
  221. data = json.loads(output.get_data(as_text=True))
  222. self.assertEqual(
  223. data, {"ready": True, "status": "finished", "successful": True}
  224. )
  225. if __name__ == "__main__":
  226. unittest.main(verbosity=2)