2
0

test_refs.py 46 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239
  1. # test_refs.py -- tests for refs.py
  2. # Copyright (C) 2013 Jelmer Vernooij <jelmer@jelmer.uk>
  3. #
  4. # SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
  5. # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
  6. # General Public License as published by the Free Software Foundation; version 2.0
  7. # or (at your option) any later version. You can redistribute it and/or
  8. # modify it under the terms of either of these two licenses.
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. #
  16. # You should have received a copy of the licenses; if not, see
  17. # <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
  18. # and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
  19. # License, Version 2.0.
  20. #
  21. """Tests for dulwich.refs."""
  22. import os
  23. import sys
  24. import tempfile
  25. from io import BytesIO
  26. from typing import ClassVar
  27. from dulwich import errors
  28. from dulwich.file import GitFile
  29. from dulwich.objects import ZERO_SHA
  30. from dulwich.refs import (
  31. DictRefsContainer,
  32. InfoRefsContainer,
  33. SymrefLoop,
  34. _split_ref_line,
  35. check_ref_format,
  36. is_per_worktree_ref,
  37. parse_remote_ref,
  38. parse_symref_value,
  39. read_packed_refs,
  40. read_packed_refs_with_peeled,
  41. shorten_ref_name,
  42. split_peeled_refs,
  43. strip_peeled_refs,
  44. write_packed_refs,
  45. )
  46. from dulwich.repo import Repo
  47. from dulwich.tests.utils import open_repo, tear_down_repo
  48. from dulwich.worktree import add_worktree
  49. from . import SkipTest, TestCase
  50. class CheckRefFormatTests(TestCase):
  51. """Tests for the check_ref_format function.
  52. These are the same tests as in the git test suite.
  53. """
  54. def test_valid(self) -> None:
  55. self.assertTrue(check_ref_format(b"heads/foo"))
  56. self.assertTrue(check_ref_format(b"foo/bar/baz"))
  57. self.assertTrue(check_ref_format(b"refs///heads/foo"))
  58. self.assertTrue(check_ref_format(b"foo./bar"))
  59. self.assertTrue(check_ref_format(b"heads/foo@bar"))
  60. self.assertTrue(check_ref_format(b"heads/fix.lock.error"))
  61. def test_invalid(self) -> None:
  62. self.assertFalse(check_ref_format(b"foo"))
  63. self.assertFalse(check_ref_format(b"heads/foo/"))
  64. self.assertFalse(check_ref_format(b"./foo"))
  65. self.assertFalse(check_ref_format(b".refs/foo"))
  66. self.assertFalse(check_ref_format(b"heads/foo..bar"))
  67. self.assertFalse(check_ref_format(b"heads/foo?bar"))
  68. self.assertFalse(check_ref_format(b"heads/foo.lock"))
  69. self.assertFalse(check_ref_format(b"heads/v@{ation"))
  70. self.assertFalse(check_ref_format(b"heads/foo\bar"))
  71. ONES = b"1" * 40
  72. TWOS = b"2" * 40
  73. THREES = b"3" * 40
  74. FOURS = b"4" * 40
  75. class PackedRefsFileTests(TestCase):
  76. def test_split_ref_line_errors(self) -> None:
  77. self.assertRaises(errors.PackedRefsException, _split_ref_line, b"singlefield")
  78. self.assertRaises(errors.PackedRefsException, _split_ref_line, b"badsha name")
  79. self.assertRaises(
  80. errors.PackedRefsException,
  81. _split_ref_line,
  82. ONES + b" bad/../refname",
  83. )
  84. def test_read_without_peeled(self) -> None:
  85. f = BytesIO(b"\n".join([b"# comment", ONES + b" ref/1", TWOS + b" ref/2"]))
  86. self.assertEqual(
  87. [(ONES, b"ref/1"), (TWOS, b"ref/2")], list(read_packed_refs(f))
  88. )
  89. def test_read_without_peeled_errors(self) -> None:
  90. f = BytesIO(b"\n".join([ONES + b" ref/1", b"^" + TWOS]))
  91. self.assertRaises(errors.PackedRefsException, list, read_packed_refs(f))
  92. def test_read_with_peeled(self) -> None:
  93. f = BytesIO(
  94. b"\n".join(
  95. [
  96. ONES + b" ref/1",
  97. TWOS + b" ref/2",
  98. b"^" + THREES,
  99. FOURS + b" ref/4",
  100. ]
  101. )
  102. )
  103. self.assertEqual(
  104. [
  105. (ONES, b"ref/1", None),
  106. (TWOS, b"ref/2", THREES),
  107. (FOURS, b"ref/4", None),
  108. ],
  109. list(read_packed_refs_with_peeled(f)),
  110. )
  111. def test_read_with_peeled_errors(self) -> None:
  112. f = BytesIO(b"\n".join([b"^" + TWOS, ONES + b" ref/1"]))
  113. self.assertRaises(errors.PackedRefsException, list, read_packed_refs(f))
  114. f = BytesIO(b"\n".join([ONES + b" ref/1", b"^" + TWOS, b"^" + THREES]))
  115. self.assertRaises(errors.PackedRefsException, list, read_packed_refs(f))
  116. def test_write_with_peeled(self) -> None:
  117. f = BytesIO()
  118. write_packed_refs(f, {b"ref/1": ONES, b"ref/2": TWOS}, {b"ref/1": THREES})
  119. self.assertEqual(
  120. b"\n".join(
  121. [
  122. b"# pack-refs with: peeled",
  123. ONES + b" ref/1",
  124. b"^" + THREES,
  125. TWOS + b" ref/2",
  126. ]
  127. )
  128. + b"\n",
  129. f.getvalue(),
  130. )
  131. def test_write_without_peeled(self) -> None:
  132. f = BytesIO()
  133. write_packed_refs(f, {b"ref/1": ONES, b"ref/2": TWOS})
  134. self.assertEqual(
  135. b"\n".join([ONES + b" ref/1", TWOS + b" ref/2"]) + b"\n",
  136. f.getvalue(),
  137. )
  138. # Dict of refs that we expect all RefsContainerTests subclasses to define.
  139. _TEST_REFS = {
  140. b"HEAD": b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  141. b"refs/heads/40-char-ref-aaaaaaaaaaaaaaaaaa": b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  142. b"refs/heads/master": b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  143. b"refs/heads/packed": b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  144. b"refs/tags/refs-0.1": b"df6800012397fb85c56e7418dd4eb9405dee075c",
  145. b"refs/tags/refs-0.2": b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8",
  146. b"refs/heads/loop": b"ref: refs/heads/loop",
  147. }
  148. class RefsContainerTests:
  149. def test_keys(self) -> None:
  150. actual_keys = set(self._refs.keys())
  151. self.assertEqual(set(self._refs.allkeys()), actual_keys)
  152. self.assertEqual(set(_TEST_REFS.keys()), actual_keys)
  153. actual_keys = self._refs.keys(b"refs/heads")
  154. actual_keys.discard(b"loop")
  155. self.assertEqual(
  156. [b"40-char-ref-aaaaaaaaaaaaaaaaaa", b"master", b"packed"],
  157. sorted(actual_keys),
  158. )
  159. self.assertEqual(
  160. [b"refs-0.1", b"refs-0.2"], sorted(self._refs.keys(b"refs/tags"))
  161. )
  162. def test_iter(self) -> None:
  163. actual_keys = set(self._refs.keys())
  164. self.assertEqual(set(self._refs), actual_keys)
  165. self.assertEqual(set(_TEST_REFS.keys()), actual_keys)
  166. def test_as_dict(self) -> None:
  167. # refs/heads/loop does not show up even if it exists
  168. expected_refs = dict(_TEST_REFS)
  169. del expected_refs[b"refs/heads/loop"]
  170. self.assertEqual(expected_refs, self._refs.as_dict())
  171. def test_get_symrefs(self) -> None:
  172. self._refs.set_symbolic_ref(b"refs/heads/src", b"refs/heads/dst")
  173. symrefs = self._refs.get_symrefs()
  174. if b"HEAD" in symrefs:
  175. symrefs.pop(b"HEAD")
  176. self.assertEqual(
  177. {
  178. b"refs/heads/src": b"refs/heads/dst",
  179. b"refs/heads/loop": b"refs/heads/loop",
  180. },
  181. symrefs,
  182. )
  183. def test_setitem(self) -> None:
  184. self._refs[b"refs/some/ref"] = b"42d06bd4b77fed026b154d16493e5deab78f02ec"
  185. self.assertEqual(
  186. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  187. self._refs[b"refs/some/ref"],
  188. )
  189. # should accept symref
  190. self._refs[b"refs/heads/symbolic"] = b"ref: refs/heads/master"
  191. self.assertEqual(
  192. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  193. self._refs[b"refs/heads/symbolic"],
  194. )
  195. # should not accept bad ref names
  196. self.assertRaises(
  197. errors.RefFormatError,
  198. self._refs.__setitem__,
  199. b"notrefs/foo",
  200. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  201. )
  202. # should not accept short sha
  203. self.assertRaises(
  204. ValueError,
  205. self._refs.__setitem__,
  206. b"refs/some/ref",
  207. b"42d06bd",
  208. )
  209. def test_set_if_equals(self) -> None:
  210. nines = b"9" * 40
  211. self.assertFalse(self._refs.set_if_equals(b"HEAD", b"c0ffee", nines))
  212. self.assertEqual(
  213. b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"HEAD"]
  214. )
  215. self.assertTrue(
  216. self._refs.set_if_equals(
  217. b"HEAD", b"42d06bd4b77fed026b154d16493e5deab78f02ec", nines
  218. )
  219. )
  220. self.assertEqual(nines, self._refs[b"HEAD"])
  221. # Setting the ref again is a no-op, but will return True.
  222. self.assertTrue(self._refs.set_if_equals(b"HEAD", nines, nines))
  223. self.assertEqual(nines, self._refs[b"HEAD"])
  224. self.assertTrue(self._refs.set_if_equals(b"refs/heads/master", None, nines))
  225. self.assertEqual(nines, self._refs[b"refs/heads/master"])
  226. self.assertTrue(
  227. self._refs.set_if_equals(b"refs/heads/nonexistent", ZERO_SHA, nines)
  228. )
  229. self.assertEqual(nines, self._refs[b"refs/heads/nonexistent"])
  230. def test_add_if_new(self) -> None:
  231. nines = b"9" * 40
  232. self.assertFalse(self._refs.add_if_new(b"refs/heads/master", nines))
  233. self.assertEqual(
  234. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  235. self._refs[b"refs/heads/master"],
  236. )
  237. self.assertTrue(self._refs.add_if_new(b"refs/some/ref", nines))
  238. self.assertEqual(nines, self._refs[b"refs/some/ref"])
  239. def test_set_symbolic_ref(self) -> None:
  240. self._refs.set_symbolic_ref(b"refs/heads/symbolic", b"refs/heads/master")
  241. self.assertEqual(
  242. b"ref: refs/heads/master",
  243. self._refs.read_loose_ref(b"refs/heads/symbolic"),
  244. )
  245. self.assertEqual(
  246. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  247. self._refs[b"refs/heads/symbolic"],
  248. )
  249. def test_set_symbolic_ref_overwrite(self) -> None:
  250. nines = b"9" * 40
  251. self.assertNotIn(b"refs/heads/symbolic", self._refs)
  252. self._refs[b"refs/heads/symbolic"] = nines
  253. self.assertEqual(nines, self._refs.read_loose_ref(b"refs/heads/symbolic"))
  254. self._refs.set_symbolic_ref(b"refs/heads/symbolic", b"refs/heads/master")
  255. self.assertEqual(
  256. b"ref: refs/heads/master",
  257. self._refs.read_loose_ref(b"refs/heads/symbolic"),
  258. )
  259. self.assertEqual(
  260. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  261. self._refs[b"refs/heads/symbolic"],
  262. )
  263. def test_check_refname(self) -> None:
  264. self._refs._check_refname(b"HEAD")
  265. self._refs._check_refname(b"refs/stash")
  266. self._refs._check_refname(b"refs/heads/foo")
  267. self.assertRaises(errors.RefFormatError, self._refs._check_refname, b"refs")
  268. self.assertRaises(
  269. errors.RefFormatError, self._refs._check_refname, b"notrefs/foo"
  270. )
  271. def test_contains(self) -> None:
  272. self.assertIn(b"refs/heads/master", self._refs)
  273. self.assertNotIn(b"refs/heads/bar", self._refs)
  274. def test_delitem(self) -> None:
  275. self.assertEqual(
  276. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  277. self._refs[b"refs/heads/master"],
  278. )
  279. del self._refs[b"refs/heads/master"]
  280. self.assertRaises(KeyError, lambda: self._refs[b"refs/heads/master"])
  281. def test_remove_if_equals(self) -> None:
  282. self.assertFalse(self._refs.remove_if_equals(b"HEAD", b"c0ffee"))
  283. self.assertEqual(
  284. b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"HEAD"]
  285. )
  286. self.assertTrue(
  287. self._refs.remove_if_equals(
  288. b"refs/tags/refs-0.2",
  289. b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8",
  290. )
  291. )
  292. self.assertTrue(self._refs.remove_if_equals(b"refs/tags/refs-0.2", ZERO_SHA))
  293. self.assertNotIn(b"refs/tags/refs-0.2", self._refs)
  294. def test_import_refs_name(self) -> None:
  295. self._refs[b"refs/remotes/origin/other"] = (
  296. b"48d01bd4b77fed026b154d16493e5deab78f02ec"
  297. )
  298. self._refs.import_refs(
  299. b"refs/remotes/origin",
  300. {b"master": b"42d06bd4b77fed026b154d16493e5deab78f02ec"},
  301. )
  302. self.assertEqual(
  303. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  304. self._refs[b"refs/remotes/origin/master"],
  305. )
  306. self.assertEqual(
  307. b"48d01bd4b77fed026b154d16493e5deab78f02ec",
  308. self._refs[b"refs/remotes/origin/other"],
  309. )
  310. def test_import_refs_name_prune(self) -> None:
  311. self._refs[b"refs/remotes/origin/other"] = (
  312. b"48d01bd4b77fed026b154d16493e5deab78f02ec"
  313. )
  314. self._refs.import_refs(
  315. b"refs/remotes/origin",
  316. {b"master": b"42d06bd4b77fed026b154d16493e5deab78f02ec"},
  317. prune=True,
  318. )
  319. self.assertEqual(
  320. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  321. self._refs[b"refs/remotes/origin/master"],
  322. )
  323. self.assertNotIn(b"refs/remotes/origin/other", self._refs)
  324. class DictRefsContainerTests(RefsContainerTests, TestCase):
  325. def setUp(self) -> None:
  326. TestCase.setUp(self)
  327. self._refs = DictRefsContainer(dict(_TEST_REFS))
  328. def test_invalid_refname(self) -> None:
  329. # FIXME: Move this test into RefsContainerTests, but requires
  330. # some way of injecting invalid refs.
  331. self._refs._refs[b"refs/stash"] = b"00" * 20
  332. expected_refs = dict(_TEST_REFS)
  333. del expected_refs[b"refs/heads/loop"]
  334. expected_refs[b"refs/stash"] = b"00" * 20
  335. self.assertEqual(expected_refs, self._refs.as_dict())
  336. def test_set_if_equals_with_symbolic_ref(self) -> None:
  337. # Test that set_if_equals only updates the requested ref,
  338. # not all refs in a symbolic reference chain
  339. # The bug in the original implementation was that when follow()
  340. # was called on a ref, it would return all refs in the chain,
  341. # and set_if_equals would update ALL of them instead of just the
  342. # requested ref.
  343. # Set up refs
  344. master_sha = b"1" * 40
  345. feature_sha = b"2" * 40
  346. new_sha = b"3" * 40
  347. self._refs[b"refs/heads/master"] = master_sha
  348. self._refs[b"refs/heads/feature"] = feature_sha
  349. # Create a second symbolic ref pointing to feature
  350. self._refs.set_symbolic_ref(b"refs/heads/other", b"refs/heads/feature")
  351. # Update refs/heads/other through set_if_equals
  352. # With the bug, this would update BOTH refs/heads/other AND refs/heads/feature
  353. # Without the bug, only refs/heads/other should be updated
  354. # Note: old_ref needs to be the actual stored value (the symref)
  355. self.assertTrue(
  356. self._refs.set_if_equals(
  357. b"refs/heads/other", b"ref: refs/heads/feature", new_sha
  358. )
  359. )
  360. # refs/heads/other should now directly point to new_sha
  361. self.assertEqual(self._refs.read_ref(b"refs/heads/other"), new_sha)
  362. # refs/heads/feature should remain unchanged
  363. # With the bug, refs/heads/feature would also be incorrectly updated to new_sha
  364. self.assertEqual(self._refs[b"refs/heads/feature"], feature_sha)
  365. self.assertEqual(self._refs[b"refs/heads/master"], master_sha)
  366. class DiskRefsContainerTests(RefsContainerTests, TestCase):
  367. def setUp(self) -> None:
  368. TestCase.setUp(self)
  369. self._repo = open_repo("refs.git")
  370. self.addCleanup(tear_down_repo, self._repo)
  371. self._refs = self._repo.refs
  372. def test_get_packed_refs(self) -> None:
  373. self.assertEqual(
  374. {
  375. b"refs/heads/packed": b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  376. b"refs/tags/refs-0.1": b"df6800012397fb85c56e7418dd4eb9405dee075c",
  377. },
  378. self._refs.get_packed_refs(),
  379. )
  380. def test_get_peeled_not_packed(self) -> None:
  381. # not packed
  382. self.assertEqual(None, self._refs.get_peeled(b"refs/tags/refs-0.2"))
  383. self.assertEqual(
  384. b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8",
  385. self._refs[b"refs/tags/refs-0.2"],
  386. )
  387. # packed, known not peelable
  388. self.assertEqual(
  389. self._refs[b"refs/heads/packed"],
  390. self._refs.get_peeled(b"refs/heads/packed"),
  391. )
  392. # packed, peeled
  393. self.assertEqual(
  394. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  395. self._refs.get_peeled(b"refs/tags/refs-0.1"),
  396. )
  397. def test_setitem(self) -> None:
  398. RefsContainerTests.test_setitem(self)
  399. path = os.path.join(self._refs.path, b"refs", b"some", b"ref")
  400. with open(path, "rb") as f:
  401. self.assertEqual(b"42d06bd4b77fed026b154d16493e5deab78f02ec", f.read()[:40])
  402. self.assertRaises(
  403. OSError,
  404. self._refs.__setitem__,
  405. b"refs/some/ref/sub",
  406. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  407. )
  408. def test_delete_refs_container(self) -> None:
  409. # We shouldn't delete the refs directory
  410. self._refs[b"refs/heads/blah"] = b"42d06bd4b77fed026b154d16493e5deab78f02ec"
  411. for ref in self._refs.allkeys():
  412. del self._refs[ref]
  413. self.assertTrue(os.path.exists(os.path.join(self._refs.path, b"refs")))
  414. def test_setitem_packed(self) -> None:
  415. with open(os.path.join(self._refs.path, b"packed-refs"), "w") as f:
  416. f.write("# pack-refs with: peeled fully-peeled sorted \n")
  417. f.write("42d06bd4b77fed026b154d16493e5deab78f02ec refs/heads/packed\n")
  418. # It's allowed to set a new ref on a packed ref, the new ref will be
  419. # placed outside on refs/
  420. self._refs[b"refs/heads/packed"] = b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8"
  421. packed_ref_path = os.path.join(self._refs.path, b"refs", b"heads", b"packed")
  422. with open(packed_ref_path, "rb") as f:
  423. self.assertEqual(b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8", f.read()[:40])
  424. self.assertRaises(
  425. OSError,
  426. self._refs.__setitem__,
  427. b"refs/heads/packed/sub",
  428. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  429. )
  430. # this shouldn't overwrite the packed refs
  431. self.assertEqual(
  432. {b"refs/heads/packed": b"42d06bd4b77fed026b154d16493e5deab78f02ec"},
  433. self._refs.get_packed_refs(),
  434. )
  435. def test_add_packed_refs(self) -> None:
  436. # first, create a non-packed ref
  437. self._refs[b"refs/heads/packed"] = b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8"
  438. packed_ref_path = os.path.join(self._refs.path, b"refs", b"heads", b"packed")
  439. self.assertTrue(os.path.exists(packed_ref_path))
  440. # now overwrite that with a packed ref
  441. packed_refs_file_path = os.path.join(self._refs.path, b"packed-refs")
  442. self._refs.add_packed_refs(
  443. {
  444. b"refs/heads/packed": b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  445. }
  446. )
  447. # that should kill the file
  448. self.assertFalse(os.path.exists(packed_ref_path))
  449. # now delete the packed ref
  450. self._refs.add_packed_refs(
  451. {
  452. b"refs/heads/packed": None,
  453. }
  454. )
  455. # and it's gone!
  456. self.assertFalse(os.path.exists(packed_ref_path))
  457. self.assertRaises(
  458. KeyError,
  459. self._refs.__getitem__,
  460. b"refs/heads/packed",
  461. )
  462. # just in case, make sure we can't pack HEAD
  463. self.assertRaises(
  464. ValueError,
  465. self._refs.add_packed_refs,
  466. {b"HEAD": "02ac81614bcdbd585a37b4b0edf8cb8a"},
  467. )
  468. # delete all packed refs
  469. self._refs.add_packed_refs({ref: None for ref in self._refs.get_packed_refs()})
  470. self.assertEqual({}, self._refs.get_packed_refs())
  471. # remove the packed ref file, and check that adding nothing doesn't affect that
  472. os.remove(packed_refs_file_path)
  473. # adding nothing doesn't make it reappear
  474. self._refs.add_packed_refs({})
  475. self.assertFalse(os.path.exists(packed_refs_file_path))
  476. def test_setitem_symbolic(self) -> None:
  477. ones = b"1" * 40
  478. self._refs[b"HEAD"] = ones
  479. self.assertEqual(ones, self._refs[b"HEAD"])
  480. # ensure HEAD was not modified
  481. f = open(os.path.join(self._refs.path, b"HEAD"), "rb")
  482. v = next(iter(f)).rstrip(b"\n\r")
  483. f.close()
  484. self.assertEqual(b"ref: refs/heads/master", v)
  485. # ensure the symbolic link was written through
  486. f = open(os.path.join(self._refs.path, b"refs", b"heads", b"master"), "rb")
  487. self.assertEqual(ones, f.read()[:40])
  488. f.close()
  489. def test_set_if_equals(self) -> None:
  490. RefsContainerTests.test_set_if_equals(self)
  491. # ensure symref was followed
  492. self.assertEqual(b"9" * 40, self._refs[b"refs/heads/master"])
  493. # ensure lockfile was deleted
  494. self.assertFalse(
  495. os.path.exists(
  496. os.path.join(self._refs.path, b"refs", b"heads", b"master.lock")
  497. )
  498. )
  499. self.assertFalse(os.path.exists(os.path.join(self._refs.path, b"HEAD.lock")))
  500. def test_add_if_new_packed(self) -> None:
  501. # don't overwrite packed ref
  502. self.assertFalse(self._refs.add_if_new(b"refs/tags/refs-0.1", b"9" * 40))
  503. self.assertEqual(
  504. b"df6800012397fb85c56e7418dd4eb9405dee075c",
  505. self._refs[b"refs/tags/refs-0.1"],
  506. )
  507. def test_add_if_new_symbolic(self) -> None:
  508. # Use an empty repo instead of the default.
  509. repo_dir = os.path.join(tempfile.mkdtemp(), "test")
  510. os.makedirs(repo_dir)
  511. repo = Repo.init(repo_dir)
  512. self.addCleanup(tear_down_repo, repo)
  513. refs = repo.refs
  514. nines = b"9" * 40
  515. self.assertEqual(b"ref: refs/heads/master", refs.read_ref(b"HEAD"))
  516. self.assertNotIn(b"refs/heads/master", refs)
  517. self.assertTrue(refs.add_if_new(b"HEAD", nines))
  518. self.assertEqual(b"ref: refs/heads/master", refs.read_ref(b"HEAD"))
  519. self.assertEqual(nines, refs[b"HEAD"])
  520. self.assertEqual(nines, refs[b"refs/heads/master"])
  521. self.assertFalse(refs.add_if_new(b"HEAD", b"1" * 40))
  522. self.assertEqual(nines, refs[b"HEAD"])
  523. self.assertEqual(nines, refs[b"refs/heads/master"])
  524. def test_follow(self) -> None:
  525. self.assertEqual(
  526. (
  527. [b"HEAD", b"refs/heads/master"],
  528. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  529. ),
  530. self._refs.follow(b"HEAD"),
  531. )
  532. self.assertEqual(
  533. (
  534. [b"refs/heads/master"],
  535. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  536. ),
  537. self._refs.follow(b"refs/heads/master"),
  538. )
  539. self.assertRaises(SymrefLoop, self._refs.follow, b"refs/heads/loop")
  540. def test_set_overwrite_loop(self) -> None:
  541. self.assertRaises(SymrefLoop, self._refs.follow, b"refs/heads/loop")
  542. self._refs[b"refs/heads/loop"] = b"42d06bd4b77fed026b154d16493e5deab78f02ec"
  543. self.assertEqual(
  544. ([b"refs/heads/loop"], b"42d06bd4b77fed026b154d16493e5deab78f02ec"),
  545. self._refs.follow(b"refs/heads/loop"),
  546. )
  547. def test_delitem(self) -> None:
  548. RefsContainerTests.test_delitem(self)
  549. ref_file = os.path.join(self._refs.path, b"refs", b"heads", b"master")
  550. self.assertFalse(os.path.exists(ref_file))
  551. self.assertNotIn(b"refs/heads/master", self._refs.get_packed_refs())
  552. def test_delitem_symbolic(self) -> None:
  553. self.assertEqual(b"ref: refs/heads/master", self._refs.read_loose_ref(b"HEAD"))
  554. del self._refs[b"HEAD"]
  555. self.assertRaises(KeyError, lambda: self._refs[b"HEAD"])
  556. self.assertEqual(
  557. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  558. self._refs[b"refs/heads/master"],
  559. )
  560. self.assertFalse(os.path.exists(os.path.join(self._refs.path, b"HEAD")))
  561. def test_remove_if_equals_symref(self) -> None:
  562. # HEAD is a symref, so shouldn't equal its dereferenced value
  563. self.assertFalse(
  564. self._refs.remove_if_equals(
  565. b"HEAD", b"42d06bd4b77fed026b154d16493e5deab78f02ec"
  566. )
  567. )
  568. self.assertTrue(
  569. self._refs.remove_if_equals(
  570. b"refs/heads/master",
  571. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  572. )
  573. )
  574. self.assertRaises(KeyError, lambda: self._refs[b"refs/heads/master"])
  575. # HEAD is now a broken symref
  576. self.assertRaises(KeyError, lambda: self._refs[b"HEAD"])
  577. self.assertEqual(b"ref: refs/heads/master", self._refs.read_loose_ref(b"HEAD"))
  578. self.assertFalse(
  579. os.path.exists(
  580. os.path.join(self._refs.path, b"refs", b"heads", b"master.lock")
  581. )
  582. )
  583. self.assertFalse(os.path.exists(os.path.join(self._refs.path, b"HEAD.lock")))
  584. def test_remove_packed_without_peeled(self) -> None:
  585. refs_file = os.path.join(self._repo.path, "packed-refs")
  586. f = GitFile(refs_file)
  587. refs_data = f.read()
  588. f.close()
  589. f = GitFile(refs_file, "wb")
  590. f.write(
  591. b"\n".join(
  592. line
  593. for line in refs_data.split(b"\n")
  594. if not line or line[0] not in b"#^"
  595. )
  596. )
  597. f.close()
  598. self._repo = Repo(self._repo.path)
  599. refs = self._repo.refs
  600. self.assertTrue(
  601. refs.remove_if_equals(
  602. b"refs/heads/packed",
  603. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  604. )
  605. )
  606. def test_remove_if_equals_packed(self) -> None:
  607. # test removing ref that is only packed
  608. self.assertEqual(
  609. b"df6800012397fb85c56e7418dd4eb9405dee075c",
  610. self._refs[b"refs/tags/refs-0.1"],
  611. )
  612. self.assertTrue(
  613. self._refs.remove_if_equals(
  614. b"refs/tags/refs-0.1",
  615. b"df6800012397fb85c56e7418dd4eb9405dee075c",
  616. )
  617. )
  618. self.assertRaises(KeyError, lambda: self._refs[b"refs/tags/refs-0.1"])
  619. def test_remove_parent(self) -> None:
  620. self._refs[b"refs/heads/foo/bar"] = b"df6800012397fb85c56e7418dd4eb9405dee075c"
  621. del self._refs[b"refs/heads/foo/bar"]
  622. ref_file = os.path.join(
  623. self._refs.path,
  624. b"refs",
  625. b"heads",
  626. b"foo",
  627. b"bar",
  628. )
  629. self.assertFalse(os.path.exists(ref_file))
  630. ref_file = os.path.join(self._refs.path, b"refs", b"heads", b"foo")
  631. self.assertFalse(os.path.exists(ref_file))
  632. ref_file = os.path.join(self._refs.path, b"refs", b"heads")
  633. self.assertTrue(os.path.exists(ref_file))
  634. self._refs[b"refs/heads/foo"] = b"df6800012397fb85c56e7418dd4eb9405dee075c"
  635. def test_read_ref(self) -> None:
  636. self.assertEqual(b"ref: refs/heads/master", self._refs.read_ref(b"HEAD"))
  637. self.assertEqual(
  638. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  639. self._refs.read_ref(b"refs/heads/packed"),
  640. )
  641. self.assertEqual(None, self._refs.read_ref(b"nonexistent"))
  642. def test_read_loose_ref(self) -> None:
  643. self._refs[b"refs/heads/foo"] = b"df6800012397fb85c56e7418dd4eb9405dee075c"
  644. self.assertEqual(None, self._refs.read_ref(b"refs/heads/foo/bar"))
  645. def test_non_ascii(self) -> None:
  646. try:
  647. encoded_ref = os.fsencode("refs/tags/schön")
  648. except UnicodeEncodeError as exc:
  649. raise SkipTest(
  650. "filesystem encoding doesn't support special character"
  651. ) from exc
  652. p = os.path.join(os.fsencode(self._repo.path), encoded_ref)
  653. with open(p, "w") as f:
  654. f.write("00" * 20)
  655. expected_refs = dict(_TEST_REFS)
  656. expected_refs[encoded_ref] = b"00" * 20
  657. del expected_refs[b"refs/heads/loop"]
  658. self.assertEqual(expected_refs, self._repo.get_refs())
  659. def test_cyrillic(self) -> None:
  660. if sys.platform in ("darwin", "win32"):
  661. raise SkipTest("filesystem encoding doesn't support arbitrary bytes")
  662. # reported in https://github.com/dulwich/dulwich/issues/608
  663. name = b"\xcd\xee\xe2\xe0\xff\xe2\xe5\xf2\xea\xe01"
  664. encoded_ref = b"refs/heads/" + name
  665. with open(os.path.join(os.fsencode(self._repo.path), encoded_ref), "w") as f:
  666. f.write("00" * 20)
  667. expected_refs = set(_TEST_REFS.keys())
  668. expected_refs.add(encoded_ref)
  669. self.assertEqual(expected_refs, set(self._repo.refs.allkeys()))
  670. self.assertEqual(
  671. {r[len(b"refs/") :] for r in expected_refs if r.startswith(b"refs/")},
  672. set(self._repo.refs.subkeys(b"refs/")),
  673. )
  674. expected_refs.remove(b"refs/heads/loop")
  675. expected_refs.add(b"HEAD")
  676. self.assertEqual(expected_refs, set(self._repo.get_refs().keys()))
  677. def test_write_unchanged_ref_optimization(self):
  678. # Test that writing unchanged ref avoids fsync but still checks locks
  679. ref_name = b"refs/heads/unchanged"
  680. ref_value = b"a" * 40
  681. # Set initial ref value
  682. self._refs[ref_name] = ref_value
  683. # Test 1: Writing same value should succeed without changes
  684. result = self._refs.set_if_equals(ref_name, ref_value, ref_value)
  685. self.assertTrue(result)
  686. # Test 2: Writing same value with wrong old_ref should fail
  687. wrong_old = b"b" * 40
  688. result = self._refs.set_if_equals(ref_name, wrong_old, ref_value)
  689. self.assertFalse(result)
  690. # Test 3: Writing different value should update normally
  691. new_value = b"c" * 40
  692. result = self._refs.set_if_equals(ref_name, ref_value, new_value)
  693. self.assertTrue(result)
  694. self.assertEqual(new_value, self._refs[ref_name])
  695. def test_write_unchanged_ref_with_lock(self):
  696. # Test that file locking is still detected when ref unchanged
  697. from dulwich.file import FileLocked
  698. ref_name = b"refs/heads/locktest"
  699. ref_value = b"d" * 40
  700. # Set initial ref value
  701. self._refs[ref_name] = ref_value
  702. # Get the actual file path
  703. ref_file = os.path.join(os.fsencode(self._refs.path), ref_name)
  704. lock_file = ref_file + b".lock"
  705. # Create lock file to simulate another process holding lock
  706. with open(lock_file, "wb") as f:
  707. f.write(b"locked by another process")
  708. # Try to write same value - should raise FileLocked
  709. with self.assertRaises(FileLocked):
  710. self._refs[ref_name] = ref_value
  711. # Clean up lock file
  712. if os.path.exists(lock_file):
  713. os.unlink(lock_file)
  714. # Now it should work
  715. self._refs[ref_name] = ref_value
  716. class IsPerWorktreeRefsTests(TestCase):
  717. def test(self) -> None:
  718. cases = [
  719. (b"HEAD", True),
  720. (b"refs/bisect/good", True),
  721. (b"refs/worktree/foo", True),
  722. (b"refs/rewritten/onto", True),
  723. (b"refs/stash", False),
  724. (b"refs/heads/main", False),
  725. (b"refs/tags/v1.0", False),
  726. (b"refs/remotes/origin/main", False),
  727. (b"refs/custom/foo", False),
  728. (b"refs/replace/aaaaaa", False),
  729. ]
  730. for ref, expected in cases:
  731. with self.subTest(ref=ref, expected=expected):
  732. self.assertEqual(is_per_worktree_ref(ref), expected)
  733. class DiskRefsContainerWorktreeRefsTest(TestCase):
  734. def setUp(self) -> None:
  735. # Create temporary directories
  736. temp_dir = tempfile.mkdtemp()
  737. test_dir = os.path.join(temp_dir, "main")
  738. os.makedirs(test_dir)
  739. repo = Repo.init(test_dir, default_branch=b"main")
  740. main_worktree = repo.get_worktree()
  741. with open(os.path.join(test_dir, "test.txt"), "wb") as f:
  742. f.write(b"test content")
  743. main_worktree.stage(["test.txt"])
  744. self.first_commit = main_worktree.commit(message=b"Initial commit")
  745. worktree_dir = os.path.join(temp_dir, "worktree")
  746. wt_repo = add_worktree(repo, worktree_dir, branch="wt-main")
  747. linked_worktree = wt_repo.get_worktree()
  748. with open(os.path.join(test_dir, "test2.txt"), "wb") as f:
  749. f.write(b"test content")
  750. linked_worktree.stage(["test2.txt"])
  751. self.second_commit = linked_worktree.commit(message=b"second commit")
  752. self.refs = repo.refs
  753. self.wt_refs = wt_repo.refs
  754. def test_refpath(self) -> None:
  755. main_path = self.refs.path
  756. common = self.wt_refs.path
  757. wt_path = self.wt_refs.worktree_path
  758. cases = [
  759. (self.refs, b"HEAD", main_path),
  760. (self.refs, b"refs/heads/main", main_path),
  761. (self.refs, b"refs/heads/wt-main", main_path),
  762. (self.refs, b"refs/worktree/foo", main_path),
  763. (self.refs, b"refs/bisect/good", main_path),
  764. (self.wt_refs, b"HEAD", wt_path),
  765. (self.wt_refs, b"refs/heads/main", common),
  766. (self.wt_refs, b"refs/heads/wt-main", common),
  767. (self.wt_refs, b"refs/worktree/foo", wt_path),
  768. (self.wt_refs, b"refs/bisect/good", wt_path),
  769. ]
  770. for refs, refname, git_dir in cases:
  771. with self.subTest(refs=refs, refname=refname, git_dir=git_dir):
  772. refpath = refs.refpath(refname)
  773. expected_path = os.path.join(
  774. git_dir, refname.replace(b"/", os.fsencode(os.sep))
  775. )
  776. self.assertEqual(refpath, expected_path)
  777. def test_shared_ref(self) -> None:
  778. self.assertEqual(self.refs[b"refs/heads/main"], self.first_commit)
  779. self.assertEqual(self.refs[b"refs/heads/wt-main"], self.second_commit)
  780. self.assertEqual(self.wt_refs[b"refs/heads/main"], self.first_commit)
  781. self.assertEqual(self.wt_refs[b"refs/heads/wt-main"], self.second_commit)
  782. expected = {b"HEAD", b"refs/heads/main", b"refs/heads/wt-main"}
  783. self.assertEqual(expected, self.refs.keys())
  784. self.assertEqual(expected, self.wt_refs.keys())
  785. self.assertEqual({b"main", b"wt-main"}, set(self.refs.keys(b"refs/heads/")))
  786. self.assertEqual({b"main", b"wt-main"}, set(self.wt_refs.keys(b"refs/heads/")))
  787. ref_path = os.path.join(self.refs.path, b"refs", b"heads", b"main")
  788. self.assertTrue(os.path.exists(ref_path))
  789. ref_path = os.path.join(self.wt_refs.worktree_path, b"refs", b"heads", b"main")
  790. self.assertFalse(os.path.exists(ref_path))
  791. def test_per_worktree_ref(self) -> None:
  792. path = self.refs.path
  793. wt_path = self.wt_refs.worktree_path
  794. self.assertEqual(self.refs[b"HEAD"], self.first_commit)
  795. self.assertEqual(self.wt_refs[b"HEAD"], self.second_commit)
  796. self.refs[b"refs/bisect/good"] = self.first_commit
  797. self.wt_refs[b"refs/bisect/good"] = self.second_commit
  798. self.refs[b"refs/bisect/start"] = self.first_commit
  799. self.wt_refs[b"refs/bisect/bad"] = self.second_commit
  800. self.assertEqual(self.refs[b"refs/bisect/good"], self.first_commit)
  801. self.assertEqual(self.wt_refs[b"refs/bisect/good"], self.second_commit)
  802. self.assertTrue(os.path.exists(os.path.join(path, b"refs", b"bisect", b"good")))
  803. self.assertTrue(
  804. os.path.exists(os.path.join(wt_path, b"refs", b"bisect", b"good"))
  805. )
  806. self.assertEqual(self.refs[b"refs/bisect/start"], self.first_commit)
  807. with self.assertRaises(KeyError):
  808. self.wt_refs[b"refs/bisect/start"]
  809. self.assertTrue(
  810. os.path.exists(os.path.join(path, b"refs", b"bisect", b"start"))
  811. )
  812. self.assertFalse(
  813. os.path.exists(os.path.join(wt_path, b"refs", b"bisect", b"start"))
  814. )
  815. with self.assertRaises(KeyError):
  816. self.refs[b"refs/bisect/bad"]
  817. self.assertEqual(self.wt_refs[b"refs/bisect/bad"], self.second_commit)
  818. self.assertFalse(os.path.exists(os.path.join(path, b"refs", b"bisect", b"bad")))
  819. self.assertTrue(
  820. os.path.exists(os.path.join(wt_path, b"refs", b"bisect", b"bad"))
  821. )
  822. expected_refs = {
  823. b"HEAD",
  824. b"refs/heads/main",
  825. b"refs/heads/wt-main",
  826. b"refs/bisect/good",
  827. b"refs/bisect/start",
  828. }
  829. self.assertEqual(self.refs.keys(), expected_refs)
  830. self.assertEqual({b"good", b"start"}, self.refs.keys(b"refs/bisect/"))
  831. expected_wt_refs = {
  832. b"HEAD",
  833. b"refs/heads/main",
  834. b"refs/heads/wt-main",
  835. b"refs/bisect/good",
  836. b"refs/bisect/bad",
  837. }
  838. self.assertEqual(self.wt_refs.keys(), expected_wt_refs)
  839. self.assertEqual({b"good", b"bad"}, self.wt_refs.keys(b"refs/bisect/"))
  840. def test_delete_per_worktree_ref(self) -> None:
  841. self.refs[b"refs/worktree/foo"] = self.first_commit
  842. self.wt_refs[b"refs/worktree/foo"] = self.second_commit
  843. del self.wt_refs[b"refs/worktree/foo"]
  844. with self.assertRaises(KeyError):
  845. self.wt_refs[b"refs/worktree/foo"]
  846. del self.refs[b"refs/worktree/foo"]
  847. with self.assertRaises(KeyError):
  848. self.refs[b"refs/worktree/foo"]
  849. def test_delete_shared_ref(self) -> None:
  850. self.refs[b"refs/heads/branch"] = self.first_commit
  851. del self.wt_refs[b"refs/heads/branch"]
  852. with self.assertRaises(KeyError):
  853. self.wt_refs[b"refs/heads/branch"]
  854. with self.assertRaises(KeyError):
  855. self.refs[b"refs/heads/branch"]
  856. def test_contains_shared_ref(self):
  857. self.assertIn(b"refs/heads/main", self.refs)
  858. self.assertIn(b"refs/heads/main", self.wt_refs)
  859. self.assertIn(b"refs/heads/wt-main", self.refs)
  860. self.assertIn(b"refs/heads/wt-main", self.wt_refs)
  861. def test_contains_per_worktree_ref(self):
  862. self.refs[b"refs/worktree/foo"] = self.first_commit
  863. self.wt_refs[b"refs/worktree/bar"] = self.second_commit
  864. self.assertIn(b"refs/worktree/foo", self.refs)
  865. self.assertNotIn(b"refs/worktree/bar", self.refs)
  866. self.assertNotIn(b"refs/worktree/foo", self.wt_refs)
  867. self.assertIn(b"refs/worktree/bar", self.wt_refs)
  868. _TEST_REFS_SERIALIZED = (
  869. b"42d06bd4b77fed026b154d16493e5deab78f02ec\t"
  870. b"refs/heads/40-char-ref-aaaaaaaaaaaaaaaaaa\n"
  871. b"42d06bd4b77fed026b154d16493e5deab78f02ec\trefs/heads/master\n"
  872. b"42d06bd4b77fed026b154d16493e5deab78f02ec\trefs/heads/packed\n"
  873. b"df6800012397fb85c56e7418dd4eb9405dee075c\trefs/tags/refs-0.1\n"
  874. b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8\trefs/tags/refs-0.2\n"
  875. )
  876. class DiskRefsContainerPathlibTests(TestCase):
  877. def test_pathlib_init(self) -> None:
  878. from pathlib import Path
  879. from dulwich.refs import DiskRefsContainer
  880. # Create a temporary directory
  881. temp_dir = tempfile.mkdtemp()
  882. self.addCleanup(os.rmdir, temp_dir)
  883. # Test with pathlib.Path
  884. path_obj = Path(temp_dir)
  885. refs = DiskRefsContainer(path_obj)
  886. self.assertEqual(refs.path, temp_dir.encode())
  887. # Test refpath with pathlib initialized container
  888. ref_path = refs.refpath(b"HEAD")
  889. self.assertTrue(isinstance(ref_path, bytes))
  890. self.assertEqual(ref_path, os.path.join(temp_dir.encode(), b"HEAD"))
  891. def test_pathlib_worktree_path(self) -> None:
  892. from pathlib import Path
  893. from dulwich.refs import DiskRefsContainer
  894. # Create temporary directories
  895. temp_dir = tempfile.mkdtemp()
  896. worktree_dir = tempfile.mkdtemp()
  897. self.addCleanup(os.rmdir, temp_dir)
  898. self.addCleanup(os.rmdir, worktree_dir)
  899. # Test with pathlib.Path for both paths
  900. path_obj = Path(temp_dir)
  901. worktree_obj = Path(worktree_dir)
  902. refs = DiskRefsContainer(path_obj, worktree_path=worktree_obj)
  903. self.assertEqual(refs.path, temp_dir.encode())
  904. self.assertEqual(refs.worktree_path, worktree_dir.encode())
  905. # Test refpath returns worktree path for HEAD
  906. ref_path = refs.refpath(b"HEAD")
  907. self.assertEqual(ref_path, os.path.join(worktree_dir.encode(), b"HEAD"))
  908. class InfoRefsContainerTests(TestCase):
  909. def test_invalid_refname(self) -> None:
  910. text = _TEST_REFS_SERIALIZED + b"00" * 20 + b"\trefs/stash\n"
  911. refs = InfoRefsContainer(BytesIO(text))
  912. expected_refs = dict(_TEST_REFS)
  913. del expected_refs[b"HEAD"]
  914. expected_refs[b"refs/stash"] = b"00" * 20
  915. del expected_refs[b"refs/heads/loop"]
  916. self.assertEqual(expected_refs, refs.as_dict())
  917. def test_keys(self) -> None:
  918. refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED))
  919. actual_keys = set(refs.keys())
  920. self.assertEqual(set(refs.allkeys()), actual_keys)
  921. expected_refs = dict(_TEST_REFS)
  922. del expected_refs[b"HEAD"]
  923. del expected_refs[b"refs/heads/loop"]
  924. self.assertEqual(set(expected_refs.keys()), actual_keys)
  925. actual_keys = refs.keys(b"refs/heads")
  926. actual_keys.discard(b"loop")
  927. self.assertEqual(
  928. [b"40-char-ref-aaaaaaaaaaaaaaaaaa", b"master", b"packed"],
  929. sorted(actual_keys),
  930. )
  931. self.assertEqual([b"refs-0.1", b"refs-0.2"], sorted(refs.keys(b"refs/tags")))
  932. def test_as_dict(self) -> None:
  933. refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED))
  934. # refs/heads/loop does not show up even if it exists
  935. expected_refs = dict(_TEST_REFS)
  936. del expected_refs[b"HEAD"]
  937. del expected_refs[b"refs/heads/loop"]
  938. self.assertEqual(expected_refs, refs.as_dict())
  939. def test_contains(self) -> None:
  940. refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED))
  941. self.assertIn(b"refs/heads/master", refs)
  942. self.assertNotIn(b"refs/heads/bar", refs)
  943. def test_get_peeled(self) -> None:
  944. refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED))
  945. # refs/heads/loop does not show up even if it exists
  946. self.assertEqual(
  947. _TEST_REFS[b"refs/heads/master"],
  948. refs.get_peeled(b"refs/heads/master"),
  949. )
  950. class ParseSymrefValueTests(TestCase):
  951. def test_valid(self) -> None:
  952. self.assertEqual(b"refs/heads/foo", parse_symref_value(b"ref: refs/heads/foo"))
  953. def test_invalid(self) -> None:
  954. self.assertRaises(ValueError, parse_symref_value, b"foobar")
  955. class ParseRemoteRefTests(TestCase):
  956. def test_valid(self) -> None:
  957. # Test simple case
  958. remote, branch = parse_remote_ref(b"refs/remotes/origin/main")
  959. self.assertEqual(b"origin", remote)
  960. self.assertEqual(b"main", branch)
  961. # Test with branch containing slashes
  962. remote, branch = parse_remote_ref(b"refs/remotes/upstream/feature/new-ui")
  963. self.assertEqual(b"upstream", remote)
  964. self.assertEqual(b"feature/new-ui", branch)
  965. def test_invalid_not_remote_ref(self) -> None:
  966. # Not a remote ref
  967. with self.assertRaises(ValueError) as cm:
  968. parse_remote_ref(b"refs/heads/main")
  969. self.assertIn("Not a remote ref", str(cm.exception))
  970. def test_invalid_format(self) -> None:
  971. # Missing branch name
  972. with self.assertRaises(ValueError) as cm:
  973. parse_remote_ref(b"refs/remotes/origin")
  974. self.assertIn("Invalid remote ref format", str(cm.exception))
  975. # Just the prefix
  976. with self.assertRaises(ValueError) as cm:
  977. parse_remote_ref(b"refs/remotes/")
  978. self.assertIn("Invalid remote ref format", str(cm.exception))
  979. class StripPeeledRefsTests(TestCase):
  980. all_refs: ClassVar[dict[bytes, bytes]] = {
  981. b"refs/heads/master": b"8843d7f92416211de9ebb963ff4ce28125932878",
  982. b"refs/heads/testing": b"186a005b134d8639a58b6731c7c1ea821a6eedba",
  983. b"refs/tags/1.0.0": b"a93db4b0360cc635a2b93675010bac8d101f73f0",
  984. b"refs/tags/1.0.0^{}": b"a93db4b0360cc635a2b93675010bac8d101f73f0",
  985. b"refs/tags/2.0.0": b"0749936d0956c661ac8f8d3483774509c165f89e",
  986. b"refs/tags/2.0.0^{}": b"0749936d0956c661ac8f8d3483774509c165f89e",
  987. }
  988. non_peeled_refs: ClassVar[dict[bytes, bytes]] = {
  989. b"refs/heads/master": b"8843d7f92416211de9ebb963ff4ce28125932878",
  990. b"refs/heads/testing": b"186a005b134d8639a58b6731c7c1ea821a6eedba",
  991. b"refs/tags/1.0.0": b"a93db4b0360cc635a2b93675010bac8d101f73f0",
  992. b"refs/tags/2.0.0": b"0749936d0956c661ac8f8d3483774509c165f89e",
  993. }
  994. def test_strip_peeled_refs(self) -> None:
  995. # Simple check of two dicts
  996. self.assertEqual(strip_peeled_refs(self.all_refs), self.non_peeled_refs)
  997. def test_split_peeled_refs(self) -> None:
  998. (regular, peeled) = split_peeled_refs(self.all_refs)
  999. self.assertEqual(regular, self.non_peeled_refs)
  1000. self.assertEqual(
  1001. peeled,
  1002. {
  1003. b"refs/tags/2.0.0": b"0749936d0956c661ac8f8d3483774509c165f89e",
  1004. b"refs/tags/1.0.0": b"a93db4b0360cc635a2b93675010bac8d101f73f0",
  1005. },
  1006. )
  1007. class ShortenRefNameTests(TestCase):
  1008. """Tests for shorten_ref_name function."""
  1009. def test_branch_ref(self) -> None:
  1010. """Test shortening branch references."""
  1011. self.assertEqual(b"master", shorten_ref_name(b"refs/heads/master"))
  1012. self.assertEqual(b"develop", shorten_ref_name(b"refs/heads/develop"))
  1013. self.assertEqual(
  1014. b"feature/new-ui", shorten_ref_name(b"refs/heads/feature/new-ui")
  1015. )
  1016. def test_remote_ref(self) -> None:
  1017. """Test shortening remote references."""
  1018. self.assertEqual(b"origin/main", shorten_ref_name(b"refs/remotes/origin/main"))
  1019. self.assertEqual(
  1020. b"upstream/master", shorten_ref_name(b"refs/remotes/upstream/master")
  1021. )
  1022. self.assertEqual(
  1023. b"origin/feature/test",
  1024. shorten_ref_name(b"refs/remotes/origin/feature/test"),
  1025. )
  1026. def test_tag_ref(self) -> None:
  1027. """Test shortening tag references."""
  1028. self.assertEqual(b"v1.0", shorten_ref_name(b"refs/tags/v1.0"))
  1029. self.assertEqual(b"release-2.0", shorten_ref_name(b"refs/tags/release-2.0"))
  1030. def test_special_refs(self) -> None:
  1031. """Test that special refs are not shortened."""
  1032. self.assertEqual(b"HEAD", shorten_ref_name(b"HEAD"))
  1033. self.assertEqual(b"FETCH_HEAD", shorten_ref_name(b"FETCH_HEAD"))
  1034. self.assertEqual(b"ORIG_HEAD", shorten_ref_name(b"ORIG_HEAD"))
  1035. def test_other_refs(self) -> None:
  1036. """Test refs that don't match standard prefixes."""
  1037. # Refs that don't match any standard prefix are returned as-is
  1038. self.assertEqual(b"refs/stash", shorten_ref_name(b"refs/stash"))
  1039. self.assertEqual(b"refs/bisect/good", shorten_ref_name(b"refs/bisect/good"))