2
0

test_repository.py 56 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566
  1. # test_repository.py -- tests for repository.py
  2. # Copyright (C) 2007 James Westby <jw+debian@jameswestby.net>
  3. #
  4. # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
  5. # General Public License as public by the Free Software Foundation; version 2.0
  6. # or (at your option) any later version. You can redistribute it and/or
  7. # modify it under the terms of either of these two licenses.
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. #
  15. # You should have received a copy of the licenses; if not, see
  16. # <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
  17. # and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
  18. # License, Version 2.0.
  19. #
  20. """Tests for the repository."""
  21. import glob
  22. import locale
  23. import os
  24. import shutil
  25. import stat
  26. import sys
  27. import tempfile
  28. import warnings
  29. from dulwich import errors, objects, porcelain
  30. from dulwich.config import Config
  31. from dulwich.errors import NotGitRepository
  32. from dulwich.object_store import tree_lookup_path
  33. from dulwich.repo import (
  34. InvalidUserIdentity,
  35. MemoryRepo,
  36. Repo,
  37. UnsupportedExtension,
  38. UnsupportedVersion,
  39. check_user_identity,
  40. )
  41. from dulwich.tests.utils import open_repo, setup_warning_catcher, tear_down_repo
  42. from . import TestCase, skipIf
  43. missing_sha = b"b91fa4d900e17e99b433218e988c4eb4a3e9a097"
  44. class CreateRepositoryTests(TestCase):
  45. def assertFileContentsEqual(self, expected, repo, path):
  46. f = repo.get_named_file(path)
  47. if not f:
  48. self.assertEqual(expected, None)
  49. else:
  50. with f:
  51. self.assertEqual(expected, f.read())
  52. def _check_repo_contents(self, repo, expect_bare):
  53. self.assertEqual(expect_bare, repo.bare)
  54. self.assertFileContentsEqual(b"Unnamed repository", repo, "description")
  55. self.assertFileContentsEqual(b"", repo, os.path.join("info", "exclude"))
  56. self.assertFileContentsEqual(None, repo, "nonexistent file")
  57. barestr = b"bare = " + str(expect_bare).lower().encode("ascii")
  58. with repo.get_named_file("config") as f:
  59. config_text = f.read()
  60. self.assertIn(barestr, config_text, f"{config_text!r}")
  61. expect_filemode = sys.platform != "win32"
  62. barestr = b"filemode = " + str(expect_filemode).lower().encode("ascii")
  63. with repo.get_named_file("config") as f:
  64. config_text = f.read()
  65. self.assertIn(barestr, config_text, f"{config_text!r}")
  66. if isinstance(repo, Repo):
  67. expected_mode = "0o100644" if expect_filemode else "0o100666"
  68. expected = {
  69. "HEAD": expected_mode,
  70. "config": expected_mode,
  71. "description": expected_mode,
  72. }
  73. actual = {
  74. f[len(repo._controldir) + 1 :]: oct(os.stat(f).st_mode)
  75. for f in glob.glob(os.path.join(repo._controldir, "*"))
  76. if os.path.isfile(f)
  77. }
  78. self.assertEqual(expected, actual)
  79. def test_create_memory(self):
  80. repo = MemoryRepo.init_bare([], {})
  81. self._check_repo_contents(repo, True)
  82. def test_create_disk_bare(self):
  83. tmp_dir = tempfile.mkdtemp()
  84. self.addCleanup(shutil.rmtree, tmp_dir)
  85. repo = Repo.init_bare(tmp_dir)
  86. self.assertEqual(tmp_dir, repo._controldir)
  87. self._check_repo_contents(repo, True)
  88. def test_create_disk_non_bare(self):
  89. tmp_dir = tempfile.mkdtemp()
  90. self.addCleanup(shutil.rmtree, tmp_dir)
  91. repo = Repo.init(tmp_dir)
  92. self.assertEqual(os.path.join(tmp_dir, ".git"), repo._controldir)
  93. self._check_repo_contents(repo, False)
  94. def test_create_disk_non_bare_mkdir(self):
  95. tmp_dir = tempfile.mkdtemp()
  96. target_dir = os.path.join(tmp_dir, "target")
  97. self.addCleanup(shutil.rmtree, tmp_dir)
  98. repo = Repo.init(target_dir, mkdir=True)
  99. self.assertEqual(os.path.join(target_dir, ".git"), repo._controldir)
  100. self._check_repo_contents(repo, False)
  101. def test_create_disk_bare_mkdir(self):
  102. tmp_dir = tempfile.mkdtemp()
  103. target_dir = os.path.join(tmp_dir, "target")
  104. self.addCleanup(shutil.rmtree, tmp_dir)
  105. repo = Repo.init_bare(target_dir, mkdir=True)
  106. self.assertEqual(target_dir, repo._controldir)
  107. self._check_repo_contents(repo, True)
  108. class MemoryRepoTests(TestCase):
  109. def test_set_description(self):
  110. r = MemoryRepo.init_bare([], {})
  111. description = b"Some description"
  112. r.set_description(description)
  113. self.assertEqual(description, r.get_description())
  114. def test_pull_into(self):
  115. r = MemoryRepo.init_bare([], {})
  116. repo = open_repo("a.git")
  117. self.addCleanup(tear_down_repo, repo)
  118. repo.fetch(r)
  119. class RepositoryRootTests(TestCase):
  120. def mkdtemp(self):
  121. return tempfile.mkdtemp()
  122. def open_repo(self, name):
  123. temp_dir = self.mkdtemp()
  124. repo = open_repo(name, temp_dir)
  125. self.addCleanup(tear_down_repo, repo)
  126. return repo
  127. def test_simple_props(self):
  128. r = self.open_repo("a.git")
  129. self.assertEqual(r.controldir(), r.path)
  130. def test_setitem(self):
  131. r = self.open_repo("a.git")
  132. r[b"refs/tags/foo"] = b"a90fa2d900a17e99b433217e988c4eb4a2e9a097"
  133. self.assertEqual(
  134. b"a90fa2d900a17e99b433217e988c4eb4a2e9a097", r[b"refs/tags/foo"].id
  135. )
  136. def test_getitem_unicode(self):
  137. r = self.open_repo("a.git")
  138. test_keys = [
  139. (b"refs/heads/master", True),
  140. (b"a90fa2d900a17e99b433217e988c4eb4a2e9a097", True),
  141. (b"11" * 19 + b"--", False),
  142. ]
  143. for k, contained in test_keys:
  144. self.assertEqual(k in r, contained)
  145. # Avoid deprecation warning under Py3.2+
  146. if getattr(self, "assertRaisesRegex", None):
  147. assertRaisesRegexp = self.assertRaisesRegex
  148. else:
  149. assertRaisesRegexp = self.assertRaisesRegexp
  150. for k, _ in test_keys:
  151. assertRaisesRegexp(
  152. TypeError,
  153. "'name' must be bytestring, not int",
  154. r.__getitem__,
  155. 12,
  156. )
  157. def test_delitem(self):
  158. r = self.open_repo("a.git")
  159. del r[b"refs/heads/master"]
  160. self.assertRaises(KeyError, lambda: r[b"refs/heads/master"])
  161. del r[b"HEAD"]
  162. self.assertRaises(KeyError, lambda: r[b"HEAD"])
  163. self.assertRaises(ValueError, r.__delitem__, b"notrefs/foo")
  164. def test_get_refs(self):
  165. r = self.open_repo("a.git")
  166. self.assertEqual(
  167. {
  168. b"HEAD": b"a90fa2d900a17e99b433217e988c4eb4a2e9a097",
  169. b"refs/heads/master": b"a90fa2d900a17e99b433217e988c4eb4a2e9a097",
  170. b"refs/tags/mytag": b"28237f4dc30d0d462658d6b937b08a0f0b6ef55a",
  171. b"refs/tags/mytag-packed": b"b0931cadc54336e78a1d980420e3268903b57a50",
  172. },
  173. r.get_refs(),
  174. )
  175. def test_head(self):
  176. r = self.open_repo("a.git")
  177. self.assertEqual(r.head(), b"a90fa2d900a17e99b433217e988c4eb4a2e9a097")
  178. def test_get_object(self):
  179. r = self.open_repo("a.git")
  180. obj = r.get_object(r.head())
  181. self.assertEqual(obj.type_name, b"commit")
  182. def test_get_object_non_existant(self):
  183. r = self.open_repo("a.git")
  184. self.assertRaises(KeyError, r.get_object, missing_sha)
  185. def test_contains_object(self):
  186. r = self.open_repo("a.git")
  187. self.assertIn(r.head(), r)
  188. self.assertNotIn(b"z" * 40, r)
  189. def test_contains_ref(self):
  190. r = self.open_repo("a.git")
  191. self.assertIn(b"HEAD", r)
  192. def test_get_no_description(self):
  193. r = self.open_repo("a.git")
  194. self.assertIs(None, r.get_description())
  195. def test_get_description(self):
  196. r = self.open_repo("a.git")
  197. with open(os.path.join(r.path, "description"), "wb") as f:
  198. f.write(b"Some description")
  199. self.assertEqual(b"Some description", r.get_description())
  200. def test_set_description(self):
  201. r = self.open_repo("a.git")
  202. description = b"Some description"
  203. r.set_description(description)
  204. self.assertEqual(description, r.get_description())
  205. def test_contains_missing(self):
  206. r = self.open_repo("a.git")
  207. self.assertNotIn(b"bar", r)
  208. def test_get_peeled(self):
  209. # unpacked ref
  210. r = self.open_repo("a.git")
  211. tag_sha = b"28237f4dc30d0d462658d6b937b08a0f0b6ef55a"
  212. self.assertNotEqual(r[tag_sha].sha().hexdigest(), r.head())
  213. self.assertEqual(r.get_peeled(b"refs/tags/mytag"), r.head())
  214. # packed ref with cached peeled value
  215. packed_tag_sha = b"b0931cadc54336e78a1d980420e3268903b57a50"
  216. parent_sha = r[r.head()].parents[0]
  217. self.assertNotEqual(r[packed_tag_sha].sha().hexdigest(), parent_sha)
  218. self.assertEqual(r.get_peeled(b"refs/tags/mytag-packed"), parent_sha)
  219. # TODO: add more corner cases to test repo
  220. def test_get_peeled_not_tag(self):
  221. r = self.open_repo("a.git")
  222. self.assertEqual(r.get_peeled(b"HEAD"), r.head())
  223. def test_get_parents(self):
  224. r = self.open_repo("a.git")
  225. self.assertEqual(
  226. [b"2a72d929692c41d8554c07f6301757ba18a65d91"],
  227. r.get_parents(b"a90fa2d900a17e99b433217e988c4eb4a2e9a097"),
  228. )
  229. r.update_shallow([b"a90fa2d900a17e99b433217e988c4eb4a2e9a097"], None)
  230. self.assertEqual([], r.get_parents(b"a90fa2d900a17e99b433217e988c4eb4a2e9a097"))
  231. def test_get_walker(self):
  232. r = self.open_repo("a.git")
  233. # include defaults to [r.head()]
  234. self.assertEqual(
  235. [e.commit.id for e in r.get_walker()],
  236. [r.head(), b"2a72d929692c41d8554c07f6301757ba18a65d91"],
  237. )
  238. self.assertEqual(
  239. [
  240. e.commit.id
  241. for e in r.get_walker([b"2a72d929692c41d8554c07f6301757ba18a65d91"])
  242. ],
  243. [b"2a72d929692c41d8554c07f6301757ba18a65d91"],
  244. )
  245. self.assertEqual(
  246. [
  247. e.commit.id
  248. for e in r.get_walker(b"2a72d929692c41d8554c07f6301757ba18a65d91")
  249. ],
  250. [b"2a72d929692c41d8554c07f6301757ba18a65d91"],
  251. )
  252. def assertFilesystemHidden(self, path):
  253. if sys.platform != "win32":
  254. return
  255. import ctypes
  256. from ctypes.wintypes import DWORD, LPCWSTR
  257. GetFileAttributesW = ctypes.WINFUNCTYPE(DWORD, LPCWSTR)(
  258. ("GetFileAttributesW", ctypes.windll.kernel32)
  259. )
  260. self.assertTrue(2 & GetFileAttributesW(path))
  261. def test_init_existing(self):
  262. tmp_dir = self.mkdtemp()
  263. self.addCleanup(shutil.rmtree, tmp_dir)
  264. t = Repo.init(tmp_dir)
  265. self.addCleanup(t.close)
  266. self.assertEqual(os.listdir(tmp_dir), [".git"])
  267. self.assertFilesystemHidden(os.path.join(tmp_dir, ".git"))
  268. def test_init_mkdir(self):
  269. tmp_dir = self.mkdtemp()
  270. self.addCleanup(shutil.rmtree, tmp_dir)
  271. repo_dir = os.path.join(tmp_dir, "a-repo")
  272. t = Repo.init(repo_dir, mkdir=True)
  273. self.addCleanup(t.close)
  274. self.assertEqual(os.listdir(repo_dir), [".git"])
  275. self.assertFilesystemHidden(os.path.join(repo_dir, ".git"))
  276. def test_init_mkdir_unicode(self):
  277. repo_name = "\xa7"
  278. try:
  279. os.fsencode(repo_name)
  280. except UnicodeEncodeError:
  281. self.skipTest("filesystem lacks unicode support")
  282. tmp_dir = self.mkdtemp()
  283. self.addCleanup(shutil.rmtree, tmp_dir)
  284. repo_dir = os.path.join(tmp_dir, repo_name)
  285. t = Repo.init(repo_dir, mkdir=True)
  286. self.addCleanup(t.close)
  287. self.assertEqual(os.listdir(repo_dir), [".git"])
  288. self.assertFilesystemHidden(os.path.join(repo_dir, ".git"))
  289. @skipIf(sys.platform == "win32", "fails on Windows")
  290. def test_fetch(self):
  291. r = self.open_repo("a.git")
  292. tmp_dir = self.mkdtemp()
  293. self.addCleanup(shutil.rmtree, tmp_dir)
  294. t = Repo.init(tmp_dir)
  295. self.addCleanup(t.close)
  296. r.fetch(t)
  297. self.assertIn(b"a90fa2d900a17e99b433217e988c4eb4a2e9a097", t)
  298. self.assertIn(b"a90fa2d900a17e99b433217e988c4eb4a2e9a097", t)
  299. self.assertIn(b"a90fa2d900a17e99b433217e988c4eb4a2e9a097", t)
  300. self.assertIn(b"28237f4dc30d0d462658d6b937b08a0f0b6ef55a", t)
  301. self.assertIn(b"b0931cadc54336e78a1d980420e3268903b57a50", t)
  302. @skipIf(sys.platform == "win32", "fails on Windows")
  303. def test_fetch_ignores_missing_refs(self):
  304. r = self.open_repo("a.git")
  305. missing = b"1234566789123456789123567891234657373833"
  306. r.refs[b"refs/heads/blah"] = missing
  307. tmp_dir = self.mkdtemp()
  308. self.addCleanup(shutil.rmtree, tmp_dir)
  309. t = Repo.init(tmp_dir)
  310. self.addCleanup(t.close)
  311. r.fetch(t)
  312. self.assertIn(b"a90fa2d900a17e99b433217e988c4eb4a2e9a097", t)
  313. self.assertIn(b"a90fa2d900a17e99b433217e988c4eb4a2e9a097", t)
  314. self.assertIn(b"a90fa2d900a17e99b433217e988c4eb4a2e9a097", t)
  315. self.assertIn(b"28237f4dc30d0d462658d6b937b08a0f0b6ef55a", t)
  316. self.assertIn(b"b0931cadc54336e78a1d980420e3268903b57a50", t)
  317. self.assertNotIn(missing, t)
  318. def test_clone(self):
  319. r = self.open_repo("a.git")
  320. tmp_dir = self.mkdtemp()
  321. self.addCleanup(shutil.rmtree, tmp_dir)
  322. with r.clone(tmp_dir, mkdir=False) as t:
  323. self.assertEqual(
  324. {
  325. b"HEAD": b"a90fa2d900a17e99b433217e988c4eb4a2e9a097",
  326. b"refs/remotes/origin/master": b"a90fa2d900a17e99b433217e988c4eb4a2e9a097",
  327. b"refs/remotes/origin/HEAD": b"a90fa2d900a17e99b433217e988c4eb4a2e9a097",
  328. b"refs/heads/master": b"a90fa2d900a17e99b433217e988c4eb4a2e9a097",
  329. b"refs/tags/mytag": b"28237f4dc30d0d462658d6b937b08a0f0b6ef55a",
  330. b"refs/tags/mytag-packed": b"b0931cadc54336e78a1d980420e3268903b57a50",
  331. },
  332. t.refs.as_dict(),
  333. )
  334. shas = [e.commit.id for e in r.get_walker()]
  335. self.assertEqual(
  336. shas, [t.head(), b"2a72d929692c41d8554c07f6301757ba18a65d91"]
  337. )
  338. c = t.get_config()
  339. encoded_path = r.path
  340. if not isinstance(encoded_path, bytes):
  341. encoded_path = os.fsencode(encoded_path)
  342. self.assertEqual(encoded_path, c.get((b"remote", b"origin"), b"url"))
  343. self.assertEqual(
  344. b"+refs/heads/*:refs/remotes/origin/*",
  345. c.get((b"remote", b"origin"), b"fetch"),
  346. )
  347. def test_clone_no_head(self):
  348. temp_dir = self.mkdtemp()
  349. self.addCleanup(shutil.rmtree, temp_dir)
  350. repo_dir = os.path.join(os.path.dirname(__file__), "..", "testdata", "repos")
  351. dest_dir = os.path.join(temp_dir, "a.git")
  352. shutil.copytree(os.path.join(repo_dir, "a.git"), dest_dir, symlinks=True)
  353. r = Repo(dest_dir)
  354. self.addCleanup(r.close)
  355. del r.refs[b"refs/heads/master"]
  356. del r.refs[b"HEAD"]
  357. t = r.clone(os.path.join(temp_dir, "b.git"), mkdir=True)
  358. self.addCleanup(t.close)
  359. self.assertEqual(
  360. {
  361. b"refs/tags/mytag": b"28237f4dc30d0d462658d6b937b08a0f0b6ef55a",
  362. b"refs/tags/mytag-packed": b"b0931cadc54336e78a1d980420e3268903b57a50",
  363. },
  364. t.refs.as_dict(),
  365. )
  366. def test_clone_empty(self):
  367. """Test clone() doesn't crash if HEAD points to a non-existing ref.
  368. This simulates cloning server-side bare repository either when it is
  369. still empty or if user renames master branch and pushes private repo
  370. to the server.
  371. Non-bare repo HEAD always points to an existing ref.
  372. """
  373. r = self.open_repo("empty.git")
  374. tmp_dir = self.mkdtemp()
  375. self.addCleanup(shutil.rmtree, tmp_dir)
  376. r.clone(tmp_dir, mkdir=False, bare=True)
  377. def test_reset_index_symlink_enabled(self):
  378. if sys.platform == "win32":
  379. self.skipTest("symlinks are not supported on Windows")
  380. tmp_dir = self.mkdtemp()
  381. self.addCleanup(shutil.rmtree, tmp_dir)
  382. o = Repo.init(os.path.join(tmp_dir, "s"), mkdir=True)
  383. os.symlink("foo", os.path.join(tmp_dir, "s", "bar"))
  384. o.stage("bar")
  385. o.do_commit(b"add symlink")
  386. t = o.clone(os.path.join(tmp_dir, "t"), symlinks=True)
  387. o.close()
  388. bar_path = os.path.join(tmp_dir, "t", "bar")
  389. if sys.platform == "win32":
  390. with open(bar_path) as f:
  391. self.assertEqual("foo", f.read())
  392. else:
  393. self.assertEqual("foo", os.readlink(bar_path))
  394. t.close()
  395. def test_reset_index_symlink_disabled(self):
  396. tmp_dir = self.mkdtemp()
  397. self.addCleanup(shutil.rmtree, tmp_dir)
  398. o = Repo.init(os.path.join(tmp_dir, "s"), mkdir=True)
  399. o.close()
  400. os.symlink("foo", os.path.join(tmp_dir, "s", "bar"))
  401. o.stage("bar")
  402. o.do_commit(b"add symlink")
  403. t = o.clone(os.path.join(tmp_dir, "t"), symlinks=False)
  404. with open(os.path.join(tmp_dir, "t", "bar")) as f:
  405. self.assertEqual("foo", f.read())
  406. t.close()
  407. def test_clone_bare(self):
  408. r = self.open_repo("a.git")
  409. tmp_dir = self.mkdtemp()
  410. self.addCleanup(shutil.rmtree, tmp_dir)
  411. t = r.clone(tmp_dir, mkdir=False)
  412. t.close()
  413. def test_clone_checkout_and_bare(self):
  414. r = self.open_repo("a.git")
  415. tmp_dir = self.mkdtemp()
  416. self.addCleanup(shutil.rmtree, tmp_dir)
  417. self.assertRaises(
  418. ValueError, r.clone, tmp_dir, mkdir=False, checkout=True, bare=True
  419. )
  420. def test_clone_branch(self):
  421. r = self.open_repo("a.git")
  422. r.refs[b"refs/heads/mybranch"] = b"28237f4dc30d0d462658d6b937b08a0f0b6ef55a"
  423. tmp_dir = self.mkdtemp()
  424. self.addCleanup(shutil.rmtree, tmp_dir)
  425. with r.clone(tmp_dir, mkdir=False, branch=b"mybranch") as t:
  426. # HEAD should point to specified branch and not origin HEAD
  427. chain, sha = t.refs.follow(b"HEAD")
  428. self.assertEqual(chain[-1], b"refs/heads/mybranch")
  429. self.assertEqual(sha, b"28237f4dc30d0d462658d6b937b08a0f0b6ef55a")
  430. self.assertEqual(
  431. t.refs[b"refs/remotes/origin/HEAD"],
  432. b"a90fa2d900a17e99b433217e988c4eb4a2e9a097",
  433. )
  434. def test_clone_tag(self):
  435. r = self.open_repo("a.git")
  436. tmp_dir = self.mkdtemp()
  437. self.addCleanup(shutil.rmtree, tmp_dir)
  438. with r.clone(tmp_dir, mkdir=False, branch=b"mytag") as t:
  439. # HEAD should be detached (and not a symbolic ref) at tag
  440. self.assertEqual(
  441. t.refs.read_ref(b"HEAD"),
  442. b"28237f4dc30d0d462658d6b937b08a0f0b6ef55a",
  443. )
  444. self.assertEqual(
  445. t.refs[b"refs/remotes/origin/HEAD"],
  446. b"a90fa2d900a17e99b433217e988c4eb4a2e9a097",
  447. )
  448. def test_clone_invalid_branch(self):
  449. r = self.open_repo("a.git")
  450. tmp_dir = self.mkdtemp()
  451. self.addCleanup(shutil.rmtree, tmp_dir)
  452. self.assertRaises(
  453. ValueError,
  454. r.clone,
  455. tmp_dir,
  456. mkdir=False,
  457. branch=b"mybranch",
  458. )
  459. def test_merge_history(self):
  460. r = self.open_repo("simple_merge.git")
  461. shas = [e.commit.id for e in r.get_walker()]
  462. self.assertEqual(
  463. shas,
  464. [
  465. b"5dac377bdded4c9aeb8dff595f0faeebcc8498cc",
  466. b"ab64bbdcc51b170d21588e5c5d391ee5c0c96dfd",
  467. b"4cffe90e0a41ad3f5190079d7c8f036bde29cbe6",
  468. b"60dacdc733de308bb77bb76ce0fb0f9b44c9769e",
  469. b"0d89f20333fbb1d2f3a94da77f4981373d8f4310",
  470. ],
  471. )
  472. def test_out_of_order_merge(self):
  473. """Test that revision history is ordered by date, not parent order."""
  474. r = self.open_repo("ooo_merge.git")
  475. shas = [e.commit.id for e in r.get_walker()]
  476. self.assertEqual(
  477. shas,
  478. [
  479. b"7601d7f6231db6a57f7bbb79ee52e4d462fd44d1",
  480. b"f507291b64138b875c28e03469025b1ea20bc614",
  481. b"fb5b0425c7ce46959bec94d54b9a157645e114f5",
  482. b"f9e39b120c68182a4ba35349f832d0e4e61f485c",
  483. ],
  484. )
  485. def test_get_tags_empty(self):
  486. r = self.open_repo("ooo_merge.git")
  487. self.assertEqual({}, r.refs.as_dict(b"refs/tags"))
  488. def test_get_config(self):
  489. r = self.open_repo("ooo_merge.git")
  490. self.assertIsInstance(r.get_config(), Config)
  491. def test_get_config_stack(self):
  492. r = self.open_repo("ooo_merge.git")
  493. self.assertIsInstance(r.get_config_stack(), Config)
  494. def test_common_revisions(self):
  495. """This test demonstrates that ``find_common_revisions()`` actually
  496. returns common heads, not revisions; dulwich already uses
  497. ``find_common_revisions()`` in such a manner (see
  498. ``Repo.find_objects()``).
  499. """
  500. expected_shas = {b"60dacdc733de308bb77bb76ce0fb0f9b44c9769e"}
  501. # Source for objects.
  502. r_base = self.open_repo("simple_merge.git")
  503. # Re-create each-side of the merge in simple_merge.git.
  504. #
  505. # Since the trees and blobs are missing, the repository created is
  506. # corrupted, but we're only checking for commits for the purpose of
  507. # this test, so it's immaterial.
  508. r1_dir = self.mkdtemp()
  509. self.addCleanup(shutil.rmtree, r1_dir)
  510. r1_commits = [
  511. b"ab64bbdcc51b170d21588e5c5d391ee5c0c96dfd", # HEAD
  512. b"60dacdc733de308bb77bb76ce0fb0f9b44c9769e",
  513. b"0d89f20333fbb1d2f3a94da77f4981373d8f4310",
  514. ]
  515. r2_dir = self.mkdtemp()
  516. self.addCleanup(shutil.rmtree, r2_dir)
  517. r2_commits = [
  518. b"4cffe90e0a41ad3f5190079d7c8f036bde29cbe6", # HEAD
  519. b"60dacdc733de308bb77bb76ce0fb0f9b44c9769e",
  520. b"0d89f20333fbb1d2f3a94da77f4981373d8f4310",
  521. ]
  522. r1 = Repo.init_bare(r1_dir)
  523. for c in r1_commits:
  524. r1.object_store.add_object(r_base.get_object(c))
  525. r1.refs[b"HEAD"] = r1_commits[0]
  526. r2 = Repo.init_bare(r2_dir)
  527. for c in r2_commits:
  528. r2.object_store.add_object(r_base.get_object(c))
  529. r2.refs[b"HEAD"] = r2_commits[0]
  530. # Finally, the 'real' testing!
  531. shas = r2.object_store.find_common_revisions(r1.get_graph_walker())
  532. self.assertEqual(set(shas), expected_shas)
  533. shas = r1.object_store.find_common_revisions(r2.get_graph_walker())
  534. self.assertEqual(set(shas), expected_shas)
  535. def test_shell_hook_pre_commit(self):
  536. if os.name != "posix":
  537. self.skipTest("shell hook tests requires POSIX shell")
  538. pre_commit_fail = """#!/bin/sh
  539. exit 1
  540. """
  541. pre_commit_success = """#!/bin/sh
  542. exit 0
  543. """
  544. repo_dir = os.path.join(self.mkdtemp())
  545. self.addCleanup(shutil.rmtree, repo_dir)
  546. r = Repo.init(repo_dir)
  547. self.addCleanup(r.close)
  548. pre_commit = os.path.join(r.controldir(), "hooks", "pre-commit")
  549. with open(pre_commit, "w") as f:
  550. f.write(pre_commit_fail)
  551. os.chmod(pre_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
  552. self.assertRaises(
  553. errors.CommitError,
  554. r.do_commit,
  555. b"failed commit",
  556. committer=b"Test Committer <test@nodomain.com>",
  557. author=b"Test Author <test@nodomain.com>",
  558. commit_timestamp=12345,
  559. commit_timezone=0,
  560. author_timestamp=12345,
  561. author_timezone=0,
  562. )
  563. with open(pre_commit, "w") as f:
  564. f.write(pre_commit_success)
  565. os.chmod(pre_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
  566. commit_sha = r.do_commit(
  567. b"empty commit",
  568. committer=b"Test Committer <test@nodomain.com>",
  569. author=b"Test Author <test@nodomain.com>",
  570. commit_timestamp=12395,
  571. commit_timezone=0,
  572. author_timestamp=12395,
  573. author_timezone=0,
  574. )
  575. self.assertEqual([], r[commit_sha].parents)
  576. def test_shell_hook_commit_msg(self):
  577. if os.name != "posix":
  578. self.skipTest("shell hook tests requires POSIX shell")
  579. commit_msg_fail = """#!/bin/sh
  580. exit 1
  581. """
  582. commit_msg_success = """#!/bin/sh
  583. exit 0
  584. """
  585. repo_dir = self.mkdtemp()
  586. self.addCleanup(shutil.rmtree, repo_dir)
  587. r = Repo.init(repo_dir)
  588. self.addCleanup(r.close)
  589. commit_msg = os.path.join(r.controldir(), "hooks", "commit-msg")
  590. with open(commit_msg, "w") as f:
  591. f.write(commit_msg_fail)
  592. os.chmod(commit_msg, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
  593. self.assertRaises(
  594. errors.CommitError,
  595. r.do_commit,
  596. b"failed commit",
  597. committer=b"Test Committer <test@nodomain.com>",
  598. author=b"Test Author <test@nodomain.com>",
  599. commit_timestamp=12345,
  600. commit_timezone=0,
  601. author_timestamp=12345,
  602. author_timezone=0,
  603. )
  604. with open(commit_msg, "w") as f:
  605. f.write(commit_msg_success)
  606. os.chmod(commit_msg, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
  607. commit_sha = r.do_commit(
  608. b"empty commit",
  609. committer=b"Test Committer <test@nodomain.com>",
  610. author=b"Test Author <test@nodomain.com>",
  611. commit_timestamp=12395,
  612. commit_timezone=0,
  613. author_timestamp=12395,
  614. author_timezone=0,
  615. )
  616. self.assertEqual([], r[commit_sha].parents)
  617. def test_shell_hook_pre_commit_add_files(self):
  618. if os.name != "posix":
  619. self.skipTest("shell hook tests requires POSIX shell")
  620. pre_commit_contents = """#!{executable}
  621. import sys
  622. sys.path.extend({path!r})
  623. from dulwich.repo import Repo
  624. with open('foo', 'w') as f:
  625. f.write('newfile')
  626. r = Repo('.')
  627. r.stage(['foo'])
  628. """.format(
  629. executable=sys.executable,
  630. path=[os.path.join(os.path.dirname(__file__), "..", ".."), *sys.path],
  631. )
  632. repo_dir = os.path.join(self.mkdtemp())
  633. self.addCleanup(shutil.rmtree, repo_dir)
  634. r = Repo.init(repo_dir)
  635. self.addCleanup(r.close)
  636. with open(os.path.join(repo_dir, "blah"), "w") as f:
  637. f.write("blah")
  638. r.stage(["blah"])
  639. pre_commit = os.path.join(r.controldir(), "hooks", "pre-commit")
  640. with open(pre_commit, "w") as f:
  641. f.write(pre_commit_contents)
  642. os.chmod(pre_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
  643. commit_sha = r.do_commit(
  644. b"new commit",
  645. committer=b"Test Committer <test@nodomain.com>",
  646. author=b"Test Author <test@nodomain.com>",
  647. commit_timestamp=12395,
  648. commit_timezone=0,
  649. author_timestamp=12395,
  650. author_timezone=0,
  651. )
  652. self.assertEqual([], r[commit_sha].parents)
  653. tree = r[r[commit_sha].tree]
  654. self.assertEqual({b"blah", b"foo"}, set(tree))
  655. def test_shell_hook_post_commit(self):
  656. if os.name != "posix":
  657. self.skipTest("shell hook tests requires POSIX shell")
  658. repo_dir = self.mkdtemp()
  659. self.addCleanup(shutil.rmtree, repo_dir)
  660. r = Repo.init(repo_dir)
  661. self.addCleanup(r.close)
  662. (fd, path) = tempfile.mkstemp(dir=repo_dir)
  663. os.close(fd)
  664. post_commit_msg = (
  665. """#!/bin/sh
  666. rm """
  667. + path
  668. + """
  669. """
  670. )
  671. root_sha = r.do_commit(
  672. b"empty commit",
  673. committer=b"Test Committer <test@nodomain.com>",
  674. author=b"Test Author <test@nodomain.com>",
  675. commit_timestamp=12345,
  676. commit_timezone=0,
  677. author_timestamp=12345,
  678. author_timezone=0,
  679. )
  680. self.assertEqual([], r[root_sha].parents)
  681. post_commit = os.path.join(r.controldir(), "hooks", "post-commit")
  682. with open(post_commit, "wb") as f:
  683. f.write(post_commit_msg.encode(locale.getpreferredencoding()))
  684. os.chmod(post_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
  685. commit_sha = r.do_commit(
  686. b"empty commit",
  687. committer=b"Test Committer <test@nodomain.com>",
  688. author=b"Test Author <test@nodomain.com>",
  689. commit_timestamp=12345,
  690. commit_timezone=0,
  691. author_timestamp=12345,
  692. author_timezone=0,
  693. )
  694. self.assertEqual([root_sha], r[commit_sha].parents)
  695. self.assertFalse(os.path.exists(path))
  696. post_commit_msg_fail = """#!/bin/sh
  697. exit 1
  698. """
  699. with open(post_commit, "w") as f:
  700. f.write(post_commit_msg_fail)
  701. os.chmod(post_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
  702. warnings.simplefilter("always", UserWarning)
  703. self.addCleanup(warnings.resetwarnings)
  704. warnings_list, restore_warnings = setup_warning_catcher()
  705. self.addCleanup(restore_warnings)
  706. commit_sha2 = r.do_commit(
  707. b"empty commit",
  708. committer=b"Test Committer <test@nodomain.com>",
  709. author=b"Test Author <test@nodomain.com>",
  710. commit_timestamp=12345,
  711. commit_timezone=0,
  712. author_timestamp=12345,
  713. author_timezone=0,
  714. )
  715. expected_warning = UserWarning(
  716. "post-commit hook failed: Hook post-commit exited with "
  717. "non-zero status 1",
  718. )
  719. for w in warnings_list:
  720. if type(w) is type(expected_warning) and w.args == expected_warning.args:
  721. break
  722. else:
  723. raise AssertionError(
  724. f"Expected warning {expected_warning!r} not in {warnings_list!r}"
  725. )
  726. self.assertEqual([commit_sha], r[commit_sha2].parents)
  727. def test_as_dict(self):
  728. def check(repo):
  729. self.assertEqual(
  730. repo.refs.subkeys(b"refs/tags"),
  731. repo.refs.subkeys(b"refs/tags/"),
  732. )
  733. self.assertEqual(
  734. repo.refs.as_dict(b"refs/tags"),
  735. repo.refs.as_dict(b"refs/tags/"),
  736. )
  737. self.assertEqual(
  738. repo.refs.as_dict(b"refs/heads"),
  739. repo.refs.as_dict(b"refs/heads/"),
  740. )
  741. bare = self.open_repo("a.git")
  742. tmp_dir = self.mkdtemp()
  743. self.addCleanup(shutil.rmtree, tmp_dir)
  744. with bare.clone(tmp_dir, mkdir=False) as nonbare:
  745. check(nonbare)
  746. check(bare)
  747. def test_working_tree(self):
  748. temp_dir = tempfile.mkdtemp()
  749. self.addCleanup(shutil.rmtree, temp_dir)
  750. worktree_temp_dir = tempfile.mkdtemp()
  751. self.addCleanup(shutil.rmtree, worktree_temp_dir)
  752. r = Repo.init(temp_dir)
  753. self.addCleanup(r.close)
  754. root_sha = r.do_commit(
  755. b"empty commit",
  756. committer=b"Test Committer <test@nodomain.com>",
  757. author=b"Test Author <test@nodomain.com>",
  758. commit_timestamp=12345,
  759. commit_timezone=0,
  760. author_timestamp=12345,
  761. author_timezone=0,
  762. )
  763. r.refs[b"refs/heads/master"] = root_sha
  764. w = Repo._init_new_working_directory(worktree_temp_dir, r)
  765. self.addCleanup(w.close)
  766. new_sha = w.do_commit(
  767. b"new commit",
  768. committer=b"Test Committer <test@nodomain.com>",
  769. author=b"Test Author <test@nodomain.com>",
  770. commit_timestamp=12345,
  771. commit_timezone=0,
  772. author_timestamp=12345,
  773. author_timezone=0,
  774. )
  775. w.refs[b"HEAD"] = new_sha
  776. self.assertEqual(
  777. os.path.abspath(r.controldir()), os.path.abspath(w.commondir())
  778. )
  779. self.assertEqual(r.refs.keys(), w.refs.keys())
  780. self.assertNotEqual(r.head(), w.head())
  781. class BuildRepoRootTests(TestCase):
  782. """Tests that build on-disk repos from scratch.
  783. Repos live in a temp dir and are torn down after each test. They start with
  784. a single commit in master having single file named 'a'.
  785. """
  786. def get_repo_dir(self):
  787. return os.path.join(tempfile.mkdtemp(), "test")
  788. def setUp(self):
  789. super().setUp()
  790. self._repo_dir = self.get_repo_dir()
  791. os.makedirs(self._repo_dir)
  792. r = self._repo = Repo.init(self._repo_dir)
  793. self.addCleanup(tear_down_repo, r)
  794. self.assertFalse(r.bare)
  795. self.assertEqual(b"ref: refs/heads/master", r.refs.read_ref(b"HEAD"))
  796. self.assertRaises(KeyError, lambda: r.refs[b"refs/heads/master"])
  797. with open(os.path.join(r.path, "a"), "wb") as f:
  798. f.write(b"file contents")
  799. r.stage(["a"])
  800. commit_sha = r.do_commit(
  801. b"msg",
  802. committer=b"Test Committer <test@nodomain.com>",
  803. author=b"Test Author <test@nodomain.com>",
  804. commit_timestamp=12345,
  805. commit_timezone=0,
  806. author_timestamp=12345,
  807. author_timezone=0,
  808. )
  809. self.assertEqual([], r[commit_sha].parents)
  810. self._root_commit = commit_sha
  811. def test_get_shallow(self):
  812. self.assertEqual(set(), self._repo.get_shallow())
  813. with open(os.path.join(self._repo.path, ".git", "shallow"), "wb") as f:
  814. f.write(b"a90fa2d900a17e99b433217e988c4eb4a2e9a097\n")
  815. self.assertEqual(
  816. {b"a90fa2d900a17e99b433217e988c4eb4a2e9a097"},
  817. self._repo.get_shallow(),
  818. )
  819. def test_update_shallow(self):
  820. self._repo.update_shallow(None, None) # no op
  821. self.assertEqual(set(), self._repo.get_shallow())
  822. self._repo.update_shallow([b"a90fa2d900a17e99b433217e988c4eb4a2e9a097"], None)
  823. self.assertEqual(
  824. {b"a90fa2d900a17e99b433217e988c4eb4a2e9a097"},
  825. self._repo.get_shallow(),
  826. )
  827. self._repo.update_shallow(
  828. [b"a90fa2d900a17e99b433217e988c4eb4a2e9a097"],
  829. [b"f9e39b120c68182a4ba35349f832d0e4e61f485c"],
  830. )
  831. self.assertEqual(
  832. {b"a90fa2d900a17e99b433217e988c4eb4a2e9a097"},
  833. self._repo.get_shallow(),
  834. )
  835. self._repo.update_shallow(None, [b"a90fa2d900a17e99b433217e988c4eb4a2e9a097"])
  836. self.assertEqual(set(), self._repo.get_shallow())
  837. self.assertEqual(
  838. False,
  839. os.path.exists(os.path.join(self._repo.controldir(), "shallow")),
  840. )
  841. def test_build_repo(self):
  842. r = self._repo
  843. self.assertEqual(b"ref: refs/heads/master", r.refs.read_ref(b"HEAD"))
  844. self.assertEqual(self._root_commit, r.refs[b"refs/heads/master"])
  845. expected_blob = objects.Blob.from_string(b"file contents")
  846. self.assertEqual(expected_blob.data, r[expected_blob.id].data)
  847. actual_commit = r[self._root_commit]
  848. self.assertEqual(b"msg", actual_commit.message)
  849. def test_commit_modified(self):
  850. r = self._repo
  851. with open(os.path.join(r.path, "a"), "wb") as f:
  852. f.write(b"new contents")
  853. r.stage(["a"])
  854. commit_sha = r.do_commit(
  855. b"modified a",
  856. committer=b"Test Committer <test@nodomain.com>",
  857. author=b"Test Author <test@nodomain.com>",
  858. commit_timestamp=12395,
  859. commit_timezone=0,
  860. author_timestamp=12395,
  861. author_timezone=0,
  862. )
  863. self.assertEqual([self._root_commit], r[commit_sha].parents)
  864. a_mode, a_id = tree_lookup_path(r.get_object, r[commit_sha].tree, b"a")
  865. self.assertEqual(stat.S_IFREG | 0o644, a_mode)
  866. self.assertEqual(b"new contents", r[a_id].data)
  867. @skipIf(not getattr(os, "symlink", None), "Requires symlink support")
  868. def test_commit_symlink(self):
  869. r = self._repo
  870. os.symlink("a", os.path.join(r.path, "b"))
  871. r.stage(["a", "b"])
  872. commit_sha = r.do_commit(
  873. b"Symlink b",
  874. committer=b"Test Committer <test@nodomain.com>",
  875. author=b"Test Author <test@nodomain.com>",
  876. commit_timestamp=12395,
  877. commit_timezone=0,
  878. author_timestamp=12395,
  879. author_timezone=0,
  880. )
  881. self.assertEqual([self._root_commit], r[commit_sha].parents)
  882. b_mode, b_id = tree_lookup_path(r.get_object, r[commit_sha].tree, b"b")
  883. self.assertTrue(stat.S_ISLNK(b_mode))
  884. self.assertEqual(b"a", r[b_id].data)
  885. def test_commit_merge_heads_file(self):
  886. tmp_dir = tempfile.mkdtemp()
  887. self.addCleanup(shutil.rmtree, tmp_dir)
  888. r = Repo.init(tmp_dir)
  889. with open(os.path.join(r.path, "a"), "w") as f:
  890. f.write("initial text")
  891. c1 = r.do_commit(
  892. b"initial commit",
  893. committer=b"Test Committer <test@nodomain.com>",
  894. author=b"Test Author <test@nodomain.com>",
  895. commit_timestamp=12395,
  896. commit_timezone=0,
  897. author_timestamp=12395,
  898. author_timezone=0,
  899. )
  900. with open(os.path.join(r.path, "a"), "w") as f:
  901. f.write("merged text")
  902. with open(os.path.join(r.path, ".git", "MERGE_HEAD"), "w") as f:
  903. f.write("c27a2d21dd136312d7fa9e8baabb82561a1727d0\n")
  904. r.stage(["a"])
  905. commit_sha = r.do_commit(
  906. b"deleted a",
  907. committer=b"Test Committer <test@nodomain.com>",
  908. author=b"Test Author <test@nodomain.com>",
  909. commit_timestamp=12395,
  910. commit_timezone=0,
  911. author_timestamp=12395,
  912. author_timezone=0,
  913. )
  914. self.assertEqual(
  915. [c1, b"c27a2d21dd136312d7fa9e8baabb82561a1727d0"],
  916. r[commit_sha].parents,
  917. )
  918. def test_commit_deleted(self):
  919. r = self._repo
  920. os.remove(os.path.join(r.path, "a"))
  921. r.stage(["a"])
  922. commit_sha = r.do_commit(
  923. b"deleted a",
  924. committer=b"Test Committer <test@nodomain.com>",
  925. author=b"Test Author <test@nodomain.com>",
  926. commit_timestamp=12395,
  927. commit_timezone=0,
  928. author_timestamp=12395,
  929. author_timezone=0,
  930. )
  931. self.assertEqual([self._root_commit], r[commit_sha].parents)
  932. self.assertEqual([], list(r.open_index()))
  933. tree = r[r[commit_sha].tree]
  934. self.assertEqual([], list(tree.iteritems()))
  935. def test_commit_follows(self):
  936. r = self._repo
  937. r.refs.set_symbolic_ref(b"HEAD", b"refs/heads/bla")
  938. commit_sha = r.do_commit(
  939. b"commit with strange character",
  940. committer=b"Test Committer <test@nodomain.com>",
  941. author=b"Test Author <test@nodomain.com>",
  942. commit_timestamp=12395,
  943. commit_timezone=0,
  944. author_timestamp=12395,
  945. author_timezone=0,
  946. ref=b"HEAD",
  947. )
  948. self.assertEqual(commit_sha, r[b"refs/heads/bla"].id)
  949. def test_commit_encoding(self):
  950. r = self._repo
  951. commit_sha = r.do_commit(
  952. b"commit with strange character \xee",
  953. committer=b"Test Committer <test@nodomain.com>",
  954. author=b"Test Author <test@nodomain.com>",
  955. commit_timestamp=12395,
  956. commit_timezone=0,
  957. author_timestamp=12395,
  958. author_timezone=0,
  959. encoding=b"iso8859-1",
  960. )
  961. self.assertEqual(b"iso8859-1", r[commit_sha].encoding)
  962. def test_compression_level(self):
  963. r = self._repo
  964. c = r.get_config()
  965. c.set(("core",), "compression", "3")
  966. c.set(("core",), "looseCompression", "4")
  967. c.write_to_path()
  968. r = Repo(self._repo_dir)
  969. self.assertEqual(r.object_store.loose_compression_level, 4)
  970. def test_repositoryformatversion_unsupported(self):
  971. r = self._repo
  972. c = r.get_config()
  973. c.set(("core",), "repositoryformatversion", "2")
  974. c.write_to_path()
  975. self.assertRaises(UnsupportedVersion, Repo, self._repo_dir)
  976. def test_repositoryformatversion_1(self):
  977. r = self._repo
  978. c = r.get_config()
  979. c.set(("core",), "repositoryformatversion", "1")
  980. c.write_to_path()
  981. Repo(self._repo_dir)
  982. def test_worktreeconfig_extension(self):
  983. r = self._repo
  984. c = r.get_config()
  985. c.set(("core",), "repositoryformatversion", "1")
  986. c.set(("extensions",), "worktreeconfig", True)
  987. c.write_to_path()
  988. c = r.get_worktree_config()
  989. c.set(("user",), "repositoryformatversion", "1")
  990. c.set((b"user",), b"name", b"Jelmer")
  991. c.write_to_path()
  992. cs = r.get_config_stack()
  993. self.assertEqual(cs.get(("user",), "name"), b"Jelmer")
  994. def test_worktreeconfig_extension_case(self):
  995. """Test that worktree code does not error for alternate case format."""
  996. r = self._repo
  997. c = r.get_config()
  998. c.set(("core",), "repositoryformatversion", "1")
  999. # Capitalize "Config"
  1000. c.set(("extensions",), "worktreeConfig", True)
  1001. c.write_to_path()
  1002. c = r.get_worktree_config()
  1003. c.set(("user",), "repositoryformatversion", "1")
  1004. c.set((b"user",), b"name", b"Jelmer")
  1005. c.write_to_path()
  1006. # The following line errored before
  1007. # https://github.com/jelmer/dulwich/issues/1285 was addressed
  1008. Repo(self._repo_dir)
  1009. def test_repositoryformatversion_1_extension(self):
  1010. r = self._repo
  1011. c = r.get_config()
  1012. c.set(("core",), "repositoryformatversion", "1")
  1013. c.set(("extensions",), "unknownextension", True)
  1014. c.write_to_path()
  1015. self.assertRaises(UnsupportedExtension, Repo, self._repo_dir)
  1016. def test_commit_encoding_from_config(self):
  1017. r = self._repo
  1018. c = r.get_config()
  1019. c.set(("i18n",), "commitEncoding", "iso8859-1")
  1020. c.write_to_path()
  1021. commit_sha = r.do_commit(
  1022. b"commit with strange character \xee",
  1023. committer=b"Test Committer <test@nodomain.com>",
  1024. author=b"Test Author <test@nodomain.com>",
  1025. commit_timestamp=12395,
  1026. commit_timezone=0,
  1027. author_timestamp=12395,
  1028. author_timezone=0,
  1029. )
  1030. self.assertEqual(b"iso8859-1", r[commit_sha].encoding)
  1031. def test_commit_config_identity(self):
  1032. # commit falls back to the users' identity if it wasn't specified
  1033. r = self._repo
  1034. c = r.get_config()
  1035. c.set((b"user",), b"name", b"Jelmer")
  1036. c.set((b"user",), b"email", b"jelmer@apache.org")
  1037. c.write_to_path()
  1038. commit_sha = r.do_commit(b"message")
  1039. self.assertEqual(b"Jelmer <jelmer@apache.org>", r[commit_sha].author)
  1040. self.assertEqual(b"Jelmer <jelmer@apache.org>", r[commit_sha].committer)
  1041. def test_commit_config_identity_strips_than(self):
  1042. # commit falls back to the users' identity if it wasn't specified,
  1043. # and strips superfluous <>
  1044. r = self._repo
  1045. c = r.get_config()
  1046. c.set((b"user",), b"name", b"Jelmer")
  1047. c.set((b"user",), b"email", b"<jelmer@apache.org>")
  1048. c.write_to_path()
  1049. commit_sha = r.do_commit(b"message")
  1050. self.assertEqual(b"Jelmer <jelmer@apache.org>", r[commit_sha].author)
  1051. self.assertEqual(b"Jelmer <jelmer@apache.org>", r[commit_sha].committer)
  1052. def test_commit_config_identity_in_memoryrepo(self):
  1053. # commit falls back to the users' identity if it wasn't specified
  1054. r = MemoryRepo.init_bare([], {})
  1055. c = r.get_config()
  1056. c.set((b"user",), b"name", b"Jelmer")
  1057. c.set((b"user",), b"email", b"jelmer@apache.org")
  1058. commit_sha = r.do_commit(b"message", tree=objects.Tree().id)
  1059. self.assertEqual(b"Jelmer <jelmer@apache.org>", r[commit_sha].author)
  1060. self.assertEqual(b"Jelmer <jelmer@apache.org>", r[commit_sha].committer)
  1061. def test_commit_config_identity_from_env(self):
  1062. # commit falls back to the users' identity if it wasn't specified
  1063. self.overrideEnv("GIT_COMMITTER_NAME", "joe")
  1064. self.overrideEnv("GIT_COMMITTER_EMAIL", "joe@example.com")
  1065. r = self._repo
  1066. c = r.get_config()
  1067. c.set((b"user",), b"name", b"Jelmer")
  1068. c.set((b"user",), b"email", b"jelmer@apache.org")
  1069. c.write_to_path()
  1070. commit_sha = r.do_commit(b"message")
  1071. self.assertEqual(b"Jelmer <jelmer@apache.org>", r[commit_sha].author)
  1072. self.assertEqual(b"joe <joe@example.com>", r[commit_sha].committer)
  1073. def test_commit_fail_ref(self):
  1074. r = self._repo
  1075. def set_if_equals(name, old_ref, new_ref, **kwargs):
  1076. return False
  1077. r.refs.set_if_equals = set_if_equals
  1078. def add_if_new(name, new_ref, **kwargs):
  1079. self.fail("Unexpected call to add_if_new")
  1080. r.refs.add_if_new = add_if_new
  1081. old_shas = set(r.object_store)
  1082. self.assertRaises(
  1083. errors.CommitError,
  1084. r.do_commit,
  1085. b"failed commit",
  1086. committer=b"Test Committer <test@nodomain.com>",
  1087. author=b"Test Author <test@nodomain.com>",
  1088. commit_timestamp=12345,
  1089. commit_timezone=0,
  1090. author_timestamp=12345,
  1091. author_timezone=0,
  1092. )
  1093. new_shas = set(r.object_store) - old_shas
  1094. self.assertEqual(1, len(new_shas))
  1095. # Check that the new commit (now garbage) was added.
  1096. new_commit = r[new_shas.pop()]
  1097. self.assertEqual(r[self._root_commit].tree, new_commit.tree)
  1098. self.assertEqual(b"failed commit", new_commit.message)
  1099. def test_commit_branch(self):
  1100. r = self._repo
  1101. commit_sha = r.do_commit(
  1102. b"commit to branch",
  1103. committer=b"Test Committer <test@nodomain.com>",
  1104. author=b"Test Author <test@nodomain.com>",
  1105. commit_timestamp=12395,
  1106. commit_timezone=0,
  1107. author_timestamp=12395,
  1108. author_timezone=0,
  1109. ref=b"refs/heads/new_branch",
  1110. )
  1111. self.assertEqual(self._root_commit, r[b"HEAD"].id)
  1112. self.assertEqual(commit_sha, r[b"refs/heads/new_branch"].id)
  1113. self.assertEqual([], r[commit_sha].parents)
  1114. self.assertIn(b"refs/heads/new_branch", r)
  1115. new_branch_head = commit_sha
  1116. commit_sha = r.do_commit(
  1117. b"commit to branch 2",
  1118. committer=b"Test Committer <test@nodomain.com>",
  1119. author=b"Test Author <test@nodomain.com>",
  1120. commit_timestamp=12395,
  1121. commit_timezone=0,
  1122. author_timestamp=12395,
  1123. author_timezone=0,
  1124. ref=b"refs/heads/new_branch",
  1125. )
  1126. self.assertEqual(self._root_commit, r[b"HEAD"].id)
  1127. self.assertEqual(commit_sha, r[b"refs/heads/new_branch"].id)
  1128. self.assertEqual([new_branch_head], r[commit_sha].parents)
  1129. def test_commit_merge_heads(self):
  1130. r = self._repo
  1131. merge_1 = r.do_commit(
  1132. b"commit to branch 2",
  1133. committer=b"Test Committer <test@nodomain.com>",
  1134. author=b"Test Author <test@nodomain.com>",
  1135. commit_timestamp=12395,
  1136. commit_timezone=0,
  1137. author_timestamp=12395,
  1138. author_timezone=0,
  1139. ref=b"refs/heads/new_branch",
  1140. )
  1141. commit_sha = r.do_commit(
  1142. b"commit with merge",
  1143. committer=b"Test Committer <test@nodomain.com>",
  1144. author=b"Test Author <test@nodomain.com>",
  1145. commit_timestamp=12395,
  1146. commit_timezone=0,
  1147. author_timestamp=12395,
  1148. author_timezone=0,
  1149. merge_heads=[merge_1],
  1150. )
  1151. self.assertEqual([self._root_commit, merge_1], r[commit_sha].parents)
  1152. def test_commit_dangling_commit(self):
  1153. r = self._repo
  1154. old_shas = set(r.object_store)
  1155. old_refs = r.get_refs()
  1156. commit_sha = r.do_commit(
  1157. b"commit with no ref",
  1158. committer=b"Test Committer <test@nodomain.com>",
  1159. author=b"Test Author <test@nodomain.com>",
  1160. commit_timestamp=12395,
  1161. commit_timezone=0,
  1162. author_timestamp=12395,
  1163. author_timezone=0,
  1164. ref=None,
  1165. )
  1166. new_shas = set(r.object_store) - old_shas
  1167. # New sha is added, but no new refs
  1168. self.assertEqual(1, len(new_shas))
  1169. new_commit = r[new_shas.pop()]
  1170. self.assertEqual(r[self._root_commit].tree, new_commit.tree)
  1171. self.assertEqual([], r[commit_sha].parents)
  1172. self.assertEqual(old_refs, r.get_refs())
  1173. def test_commit_dangling_commit_with_parents(self):
  1174. r = self._repo
  1175. old_shas = set(r.object_store)
  1176. old_refs = r.get_refs()
  1177. commit_sha = r.do_commit(
  1178. b"commit with no ref",
  1179. committer=b"Test Committer <test@nodomain.com>",
  1180. author=b"Test Author <test@nodomain.com>",
  1181. commit_timestamp=12395,
  1182. commit_timezone=0,
  1183. author_timestamp=12395,
  1184. author_timezone=0,
  1185. ref=None,
  1186. merge_heads=[self._root_commit],
  1187. )
  1188. new_shas = set(r.object_store) - old_shas
  1189. # New sha is added, but no new refs
  1190. self.assertEqual(1, len(new_shas))
  1191. new_commit = r[new_shas.pop()]
  1192. self.assertEqual(r[self._root_commit].tree, new_commit.tree)
  1193. self.assertEqual([self._root_commit], r[commit_sha].parents)
  1194. self.assertEqual(old_refs, r.get_refs())
  1195. def test_stage_absolute(self):
  1196. r = self._repo
  1197. os.remove(os.path.join(r.path, "a"))
  1198. self.assertRaises(ValueError, r.stage, [os.path.join(r.path, "a")])
  1199. def test_stage_deleted(self):
  1200. r = self._repo
  1201. os.remove(os.path.join(r.path, "a"))
  1202. r.stage(["a"])
  1203. r.stage(["a"]) # double-stage a deleted path
  1204. self.assertEqual([], list(r.open_index()))
  1205. def test_stage_directory(self):
  1206. r = self._repo
  1207. os.mkdir(os.path.join(r.path, "c"))
  1208. r.stage(["c"])
  1209. self.assertEqual([b"a"], list(r.open_index()))
  1210. def test_stage_submodule(self):
  1211. r = self._repo
  1212. s = Repo.init(os.path.join(r.path, "sub"), mkdir=True)
  1213. s.do_commit(b"message")
  1214. r.stage(["sub"])
  1215. self.assertEqual([b"a", b"sub"], list(r.open_index()))
  1216. def test_unstage_midify_file_with_dir(self):
  1217. os.mkdir(os.path.join(self._repo.path, "new_dir"))
  1218. full_path = os.path.join(self._repo.path, "new_dir", "foo")
  1219. with open(full_path, "w") as f:
  1220. f.write("hello")
  1221. porcelain.add(self._repo, paths=[full_path])
  1222. porcelain.commit(
  1223. self._repo,
  1224. message=b"unitest",
  1225. committer=b"Jane <jane@example.com>",
  1226. author=b"John <john@example.com>",
  1227. )
  1228. with open(full_path, "a") as f:
  1229. f.write("something new")
  1230. self._repo.unstage(["new_dir/foo"])
  1231. status = list(porcelain.status(self._repo))
  1232. self.assertEqual(
  1233. [{"add": [], "delete": [], "modify": []}, [b"new_dir/foo"], []], status
  1234. )
  1235. def test_unstage_while_no_commit(self):
  1236. file = "foo"
  1237. full_path = os.path.join(self._repo.path, file)
  1238. with open(full_path, "w") as f:
  1239. f.write("hello")
  1240. porcelain.add(self._repo, paths=[full_path])
  1241. self._repo.unstage([file])
  1242. status = list(porcelain.status(self._repo))
  1243. self.assertEqual([{"add": [], "delete": [], "modify": []}, [], ["foo"]], status)
  1244. def test_unstage_add_file(self):
  1245. file = "foo"
  1246. full_path = os.path.join(self._repo.path, file)
  1247. porcelain.commit(
  1248. self._repo,
  1249. message=b"unitest",
  1250. committer=b"Jane <jane@example.com>",
  1251. author=b"John <john@example.com>",
  1252. )
  1253. with open(full_path, "w") as f:
  1254. f.write("hello")
  1255. porcelain.add(self._repo, paths=[full_path])
  1256. self._repo.unstage([file])
  1257. status = list(porcelain.status(self._repo))
  1258. self.assertEqual([{"add": [], "delete": [], "modify": []}, [], ["foo"]], status)
  1259. def test_unstage_modify_file(self):
  1260. file = "foo"
  1261. full_path = os.path.join(self._repo.path, file)
  1262. with open(full_path, "w") as f:
  1263. f.write("hello")
  1264. porcelain.add(self._repo, paths=[full_path])
  1265. porcelain.commit(
  1266. self._repo,
  1267. message=b"unitest",
  1268. committer=b"Jane <jane@example.com>",
  1269. author=b"John <john@example.com>",
  1270. )
  1271. with open(full_path, "a") as f:
  1272. f.write("broken")
  1273. porcelain.add(self._repo, paths=[full_path])
  1274. self._repo.unstage([file])
  1275. status = list(porcelain.status(self._repo))
  1276. self.assertEqual(
  1277. [{"add": [], "delete": [], "modify": []}, [b"foo"], []], status
  1278. )
  1279. def test_unstage_remove_file(self):
  1280. file = "foo"
  1281. full_path = os.path.join(self._repo.path, file)
  1282. with open(full_path, "w") as f:
  1283. f.write("hello")
  1284. porcelain.add(self._repo, paths=[full_path])
  1285. porcelain.commit(
  1286. self._repo,
  1287. message=b"unitest",
  1288. committer=b"Jane <jane@example.com>",
  1289. author=b"John <john@example.com>",
  1290. )
  1291. os.remove(full_path)
  1292. self._repo.unstage([file])
  1293. status = list(porcelain.status(self._repo))
  1294. self.assertEqual(
  1295. [{"add": [], "delete": [], "modify": []}, [b"foo"], []], status
  1296. )
  1297. def test_reset_index(self):
  1298. r = self._repo
  1299. with open(os.path.join(r.path, "a"), "wb") as f:
  1300. f.write(b"changed")
  1301. with open(os.path.join(r.path, "b"), "wb") as f:
  1302. f.write(b"added")
  1303. r.stage(["a", "b"])
  1304. status = list(porcelain.status(self._repo))
  1305. self.assertEqual(
  1306. [{"add": [b"b"], "delete": [], "modify": [b"a"]}, [], []], status
  1307. )
  1308. r.reset_index()
  1309. status = list(porcelain.status(self._repo))
  1310. self.assertEqual([{"add": [], "delete": [], "modify": []}, [], ["b"]], status)
  1311. @skipIf(
  1312. sys.platform in ("win32", "darwin"),
  1313. "tries to implicitly decode as utf8",
  1314. )
  1315. def test_commit_no_encode_decode(self):
  1316. r = self._repo
  1317. repo_path_bytes = os.fsencode(r.path)
  1318. encodings = ("utf8", "latin1")
  1319. names = ["À".encode(encoding) for encoding in encodings]
  1320. for name, encoding in zip(names, encodings):
  1321. full_path = os.path.join(repo_path_bytes, name)
  1322. with open(full_path, "wb") as f:
  1323. f.write(encoding.encode("ascii"))
  1324. # These files are break tear_down_repo, so cleanup these files
  1325. # ourselves.
  1326. self.addCleanup(os.remove, full_path)
  1327. r.stage(names)
  1328. commit_sha = r.do_commit(
  1329. b"Files with different encodings",
  1330. committer=b"Test Committer <test@nodomain.com>",
  1331. author=b"Test Author <test@nodomain.com>",
  1332. commit_timestamp=12395,
  1333. commit_timezone=0,
  1334. author_timestamp=12395,
  1335. author_timezone=0,
  1336. ref=None,
  1337. merge_heads=[self._root_commit],
  1338. )
  1339. for name, encoding in zip(names, encodings):
  1340. mode, id = tree_lookup_path(r.get_object, r[commit_sha].tree, name)
  1341. self.assertEqual(stat.S_IFREG | 0o644, mode)
  1342. self.assertEqual(encoding.encode("ascii"), r[id].data)
  1343. def test_discover_intended(self):
  1344. path = os.path.join(self._repo_dir, "b/c")
  1345. r = Repo.discover(path)
  1346. self.assertEqual(r.head(), self._repo.head())
  1347. def test_discover_isrepo(self):
  1348. r = Repo.discover(self._repo_dir)
  1349. self.assertEqual(r.head(), self._repo.head())
  1350. def test_discover_notrepo(self):
  1351. with self.assertRaises(NotGitRepository):
  1352. Repo.discover("/")
  1353. class CheckUserIdentityTests(TestCase):
  1354. def test_valid(self):
  1355. check_user_identity(b"Me <me@example.com>")
  1356. def test_invalid(self):
  1357. self.assertRaises(InvalidUserIdentity, check_user_identity, b"No Email")
  1358. self.assertRaises(
  1359. InvalidUserIdentity, check_user_identity, b"Fullname <missing"
  1360. )
  1361. self.assertRaises(
  1362. InvalidUserIdentity, check_user_identity, b"Fullname missing>"
  1363. )
  1364. self.assertRaises(
  1365. InvalidUserIdentity, check_user_identity, b"Fullname >order<>"
  1366. )
  1367. self.assertRaises(
  1368. InvalidUserIdentity, check_user_identity, b"Contains\0null byte <>"
  1369. )
  1370. self.assertRaises(
  1371. InvalidUserIdentity, check_user_identity, b"Contains\nnewline byte <>"
  1372. )