37
37
metalink_checksum_directory,
39
39
from cdimage.config import Config
40
from cdimage.tests.helpers import TestCase, touch
40
from cdimage.tests.helpers import TestCase, mkfile, touch
43
43
class TestApplySed(TestCase):
73
73
def test_checksum_small_file(self):
74
74
entry_path = os.path.join(self.temp_dir, "entry")
76
with open(entry_path, "wb") as entry:
76
with mkfile(entry_path, mode="wb") as entry:
78
78
checksum_file = ChecksumFile(
79
79
self.config, self.temp_dir, "MD5SUMS", hashlib.md5)
83
83
def test_checksum_large_file(self):
84
84
entry_path = os.path.join(self.temp_dir, "entry")
85
85
data = b"a" * 1048576
86
with open(entry_path, "wb") as entry:
86
with mkfile(entry_path, mode="wb") as entry:
88
88
checksum_file = ChecksumFile(
89
89
self.config, self.temp_dir, "SHA1SUMS", hashlib.sha1)
93
93
def test_add(self):
94
94
entry_path = os.path.join(self.temp_dir, "entry")
96
with open(entry_path, "wb") as entry:
96
with mkfile(entry_path, mode="wb") as entry:
98
98
checksum_file = ChecksumFile(
99
99
self.config, self.temp_dir, "MD5SUMS", hashlib.md5)
107
107
# an existing checksum.)
108
108
entry_path = os.path.join(self.temp_dir, "entry")
110
with open(entry_path, "w") as entry:
110
with mkfile(entry_path) as entry:
111
111
entry.write(data)
112
112
checksum_file = ChecksumFile(
113
113
self.config, self.temp_dir, "MD5SUMS", hashlib.md5)
126
126
# Adding an existing file with an mtime newer than that of the
127
127
# checksums file causes its checksum to be updated.
128
128
path = os.path.join(self.temp_dir, "entry")
129
with open(path, "w") as entry:
129
with mkfile(path) as entry:
131
131
checksum_file = ChecksumFile(
132
self.config, self.temp_dir, "MD5SUMS", hashlib.md5)
132
self.config, self.temp_dir, "MD5SUMS", hashlib.md5, sign=False)
133
133
checksum_file.add("entry")
134
134
checksum_file.write()
135
135
self.rewind_mtime(checksum_file.path)
136
with open(path, "w") as entry:
136
with mkfile(path) as entry:
137
137
print("mtime", end="", file=entry)
138
138
checksum_file.add("entry")
139
139
self.assertEqual(
143
143
# Adding an existing file with a ctime newer than that of the
144
144
# checksums file causes its checksum to be updated.
145
145
path = os.path.join(self.temp_dir, "entry")
146
with open(path, "w") as entry:
146
with mkfile(path) as entry:
147
147
print("ctime", end="", file=entry)
148
148
checksum_file = ChecksumFile(
149
self.config, self.temp_dir, "MD5SUMS", hashlib.md5)
149
self.config, self.temp_dir, "MD5SUMS", hashlib.md5, sign=False)
150
150
checksum_file.entries["entry"] = ""
151
checksum_file.changed = True
151
152
checksum_file.write()
152
153
# We can simulate a ctime change by rewinding the mtime of both
153
154
# entry and the checksums file.
167
168
def test_merge_takes_valid_checksums(self):
168
169
old_dir = os.path.join(self.temp_dir, "old")
170
170
touch(os.path.join(self.temp_dir, "entry"))
171
with open(os.path.join(old_dir, "MD5SUMS"), "w") as old_md5sums:
171
with mkfile(os.path.join(old_dir, "MD5SUMS")) as old_md5sums:
172
172
print("checksum *entry", file=old_md5sums)
173
173
checksum_file = ChecksumFile(
174
174
self.config, self.temp_dir, "MD5SUMS", hashlib.md5)
178
178
def test_merge_ignores_stale_checksums(self):
179
179
old_dir = os.path.join(self.temp_dir, "old")
181
with open(os.path.join(old_dir, "MD5SUMS"), "w") as old_md5sums:
180
with mkfile(os.path.join(old_dir, "MD5SUMS")) as old_md5sums:
182
181
print("checksum *entry", file=old_md5sums)
183
182
entry_path = os.path.join(self.temp_dir, "entry")
184
183
touch(entry_path)
192
191
def test_merge_takes_other_names(self):
193
192
old_dir = os.path.join(self.temp_dir, "old")
195
193
touch(os.path.join(self.temp_dir, "entry"))
196
with open(os.path.join(old_dir, "MD5SUMS"), "w") as old_md5sums:
194
with mkfile(os.path.join(old_dir, "MD5SUMS")) as old_md5sums:
197
195
print("checksum *other-entry", file=old_md5sums)
198
196
checksum_file = ChecksumFile(
199
197
self.config, self.temp_dir, "MD5SUMS", hashlib.md5)
200
198
checksum_file.merge([old_dir], "entry", ["other-entry"])
201
199
self.assertEqual({"entry": "checksum"}, checksum_file.entries)
201
def test_merge_handles_symlinks(self):
202
old_dir = os.path.join(self.temp_dir, "old")
203
touch(os.path.join(old_dir, "entry"))
204
with mkfile(os.path.join(old_dir, "MD5SUMS")) as old_md5sums:
205
print("correct-checksum *entry", file=old_md5sums)
206
new_dir = os.path.join(self.temp_dir, "new")
209
os.path.join(os.pardir, "old", "entry"),
210
os.path.join(new_dir, "entry"))
211
with mkfile(os.path.join(new_dir, "MD5SUMS")) as new_md5sums:
212
print("wrong-checksum *entry", file=new_md5sums)
213
checksum_file = ChecksumFile(
214
self.config, new_dir, "MD5SUMS", hashlib.md5)
215
checksum_file.merge([new_dir, old_dir], "entry", ["entry"])
216
self.assertEqual({"entry": "correct-checksum"}, checksum_file.entries)
203
218
def test_write(self):
204
219
checksum_file = ChecksumFile(
205
220
self.config, self.temp_dir, "MD5SUMS", hashlib.md5, sign=False)
206
221
for name in "1", "2":
207
222
entry_path = os.path.join(self.temp_dir, name)
208
with open(entry_path, "w") as entry:
223
with mkfile(entry_path) as entry:
209
224
print(name, end="", file=entry)
210
225
checksum_file.add(name)
211
226
checksum_file.write()
224
239
def test_context_manager(self):
225
240
for name in "1", "2":
226
241
entry_path = os.path.join(self.temp_dir, name)
227
with open(entry_path, "w") as entry:
242
with mkfile(entry_path) as entry:
228
243
print(name, end="", file=entry)
229
244
md5sums_path = os.path.join(self.temp_dir, "MD5SUMS")
230
with open(md5sums_path, "w") as md5sums:
245
with mkfile(md5sums_path) as md5sums:
232
247
["md5sum", "-b", "1", "2"], stdout=md5sums, cwd=self.temp_dir)
233
248
with ChecksumFile(
234
249
self.config, self.temp_dir, "MD5SUMS", hashlib.md5,
235
250
sign=False) as checksum_file:
236
self.assertEqual(["1", "2"], sorted(checksum_file.entries))
251
self.assertCountEqual(["1", "2"], checksum_file.entries)
237
252
checksum_file.remove("1")
238
253
with open(md5sums_path) as md5sums:
239
254
self.assertEqual(
256
271
if directory is None:
257
272
directory = self.temp_dir
258
273
for base, command in self.files_and_commands.items():
259
with open(os.path.join(directory, base), "w") as f:
274
with mkfile(os.path.join(directory, base)) as f:
261
276
[command, "-b"] + names, stdout=f, cwd=directory)
279
294
def test_read(self):
280
295
entry_path = os.path.join(self.temp_dir, "entry")
281
with open(entry_path, "w") as entry:
296
with mkfile(entry_path) as entry:
282
297
print("data", end="", file=entry)
283
298
self.create_checksum_files(["entry"])
284
299
for base, command in self.files_and_commands.items():
285
with open(os.path.join(self.temp_dir, base), "w") as f:
300
with mkfile(os.path.join(self.temp_dir, base)) as f:
287
302
[command, "-b", "entry"], stdout=f, cwd=self.temp_dir)
288
303
checksum_files = self.cls(self.config, self.temp_dir)
292
307
def test_add(self):
293
308
entry_path = os.path.join(self.temp_dir, "entry")
295
with open(entry_path, "w") as entry:
310
with mkfile(entry_path) as entry:
296
311
print(data, end="", file=entry)
297
312
checksum_files = self.cls(self.config, self.temp_dir)
298
313
checksum_files.add("entry")
301
316
def test_remove(self):
302
317
entry_path = os.path.join(self.temp_dir, "entry")
304
with open(entry_path, "w") as entry:
319
with mkfile(entry_path) as entry:
305
320
print(data, end="", file=entry)
306
321
self.create_checksum_files(["entry"])
307
322
checksum_files = self.cls(self.config, self.temp_dir)
313
328
old_dir = os.path.join(self.temp_dir, "old")
314
329
os.mkdir(old_dir)
315
330
entry_path = os.path.join(self.temp_dir, "entry")
316
with open(entry_path, "w") as entry:
331
with mkfile(entry_path) as entry:
317
332
print("data", end="", file=entry)
318
333
shutil.copy(entry_path, os.path.join(old_dir, "entry"))
319
334
self.create_checksum_files(["entry"], directory=old_dir)
330
345
def test_merge_all(self):
331
346
old_dir = os.path.join(self.temp_dir, "old")
333
old_iso_hppa_path = os.path.join(self.temp_dir, "old", "foo-hppa.raw")
334
with open(old_iso_hppa_path, "w") as old_iso_hppa:
347
old_iso_hppa_path = os.path.join(old_dir, "foo-hppa.raw")
348
with mkfile(old_iso_hppa_path) as old_iso_hppa:
335
349
print("foo-hppa.raw", end="", file=old_iso_hppa)
336
old_iso_i386_path = os.path.join(self.temp_dir, "old", "foo-i386.raw")
337
with open(old_iso_i386_path, "w") as old_iso_i386:
350
old_iso_i386_path = os.path.join(old_dir, "foo-i386.raw")
351
with mkfile(old_iso_i386_path) as old_iso_i386:
338
352
print("foo-i386.raw", end="", file=old_iso_i386)
339
353
self.create_checksum_files(
340
354
["foo-hppa.raw", "foo-i386.raw"], directory=old_dir)
341
355
iso_amd64_path = os.path.join(self.temp_dir, "foo-amd64.iso")
342
with open(iso_amd64_path, "w") as iso_amd64:
356
with mkfile(iso_amd64_path) as iso_amd64:
343
357
print("foo-amd64.iso", end="", file=iso_amd64)
344
358
touch(os.path.join(self.temp_dir, "foo-amd64.list"))
359
373
self.config, self.temp_dir, sign=False)
360
374
for name in "1", "2":
361
375
entry_path = os.path.join(self.temp_dir, name)
362
with open(entry_path, "w") as entry:
376
with mkfile(entry_path) as entry:
363
377
print(name, end="", file=entry)
364
378
checksum_files.add(name)
365
379
checksum_files.write()
373
387
def test_context_manager(self):
374
388
for name in "1", "2":
375
389
entry_path = os.path.join(self.temp_dir, name)
376
with open(entry_path, "w") as entry:
390
with mkfile(entry_path) as entry:
377
391
print(name, end="", file=entry)
378
392
self.create_checksum_files(["1", "2"])
394
408
def test_checksum_directory(self):
395
409
old_dir = os.path.join(self.temp_dir, "old")
397
old_iso_hppa_path = os.path.join(self.temp_dir, "old", "foo-hppa.raw")
398
with open(old_iso_hppa_path, "w") as old_iso_hppa:
410
old_iso_hppa_path = os.path.join(old_dir, "foo-hppa.raw")
411
with mkfile(old_iso_hppa_path) as old_iso_hppa:
399
412
print("foo-hppa.raw", end="", file=old_iso_hppa)
400
old_iso_i386_path = os.path.join(self.temp_dir, "old", "foo-i386.raw")
401
with open(old_iso_i386_path, "w") as old_iso_i386:
413
old_iso_i386_path = os.path.join(old_dir, "foo-i386.raw")
414
with mkfile(old_iso_i386_path) as old_iso_i386:
402
415
print("foo-i386.raw", end="", file=old_iso_i386)
403
416
self.create_checksum_files(
404
417
["foo-hppa.raw", "foo-i386.raw"], directory=old_dir)
405
418
iso_amd64_path = os.path.join(self.temp_dir, "foo-amd64.iso")
406
with open(iso_amd64_path, "w") as iso_amd64:
419
with mkfile(iso_amd64_path) as iso_amd64:
407
420
print("foo-amd64.iso", end="", file=iso_amd64)
408
421
touch(os.path.join(self.temp_dir, "foo-amd64.list"))
412
425
old_iso_i386_path, os.path.join(self.temp_dir, "foo-i386.iso"))
413
426
checksum_directory(
414
self.config, self.temp_dir, old_directories=[old_dir],
427
self.config, self.temp_dir, old_directories=[old_dir], sign=False,
415
428
map_expr=r"s/\.iso$/.raw/")
416
429
with open(os.path.join(self.temp_dir, "MD5SUMS")) as md5sums:
452
465
def test_merge_all(self):
453
466
old_dir = os.path.join(self.temp_dir, "old")
455
old_iso_i386_path = os.path.join(self.temp_dir, "old", "foo-i386.iso")
456
with open(old_iso_i386_path, "w") as old_iso_i386:
467
old_iso_i386_path = os.path.join(old_dir, "foo-i386.iso")
468
with mkfile(old_iso_i386_path) as old_iso_i386:
457
469
print("foo-i386.iso", end="", file=old_iso_i386)
458
old_metalink_i386_path = os.path.join(
459
self.temp_dir, "old", "foo-i386.metalink")
460
with open(old_metalink_i386_path, "w") as old_metalink_i386:
470
old_metalink_i386_path = os.path.join(old_dir, "foo-i386.metalink")
471
with mkfile(old_metalink_i386_path) as old_metalink_i386:
461
472
print("foo-i386.metalink", end="", file=old_metalink_i386)
462
473
self.create_checksum_files(
463
474
["foo-i386.metalink"], directory=old_dir)
464
475
metalink_amd64_path = os.path.join(self.temp_dir, "foo-amd64.metalink")
465
with open(metalink_amd64_path, "w") as metalink_amd64:
476
with mkfile(metalink_amd64_path) as metalink_amd64:
466
477
print("foo-amd64.metalink", end="", file=metalink_amd64)
467
478
touch(os.path.join(self.temp_dir, "foo-amd64.list"))
478
489
def test_context_manager(self):
479
490
for name in "1", "2":
480
491
entry_path = os.path.join(self.temp_dir, name)
481
with open(entry_path, "w") as entry:
492
with mkfile(entry_path) as entry:
482
493
print(name, end="", file=entry)
483
494
self.create_checksum_files(["1", "2"])
492
503
def test_checksum_directory(self):
493
504
old_dir = os.path.join(self.temp_dir, "old")
495
old_iso_i386_path = os.path.join(self.temp_dir, "old", "foo-i386.iso")
496
with open(old_iso_i386_path, "w") as old_iso_i386:
505
old_iso_i386_path = os.path.join(old_dir, "foo-i386.iso")
506
with mkfile(old_iso_i386_path) as old_iso_i386:
497
507
print("foo-i386.iso", end="", file=old_iso_i386)
498
old_metalink_i386_path = os.path.join(
499
self.temp_dir, "old", "foo-i386.metalink")
500
with open(old_metalink_i386_path, "w") as old_metalink_i386:
508
old_metalink_i386_path = os.path.join(old_dir, "foo-i386.metalink")
509
with mkfile(old_metalink_i386_path) as old_metalink_i386:
501
510
print("foo-i386.metalink", end="", file=old_metalink_i386)
502
511
self.create_checksum_files(
503
512
["foo-i386.metalink"], directory=old_dir)
504
513
metalink_amd64_path = os.path.join(self.temp_dir, "foo-amd64.metalink")
505
with open(metalink_amd64_path, "w") as metalink_amd64:
514
with mkfile(metalink_amd64_path) as metalink_amd64:
506
515
print("foo-amd64.metalink", end="", file=metalink_amd64)
507
516
touch(os.path.join(self.temp_dir, "foo-amd64.list"))
509
518
old_metalink_i386_path,
510
519
os.path.join(self.temp_dir, "foo-i386.metalink"))
511
520
metalink_checksum_directory(
512
self.config, self.temp_dir, old_directories=[old_dir])
521
self.config, self.temp_dir, old_directories=[old_dir], sign=False)
513
522
with open(os.path.join(self.temp_dir, "MD5SUMS-metalink")) as md5sums:
515
524
hashlib.md5(b"foo-amd64.metalink").hexdigest(),