test_pagure_flask_ui_pr_no_sources.py 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304
  1. # -*- coding: utf-8 -*-
  2. """
  3. (c) 2015-2018 - Copyright Red Hat Inc
  4. Authors:
  5. Pierre-Yves Chibon <pingou@pingoured.fr>
  6. """
  7. from __future__ import unicode_literals, absolute_import
  8. import json
  9. import unittest
  10. import shutil
  11. import sys
  12. import tempfile
  13. import time
  14. import os
  15. import pygit2
  16. from mock import patch, MagicMock
  17. sys.path.insert(
  18. 0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
  19. )
  20. import pagure.lib.query
  21. import tests
  22. from pagure.lib.repo import PagureRepo
  23. class PagureFlaskPrNoSourcestests(tests.Modeltests):
  24. """ Tests PR in pagure when the source is gone """
  25. maxDiff = None
  26. @patch("pagure.lib.notify.send_email", MagicMock(return_value=True))
  27. @patch("pagure.lib.notify.fedmsg_publish", MagicMock(return_value=True))
  28. def setUp(self):
  29. """ Set up the environnment, ran before every tests. """
  30. super(PagureFlaskPrNoSourcestests, self).setUp()
  31. tests.create_projects(self.session)
  32. tests.create_projects_git(os.path.join(self.path, "repos"), bare=True)
  33. # Create foo's fork of pingou's test project
  34. item = pagure.lib.model.Project(
  35. user_id=2, # foo
  36. name="test",
  37. description="test project #1",
  38. hook_token="aaabbb",
  39. is_fork=True,
  40. parent_id=1,
  41. )
  42. self.session.add(item)
  43. self.session.commit()
  44. # Create the fork's git repo
  45. repo_path = os.path.join(self.path, "repos", item.path)
  46. pygit2.init_repository(repo_path, bare=True)
  47. project = pagure.lib.query.get_authorized_project(self.session, "test")
  48. fork = pagure.lib.query.get_authorized_project(
  49. self.session, "test", user="foo"
  50. )
  51. self.set_up_git_repo(repo=project, fork=fork)
  52. # Ensure things got setup straight
  53. project = pagure.lib.query.get_authorized_project(self.session, "test")
  54. self.assertEqual(len(project.requests), 1)
  55. # wait for the worker to process the task
  56. path = os.path.join(
  57. self.path, "repos", "test.git", "refs", "pull", "1", "head"
  58. )
  59. self.assertTrue(os.path.exists(path))
  60. def set_up_git_repo(self, repo, fork, branch_from="feature"):
  61. """ Set up the git repo and create the corresponding PullRequest
  62. object.
  63. """
  64. # Clone the main repo
  65. gitrepo = os.path.join(self.path, "repos", repo.path)
  66. newpath = tempfile.mkdtemp(prefix="pagure-fork-test")
  67. repopath = os.path.join(newpath, "test")
  68. clone_repo = pygit2.clone_repository(gitrepo, repopath)
  69. # Create a file in that git repo
  70. with open(os.path.join(repopath, "sources"), "w") as stream:
  71. stream.write("foo\n bar")
  72. clone_repo.index.add("sources")
  73. clone_repo.index.write()
  74. # Commits the files added
  75. tree = clone_repo.index.write_tree()
  76. author = pygit2.Signature("Alice Author", "alice@authors.tld")
  77. committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld")
  78. clone_repo.create_commit(
  79. "refs/heads/master", # the name of the reference to update
  80. author,
  81. committer,
  82. "Add sources file for testing",
  83. # binary string representing the tree object ID
  84. tree,
  85. # list of binary strings representing parents of the new commit
  86. [],
  87. )
  88. refname = "refs/heads/master:refs/heads/master"
  89. ori_remote = clone_repo.remotes[0]
  90. PagureRepo.push(ori_remote, refname)
  91. first_commit = clone_repo.revparse_single("HEAD")
  92. # Set the second repo
  93. repopath = os.path.join(self.path, "repos", fork.path)
  94. new_gitrepo = os.path.join(newpath, "fork_test")
  95. clone_repo = pygit2.clone_repository(repopath, new_gitrepo)
  96. # Add the main project as remote repo
  97. upstream_path = os.path.join(self.path, "repos", repo.path)
  98. remote = clone_repo.create_remote("upstream", upstream_path)
  99. remote.fetch()
  100. # Edit the sources file again
  101. with open(os.path.join(new_gitrepo, "sources"), "w") as stream:
  102. stream.write("foo\n bar\nbaz\n boose")
  103. clone_repo.index.add("sources")
  104. clone_repo.index.write()
  105. # Commits the files added
  106. tree = clone_repo.index.write_tree()
  107. author = pygit2.Signature("Alice Author", "alice@authors.tld")
  108. committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld")
  109. clone_repo.create_commit(
  110. "refs/heads/%s" % branch_from,
  111. author,
  112. committer,
  113. "A commit on branch %s" % branch_from,
  114. tree,
  115. [first_commit.oid.hex],
  116. )
  117. refname = "refs/heads/%s" % (branch_from)
  118. ori_remote = clone_repo.remotes[0]
  119. PagureRepo.push(ori_remote, refname)
  120. # Create a PR for these changes
  121. project = pagure.lib.query.get_authorized_project(self.session, "test")
  122. req = pagure.lib.query.new_pull_request(
  123. session=self.session,
  124. repo_from=fork,
  125. branch_from=branch_from,
  126. repo_to=project,
  127. branch_to="master",
  128. title="PR from the %s branch" % branch_from,
  129. user="pingou",
  130. )
  131. self.session.commit()
  132. self.assertEqual(req.id, 1)
  133. self.assertEqual(req.title, "PR from the %s branch" % branch_from)
  134. shutil.rmtree(newpath)
  135. def test_request_pull_reference(self):
  136. """ Test if there is a reference created for a new PR. """
  137. project = pagure.lib.query.get_authorized_project(self.session, "test")
  138. self.assertEqual(len(project.requests), 1)
  139. gitrepo = os.path.join(self.path, "repos", "test.git")
  140. repo = pygit2.Repository(gitrepo)
  141. self.assertEqual(
  142. list(repo.listall_references()),
  143. ["refs/heads/master", "refs/pull/1/head"],
  144. )
  145. def test_request_pull_fork_reference(self):
  146. """ Test if there the references created on the fork. """
  147. project = pagure.lib.query.get_authorized_project(
  148. self.session, "test", user="foo"
  149. )
  150. self.assertEqual(len(project.requests), 0)
  151. gitrepo = os.path.join(self.path, "repos", project.path)
  152. repo = pygit2.Repository(gitrepo)
  153. self.assertEqual(
  154. list(repo.listall_references()), ["refs/heads/feature"]
  155. )
  156. def test_accessing_pr_fork_deleted(self):
  157. """ Test accessing the PR if the fork has been deleted. """
  158. # Delete fork on disk
  159. project = pagure.lib.query.get_authorized_project(
  160. self.session, "test", user="foo"
  161. )
  162. repo_path = os.path.join(self.path, "repos", project.path)
  163. self.assertTrue(os.path.exists(repo_path))
  164. shutil.rmtree(repo_path)
  165. self.assertFalse(os.path.exists(repo_path))
  166. # Delete fork in the DB
  167. self.session.delete(project)
  168. self.session.commit()
  169. # View the pull-request
  170. output2 = self.app.get("/test/pull-request/1")
  171. self.assertEqual(output2.status_code, 200)
  172. def test_accessing_pr_patch_fork_deleted(self):
  173. """ Test accessing the PR's patch if the fork has been deleted. """
  174. # Delete fork on disk
  175. project = pagure.lib.query.get_authorized_project(
  176. self.session, "test", user="foo"
  177. )
  178. repo_path = os.path.join(self.path, "repos", project.path)
  179. self.assertTrue(os.path.exists(repo_path))
  180. shutil.rmtree(repo_path)
  181. self.assertFalse(os.path.exists(repo_path))
  182. # Delete fork in the DB
  183. self.session.delete(project)
  184. self.session.commit()
  185. # View the pull-request
  186. output = self.app.get("/test/pull-request/1.patch")
  187. self.assertEqual(output.status_code, 200)
  188. self.assertIn(
  189. "--- a/sources\n+++ b/sources\n@@ -1,2 +1,4 @@",
  190. output.get_data(as_text=True),
  191. )
  192. def test_accessing_pr_branch_deleted(self):
  193. """ Test accessing the PR if branch it originates from has been
  194. deleted. """
  195. project = pagure.lib.query.get_authorized_project(
  196. self.session, "test", user="foo"
  197. )
  198. # Check the branches before
  199. gitrepo = os.path.join(self.path, "repos", project.path)
  200. repo = pygit2.Repository(gitrepo)
  201. self.assertEqual(
  202. list(repo.listall_references()), ["refs/heads/feature"]
  203. )
  204. # Delete branch of the fork
  205. user = tests.FakeUser(username="foo")
  206. with tests.user_set(self.app.application, user):
  207. output = self.app.post(
  208. "/fork/foo/test/b/feature/delete", follow_redirects=True
  209. )
  210. self.assertEqual(output.status_code, 200)
  211. # Check the branches after
  212. gitrepo = os.path.join(self.path, "repos", project.path)
  213. repo = pygit2.Repository(gitrepo)
  214. self.assertEqual(list(repo.listall_references()), [])
  215. # View the pull-request
  216. output2 = self.app.get("/test/pull-request/1")
  217. self.assertEqual(output2.status_code, 200)
  218. def test_accessing_pr_patch_branch_deleted(self):
  219. """ Test accessing the PR's patch if branch it originates from has
  220. been deleted. """
  221. project = pagure.lib.query.get_authorized_project(
  222. self.session, "test", user="foo"
  223. )
  224. # Check the branches before
  225. gitrepo = os.path.join(self.path, "repos", project.path)
  226. repo = pygit2.Repository(gitrepo)
  227. self.assertEqual(
  228. list(repo.listall_references()), ["refs/heads/feature"]
  229. )
  230. # Delete branch of the fork
  231. user = tests.FakeUser(username="foo")
  232. with tests.user_set(self.app.application, user):
  233. output = self.app.post(
  234. "/fork/foo/test/b/feature/delete", follow_redirects=True
  235. )
  236. self.assertEqual(output.status_code, 200)
  237. # Check the branches after
  238. gitrepo = os.path.join(self.path, "repos", project.path)
  239. repo = pygit2.Repository(gitrepo)
  240. self.assertEqual(list(repo.listall_references()), [])
  241. # View the pull-request
  242. output = self.app.get("/test/pull-request/1.patch")
  243. self.assertEqual(output.status_code, 200)
  244. self.assertIn(
  245. "--- a/sources\n+++ b/sources\n@@ -1,2 +1,4 @@",
  246. output.get_data(as_text=True),
  247. )
  248. if __name__ == "__main__":
  249. unittest.main(verbosity=2)