test_refs.py 35 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957
  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 public 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. parse_remote_ref,
  37. parse_symref_value,
  38. read_packed_refs,
  39. read_packed_refs_with_peeled,
  40. split_peeled_refs,
  41. strip_peeled_refs,
  42. write_packed_refs,
  43. )
  44. from dulwich.repo import Repo
  45. from dulwich.tests.utils import open_repo, tear_down_repo
  46. from . import SkipTest, TestCase
  47. class CheckRefFormatTests(TestCase):
  48. """Tests for the check_ref_format function.
  49. These are the same tests as in the git test suite.
  50. """
  51. def test_valid(self) -> None:
  52. self.assertTrue(check_ref_format(b"heads/foo"))
  53. self.assertTrue(check_ref_format(b"foo/bar/baz"))
  54. self.assertTrue(check_ref_format(b"refs///heads/foo"))
  55. self.assertTrue(check_ref_format(b"foo./bar"))
  56. self.assertTrue(check_ref_format(b"heads/foo@bar"))
  57. self.assertTrue(check_ref_format(b"heads/fix.lock.error"))
  58. def test_invalid(self) -> None:
  59. self.assertFalse(check_ref_format(b"foo"))
  60. self.assertFalse(check_ref_format(b"heads/foo/"))
  61. self.assertFalse(check_ref_format(b"./foo"))
  62. self.assertFalse(check_ref_format(b".refs/foo"))
  63. self.assertFalse(check_ref_format(b"heads/foo..bar"))
  64. self.assertFalse(check_ref_format(b"heads/foo?bar"))
  65. self.assertFalse(check_ref_format(b"heads/foo.lock"))
  66. self.assertFalse(check_ref_format(b"heads/v@{ation"))
  67. self.assertFalse(check_ref_format(b"heads/foo\bar"))
  68. ONES = b"1" * 40
  69. TWOS = b"2" * 40
  70. THREES = b"3" * 40
  71. FOURS = b"4" * 40
  72. class PackedRefsFileTests(TestCase):
  73. def test_split_ref_line_errors(self) -> None:
  74. self.assertRaises(errors.PackedRefsException, _split_ref_line, b"singlefield")
  75. self.assertRaises(errors.PackedRefsException, _split_ref_line, b"badsha name")
  76. self.assertRaises(
  77. errors.PackedRefsException,
  78. _split_ref_line,
  79. ONES + b" bad/../refname",
  80. )
  81. def test_read_without_peeled(self) -> None:
  82. f = BytesIO(b"\n".join([b"# comment", ONES + b" ref/1", TWOS + b" ref/2"]))
  83. self.assertEqual(
  84. [(ONES, b"ref/1"), (TWOS, b"ref/2")], list(read_packed_refs(f))
  85. )
  86. def test_read_without_peeled_errors(self) -> None:
  87. f = BytesIO(b"\n".join([ONES + b" ref/1", b"^" + TWOS]))
  88. self.assertRaises(errors.PackedRefsException, list, read_packed_refs(f))
  89. def test_read_with_peeled(self) -> None:
  90. f = BytesIO(
  91. b"\n".join(
  92. [
  93. ONES + b" ref/1",
  94. TWOS + b" ref/2",
  95. b"^" + THREES,
  96. FOURS + b" ref/4",
  97. ]
  98. )
  99. )
  100. self.assertEqual(
  101. [
  102. (ONES, b"ref/1", None),
  103. (TWOS, b"ref/2", THREES),
  104. (FOURS, b"ref/4", None),
  105. ],
  106. list(read_packed_refs_with_peeled(f)),
  107. )
  108. def test_read_with_peeled_errors(self) -> None:
  109. f = BytesIO(b"\n".join([b"^" + TWOS, ONES + b" ref/1"]))
  110. self.assertRaises(errors.PackedRefsException, list, read_packed_refs(f))
  111. f = BytesIO(b"\n".join([ONES + b" ref/1", b"^" + TWOS, b"^" + THREES]))
  112. self.assertRaises(errors.PackedRefsException, list, read_packed_refs(f))
  113. def test_write_with_peeled(self) -> None:
  114. f = BytesIO()
  115. write_packed_refs(f, {b"ref/1": ONES, b"ref/2": TWOS}, {b"ref/1": THREES})
  116. self.assertEqual(
  117. b"\n".join(
  118. [
  119. b"# pack-refs with: peeled",
  120. ONES + b" ref/1",
  121. b"^" + THREES,
  122. TWOS + b" ref/2",
  123. ]
  124. )
  125. + b"\n",
  126. f.getvalue(),
  127. )
  128. def test_write_without_peeled(self) -> None:
  129. f = BytesIO()
  130. write_packed_refs(f, {b"ref/1": ONES, b"ref/2": TWOS})
  131. self.assertEqual(
  132. b"\n".join([ONES + b" ref/1", TWOS + b" ref/2"]) + b"\n",
  133. f.getvalue(),
  134. )
  135. # Dict of refs that we expect all RefsContainerTests subclasses to define.
  136. _TEST_REFS = {
  137. b"HEAD": b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  138. b"refs/heads/40-char-ref-aaaaaaaaaaaaaaaaaa": b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  139. b"refs/heads/master": b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  140. b"refs/heads/packed": b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  141. b"refs/tags/refs-0.1": b"df6800012397fb85c56e7418dd4eb9405dee075c",
  142. b"refs/tags/refs-0.2": b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8",
  143. b"refs/heads/loop": b"ref: refs/heads/loop",
  144. }
  145. class RefsContainerTests:
  146. def test_keys(self) -> None:
  147. actual_keys = set(self._refs.keys())
  148. self.assertEqual(set(self._refs.allkeys()), actual_keys)
  149. self.assertEqual(set(_TEST_REFS.keys()), actual_keys)
  150. actual_keys = self._refs.keys(b"refs/heads")
  151. actual_keys.discard(b"loop")
  152. self.assertEqual(
  153. [b"40-char-ref-aaaaaaaaaaaaaaaaaa", b"master", b"packed"],
  154. sorted(actual_keys),
  155. )
  156. self.assertEqual(
  157. [b"refs-0.1", b"refs-0.2"], sorted(self._refs.keys(b"refs/tags"))
  158. )
  159. def test_iter(self) -> None:
  160. actual_keys = set(self._refs.keys())
  161. self.assertEqual(set(self._refs), actual_keys)
  162. self.assertEqual(set(_TEST_REFS.keys()), actual_keys)
  163. def test_as_dict(self) -> None:
  164. # refs/heads/loop does not show up even if it exists
  165. expected_refs = dict(_TEST_REFS)
  166. del expected_refs[b"refs/heads/loop"]
  167. self.assertEqual(expected_refs, self._refs.as_dict())
  168. def test_get_symrefs(self) -> None:
  169. self._refs.set_symbolic_ref(b"refs/heads/src", b"refs/heads/dst")
  170. symrefs = self._refs.get_symrefs()
  171. if b"HEAD" in symrefs:
  172. symrefs.pop(b"HEAD")
  173. self.assertEqual(
  174. {
  175. b"refs/heads/src": b"refs/heads/dst",
  176. b"refs/heads/loop": b"refs/heads/loop",
  177. },
  178. symrefs,
  179. )
  180. def test_setitem(self) -> None:
  181. self._refs[b"refs/some/ref"] = b"42d06bd4b77fed026b154d16493e5deab78f02ec"
  182. self.assertEqual(
  183. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  184. self._refs[b"refs/some/ref"],
  185. )
  186. # should accept symref
  187. self._refs[b"refs/heads/symbolic"] = b"ref: refs/heads/master"
  188. self.assertEqual(
  189. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  190. self._refs[b"refs/heads/symbolic"],
  191. )
  192. # should not accept bad ref names
  193. self.assertRaises(
  194. errors.RefFormatError,
  195. self._refs.__setitem__,
  196. b"notrefs/foo",
  197. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  198. )
  199. # should not accept short sha
  200. self.assertRaises(
  201. ValueError,
  202. self._refs.__setitem__,
  203. b"refs/some/ref",
  204. b"42d06bd",
  205. )
  206. def test_set_if_equals(self) -> None:
  207. nines = b"9" * 40
  208. self.assertFalse(self._refs.set_if_equals(b"HEAD", b"c0ffee", nines))
  209. self.assertEqual(
  210. b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"HEAD"]
  211. )
  212. self.assertTrue(
  213. self._refs.set_if_equals(
  214. b"HEAD", b"42d06bd4b77fed026b154d16493e5deab78f02ec", nines
  215. )
  216. )
  217. self.assertEqual(nines, self._refs[b"HEAD"])
  218. # Setting the ref again is a no-op, but will return True.
  219. self.assertTrue(self._refs.set_if_equals(b"HEAD", nines, nines))
  220. self.assertEqual(nines, self._refs[b"HEAD"])
  221. self.assertTrue(self._refs.set_if_equals(b"refs/heads/master", None, nines))
  222. self.assertEqual(nines, self._refs[b"refs/heads/master"])
  223. self.assertTrue(
  224. self._refs.set_if_equals(b"refs/heads/nonexistent", ZERO_SHA, nines)
  225. )
  226. self.assertEqual(nines, self._refs[b"refs/heads/nonexistent"])
  227. def test_add_if_new(self) -> None:
  228. nines = b"9" * 40
  229. self.assertFalse(self._refs.add_if_new(b"refs/heads/master", nines))
  230. self.assertEqual(
  231. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  232. self._refs[b"refs/heads/master"],
  233. )
  234. self.assertTrue(self._refs.add_if_new(b"refs/some/ref", nines))
  235. self.assertEqual(nines, self._refs[b"refs/some/ref"])
  236. def test_set_symbolic_ref(self) -> None:
  237. self._refs.set_symbolic_ref(b"refs/heads/symbolic", b"refs/heads/master")
  238. self.assertEqual(
  239. b"ref: refs/heads/master",
  240. self._refs.read_loose_ref(b"refs/heads/symbolic"),
  241. )
  242. self.assertEqual(
  243. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  244. self._refs[b"refs/heads/symbolic"],
  245. )
  246. def test_set_symbolic_ref_overwrite(self) -> None:
  247. nines = b"9" * 40
  248. self.assertNotIn(b"refs/heads/symbolic", self._refs)
  249. self._refs[b"refs/heads/symbolic"] = nines
  250. self.assertEqual(nines, self._refs.read_loose_ref(b"refs/heads/symbolic"))
  251. self._refs.set_symbolic_ref(b"refs/heads/symbolic", b"refs/heads/master")
  252. self.assertEqual(
  253. b"ref: refs/heads/master",
  254. self._refs.read_loose_ref(b"refs/heads/symbolic"),
  255. )
  256. self.assertEqual(
  257. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  258. self._refs[b"refs/heads/symbolic"],
  259. )
  260. def test_check_refname(self) -> None:
  261. self._refs._check_refname(b"HEAD")
  262. self._refs._check_refname(b"refs/stash")
  263. self._refs._check_refname(b"refs/heads/foo")
  264. self.assertRaises(errors.RefFormatError, self._refs._check_refname, b"refs")
  265. self.assertRaises(
  266. errors.RefFormatError, self._refs._check_refname, b"notrefs/foo"
  267. )
  268. def test_contains(self) -> None:
  269. self.assertIn(b"refs/heads/master", self._refs)
  270. self.assertNotIn(b"refs/heads/bar", self._refs)
  271. def test_delitem(self) -> None:
  272. self.assertEqual(
  273. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  274. self._refs[b"refs/heads/master"],
  275. )
  276. del self._refs[b"refs/heads/master"]
  277. self.assertRaises(KeyError, lambda: self._refs[b"refs/heads/master"])
  278. def test_remove_if_equals(self) -> None:
  279. self.assertFalse(self._refs.remove_if_equals(b"HEAD", b"c0ffee"))
  280. self.assertEqual(
  281. b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"HEAD"]
  282. )
  283. self.assertTrue(
  284. self._refs.remove_if_equals(
  285. b"refs/tags/refs-0.2",
  286. b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8",
  287. )
  288. )
  289. self.assertTrue(self._refs.remove_if_equals(b"refs/tags/refs-0.2", ZERO_SHA))
  290. self.assertNotIn(b"refs/tags/refs-0.2", self._refs)
  291. def test_import_refs_name(self) -> None:
  292. self._refs[b"refs/remotes/origin/other"] = (
  293. b"48d01bd4b77fed026b154d16493e5deab78f02ec"
  294. )
  295. self._refs.import_refs(
  296. b"refs/remotes/origin",
  297. {b"master": b"42d06bd4b77fed026b154d16493e5deab78f02ec"},
  298. )
  299. self.assertEqual(
  300. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  301. self._refs[b"refs/remotes/origin/master"],
  302. )
  303. self.assertEqual(
  304. b"48d01bd4b77fed026b154d16493e5deab78f02ec",
  305. self._refs[b"refs/remotes/origin/other"],
  306. )
  307. def test_import_refs_name_prune(self) -> None:
  308. self._refs[b"refs/remotes/origin/other"] = (
  309. b"48d01bd4b77fed026b154d16493e5deab78f02ec"
  310. )
  311. self._refs.import_refs(
  312. b"refs/remotes/origin",
  313. {b"master": b"42d06bd4b77fed026b154d16493e5deab78f02ec"},
  314. prune=True,
  315. )
  316. self.assertEqual(
  317. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  318. self._refs[b"refs/remotes/origin/master"],
  319. )
  320. self.assertNotIn(b"refs/remotes/origin/other", self._refs)
  321. class DictRefsContainerTests(RefsContainerTests, TestCase):
  322. def setUp(self) -> None:
  323. TestCase.setUp(self)
  324. self._refs = DictRefsContainer(dict(_TEST_REFS))
  325. def test_invalid_refname(self) -> None:
  326. # FIXME: Move this test into RefsContainerTests, but requires
  327. # some way of injecting invalid refs.
  328. self._refs._refs[b"refs/stash"] = b"00" * 20
  329. expected_refs = dict(_TEST_REFS)
  330. del expected_refs[b"refs/heads/loop"]
  331. expected_refs[b"refs/stash"] = b"00" * 20
  332. self.assertEqual(expected_refs, self._refs.as_dict())
  333. def test_set_if_equals_with_symbolic_ref(self) -> None:
  334. # Test that set_if_equals only updates the requested ref,
  335. # not all refs in a symbolic reference chain
  336. # The bug in the original implementation was that when follow()
  337. # was called on a ref, it would return all refs in the chain,
  338. # and set_if_equals would update ALL of them instead of just the
  339. # requested ref.
  340. # Set up refs
  341. master_sha = b"1" * 40
  342. feature_sha = b"2" * 40
  343. new_sha = b"3" * 40
  344. self._refs[b"refs/heads/master"] = master_sha
  345. self._refs[b"refs/heads/feature"] = feature_sha
  346. # Create a second symbolic ref pointing to feature
  347. self._refs.set_symbolic_ref(b"refs/heads/other", b"refs/heads/feature")
  348. # Update refs/heads/other through set_if_equals
  349. # With the bug, this would update BOTH refs/heads/other AND refs/heads/feature
  350. # Without the bug, only refs/heads/other should be updated
  351. # Note: old_ref needs to be the actual stored value (the symref)
  352. self.assertTrue(
  353. self._refs.set_if_equals(
  354. b"refs/heads/other", b"ref: refs/heads/feature", new_sha
  355. )
  356. )
  357. # refs/heads/other should now directly point to new_sha
  358. self.assertEqual(self._refs.read_ref(b"refs/heads/other"), new_sha)
  359. # refs/heads/feature should remain unchanged
  360. # With the bug, refs/heads/feature would also be incorrectly updated to new_sha
  361. self.assertEqual(self._refs[b"refs/heads/feature"], feature_sha)
  362. self.assertEqual(self._refs[b"refs/heads/master"], master_sha)
  363. class DiskRefsContainerTests(RefsContainerTests, TestCase):
  364. def setUp(self) -> None:
  365. TestCase.setUp(self)
  366. self._repo = open_repo("refs.git")
  367. self.addCleanup(tear_down_repo, self._repo)
  368. self._refs = self._repo.refs
  369. def test_get_packed_refs(self) -> None:
  370. self.assertEqual(
  371. {
  372. b"refs/heads/packed": b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  373. b"refs/tags/refs-0.1": b"df6800012397fb85c56e7418dd4eb9405dee075c",
  374. },
  375. self._refs.get_packed_refs(),
  376. )
  377. def test_get_peeled_not_packed(self) -> None:
  378. # not packed
  379. self.assertEqual(None, self._refs.get_peeled(b"refs/tags/refs-0.2"))
  380. self.assertEqual(
  381. b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8",
  382. self._refs[b"refs/tags/refs-0.2"],
  383. )
  384. # packed, known not peelable
  385. self.assertEqual(
  386. self._refs[b"refs/heads/packed"],
  387. self._refs.get_peeled(b"refs/heads/packed"),
  388. )
  389. # packed, peeled
  390. self.assertEqual(
  391. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  392. self._refs.get_peeled(b"refs/tags/refs-0.1"),
  393. )
  394. def test_setitem(self) -> None:
  395. RefsContainerTests.test_setitem(self)
  396. path = os.path.join(self._refs.path, b"refs", b"some", b"ref")
  397. with open(path, "rb") as f:
  398. self.assertEqual(b"42d06bd4b77fed026b154d16493e5deab78f02ec", f.read()[:40])
  399. self.assertRaises(
  400. OSError,
  401. self._refs.__setitem__,
  402. b"refs/some/ref/sub",
  403. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  404. )
  405. def test_delete_refs_container(self) -> None:
  406. # We shouldn't delete the refs directory
  407. self._refs[b"refs/heads/blah"] = b"42d06bd4b77fed026b154d16493e5deab78f02ec"
  408. for ref in self._refs.allkeys():
  409. del self._refs[ref]
  410. self.assertTrue(os.path.exists(os.path.join(self._refs.path, b"refs")))
  411. def test_setitem_packed(self) -> None:
  412. with open(os.path.join(self._refs.path, b"packed-refs"), "w") as f:
  413. f.write("# pack-refs with: peeled fully-peeled sorted \n")
  414. f.write("42d06bd4b77fed026b154d16493e5deab78f02ec refs/heads/packed\n")
  415. # It's allowed to set a new ref on a packed ref, the new ref will be
  416. # placed outside on refs/
  417. self._refs[b"refs/heads/packed"] = b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8"
  418. packed_ref_path = os.path.join(self._refs.path, b"refs", b"heads", b"packed")
  419. with open(packed_ref_path, "rb") as f:
  420. self.assertEqual(b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8", f.read()[:40])
  421. self.assertRaises(
  422. OSError,
  423. self._refs.__setitem__,
  424. b"refs/heads/packed/sub",
  425. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  426. )
  427. # this shouldn't overwrite the packed refs
  428. self.assertEqual(
  429. {b"refs/heads/packed": b"42d06bd4b77fed026b154d16493e5deab78f02ec"},
  430. self._refs.get_packed_refs(),
  431. )
  432. def test_add_packed_refs(self) -> None:
  433. # first, create a non-packed ref
  434. self._refs[b"refs/heads/packed"] = b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8"
  435. packed_ref_path = os.path.join(self._refs.path, b"refs", b"heads", b"packed")
  436. self.assertTrue(os.path.exists(packed_ref_path))
  437. # now overwrite that with a packed ref
  438. packed_refs_file_path = os.path.join(self._refs.path, b"packed-refs")
  439. self._refs.add_packed_refs(
  440. {
  441. b"refs/heads/packed": b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  442. }
  443. )
  444. # that should kill the file
  445. self.assertFalse(os.path.exists(packed_ref_path))
  446. # now delete the packed ref
  447. self._refs.add_packed_refs(
  448. {
  449. b"refs/heads/packed": None,
  450. }
  451. )
  452. # and it's gone!
  453. self.assertFalse(os.path.exists(packed_ref_path))
  454. self.assertRaises(
  455. KeyError,
  456. self._refs.__getitem__,
  457. b"refs/heads/packed",
  458. )
  459. # just in case, make sure we can't pack HEAD
  460. self.assertRaises(
  461. ValueError,
  462. self._refs.add_packed_refs,
  463. {b"HEAD": "02ac81614bcdbd585a37b4b0edf8cb8a"},
  464. )
  465. # delete all packed refs
  466. self._refs.add_packed_refs({ref: None for ref in self._refs.get_packed_refs()})
  467. self.assertEqual({}, self._refs.get_packed_refs())
  468. # remove the packed ref file, and check that adding nothing doesn't affect that
  469. os.remove(packed_refs_file_path)
  470. # adding nothing doesn't make it reappear
  471. self._refs.add_packed_refs({})
  472. self.assertFalse(os.path.exists(packed_refs_file_path))
  473. def test_setitem_symbolic(self) -> None:
  474. ones = b"1" * 40
  475. self._refs[b"HEAD"] = ones
  476. self.assertEqual(ones, self._refs[b"HEAD"])
  477. # ensure HEAD was not modified
  478. f = open(os.path.join(self._refs.path, b"HEAD"), "rb")
  479. v = next(iter(f)).rstrip(b"\n\r")
  480. f.close()
  481. self.assertEqual(b"ref: refs/heads/master", v)
  482. # ensure the symbolic link was written through
  483. f = open(os.path.join(self._refs.path, b"refs", b"heads", b"master"), "rb")
  484. self.assertEqual(ones, f.read()[:40])
  485. f.close()
  486. def test_set_if_equals(self) -> None:
  487. RefsContainerTests.test_set_if_equals(self)
  488. # ensure symref was followed
  489. self.assertEqual(b"9" * 40, self._refs[b"refs/heads/master"])
  490. # ensure lockfile was deleted
  491. self.assertFalse(
  492. os.path.exists(
  493. os.path.join(self._refs.path, b"refs", b"heads", b"master.lock")
  494. )
  495. )
  496. self.assertFalse(os.path.exists(os.path.join(self._refs.path, b"HEAD.lock")))
  497. def test_add_if_new_packed(self) -> None:
  498. # don't overwrite packed ref
  499. self.assertFalse(self._refs.add_if_new(b"refs/tags/refs-0.1", b"9" * 40))
  500. self.assertEqual(
  501. b"df6800012397fb85c56e7418dd4eb9405dee075c",
  502. self._refs[b"refs/tags/refs-0.1"],
  503. )
  504. def test_add_if_new_symbolic(self) -> None:
  505. # Use an empty repo instead of the default.
  506. repo_dir = os.path.join(tempfile.mkdtemp(), "test")
  507. os.makedirs(repo_dir)
  508. repo = Repo.init(repo_dir)
  509. self.addCleanup(tear_down_repo, repo)
  510. refs = repo.refs
  511. nines = b"9" * 40
  512. self.assertEqual(b"ref: refs/heads/master", refs.read_ref(b"HEAD"))
  513. self.assertNotIn(b"refs/heads/master", refs)
  514. self.assertTrue(refs.add_if_new(b"HEAD", nines))
  515. self.assertEqual(b"ref: refs/heads/master", refs.read_ref(b"HEAD"))
  516. self.assertEqual(nines, refs[b"HEAD"])
  517. self.assertEqual(nines, refs[b"refs/heads/master"])
  518. self.assertFalse(refs.add_if_new(b"HEAD", b"1" * 40))
  519. self.assertEqual(nines, refs[b"HEAD"])
  520. self.assertEqual(nines, refs[b"refs/heads/master"])
  521. def test_follow(self) -> None:
  522. self.assertEqual(
  523. (
  524. [b"HEAD", b"refs/heads/master"],
  525. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  526. ),
  527. self._refs.follow(b"HEAD"),
  528. )
  529. self.assertEqual(
  530. (
  531. [b"refs/heads/master"],
  532. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  533. ),
  534. self._refs.follow(b"refs/heads/master"),
  535. )
  536. self.assertRaises(SymrefLoop, self._refs.follow, b"refs/heads/loop")
  537. def test_set_overwrite_loop(self) -> None:
  538. self.assertRaises(SymrefLoop, self._refs.follow, b"refs/heads/loop")
  539. self._refs[b"refs/heads/loop"] = b"42d06bd4b77fed026b154d16493e5deab78f02ec"
  540. self.assertEqual(
  541. ([b"refs/heads/loop"], b"42d06bd4b77fed026b154d16493e5deab78f02ec"),
  542. self._refs.follow(b"refs/heads/loop"),
  543. )
  544. def test_delitem(self) -> None:
  545. RefsContainerTests.test_delitem(self)
  546. ref_file = os.path.join(self._refs.path, b"refs", b"heads", b"master")
  547. self.assertFalse(os.path.exists(ref_file))
  548. self.assertNotIn(b"refs/heads/master", self._refs.get_packed_refs())
  549. def test_delitem_symbolic(self) -> None:
  550. self.assertEqual(b"ref: refs/heads/master", self._refs.read_loose_ref(b"HEAD"))
  551. del self._refs[b"HEAD"]
  552. self.assertRaises(KeyError, lambda: self._refs[b"HEAD"])
  553. self.assertEqual(
  554. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  555. self._refs[b"refs/heads/master"],
  556. )
  557. self.assertFalse(os.path.exists(os.path.join(self._refs.path, b"HEAD")))
  558. def test_remove_if_equals_symref(self) -> None:
  559. # HEAD is a symref, so shouldn't equal its dereferenced value
  560. self.assertFalse(
  561. self._refs.remove_if_equals(
  562. b"HEAD", b"42d06bd4b77fed026b154d16493e5deab78f02ec"
  563. )
  564. )
  565. self.assertTrue(
  566. self._refs.remove_if_equals(
  567. b"refs/heads/master",
  568. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  569. )
  570. )
  571. self.assertRaises(KeyError, lambda: self._refs[b"refs/heads/master"])
  572. # HEAD is now a broken symref
  573. self.assertRaises(KeyError, lambda: self._refs[b"HEAD"])
  574. self.assertEqual(b"ref: refs/heads/master", self._refs.read_loose_ref(b"HEAD"))
  575. self.assertFalse(
  576. os.path.exists(
  577. os.path.join(self._refs.path, b"refs", b"heads", b"master.lock")
  578. )
  579. )
  580. self.assertFalse(os.path.exists(os.path.join(self._refs.path, b"HEAD.lock")))
  581. def test_remove_packed_without_peeled(self) -> None:
  582. refs_file = os.path.join(self._repo.path, "packed-refs")
  583. f = GitFile(refs_file)
  584. refs_data = f.read()
  585. f.close()
  586. f = GitFile(refs_file, "wb")
  587. f.write(
  588. b"\n".join(
  589. line
  590. for line in refs_data.split(b"\n")
  591. if not line or line[0] not in b"#^"
  592. )
  593. )
  594. f.close()
  595. self._repo = Repo(self._repo.path)
  596. refs = self._repo.refs
  597. self.assertTrue(
  598. refs.remove_if_equals(
  599. b"refs/heads/packed",
  600. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  601. )
  602. )
  603. def test_remove_if_equals_packed(self) -> None:
  604. # test removing ref that is only packed
  605. self.assertEqual(
  606. b"df6800012397fb85c56e7418dd4eb9405dee075c",
  607. self._refs[b"refs/tags/refs-0.1"],
  608. )
  609. self.assertTrue(
  610. self._refs.remove_if_equals(
  611. b"refs/tags/refs-0.1",
  612. b"df6800012397fb85c56e7418dd4eb9405dee075c",
  613. )
  614. )
  615. self.assertRaises(KeyError, lambda: self._refs[b"refs/tags/refs-0.1"])
  616. def test_remove_parent(self) -> None:
  617. self._refs[b"refs/heads/foo/bar"] = b"df6800012397fb85c56e7418dd4eb9405dee075c"
  618. del self._refs[b"refs/heads/foo/bar"]
  619. ref_file = os.path.join(
  620. self._refs.path,
  621. b"refs",
  622. b"heads",
  623. b"foo",
  624. b"bar",
  625. )
  626. self.assertFalse(os.path.exists(ref_file))
  627. ref_file = os.path.join(self._refs.path, b"refs", b"heads", b"foo")
  628. self.assertFalse(os.path.exists(ref_file))
  629. ref_file = os.path.join(self._refs.path, b"refs", b"heads")
  630. self.assertTrue(os.path.exists(ref_file))
  631. self._refs[b"refs/heads/foo"] = b"df6800012397fb85c56e7418dd4eb9405dee075c"
  632. def test_read_ref(self) -> None:
  633. self.assertEqual(b"ref: refs/heads/master", self._refs.read_ref(b"HEAD"))
  634. self.assertEqual(
  635. b"42d06bd4b77fed026b154d16493e5deab78f02ec",
  636. self._refs.read_ref(b"refs/heads/packed"),
  637. )
  638. self.assertEqual(None, self._refs.read_ref(b"nonexistent"))
  639. def test_read_loose_ref(self) -> None:
  640. self._refs[b"refs/heads/foo"] = b"df6800012397fb85c56e7418dd4eb9405dee075c"
  641. self.assertEqual(None, self._refs.read_ref(b"refs/heads/foo/bar"))
  642. def test_non_ascii(self) -> None:
  643. try:
  644. encoded_ref = os.fsencode("refs/tags/schön")
  645. except UnicodeEncodeError as exc:
  646. raise SkipTest(
  647. "filesystem encoding doesn't support special character"
  648. ) from exc
  649. p = os.path.join(os.fsencode(self._repo.path), encoded_ref)
  650. with open(p, "w") as f:
  651. f.write("00" * 20)
  652. expected_refs = dict(_TEST_REFS)
  653. expected_refs[encoded_ref] = b"00" * 20
  654. del expected_refs[b"refs/heads/loop"]
  655. self.assertEqual(expected_refs, self._repo.get_refs())
  656. def test_cyrillic(self) -> None:
  657. if sys.platform in ("darwin", "win32"):
  658. raise SkipTest("filesystem encoding doesn't support arbitrary bytes")
  659. # reported in https://github.com/dulwich/dulwich/issues/608
  660. name = b"\xcd\xee\xe2\xe0\xff\xe2\xe5\xf2\xea\xe01"
  661. encoded_ref = b"refs/heads/" + name
  662. with open(os.path.join(os.fsencode(self._repo.path), encoded_ref), "w") as f:
  663. f.write("00" * 20)
  664. expected_refs = set(_TEST_REFS.keys())
  665. expected_refs.add(encoded_ref)
  666. self.assertEqual(expected_refs, set(self._repo.refs.allkeys()))
  667. self.assertEqual(
  668. {r[len(b"refs/") :] for r in expected_refs if r.startswith(b"refs/")},
  669. set(self._repo.refs.subkeys(b"refs/")),
  670. )
  671. expected_refs.remove(b"refs/heads/loop")
  672. expected_refs.add(b"HEAD")
  673. self.assertEqual(expected_refs, set(self._repo.get_refs().keys()))
  674. _TEST_REFS_SERIALIZED = (
  675. b"42d06bd4b77fed026b154d16493e5deab78f02ec\t"
  676. b"refs/heads/40-char-ref-aaaaaaaaaaaaaaaaaa\n"
  677. b"42d06bd4b77fed026b154d16493e5deab78f02ec\trefs/heads/master\n"
  678. b"42d06bd4b77fed026b154d16493e5deab78f02ec\trefs/heads/packed\n"
  679. b"df6800012397fb85c56e7418dd4eb9405dee075c\trefs/tags/refs-0.1\n"
  680. b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8\trefs/tags/refs-0.2\n"
  681. )
  682. class DiskRefsContainerPathlibTests(TestCase):
  683. def test_pathlib_init(self) -> None:
  684. from pathlib import Path
  685. from dulwich.refs import DiskRefsContainer
  686. # Create a temporary directory
  687. temp_dir = tempfile.mkdtemp()
  688. self.addCleanup(os.rmdir, temp_dir)
  689. # Test with pathlib.Path
  690. path_obj = Path(temp_dir)
  691. refs = DiskRefsContainer(path_obj)
  692. self.assertEqual(refs.path, temp_dir.encode())
  693. # Test refpath with pathlib initialized container
  694. ref_path = refs.refpath(b"HEAD")
  695. self.assertTrue(isinstance(ref_path, bytes))
  696. self.assertEqual(ref_path, os.path.join(temp_dir.encode(), b"HEAD"))
  697. def test_pathlib_worktree_path(self) -> None:
  698. from pathlib import Path
  699. from dulwich.refs import DiskRefsContainer
  700. # Create temporary directories
  701. temp_dir = tempfile.mkdtemp()
  702. worktree_dir = tempfile.mkdtemp()
  703. self.addCleanup(os.rmdir, temp_dir)
  704. self.addCleanup(os.rmdir, worktree_dir)
  705. # Test with pathlib.Path for both paths
  706. path_obj = Path(temp_dir)
  707. worktree_obj = Path(worktree_dir)
  708. refs = DiskRefsContainer(path_obj, worktree_path=worktree_obj)
  709. self.assertEqual(refs.path, temp_dir.encode())
  710. self.assertEqual(refs.worktree_path, worktree_dir.encode())
  711. # Test refpath returns worktree path for HEAD
  712. ref_path = refs.refpath(b"HEAD")
  713. self.assertEqual(ref_path, os.path.join(worktree_dir.encode(), b"HEAD"))
  714. class InfoRefsContainerTests(TestCase):
  715. def test_invalid_refname(self) -> None:
  716. text = _TEST_REFS_SERIALIZED + b"00" * 20 + b"\trefs/stash\n"
  717. refs = InfoRefsContainer(BytesIO(text))
  718. expected_refs = dict(_TEST_REFS)
  719. del expected_refs[b"HEAD"]
  720. expected_refs[b"refs/stash"] = b"00" * 20
  721. del expected_refs[b"refs/heads/loop"]
  722. self.assertEqual(expected_refs, refs.as_dict())
  723. def test_keys(self) -> None:
  724. refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED))
  725. actual_keys = set(refs.keys())
  726. self.assertEqual(set(refs.allkeys()), actual_keys)
  727. expected_refs = dict(_TEST_REFS)
  728. del expected_refs[b"HEAD"]
  729. del expected_refs[b"refs/heads/loop"]
  730. self.assertEqual(set(expected_refs.keys()), actual_keys)
  731. actual_keys = refs.keys(b"refs/heads")
  732. actual_keys.discard(b"loop")
  733. self.assertEqual(
  734. [b"40-char-ref-aaaaaaaaaaaaaaaaaa", b"master", b"packed"],
  735. sorted(actual_keys),
  736. )
  737. self.assertEqual([b"refs-0.1", b"refs-0.2"], sorted(refs.keys(b"refs/tags")))
  738. def test_as_dict(self) -> None:
  739. refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED))
  740. # refs/heads/loop does not show up even if it exists
  741. expected_refs = dict(_TEST_REFS)
  742. del expected_refs[b"HEAD"]
  743. del expected_refs[b"refs/heads/loop"]
  744. self.assertEqual(expected_refs, refs.as_dict())
  745. def test_contains(self) -> None:
  746. refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED))
  747. self.assertIn(b"refs/heads/master", refs)
  748. self.assertNotIn(b"refs/heads/bar", refs)
  749. def test_get_peeled(self) -> None:
  750. refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED))
  751. # refs/heads/loop does not show up even if it exists
  752. self.assertEqual(
  753. _TEST_REFS[b"refs/heads/master"],
  754. refs.get_peeled(b"refs/heads/master"),
  755. )
  756. class ParseSymrefValueTests(TestCase):
  757. def test_valid(self) -> None:
  758. self.assertEqual(b"refs/heads/foo", parse_symref_value(b"ref: refs/heads/foo"))
  759. def test_invalid(self) -> None:
  760. self.assertRaises(ValueError, parse_symref_value, b"foobar")
  761. class ParseRemoteRefTests(TestCase):
  762. def test_valid(self) -> None:
  763. # Test simple case
  764. remote, branch = parse_remote_ref(b"refs/remotes/origin/main")
  765. self.assertEqual(b"origin", remote)
  766. self.assertEqual(b"main", branch)
  767. # Test with branch containing slashes
  768. remote, branch = parse_remote_ref(b"refs/remotes/upstream/feature/new-ui")
  769. self.assertEqual(b"upstream", remote)
  770. self.assertEqual(b"feature/new-ui", branch)
  771. def test_invalid_not_remote_ref(self) -> None:
  772. # Not a remote ref
  773. with self.assertRaises(ValueError) as cm:
  774. parse_remote_ref(b"refs/heads/main")
  775. self.assertIn("Not a remote ref", str(cm.exception))
  776. def test_invalid_format(self) -> None:
  777. # Missing branch name
  778. with self.assertRaises(ValueError) as cm:
  779. parse_remote_ref(b"refs/remotes/origin")
  780. self.assertIn("Invalid remote ref format", str(cm.exception))
  781. # Just the prefix
  782. with self.assertRaises(ValueError) as cm:
  783. parse_remote_ref(b"refs/remotes/")
  784. self.assertIn("Invalid remote ref format", str(cm.exception))
  785. class StripPeeledRefsTests(TestCase):
  786. all_refs: ClassVar[dict[bytes, bytes]] = {
  787. b"refs/heads/master": b"8843d7f92416211de9ebb963ff4ce28125932878",
  788. b"refs/heads/testing": b"186a005b134d8639a58b6731c7c1ea821a6eedba",
  789. b"refs/tags/1.0.0": b"a93db4b0360cc635a2b93675010bac8d101f73f0",
  790. b"refs/tags/1.0.0^{}": b"a93db4b0360cc635a2b93675010bac8d101f73f0",
  791. b"refs/tags/2.0.0": b"0749936d0956c661ac8f8d3483774509c165f89e",
  792. b"refs/tags/2.0.0^{}": b"0749936d0956c661ac8f8d3483774509c165f89e",
  793. }
  794. non_peeled_refs: ClassVar[dict[bytes, bytes]] = {
  795. b"refs/heads/master": b"8843d7f92416211de9ebb963ff4ce28125932878",
  796. b"refs/heads/testing": b"186a005b134d8639a58b6731c7c1ea821a6eedba",
  797. b"refs/tags/1.0.0": b"a93db4b0360cc635a2b93675010bac8d101f73f0",
  798. b"refs/tags/2.0.0": b"0749936d0956c661ac8f8d3483774509c165f89e",
  799. }
  800. def test_strip_peeled_refs(self) -> None:
  801. # Simple check of two dicts
  802. self.assertEqual(strip_peeled_refs(self.all_refs), self.non_peeled_refs)
  803. def test_split_peeled_refs(self) -> None:
  804. (regular, peeled) = split_peeled_refs(self.all_refs)
  805. self.assertEqual(regular, self.non_peeled_refs)
  806. self.assertEqual(
  807. peeled,
  808. {
  809. b"refs/tags/2.0.0": b"0749936d0956c661ac8f8d3483774509c165f89e",
  810. b"refs/tags/1.0.0": b"a93db4b0360cc635a2b93675010bac8d101f73f0",
  811. },
  812. )