# test_refs.py -- tests for refs.py # Copyright (C) 2013 Jelmer Vernooij # # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU # General Public License as public by the Free Software Foundation; version 2.0 # or (at your option) any later version. You can redistribute it and/or # modify it under the terms of either of these two licenses. # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # You should have received a copy of the licenses; if not, see # for a copy of the GNU General Public License # and for a copy of the Apache # License, Version 2.0. # """Tests for dulwich.refs.""" import os import sys import tempfile from io import BytesIO from typing import ClassVar, Dict from dulwich import errors from dulwich.file import GitFile from dulwich.objects import ZERO_SHA from dulwich.refs import ( DictRefsContainer, InfoRefsContainer, SymrefLoop, _split_ref_line, check_ref_format, parse_symref_value, read_packed_refs, read_packed_refs_with_peeled, strip_peeled_refs, write_packed_refs, ) from dulwich.repo import Repo from dulwich.tests.utils import open_repo, tear_down_repo from . import SkipTest, TestCase class CheckRefFormatTests(TestCase): """Tests for the check_ref_format function. These are the same tests as in the git test suite. """ def test_valid(self): self.assertTrue(check_ref_format(b"heads/foo")) self.assertTrue(check_ref_format(b"foo/bar/baz")) self.assertTrue(check_ref_format(b"refs///heads/foo")) self.assertTrue(check_ref_format(b"foo./bar")) self.assertTrue(check_ref_format(b"heads/foo@bar")) self.assertTrue(check_ref_format(b"heads/fix.lock.error")) def test_invalid(self): self.assertFalse(check_ref_format(b"foo")) self.assertFalse(check_ref_format(b"heads/foo/")) self.assertFalse(check_ref_format(b"./foo")) self.assertFalse(check_ref_format(b".refs/foo")) self.assertFalse(check_ref_format(b"heads/foo..bar")) self.assertFalse(check_ref_format(b"heads/foo?bar")) self.assertFalse(check_ref_format(b"heads/foo.lock")) self.assertFalse(check_ref_format(b"heads/v@{ation")) self.assertFalse(check_ref_format(b"heads/foo\bar")) ONES = b"1" * 40 TWOS = b"2" * 40 THREES = b"3" * 40 FOURS = b"4" * 40 class PackedRefsFileTests(TestCase): def test_split_ref_line_errors(self): self.assertRaises(errors.PackedRefsException, _split_ref_line, b"singlefield") self.assertRaises(errors.PackedRefsException, _split_ref_line, b"badsha name") self.assertRaises( errors.PackedRefsException, _split_ref_line, ONES + b" bad/../refname", ) def test_read_without_peeled(self): f = BytesIO(b"\n".join([b"# comment", ONES + b" ref/1", TWOS + b" ref/2"])) self.assertEqual( [(ONES, b"ref/1"), (TWOS, b"ref/2")], list(read_packed_refs(f)) ) def test_read_without_peeled_errors(self): f = BytesIO(b"\n".join([ONES + b" ref/1", b"^" + TWOS])) self.assertRaises(errors.PackedRefsException, list, read_packed_refs(f)) def test_read_with_peeled(self): f = BytesIO( b"\n".join( [ ONES + b" ref/1", TWOS + b" ref/2", b"^" + THREES, FOURS + b" ref/4", ] ) ) self.assertEqual( [ (ONES, b"ref/1", None), (TWOS, b"ref/2", THREES), (FOURS, b"ref/4", None), ], list(read_packed_refs_with_peeled(f)), ) def test_read_with_peeled_errors(self): f = BytesIO(b"\n".join([b"^" + TWOS, ONES + b" ref/1"])) self.assertRaises(errors.PackedRefsException, list, read_packed_refs(f)) f = BytesIO(b"\n".join([ONES + b" ref/1", b"^" + TWOS, b"^" + THREES])) self.assertRaises(errors.PackedRefsException, list, read_packed_refs(f)) def test_write_with_peeled(self): f = BytesIO() write_packed_refs(f, {b"ref/1": ONES, b"ref/2": TWOS}, {b"ref/1": THREES}) self.assertEqual( b"\n".join( [ b"# pack-refs with: peeled", ONES + b" ref/1", b"^" + THREES, TWOS + b" ref/2", ] ) + b"\n", f.getvalue(), ) def test_write_without_peeled(self): f = BytesIO() write_packed_refs(f, {b"ref/1": ONES, b"ref/2": TWOS}) self.assertEqual( b"\n".join([ONES + b" ref/1", TWOS + b" ref/2"]) + b"\n", f.getvalue(), ) # Dict of refs that we expect all RefsContainerTests subclasses to define. _TEST_REFS = { b"HEAD": b"42d06bd4b77fed026b154d16493e5deab78f02ec", b"refs/heads/40-char-ref-aaaaaaaaaaaaaaaaaa": b"42d06bd4b77fed026b154d16493e5deab78f02ec", b"refs/heads/master": b"42d06bd4b77fed026b154d16493e5deab78f02ec", b"refs/heads/packed": b"42d06bd4b77fed026b154d16493e5deab78f02ec", b"refs/tags/refs-0.1": b"df6800012397fb85c56e7418dd4eb9405dee075c", b"refs/tags/refs-0.2": b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8", b"refs/heads/loop": b"ref: refs/heads/loop", } class RefsContainerTests: def test_keys(self): actual_keys = set(self._refs.keys()) self.assertEqual(set(self._refs.allkeys()), actual_keys) self.assertEqual(set(_TEST_REFS.keys()), actual_keys) actual_keys = self._refs.keys(b"refs/heads") actual_keys.discard(b"loop") self.assertEqual( [b"40-char-ref-aaaaaaaaaaaaaaaaaa", b"master", b"packed"], sorted(actual_keys), ) self.assertEqual( [b"refs-0.1", b"refs-0.2"], sorted(self._refs.keys(b"refs/tags")) ) def test_iter(self): actual_keys = set(self._refs.keys()) self.assertEqual(set(self._refs), actual_keys) self.assertEqual(set(_TEST_REFS.keys()), actual_keys) def test_as_dict(self): # refs/heads/loop does not show up even if it exists expected_refs = dict(_TEST_REFS) del expected_refs[b"refs/heads/loop"] self.assertEqual(expected_refs, self._refs.as_dict()) def test_get_symrefs(self): self._refs.set_symbolic_ref(b"refs/heads/src", b"refs/heads/dst") symrefs = self._refs.get_symrefs() if b"HEAD" in symrefs: symrefs.pop(b"HEAD") self.assertEqual( { b"refs/heads/src": b"refs/heads/dst", b"refs/heads/loop": b"refs/heads/loop", }, symrefs, ) def test_setitem(self): self._refs[b"refs/some/ref"] = b"42d06bd4b77fed026b154d16493e5deab78f02ec" self.assertEqual( b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"refs/some/ref"], ) self.assertRaises( errors.RefFormatError, self._refs.__setitem__, b"notrefs/foo", b"42d06bd4b77fed026b154d16493e5deab78f02ec", ) def test_set_if_equals(self): nines = b"9" * 40 self.assertFalse(self._refs.set_if_equals(b"HEAD", b"c0ffee", nines)) self.assertEqual( b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"HEAD"] ) self.assertTrue( self._refs.set_if_equals( b"HEAD", b"42d06bd4b77fed026b154d16493e5deab78f02ec", nines ) ) self.assertEqual(nines, self._refs[b"HEAD"]) # Setting the ref again is a no-op, but will return True. self.assertTrue(self._refs.set_if_equals(b"HEAD", nines, nines)) self.assertEqual(nines, self._refs[b"HEAD"]) self.assertTrue(self._refs.set_if_equals(b"refs/heads/master", None, nines)) self.assertEqual(nines, self._refs[b"refs/heads/master"]) self.assertTrue( self._refs.set_if_equals(b"refs/heads/nonexistent", ZERO_SHA, nines) ) self.assertEqual(nines, self._refs[b"refs/heads/nonexistent"]) def test_add_if_new(self): nines = b"9" * 40 self.assertFalse(self._refs.add_if_new(b"refs/heads/master", nines)) self.assertEqual( b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"refs/heads/master"], ) self.assertTrue(self._refs.add_if_new(b"refs/some/ref", nines)) self.assertEqual(nines, self._refs[b"refs/some/ref"]) def test_set_symbolic_ref(self): self._refs.set_symbolic_ref(b"refs/heads/symbolic", b"refs/heads/master") self.assertEqual( b"ref: refs/heads/master", self._refs.read_loose_ref(b"refs/heads/symbolic"), ) self.assertEqual( b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"refs/heads/symbolic"], ) def test_set_symbolic_ref_overwrite(self): nines = b"9" * 40 self.assertNotIn(b"refs/heads/symbolic", self._refs) self._refs[b"refs/heads/symbolic"] = nines self.assertEqual(nines, self._refs.read_loose_ref(b"refs/heads/symbolic")) self._refs.set_symbolic_ref(b"refs/heads/symbolic", b"refs/heads/master") self.assertEqual( b"ref: refs/heads/master", self._refs.read_loose_ref(b"refs/heads/symbolic"), ) self.assertEqual( b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"refs/heads/symbolic"], ) def test_check_refname(self): self._refs._check_refname(b"HEAD") self._refs._check_refname(b"refs/stash") self._refs._check_refname(b"refs/heads/foo") self.assertRaises(errors.RefFormatError, self._refs._check_refname, b"refs") self.assertRaises( errors.RefFormatError, self._refs._check_refname, b"notrefs/foo" ) def test_contains(self): self.assertIn(b"refs/heads/master", self._refs) self.assertNotIn(b"refs/heads/bar", self._refs) def test_delitem(self): self.assertEqual( b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"refs/heads/master"], ) del self._refs[b"refs/heads/master"] self.assertRaises(KeyError, lambda: self._refs[b"refs/heads/master"]) def test_remove_if_equals(self): self.assertFalse(self._refs.remove_if_equals(b"HEAD", b"c0ffee")) self.assertEqual( b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"HEAD"] ) self.assertTrue( self._refs.remove_if_equals( b"refs/tags/refs-0.2", b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8", ) ) self.assertTrue(self._refs.remove_if_equals(b"refs/tags/refs-0.2", ZERO_SHA)) self.assertNotIn(b"refs/tags/refs-0.2", self._refs) def test_import_refs_name(self): self._refs[b"refs/remotes/origin/other"] = ( b"48d01bd4b77fed026b154d16493e5deab78f02ec" ) self._refs.import_refs( b"refs/remotes/origin", {b"master": b"42d06bd4b77fed026b154d16493e5deab78f02ec"}, ) self.assertEqual( b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"refs/remotes/origin/master"], ) self.assertEqual( b"48d01bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"refs/remotes/origin/other"], ) def test_import_refs_name_prune(self): self._refs[b"refs/remotes/origin/other"] = ( b"48d01bd4b77fed026b154d16493e5deab78f02ec" ) self._refs.import_refs( b"refs/remotes/origin", {b"master": b"42d06bd4b77fed026b154d16493e5deab78f02ec"}, prune=True, ) self.assertEqual( b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"refs/remotes/origin/master"], ) self.assertNotIn(b"refs/remotes/origin/other", self._refs) class DictRefsContainerTests(RefsContainerTests, TestCase): def setUp(self): TestCase.setUp(self) self._refs = DictRefsContainer(dict(_TEST_REFS)) def test_invalid_refname(self): # FIXME: Move this test into RefsContainerTests, but requires # some way of injecting invalid refs. self._refs._refs[b"refs/stash"] = b"00" * 20 expected_refs = dict(_TEST_REFS) del expected_refs[b"refs/heads/loop"] expected_refs[b"refs/stash"] = b"00" * 20 self.assertEqual(expected_refs, self._refs.as_dict()) class DiskRefsContainerTests(RefsContainerTests, TestCase): def setUp(self): TestCase.setUp(self) self._repo = open_repo("refs.git") self.addCleanup(tear_down_repo, self._repo) self._refs = self._repo.refs def test_get_packed_refs(self): self.assertEqual( { b"refs/heads/packed": b"42d06bd4b77fed026b154d16493e5deab78f02ec", b"refs/tags/refs-0.1": b"df6800012397fb85c56e7418dd4eb9405dee075c", }, self._refs.get_packed_refs(), ) def test_get_peeled_not_packed(self): # not packed self.assertEqual(None, self._refs.get_peeled(b"refs/tags/refs-0.2")) self.assertEqual( b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8", self._refs[b"refs/tags/refs-0.2"], ) # packed, known not peelable self.assertEqual( self._refs[b"refs/heads/packed"], self._refs.get_peeled(b"refs/heads/packed"), ) # packed, peeled self.assertEqual( b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs.get_peeled(b"refs/tags/refs-0.1"), ) def test_setitem(self): RefsContainerTests.test_setitem(self) path = os.path.join(self._refs.path, b"refs", b"some", b"ref") with open(path, "rb") as f: self.assertEqual(b"42d06bd4b77fed026b154d16493e5deab78f02ec", f.read()[:40]) self.assertRaises( OSError, self._refs.__setitem__, b"refs/some/ref/sub", b"42d06bd4b77fed026b154d16493e5deab78f02ec", ) def test_delete_refs_container(self): # We shouldn't delete the refs directory self._refs[b"refs/heads/blah"] = b"42d06bd4b77fed026b154d16493e5deab78f02ec" for ref in self._refs.allkeys(): del self._refs[ref] self.assertTrue(os.path.exists(os.path.join(self._refs.path, b"refs"))) def test_setitem_packed(self): with open(os.path.join(self._refs.path, b"packed-refs"), "w") as f: f.write("# pack-refs with: peeled fully-peeled sorted \n") f.write("42d06bd4b77fed026b154d16493e5deab78f02ec refs/heads/packed\n") # It's allowed to set a new ref on a packed ref, the new ref will be # placed outside on refs/ self._refs[b"refs/heads/packed"] = b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8" packed_ref_path = os.path.join(self._refs.path, b"refs", b"heads", b"packed") with open(packed_ref_path, "rb") as f: self.assertEqual(b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8", f.read()[:40]) self.assertRaises( OSError, self._refs.__setitem__, b"refs/heads/packed/sub", b"42d06bd4b77fed026b154d16493e5deab78f02ec", ) # this shouldn't overwrite the packed refs self.assertEqual( {b"refs/heads/packed": b"42d06bd4b77fed026b154d16493e5deab78f02ec"}, self._refs.get_packed_refs(), ) def test_add_packed_refs(self): # first, create a non-packed ref self._refs[b"refs/heads/packed"] = b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8" packed_ref_path = os.path.join(self._refs.path, b"refs", b"heads", b"packed") self.assertTrue(os.path.exists(packed_ref_path)) # now overwrite that with a packed ref packed_refs_file_path = os.path.join(self._refs.path, b"packed-refs") self._refs.add_packed_refs( { b"refs/heads/packed": b"42d06bd4b77fed026b154d16493e5deab78f02ec", } ) # that should kill the file self.assertFalse(os.path.exists(packed_ref_path)) # now delete the packed ref self._refs.add_packed_refs( { b"refs/heads/packed": None, } ) # and it's gone! self.assertFalse(os.path.exists(packed_ref_path)) self.assertRaises( KeyError, self._refs.__getitem__, b"refs/heads/packed", ) # just in case, make sure we can't pack HEAD self.assertRaises( ValueError, self._refs.add_packed_refs, {b"HEAD": "02ac81614bcdbd585a37b4b0edf8cb8a"}, ) # delete all packed refs self._refs.add_packed_refs({ref: None for ref in self._refs.get_packed_refs()}) self.assertEqual({}, self._refs.get_packed_refs()) # remove the packed ref file, and check that adding nothing doesn't affect that os.remove(packed_refs_file_path) # adding nothing doesn't make it reappear self._refs.add_packed_refs({}) self.assertFalse(os.path.exists(packed_refs_file_path)) def test_setitem_symbolic(self): ones = b"1" * 40 self._refs[b"HEAD"] = ones self.assertEqual(ones, self._refs[b"HEAD"]) # ensure HEAD was not modified f = open(os.path.join(self._refs.path, b"HEAD"), "rb") v = next(iter(f)).rstrip(b"\n\r") f.close() self.assertEqual(b"ref: refs/heads/master", v) # ensure the symbolic link was written through f = open(os.path.join(self._refs.path, b"refs", b"heads", b"master"), "rb") self.assertEqual(ones, f.read()[:40]) f.close() def test_set_if_equals(self): RefsContainerTests.test_set_if_equals(self) # ensure symref was followed self.assertEqual(b"9" * 40, self._refs[b"refs/heads/master"]) # ensure lockfile was deleted self.assertFalse( os.path.exists( os.path.join(self._refs.path, b"refs", b"heads", b"master.lock") ) ) self.assertFalse(os.path.exists(os.path.join(self._refs.path, b"HEAD.lock"))) def test_add_if_new_packed(self): # don't overwrite packed ref self.assertFalse(self._refs.add_if_new(b"refs/tags/refs-0.1", b"9" * 40)) self.assertEqual( b"df6800012397fb85c56e7418dd4eb9405dee075c", self._refs[b"refs/tags/refs-0.1"], ) def test_add_if_new_symbolic(self): # Use an empty repo instead of the default. repo_dir = os.path.join(tempfile.mkdtemp(), "test") os.makedirs(repo_dir) repo = Repo.init(repo_dir) self.addCleanup(tear_down_repo, repo) refs = repo.refs nines = b"9" * 40 self.assertEqual(b"ref: refs/heads/master", refs.read_ref(b"HEAD")) self.assertNotIn(b"refs/heads/master", refs) self.assertTrue(refs.add_if_new(b"HEAD", nines)) self.assertEqual(b"ref: refs/heads/master", refs.read_ref(b"HEAD")) self.assertEqual(nines, refs[b"HEAD"]) self.assertEqual(nines, refs[b"refs/heads/master"]) self.assertFalse(refs.add_if_new(b"HEAD", b"1" * 40)) self.assertEqual(nines, refs[b"HEAD"]) self.assertEqual(nines, refs[b"refs/heads/master"]) def test_follow(self): self.assertEqual( ( [b"HEAD", b"refs/heads/master"], b"42d06bd4b77fed026b154d16493e5deab78f02ec", ), self._refs.follow(b"HEAD"), ) self.assertEqual( ( [b"refs/heads/master"], b"42d06bd4b77fed026b154d16493e5deab78f02ec", ), self._refs.follow(b"refs/heads/master"), ) self.assertRaises(SymrefLoop, self._refs.follow, b"refs/heads/loop") def test_set_overwrite_loop(self): self.assertRaises(SymrefLoop, self._refs.follow, b"refs/heads/loop") self._refs[b"refs/heads/loop"] = b"42d06bd4b77fed026b154d16493e5deab78f02ec" self.assertEqual( ([b"refs/heads/loop"], b"42d06bd4b77fed026b154d16493e5deab78f02ec"), self._refs.follow(b"refs/heads/loop"), ) def test_delitem(self): RefsContainerTests.test_delitem(self) ref_file = os.path.join(self._refs.path, b"refs", b"heads", b"master") self.assertFalse(os.path.exists(ref_file)) self.assertNotIn(b"refs/heads/master", self._refs.get_packed_refs()) def test_delitem_symbolic(self): self.assertEqual(b"ref: refs/heads/master", self._refs.read_loose_ref(b"HEAD")) del self._refs[b"HEAD"] self.assertRaises(KeyError, lambda: self._refs[b"HEAD"]) self.assertEqual( b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs[b"refs/heads/master"], ) self.assertFalse(os.path.exists(os.path.join(self._refs.path, b"HEAD"))) def test_remove_if_equals_symref(self): # HEAD is a symref, so shouldn't equal its dereferenced value self.assertFalse( self._refs.remove_if_equals( b"HEAD", b"42d06bd4b77fed026b154d16493e5deab78f02ec" ) ) self.assertTrue( self._refs.remove_if_equals( b"refs/heads/master", b"42d06bd4b77fed026b154d16493e5deab78f02ec", ) ) self.assertRaises(KeyError, lambda: self._refs[b"refs/heads/master"]) # HEAD is now a broken symref self.assertRaises(KeyError, lambda: self._refs[b"HEAD"]) self.assertEqual(b"ref: refs/heads/master", self._refs.read_loose_ref(b"HEAD")) self.assertFalse( os.path.exists( os.path.join(self._refs.path, b"refs", b"heads", b"master.lock") ) ) self.assertFalse(os.path.exists(os.path.join(self._refs.path, b"HEAD.lock"))) def test_remove_packed_without_peeled(self): refs_file = os.path.join(self._repo.path, "packed-refs") f = GitFile(refs_file) refs_data = f.read() f.close() f = GitFile(refs_file, "wb") f.write( b"\n".join( line for line in refs_data.split(b"\n") if not line or line[0] not in b"#^" ) ) f.close() self._repo = Repo(self._repo.path) refs = self._repo.refs self.assertTrue( refs.remove_if_equals( b"refs/heads/packed", b"42d06bd4b77fed026b154d16493e5deab78f02ec", ) ) def test_remove_if_equals_packed(self): # test removing ref that is only packed self.assertEqual( b"df6800012397fb85c56e7418dd4eb9405dee075c", self._refs[b"refs/tags/refs-0.1"], ) self.assertTrue( self._refs.remove_if_equals( b"refs/tags/refs-0.1", b"df6800012397fb85c56e7418dd4eb9405dee075c", ) ) self.assertRaises(KeyError, lambda: self._refs[b"refs/tags/refs-0.1"]) def test_remove_parent(self): self._refs[b"refs/heads/foo/bar"] = b"df6800012397fb85c56e7418dd4eb9405dee075c" del self._refs[b"refs/heads/foo/bar"] ref_file = os.path.join( self._refs.path, b"refs", b"heads", b"foo", b"bar", ) self.assertFalse(os.path.exists(ref_file)) ref_file = os.path.join(self._refs.path, b"refs", b"heads", b"foo") self.assertFalse(os.path.exists(ref_file)) ref_file = os.path.join(self._refs.path, b"refs", b"heads") self.assertTrue(os.path.exists(ref_file)) self._refs[b"refs/heads/foo"] = b"df6800012397fb85c56e7418dd4eb9405dee075c" def test_read_ref(self): self.assertEqual(b"ref: refs/heads/master", self._refs.read_ref(b"HEAD")) self.assertEqual( b"42d06bd4b77fed026b154d16493e5deab78f02ec", self._refs.read_ref(b"refs/heads/packed"), ) self.assertEqual(None, self._refs.read_ref(b"nonexistent")) def test_read_loose_ref(self): self._refs[b"refs/heads/foo"] = b"df6800012397fb85c56e7418dd4eb9405dee075c" self.assertEqual(None, self._refs.read_ref(b"refs/heads/foo/bar")) def test_non_ascii(self): try: encoded_ref = os.fsencode("refs/tags/schön") except UnicodeEncodeError as exc: raise SkipTest( "filesystem encoding doesn't support special character" ) from exc p = os.path.join(os.fsencode(self._repo.path), encoded_ref) with open(p, "w") as f: f.write("00" * 20) expected_refs = dict(_TEST_REFS) expected_refs[encoded_ref] = b"00" * 20 del expected_refs[b"refs/heads/loop"] self.assertEqual(expected_refs, self._repo.get_refs()) def test_cyrillic(self): if sys.platform in ("darwin", "win32"): raise SkipTest("filesystem encoding doesn't support arbitrary bytes") # reported in https://github.com/dulwich/dulwich/issues/608 name = b"\xcd\xee\xe2\xe0\xff\xe2\xe5\xf2\xea\xe01" encoded_ref = b"refs/heads/" + name with open(os.path.join(os.fsencode(self._repo.path), encoded_ref), "w") as f: f.write("00" * 20) expected_refs = set(_TEST_REFS.keys()) expected_refs.add(encoded_ref) self.assertEqual(expected_refs, set(self._repo.refs.allkeys())) self.assertEqual( {r[len(b"refs/") :] for r in expected_refs if r.startswith(b"refs/")}, set(self._repo.refs.subkeys(b"refs/")), ) expected_refs.remove(b"refs/heads/loop") expected_refs.add(b"HEAD") self.assertEqual(expected_refs, set(self._repo.get_refs().keys())) _TEST_REFS_SERIALIZED = ( b"42d06bd4b77fed026b154d16493e5deab78f02ec\t" b"refs/heads/40-char-ref-aaaaaaaaaaaaaaaaaa\n" b"42d06bd4b77fed026b154d16493e5deab78f02ec\trefs/heads/master\n" b"42d06bd4b77fed026b154d16493e5deab78f02ec\trefs/heads/packed\n" b"df6800012397fb85c56e7418dd4eb9405dee075c\trefs/tags/refs-0.1\n" b"3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8\trefs/tags/refs-0.2\n" ) class InfoRefsContainerTests(TestCase): def test_invalid_refname(self): text = _TEST_REFS_SERIALIZED + b"00" * 20 + b"\trefs/stash\n" refs = InfoRefsContainer(BytesIO(text)) expected_refs = dict(_TEST_REFS) del expected_refs[b"HEAD"] expected_refs[b"refs/stash"] = b"00" * 20 del expected_refs[b"refs/heads/loop"] self.assertEqual(expected_refs, refs.as_dict()) def test_keys(self): refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED)) actual_keys = set(refs.keys()) self.assertEqual(set(refs.allkeys()), actual_keys) expected_refs = dict(_TEST_REFS) del expected_refs[b"HEAD"] del expected_refs[b"refs/heads/loop"] self.assertEqual(set(expected_refs.keys()), actual_keys) actual_keys = refs.keys(b"refs/heads") actual_keys.discard(b"loop") self.assertEqual( [b"40-char-ref-aaaaaaaaaaaaaaaaaa", b"master", b"packed"], sorted(actual_keys), ) self.assertEqual([b"refs-0.1", b"refs-0.2"], sorted(refs.keys(b"refs/tags"))) def test_as_dict(self): refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED)) # refs/heads/loop does not show up even if it exists expected_refs = dict(_TEST_REFS) del expected_refs[b"HEAD"] del expected_refs[b"refs/heads/loop"] self.assertEqual(expected_refs, refs.as_dict()) def test_contains(self): refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED)) self.assertIn(b"refs/heads/master", refs) self.assertNotIn(b"refs/heads/bar", refs) def test_get_peeled(self): refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED)) # refs/heads/loop does not show up even if it exists self.assertEqual( _TEST_REFS[b"refs/heads/master"], refs.get_peeled(b"refs/heads/master"), ) class ParseSymrefValueTests(TestCase): def test_valid(self): self.assertEqual(b"refs/heads/foo", parse_symref_value(b"ref: refs/heads/foo")) def test_invalid(self): self.assertRaises(ValueError, parse_symref_value, b"foobar") class StripPeeledRefsTests(TestCase): all_refs: ClassVar[Dict[bytes, bytes]] = { b"refs/heads/master": b"8843d7f92416211de9ebb963ff4ce28125932878", b"refs/heads/testing": b"186a005b134d8639a58b6731c7c1ea821a6eedba", b"refs/tags/1.0.0": b"a93db4b0360cc635a2b93675010bac8d101f73f0", b"refs/tags/1.0.0^{}": b"a93db4b0360cc635a2b93675010bac8d101f73f0", b"refs/tags/2.0.0": b"0749936d0956c661ac8f8d3483774509c165f89e", b"refs/tags/2.0.0^{}": b"0749936d0956c661ac8f8d3483774509c165f89e", } non_peeled_refs: ClassVar[Dict[bytes, bytes]] = { b"refs/heads/master": b"8843d7f92416211de9ebb963ff4ce28125932878", b"refs/heads/testing": b"186a005b134d8639a58b6731c7c1ea821a6eedba", b"refs/tags/1.0.0": b"a93db4b0360cc635a2b93675010bac8d101f73f0", b"refs/tags/2.0.0": b"0749936d0956c661ac8f8d3483774509c165f89e", } def test_strip_peeled_refs(self): # Simple check of two dicts self.assertEqual(strip_peeled_refs(self.all_refs), self.non_peeled_refs)