Coverage for tests/test_compression.py: 100%

112 statements  

« prev     ^ index     » next       coverage.py v7.6.10, created at 2025-02-28 11:02 +0100

1import tempfile 

2from contextlib import chdir 

3from pathlib import Path 

4 

5from edwh_files_plugin.compression import Compression, Gzip, Pigz, Zip, Nocompression 

6 

7DATA = "x" * int(1e9) 

8 

9 

10def run_test_with_file(compressor: Compression, extension: str, decompressor: Compression = None): 

11 decompressor = decompressor or compressor 

12 assert compressor.is_available() 

13 assert decompressor.is_available() 

14 

15 with tempfile.TemporaryDirectory(prefix="pytest_file") as d: 

16 dir_path = Path(d) 

17 bigfile = dir_path / "myfile.txt" 

18 bigfile.write_text(DATA) 

19 

20 lilfile = dir_path / f"myfile.{extension}" 

21 

22 assert compressor.compress(bigfile, lilfile) 

23 

24 assert lilfile.exists() 

25 assert lilfile.stat().st_size < bigfile.stat().st_size 

26 

27 unzipped = dir_path / "unzipped.txt" 

28 assert decompressor.decompress(lilfile, unzipped) 

29 

30 assert unzipped.read_text() == DATA 

31 

32 

33def run_test_with_folder(compressor: Compression, extension: str, decompressor: Compression = None): 

34 decompressor = decompressor or compressor 

35 assert compressor.is_available() 

36 assert decompressor.is_available() 

37 

38 with tempfile.TemporaryDirectory(prefix="pytest_folder") as d: 

39 parent_d = Path(d) 

40 child_d = parent_d / "somefolder" 

41 child_d.mkdir() 

42 

43 bigfile = child_d / "raw.txt" 

44 bigfile.write_text(DATA) 

45 

46 file2 = child_d / "small.txt" 

47 file2.write_text("-") 

48 

49 lilfile = parent_d / f"compressed.{extension}" 

50 

51 assert compressor.compress(child_d, lilfile) 

52 

53 assert lilfile.exists() 

54 assert lilfile.stat().st_size < bigfile.stat().st_size 

55 

56 unzipped_d = parent_d / "somefolder2" 

57 unzipped_d.mkdir() 

58 assert decompressor.decompress(lilfile, unzipped_d) 

59 

60 unzipped = unzipped_d / "raw.txt" 

61 assert unzipped.read_text() == DATA 

62 

63 

64def test_zip(): 

65 zip_compression = Compression.for_extension("zip") 

66 assert isinstance(zip_compression, Zip) 

67 run_test_with_file(zip_compression, "zip") 

68 run_test_with_folder(zip_compression, "zip") 

69 

70 

71def test_gzip(): 

72 gzip_compression = Gzip() 

73 assert isinstance(gzip_compression, Gzip) 

74 run_test_with_file(gzip_compression, "gz") 

75 run_test_with_folder(gzip_compression, "tgz") 

76 run_test_with_folder(gzip_compression, "tar.gz") 

77 

78 

79def test_pigz(): 

80 pigz_compression = Compression.for_extension("gz") 

81 assert isinstance(pigz_compression, Pigz) # pigz > gz 

82 run_test_with_file(pigz_compression, "gz") 

83 run_test_with_folder(pigz_compression, "tgz") 

84 run_test_with_folder(pigz_compression, "tar.gz") 

85 

86 

87def test_gzip_pigz_cross(): 

88 gz_compression = Gzip() 

89 pigz_compression = Pigz() 

90 run_test_with_file(pigz_compression, "gz", decompressor=gz_compression) 

91 run_test_with_folder(pigz_compression, "tgz", decompressor=gz_compression) 

92 

93 run_test_with_file(gz_compression, "gz", decompressor=pigz_compression) 

94 run_test_with_folder(gz_compression, "tgz", decompressor=pigz_compression) 

95 

96 

97def test_noop(): 

98 class Noop(Compression, extension="noop"): ... 

99 

100 # false because it is not available 

101 assert not Compression.for_extension("noop") 

102 assert not Noop.is_available() 

103 

104 assert not Compression.for_extension("fake") 

105 

106def test_nocompression(): 

107 compressor = Nocompression() 

108 

109 # test file: 

110 assert compressor.is_available() 

111 

112 with tempfile.TemporaryDirectory(prefix="pytest_file") as d: 

113 dir_path = Path(d) 

114 bigfile = dir_path / "myfile.txt" 

115 bigfile.write_text(DATA) 

116 

117 lilfile = dir_path / f"myfile.txt" 

118 

119 assert compressor.compress(bigfile) 

120 

121 assert lilfile.exists() 

122 assert lilfile.stat().st_size == bigfile.stat().st_size 

123 

124 unzipped = dir_path / "myfile.txt" 

125 assert compressor.decompress(lilfile) 

126 

127 assert unzipped.read_text() == DATA 

128 

129 # test folder: 

130 with tempfile.TemporaryDirectory(prefix="pytest_folder") as d: 

131 parent_d = Path(d) 

132 child_d = parent_d / "somefolder" 

133 child_d.mkdir() 

134 

135 bigfile = child_d / "raw.txt" 

136 bigfile.write_text(DATA) 

137 

138 file2 = child_d / "small.txt" 

139 file2.write_text("-") 

140 

141 lilfile = parent_d / f"somefolder.tar" 

142 

143 assert compressor.compress(child_d) 

144 

145 assert lilfile.exists() 

146 

147 unzipped_d = parent_d / "somefolder" 

148 assert compressor.decompress(lilfile) 

149 

150 unzipped = unzipped_d / "raw.txt" 

151 assert unzipped.read_text() == DATA 

152 

153def test_best(): 

154 compressor = Compression.best() 

155 assert isinstance(compressor, Pigz) 

156 

157 

158def test_compress_decompress_without_filename(): 

159 c = Compression.best() 

160 

161 with tempfile.TemporaryDirectory() as d, chdir(d): 

162 p = Path(d) 

163 t = p / "file.txt" 

164 t.write_text("--------------------") 

165 

166 assert c.compress(".") 

167 assert c.compress(t) 

168 

169 assert c.decompress(p.with_suffix(".tgz")) 

170 assert c.decompress(t.with_suffix(".txt.gz"))