Changeset View
Changeset View
Standalone View
Standalone View
docker/tests/test_vault.py
Show All 36 Lines | def test_vault_directory(scheduler_host, git_origin): | ||||
# and check the tarfile seems ok wrt. 'directory' | # and check the tarfile seems ok wrt. 'directory' | ||||
assert tarf.getnames()[0] == dir_id | assert tarf.getnames()[0] == dir_id | ||||
tarfiles = {t.name: t for t in tarf.getmembers()} | tarfiles = {t.name: t for t in tarf.getmembers()} | ||||
for fname, fdesc in directory: | for fname, fdesc in directory: | ||||
tfinfo = tarfiles.get(join(dir_id, fname)) | tfinfo = tarfiles.get(join(dir_id, fname)) | ||||
assert tfinfo, f"Missing path {fname} in retrieved tarfile" | assert tfinfo, f"Missing path {fname} in retrieved tarfile" | ||||
if fdesc["type"] == "file": | if fdesc["type"] == "file": | ||||
if tfinfo.issym(): | |||||
# checksum was computed from targeted path for symlink | |||||
fdata = tfinfo.linkname.encode() | |||||
else: | |||||
# symlink has no size in tar archive so this test fails | |||||
assert fdesc["length"] == tfinfo.size, f"File {fname}: length mismatch" | assert fdesc["length"] == tfinfo.size, f"File {fname}: length mismatch" | ||||
fdata = tarf.extractfile(tfinfo).read() | fdata = tarf.extractfile(tfinfo).read() | ||||
for algo in fdesc["checksums"]: | for algo in fdesc["checksums"]: | ||||
vlorentz: please,
```
if C: A
else: B
```
instead of
```
if not C: B
else: A
``` | |||||
if algo not in hashlib.algorithms_available: | if algo not in hashlib.algorithms_available: | ||||
continue | continue | ||||
hash = hashlib.new(algo, fdata).hexdigest() | hash = hashlib.new(algo, fdata).hexdigest() | ||||
assert ( | assert ( | ||||
hash == fdesc["checksums"][algo] | hash == fdesc["checksums"][algo] | ||||
), f"File {fname}: {algo} mismatch" | ), f"File {fname}: {algo} mismatch" | ||||
# XXX what to check for dir? symlink? (other?) | # XXX what to check for dir? symlink? (other?) | ||||
# check that if we ask a second time this directory, it returns the same | # check that if we ask a second time this directory, it returns the same | ||||
# and does not cook it again | # and does not cook it again | ||||
recook = apiget(f"vault/directory/{dir_id}/", "POST") | recook = apiget(f"vault/directory/{dir_id}/", "POST") | ||||
assert recook["obj_type"] == "directory" | assert recook["obj_type"] == "directory" | ||||
assert recook["obj_id"] == dir_id | assert recook["obj_id"] == dir_id | ||||
assert recook["id"] == cook["id"] | assert recook["id"] == cook["id"] | ||||
assert recook["status"] == "done" # no need to wait for this to be true | assert recook["status"] == "done" # no need to wait for this to be true |
please,
instead of