test_pagure_lib_task_mirror.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341
  1. # -*- coding: utf-8 -*-
  2. """
  3. (c) 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 datetime
  9. import os
  10. import shutil
  11. import sys
  12. import tempfile
  13. import time
  14. import unittest
  15. import pygit2
  16. import six
  17. from mock import patch, MagicMock, call
  18. sys.path.insert(
  19. 0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
  20. )
  21. import pagure.lib.plugins
  22. import pagure.lib.query
  23. import pagure.lib.tasks_mirror
  24. import tests
  25. import pagure.lib.tasks_mirror
  26. class PagureLibTaskMirrortests(tests.Modeltests):
  27. """Tests for pagure.lib.task_mirror"""
  28. maxDiff = None
  29. def setUp(self):
  30. """Set up the environnment, ran before every tests."""
  31. super(PagureLibTaskMirrortests, self).setUp()
  32. pagure.config.config["REQUESTS_FOLDER"] = None
  33. self.sshkeydir = os.path.join(self.path, "sshkeys")
  34. pagure.config.config["MIRROR_SSHKEYS_FOLDER"] = self.sshkeydir
  35. tests.create_projects(self.session)
  36. def test_create_ssh_key(self):
  37. """Test the _create_ssh_key method."""
  38. # before
  39. self.assertFalse(os.path.exists(self.sshkeydir))
  40. os.mkdir(self.sshkeydir)
  41. self.assertEqual(sorted(os.listdir(self.sshkeydir)), [])
  42. keyfile = os.path.join(self.sshkeydir, "testkey")
  43. pagure.lib.tasks_mirror._create_ssh_key(keyfile)
  44. # after
  45. self.assertEqual(
  46. sorted(os.listdir(self.sshkeydir)), ["testkey", "testkey.pub"]
  47. )
  48. def test_setup_mirroring(self):
  49. """Test the setup_mirroring method."""
  50. # before
  51. self.assertFalse(os.path.exists(self.sshkeydir))
  52. project = pagure.lib.query.get_authorized_project(self.session, "test")
  53. self.assertIsNone(project.mirror_hook)
  54. # Install the plugin at the DB level
  55. plugin = pagure.lib.plugins.get_plugin("Mirroring")
  56. dbobj = plugin.db_object()
  57. dbobj.project_id = project.id
  58. self.session.add(dbobj)
  59. self.session.commit()
  60. pagure.lib.tasks_mirror.setup_mirroring(
  61. username=None, namespace=None, name="test"
  62. )
  63. # after
  64. self.assertEqual(
  65. sorted(os.listdir(self.sshkeydir)), ["test", "test.pub"]
  66. )
  67. project = pagure.lib.query.get_authorized_project(self.session, "test")
  68. self.assertIsNotNone(project.mirror_hook.public_key)
  69. self.assertTrue(project.mirror_hook.public_key.startswith("ssh-rsa "))
  70. def test_setup_mirroring_ssh_folder_exists_wrong_permissions(self):
  71. """Test the setup_mirroring method."""
  72. os.makedirs(self.sshkeydir)
  73. # before
  74. self.assertEqual(sorted(os.listdir(self.sshkeydir)), [])
  75. project = pagure.lib.query.get_authorized_project(self.session, "test")
  76. self.assertIsNone(project.mirror_hook)
  77. # Install the plugin at the DB level
  78. plugin = pagure.lib.plugins.get_plugin("Mirroring")
  79. dbobj = plugin.db_object()
  80. dbobj.project_id = project.id
  81. self.session.add(dbobj)
  82. self.session.commit()
  83. self.assertRaises(
  84. pagure.exceptions.PagureException,
  85. pagure.lib.tasks_mirror.setup_mirroring,
  86. username=None,
  87. namespace=None,
  88. name="test",
  89. )
  90. # after
  91. self.assertEqual(sorted(os.listdir(self.sshkeydir)), [])
  92. project = pagure.lib.query.get_authorized_project(self.session, "test")
  93. self.assertIsNone(project.mirror_hook.public_key)
  94. def test_setup_mirroring_ssh_folder_symlink(self):
  95. """Test the setup_mirroring method."""
  96. os.symlink(self.path, self.sshkeydir)
  97. # before
  98. self.assertEqual(
  99. sorted(os.listdir(self.sshkeydir)),
  100. [
  101. "attachments",
  102. "config",
  103. "forks",
  104. "releases",
  105. "remotes",
  106. "repos",
  107. "sshkeys",
  108. ],
  109. )
  110. project = pagure.lib.query.get_authorized_project(self.session, "test")
  111. self.assertIsNone(project.mirror_hook)
  112. # Install the plugin at the DB level
  113. plugin = pagure.lib.plugins.get_plugin("Mirroring")
  114. dbobj = plugin.db_object()
  115. dbobj.project_id = project.id
  116. self.session.add(dbobj)
  117. self.session.commit()
  118. self.assertRaises(
  119. pagure.exceptions.PagureException,
  120. pagure.lib.tasks_mirror.setup_mirroring,
  121. username=None,
  122. namespace=None,
  123. name="test",
  124. )
  125. # after
  126. self.assertEqual(
  127. sorted(os.listdir(self.sshkeydir)),
  128. [
  129. "attachments",
  130. "config",
  131. "forks",
  132. "releases",
  133. "remotes",
  134. "repos",
  135. "sshkeys",
  136. ],
  137. )
  138. project = pagure.lib.query.get_authorized_project(self.session, "test")
  139. self.assertIsNone(project.mirror_hook.public_key)
  140. @patch("os.getuid", MagicMock(return_value=450))
  141. def test_setup_mirroring_ssh_folder_owner(self):
  142. """Test the setup_mirroring method."""
  143. os.makedirs(self.sshkeydir, mode=0o700)
  144. # before
  145. self.assertEqual(sorted(os.listdir(self.sshkeydir)), [])
  146. project = pagure.lib.query.get_authorized_project(self.session, "test")
  147. self.assertIsNone(project.mirror_hook)
  148. # Install the plugin at the DB level
  149. plugin = pagure.lib.plugins.get_plugin("Mirroring")
  150. dbobj = plugin.db_object()
  151. dbobj.project_id = project.id
  152. self.session.add(dbobj)
  153. self.session.commit()
  154. self.assertRaises(
  155. pagure.exceptions.PagureException,
  156. pagure.lib.tasks_mirror.setup_mirroring,
  157. username=None,
  158. namespace=None,
  159. name="test",
  160. )
  161. # after
  162. self.assertEqual(sorted(os.listdir(self.sshkeydir)), [])
  163. project = pagure.lib.query.get_authorized_project(self.session, "test")
  164. self.assertIsNone(project.mirror_hook.public_key)
  165. class PagureLibTaskMirrorSetuptests(tests.Modeltests):
  166. """Tests for pagure.lib.task_mirror"""
  167. maxDiff = None
  168. def setUp(self):
  169. """Set up the environnment, ran before every tests."""
  170. super(PagureLibTaskMirrorSetuptests, self).setUp()
  171. pagure.config.config["REQUESTS_FOLDER"] = None
  172. self.sshkeydir = os.path.join(self.path, "sshkeys")
  173. pagure.config.config["MIRROR_SSHKEYS_FOLDER"] = self.sshkeydir
  174. tests.create_projects(self.session)
  175. project = pagure.lib.query.get_authorized_project(self.session, "test")
  176. # Install the plugin at the DB level
  177. plugin = pagure.lib.plugins.get_plugin("Mirroring")
  178. dbobj = plugin.db_object()
  179. dbobj.target = "ssh://user@localhost.localdomain/foobar.git"
  180. dbobj.project_id = project.id
  181. self.session.add(dbobj)
  182. self.session.commit()
  183. pagure.lib.tasks_mirror.setup_mirroring(
  184. username=None, namespace=None, name="test"
  185. )
  186. def test_setup_mirroring_twice(self):
  187. """Test the setup_mirroring method."""
  188. # before
  189. self.assertEqual(
  190. sorted(os.listdir(self.sshkeydir)), ["test", "test.pub"]
  191. )
  192. project = pagure.lib.query.get_authorized_project(self.session, "test")
  193. self.assertIsNotNone(project.mirror_hook.public_key)
  194. before_key = project.mirror_hook.public_key
  195. self.assertTrue(project.mirror_hook.public_key.startswith("ssh-rsa "))
  196. self.assertRaises(
  197. pagure.exceptions.PagureException,
  198. pagure.lib.tasks_mirror.setup_mirroring,
  199. username=None,
  200. namespace=None,
  201. name="test",
  202. )
  203. # after
  204. self.assertEqual(
  205. sorted(os.listdir(self.sshkeydir)), ["test", "test.pub"]
  206. )
  207. project = pagure.lib.query.get_authorized_project(self.session, "test")
  208. self.assertIsNotNone(project.mirror_hook.public_key)
  209. self.assertEqual(project.mirror_hook.public_key, before_key)
  210. def test_teardown_mirroring(self):
  211. """Test the teardown_mirroring method."""
  212. # before
  213. self.assertEqual(
  214. sorted(os.listdir(self.sshkeydir)), ["test", "test.pub"]
  215. )
  216. project = pagure.lib.query.get_authorized_project(self.session, "test")
  217. self.assertIsNotNone(project.mirror_hook.public_key)
  218. self.assertTrue(project.mirror_hook.public_key.startswith("ssh-rsa "))
  219. pagure.lib.tasks_mirror.teardown_mirroring(
  220. username=None, namespace=None, name="test"
  221. )
  222. # after
  223. self.session = pagure.lib.query.create_session(self.dbpath)
  224. self.assertEqual(sorted(os.listdir(self.sshkeydir)), [])
  225. project = pagure.lib.query.get_authorized_project(self.session, "test")
  226. self.assertIsNone(project.mirror_hook.public_key)
  227. @patch("pagure.lib.git.read_git_lines")
  228. def test_mirror_project(self, rgl):
  229. """Test the mirror_project method."""
  230. rgl.return_value = ("stdout", "stderr")
  231. tests.create_projects_git(os.path.join(self.path, "repos"), bare=True)
  232. # before
  233. self.assertEqual(
  234. sorted(os.listdir(self.sshkeydir)), ["test", "test.pub"]
  235. )
  236. project = pagure.lib.query.get_authorized_project(self.session, "test")
  237. self.assertIsNotNone(project.mirror_hook.public_key)
  238. self.assertTrue(project.mirror_hook.public_key.startswith("ssh-rsa "))
  239. pagure.lib.tasks_mirror.mirror_project(
  240. username=None, namespace=None, name="test"
  241. )
  242. # after
  243. self.assertEqual(
  244. sorted(os.listdir(self.sshkeydir)), ["test", "test.pub"]
  245. )
  246. project = pagure.lib.query.get_authorized_project(self.session, "test")
  247. self.assertIsNotNone(project.mirror_hook.public_key)
  248. self.assertTrue(project.mirror_hook.public_key.startswith("ssh-rsa "))
  249. ssh_script = os.path.abspath(
  250. os.path.join(
  251. os.path.dirname(os.path.abspath(__file__)),
  252. "..",
  253. "pagure",
  254. "lib",
  255. "ssh_script.sh",
  256. )
  257. )
  258. calls = [
  259. call(
  260. [
  261. "push",
  262. "--mirror",
  263. "ssh://user@localhost.localdomain/foobar.git",
  264. ],
  265. abspath=os.path.join(self.path, "repos", "test.git"),
  266. env={
  267. "GIT_SSH": ssh_script,
  268. "SSHKEY": "%s/sshkeys/test" % self.path,
  269. },
  270. error=True,
  271. )
  272. ]
  273. self.assertEqual(rgl.call_count, 1)
  274. self.assertEqual(calls, rgl.mock_calls)
  275. if __name__ == "__main__":
  276. unittest.main(verbosity=2)