1
0

test_stream_server.py 8.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280
  1. #!/usr/bin/env python
  2. # -*- coding: utf-8 -*-
  3. """
  4. (c) 2016 - Copyright Red Hat Inc
  5. Authors:
  6. Adam Williamson <awilliam@redhat.com>
  7. Tests for the Pagure streaming server.
  8. """
  9. # obviously this is fine for testing.
  10. # pylint: disable=locally-disabled, protected-access
  11. from __future__ import unicode_literals, absolute_import
  12. import logging
  13. import os
  14. import sys
  15. import unittest
  16. import mock
  17. import six
  18. sys.path.insert(
  19. 0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
  20. )
  21. sys.path.insert(
  22. 0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "../pagure-ev")
  23. )
  24. if six.PY3:
  25. raise unittest.case.SkipTest("Skipping on python3")
  26. import pagure.lib.query # pylint: disable=wrong-import-position
  27. from pagure.exceptions import (
  28. PagureException,
  29. PagureEvException,
  30. ) # pylint: disable=wrong-import-position
  31. import tests # pylint: disable=wrong-import-position
  32. # comes from ev-server/
  33. import pagure_stream_server as pss # pylint: disable=wrong-import-position, import-error
  34. logging.basicConfig(stream=sys.stderr)
  35. class StreamingServerTests(tests.Modeltests):
  36. """Tests for the streaming server."""
  37. def setUp(self):
  38. """Set up the environnment, run before every test."""
  39. super(StreamingServerTests, self).setUp()
  40. # Make sure the server uses the existing session
  41. pss.SESSION = self.session
  42. # Mock send_email, we never want to send or see emails here.
  43. self.mailpatcher = mock.patch("pagure.lib.notify.send_email")
  44. self.mailpatcher.start()
  45. # Setup projects
  46. tests.create_projects(self.session)
  47. self.repo = pagure.lib.query._get_project(self.session, "test")
  48. self.repo2 = pagure.lib.query._get_project(self.session, "test2")
  49. # Disable repo 2's issue tracker and PR tracker
  50. pagure.lib.query.update_project_settings(
  51. session=self.session,
  52. repo=self.repo2,
  53. user="pingou",
  54. settings={"issue_tracker": False, "pull_requests": False},
  55. )
  56. # Create a public issue
  57. pagure.lib.query.new_issue(
  58. session=self.session,
  59. repo=self.repo,
  60. title="Test issue",
  61. content="We should work on this",
  62. user="pingou",
  63. )
  64. # Create a private issue
  65. pagure.lib.query.new_issue(
  66. session=self.session,
  67. repo=self.repo,
  68. title="Private issue #2",
  69. content="The world can see my porn folder",
  70. user="pingou",
  71. private=True,
  72. )
  73. # Create a PR
  74. pagure.lib.query.new_pull_request(
  75. session=self.session,
  76. repo_from=self.repo,
  77. repo_to=self.repo,
  78. branch_from="feature",
  79. branch_to="master",
  80. title="Test PR",
  81. user="pingou",
  82. )
  83. def tearDown(self):
  84. "Stop the patchers, as well as calling super." ""
  85. super(StreamingServerTests, self).tearDown()
  86. self.mailpatcher.stop()
  87. def test_parse_path(self):
  88. """Tests for _parse_path."""
  89. # Result format is: (username, namespace, repo, objtype, objid)
  90. # Simple case: issue for non-namespaced, non-forked repo.
  91. result = pagure.utils.parse_path("/pagure/issue/1")
  92. self.assertEqual(result, (None, None, "pagure", "issue", "1"))
  93. # Pull request for namespaced repo.
  94. result = pagure.utils.parse_path("/fedora-qa/fedfind/pull-request/2")
  95. self.assertEqual(
  96. result, (None, "fedora-qa", "fedfind", "pull-request", "2")
  97. )
  98. # Issue for forked repo.
  99. result = pagure.utils.parse_path("/fork/adamwill/pagure/issue/3")
  100. self.assertEqual(result, ("adamwill", None, "pagure", "issue", "3"))
  101. # Issue for forked, namespaced repo.
  102. result = pagure.utils.parse_path(
  103. "/fork/pingou/fedora-qa/fedfind/issue/4"
  104. )
  105. self.assertEqual(
  106. result, ("pingou", "fedora-qa", "fedfind", "issue", "4")
  107. )
  108. # Issue for repo named 'pull-request' (yeah, now we're getting tricksy).
  109. result = pagure.utils.parse_path("/pull-request/issue/5")
  110. self.assertEqual(result, (None, None, "pull-request", "issue", "5"))
  111. # Unknown object type.
  112. six.assertRaisesRegex(
  113. self,
  114. PagureException,
  115. r"No known object",
  116. pagure.utils.parse_path,
  117. "/pagure/unexpected/1",
  118. )
  119. # No object ID.
  120. six.assertRaisesRegex(
  121. self,
  122. PagureException,
  123. r"No project or object ID",
  124. pagure.utils.parse_path,
  125. "/pagure/issue",
  126. )
  127. # No repo name. Note: we cannot catch 'namespace but no repo name',
  128. # but that should fail later in pagure.lib.query.get_project
  129. six.assertRaisesRegex(
  130. self,
  131. PagureException,
  132. r"No project or object ID",
  133. pagure.utils.parse_path,
  134. "/issue/1",
  135. )
  136. # /fork but no user name.
  137. six.assertRaisesRegex(
  138. self,
  139. PagureException,
  140. r"no user found!",
  141. pagure.utils.parse_path,
  142. "/fork/pagure/issue/1",
  143. )
  144. # Too many path components before object type.
  145. six.assertRaisesRegex(
  146. self,
  147. PagureException,
  148. r"More path components",
  149. pagure.utils.parse_path,
  150. "/fork/adamwill/fedora-qa/fedfind/unexpected/issue/1",
  151. )
  152. six.assertRaisesRegex(
  153. self,
  154. PagureException,
  155. r"More path components",
  156. pagure.utils.parse_path,
  157. "/fedora-qa/fedfind/unexpected/issue/1",
  158. )
  159. def test_get_issue(self):
  160. """Tests for _get_issue."""
  161. # Simple case: get the existing issue from the existing repo.
  162. result = pss._get_issue(self.repo, "1")
  163. self.assertEqual(result.id, 1)
  164. # Issue that doesn't exist.
  165. six.assertRaisesRegex(
  166. self,
  167. PagureEvException,
  168. r"Issue '3' not found",
  169. pss._get_issue,
  170. self.repo,
  171. "3",
  172. )
  173. # Private issue (for now we don't handle auth).
  174. six.assertRaisesRegex(
  175. self,
  176. PagureEvException,
  177. r"issue is private",
  178. pss._get_issue,
  179. self.repo,
  180. "2",
  181. )
  182. # Issue from a project with no issue tracker.
  183. six.assertRaisesRegex(
  184. self,
  185. PagureEvException,
  186. r"No issue tracker found",
  187. pss._get_issue,
  188. self.repo2,
  189. "1",
  190. )
  191. def test_get_pull_request(self):
  192. """Tests for _get_pull_request."""
  193. # Simple case: get the existing PR from the existing repo.
  194. result = pss._get_pull_request(self.repo, "3")
  195. self.assertEqual(result.id, 3)
  196. # PR that doesn't exist.
  197. six.assertRaisesRegex(
  198. self,
  199. PagureEvException,
  200. r"Pull-Request '2' not found",
  201. pss._get_pull_request,
  202. self.repo,
  203. "2",
  204. )
  205. # PR from a project with no PR tracker.
  206. six.assertRaisesRegex(
  207. self,
  208. PagureEvException,
  209. r"No pull-request tracker found",
  210. pss._get_pull_request,
  211. self.repo2,
  212. "1",
  213. )
  214. def test_get_obj_from_path(self):
  215. """Tests for get_obj_from_path."""
  216. # Simple issue case.
  217. result = pss.get_obj_from_path("/test/issue/1")
  218. self.assertEqual(result.id, 1)
  219. # Simple PR case.
  220. result = pss.get_obj_from_path("/test/pull-request/3")
  221. self.assertEqual(result.id, 3)
  222. # Non-existent repo.
  223. six.assertRaisesRegex(
  224. self,
  225. PagureEvException,
  226. r"Project 'foo' not found",
  227. pss.get_obj_from_path,
  228. "/foo/issue/1",
  229. )
  230. # NOTE: we cannot test the 'Invalid object provided' exception
  231. # as it's a backup (current code will never hit it)
  232. if __name__ == "__main__":
  233. unittest.main(verbosity=2)