|
| 1 | +from tempfile import NamedTemporaryFile, TemporaryDirectory |
| 2 | +from unittest import TestCase, main |
| 3 | +from uuid import uuid4 |
| 4 | + |
| 5 | +from pycpio import PyCPIO |
| 6 | +from zenlib.logging import loggify |
| 7 | + |
| 8 | + |
| 9 | +@loggify |
| 10 | +class TestCpio(TestCase): |
| 11 | + def setUp(self): |
| 12 | + self.cpio = PyCPIO(logger=self.logger) |
| 13 | + self.make_workdir() |
| 14 | + |
| 15 | + def tearDown(self): |
| 16 | + for file in self.test_files: |
| 17 | + file.close() |
| 18 | + for directory in self.test_dirs: |
| 19 | + directory.cleanup() |
| 20 | + self.workdir.cleanup() |
| 21 | + del self.cpio |
| 22 | + |
| 23 | + def make_workdir(self): |
| 24 | + """ |
| 25 | + Create a temporary directory for testing. |
| 26 | + sets self.workdir to the Path object of the directory |
| 27 | + initializes self.test_files as an empty list |
| 28 | + """ |
| 29 | + self.workdir = TemporaryDirectory(prefix="pycpio-test-") |
| 30 | + self.test_files = [] |
| 31 | + self.test_dirs = [] |
| 32 | + |
| 33 | + def make_test_file(self, subdir=None, data=None): |
| 34 | + """Creates a test file in the workdir""" |
| 35 | + base_dir = self.workdir.name |
| 36 | + if subdir is True: |
| 37 | + d = TemporaryDirectory(dir=base_dir) |
| 38 | + self.test_dirs.append(d) |
| 39 | + base_dir = d.name |
| 40 | + elif subdir is not None and subdir in self.test_dirs: |
| 41 | + base_dir = subdir.name |
| 42 | + |
| 43 | + file = NamedTemporaryFile(dir=base_dir) |
| 44 | + file_data = data.encode() if data is not None else bytes(str(uuid4()), "utf-8") |
| 45 | + file.file.write(file_data) |
| 46 | + file.file.flush() |
| 47 | + |
| 48 | + self.test_files.append(file) |
| 49 | + return file |
| 50 | + |
| 51 | + def make_test_files(self, count, subdir=None, data=None): |
| 52 | + """Creates count test files in the workdir""" |
| 53 | + for _ in range(count): |
| 54 | + self.make_test_file(subdir=subdir, data=data) |
| 55 | + |
| 56 | + def test_write_no_compress(self): |
| 57 | + self.make_test_files(100) |
| 58 | + self.cpio.append_cpio(self.workdir.name) |
| 59 | + out_file = NamedTemporaryFile() # Out file for the cpio |
| 60 | + self.cpio.write_cpio_file(out_file.file.name) |
| 61 | + out_file.file.flush() |
| 62 | + |
| 63 | + def test_write_xz_compress(self): |
| 64 | + self.make_test_files(100) |
| 65 | + self.cpio.append_cpio(self.workdir.name) |
| 66 | + out_file = NamedTemporaryFile() |
| 67 | + self.cpio.write_cpio_file(out_file.file.name, compress="xz") |
| 68 | + out_file.file.flush() |
| 69 | + |
| 70 | + def test_write_zstd_compress(self): |
| 71 | + self.make_test_files(100) |
| 72 | + self.cpio.append_cpio(self.workdir.name) |
| 73 | + out_file = NamedTemporaryFile() |
| 74 | + self.cpio.write_cpio_file(out_file.file.name, compress="zstd") |
| 75 | + out_file.file.flush() |
| 76 | + |
| 77 | + |
| 78 | +if __name__ == "__main__": |
| 79 | + main() |
0 commit comments