Changeset View
Changeset View
Standalone View
Standalone View
swh/vault/tests/test_cookers.py
Show First 20 Lines • Show All 145 Lines • ▼ Show 20 Lines | def test_directory_simple(self): | ||||
self.assertTrue((p / 'link').is_symlink) | self.assertTrue((p / 'link').is_symlink) | ||||
self.assertEqual(os.readlink(str(p / 'link')), 'file') | self.assertEqual(os.readlink(str(p / 'link')), 'file') | ||||
self.assertEqual((p / 'dir1/dir2/file').stat().st_mode, 0o100644) | self.assertEqual((p / 'dir1/dir2/file').stat().st_mode, 0o100644) | ||||
self.assertEqual((p / 'dir1/dir2/file').read_text(), TEST_CONTENT) | self.assertEqual((p / 'dir1/dir2/file').read_text(), TEST_CONTENT) | ||||
directory = Directory.from_disk(path=bytes(p)) | directory = Directory.from_disk(path=bytes(p)) | ||||
self.assertEqual(obj_id_hex, hashutil.hash_to_hex(directory.hash)) | self.assertEqual(obj_id_hex, hashutil.hash_to_hex(directory.hash)) | ||||
def test_filtered_objects(self): | def test_directory_filtered_objects(self): | ||||
repo = TestRepo() | repo = TestRepo() | ||||
with repo as rp: | with repo as rp: | ||||
file_1, id_1 = hash_content(b'test1') | file_1, id_1 = hash_content(b'test1') | ||||
file_2, id_2 = hash_content(b'test2') | file_2, id_2 = hash_content(b'test2') | ||||
file_3, id_3 = hash_content(b'test3') | file_3, id_3 = hash_content(b'test3') | ||||
(rp / 'file').write_bytes(file_1) | (rp / 'file').write_bytes(file_1) | ||||
(rp / 'hidden_file').write_bytes(file_2) | (rp / 'hidden_file').write_bytes(file_2) | ||||
Show All 16 Lines | def test_directory_filtered_objects(self): | ||||
where sha1 = %s""", (id_3,)) | where sha1 = %s""", (id_3,)) | ||||
cur.close() | cur.close() | ||||
with self.cook_extract_directory(obj_id) as p: | with self.cook_extract_directory(obj_id) as p: | ||||
self.assertEqual((p / 'file').read_bytes(), b'test1') | self.assertEqual((p / 'file').read_bytes(), b'test1') | ||||
self.assertEqual((p / 'hidden_file').read_bytes(), HIDDEN_MESSAGE) | self.assertEqual((p / 'hidden_file').read_bytes(), HIDDEN_MESSAGE) | ||||
self.assertEqual((p / 'absent_file').read_bytes(), SKIPPED_MESSAGE) | self.assertEqual((p / 'absent_file').read_bytes(), SKIPPED_MESSAGE) | ||||
def test_directory_bogus_perms(self): | |||||
# Some early git repositories have 664/775 permissions... let's check | |||||
# if all the weird modes are properly normalized in the directory | |||||
# cooker. | |||||
repo = TestRepo() | |||||
with repo as rp: | |||||
(rp / 'file').write_text(TEST_CONTENT) | |||||
(rp / 'file').chmod(0o664) | |||||
(rp / 'executable').write_bytes(TEST_EXECUTABLE) | |||||
(rp / 'executable').chmod(0o775) | |||||
(rp / 'wat').write_text(TEST_CONTENT) | |||||
(rp / 'wat').chmod(0o604) | |||||
c = repo.commit() | |||||
self.load(str(rp)) | |||||
obj_id_hex = repo.repo[c].tree.decode() | |||||
obj_id = hashutil.hash_to_bytes(obj_id_hex) | |||||
with self.cook_extract_directory(obj_id) as p: | |||||
self.assertEqual((p / 'file').stat().st_mode, 0o100644) | |||||
self.assertEqual((p / 'executable').stat().st_mode, 0o100755) | |||||
self.assertEqual((p / 'wat').stat().st_mode, 0o100644) | |||||
class TestRevisionGitfastCooker(BaseTestCookers, unittest.TestCase): | class TestRevisionGitfastCooker(BaseTestCookers, unittest.TestCase): | ||||
def test_revision_simple(self): | def test_revision_simple(self): | ||||
# | # | ||||
# 1--2--3--4--5--6--7 | # 1--2--3--4--5--6--7 | ||||
# | # | ||||
repo = TestRepo() | repo = TestRepo() | ||||
with repo as rp: | with repo as rp: | ||||
▲ Show 20 Lines • Show All 107 Lines • ▼ Show 20 Lines | def test_revision_triple_merge(self): | ||||
obj_id_hex = repo.repo.refs[b'HEAD'].decode() | obj_id_hex = repo.repo.refs[b'HEAD'].decode() | ||||
obj_id = hashutil.hash_to_bytes(obj_id_hex) | obj_id = hashutil.hash_to_bytes(obj_id_hex) | ||||
self.load(str(rp)) | self.load(str(rp)) | ||||
with self.cook_extract_revision_gitfast(obj_id) as (ert, p): | with self.cook_extract_revision_gitfast(obj_id) as (ert, p): | ||||
self.assertEqual(ert.repo.refs[b'HEAD'].decode(), obj_id_hex) | self.assertEqual(ert.repo.refs[b'HEAD'].decode(), obj_id_hex) | ||||
def test_filtered_objects(self): | def test_revision_filtered_objects(self): | ||||
repo = TestRepo() | repo = TestRepo() | ||||
with repo as rp: | with repo as rp: | ||||
file_1, id_1 = hash_content(b'test1') | file_1, id_1 = hash_content(b'test1') | ||||
file_2, id_2 = hash_content(b'test2') | file_2, id_2 = hash_content(b'test2') | ||||
file_3, id_3 = hash_content(b'test3') | file_3, id_3 = hash_content(b'test3') | ||||
(rp / 'file').write_bytes(file_1) | (rp / 'file').write_bytes(file_1) | ||||
(rp / 'hidden_file').write_bytes(file_2) | (rp / 'hidden_file').write_bytes(file_2) | ||||
Show All 15 Lines | def test_revision_filtered_objects(self): | ||||
where sha1 = %s""", (id_3,)) | where sha1 = %s""", (id_3,)) | ||||
cur.close() | cur.close() | ||||
with self.cook_extract_revision_gitfast(obj_id) as (ert, p): | with self.cook_extract_revision_gitfast(obj_id) as (ert, p): | ||||
ert.checkout(b'HEAD') | ert.checkout(b'HEAD') | ||||
self.assertEqual((p / 'file').read_bytes(), b'test1') | self.assertEqual((p / 'file').read_bytes(), b'test1') | ||||
self.assertEqual((p / 'hidden_file').read_bytes(), HIDDEN_MESSAGE) | self.assertEqual((p / 'hidden_file').read_bytes(), HIDDEN_MESSAGE) | ||||
self.assertEqual((p / 'absent_file').read_bytes(), SKIPPED_MESSAGE) | self.assertEqual((p / 'absent_file').read_bytes(), SKIPPED_MESSAGE) | ||||
def test_revision_bogus_perms(self): | |||||
# Some early git repositories have 664/775 permissions... let's check | |||||
# if all the weird modes are properly normalized in the revision | |||||
# cooker. | |||||
repo = TestRepo() | |||||
with repo as rp: | |||||
(rp / 'file').write_text(TEST_CONTENT) | |||||
(rp / 'file').chmod(0o664) | |||||
(rp / 'executable').write_bytes(TEST_EXECUTABLE) | |||||
(rp / 'executable').chmod(0o775) | |||||
(rp / 'wat').write_text(TEST_CONTENT) | |||||
(rp / 'wat').chmod(0o604) | |||||
repo.commit('initial commit') | |||||
self.load(str(rp)) | |||||
obj_id_hex = repo.repo.refs[b'HEAD'].decode() | |||||
obj_id = hashutil.hash_to_bytes(obj_id_hex) | |||||
with self.cook_extract_revision_gitfast(obj_id) as (ert, p): | |||||
ert.checkout(b'HEAD') | |||||
self.assertEqual((p / 'file').stat().st_mode, 0o100644) | |||||
self.assertEqual((p / 'executable').stat().st_mode, 0o100755) | |||||
self.assertEqual((p / 'wat').stat().st_mode, 0o100644) |