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