53 lines
1.6 KiB
Python
53 lines
1.6 KiB
Python
|
import io
|
||
|
|
import tarfile
|
||
|
|
|
||
|
|
from jw.pkg.lib.TarIo import TarIoTarFile
|
||
|
|
|
||
|
|
def make_tar(files: dict[str, bytes]) -> bytes:
|
||
|
|
buf = io.BytesIO()
|
||
|
|
with tarfile.open(fileobj = buf, mode = 'w') as tf:
|
||
|
|
for name, content in files.items():
|
||
|
|
info = tarfile.TarInfo(name)
|
||
|
|
info.size = len(content)
|
||
|
|
tf.addfile(info, io.BytesIO(content))
|
||
|
|
return buf.getvalue()
|
||
|
|
|
||
|
|
def members(blob: bytes) -> dict[str, bytes | None]:
|
||
|
|
ret = {}
|
||
|
|
with tarfile.open(fileobj = io.BytesIO(blob)) as tf:
|
||
|
|
for info in tf.getmembers():
|
||
|
|
f = tf.extractfile(info)
|
||
|
|
ret[info.name] = f.read() if f is not None else None
|
||
|
|
return ret
|
||
|
|
|
||
|
|
# -- Bypass the constructor: _filter_tar_file() and _match() do not use
|
||
|
|
# the copy contexts, so a bare instance is sufficient
|
||
|
|
tar = object.__new__(TarIoTarFile)
|
||
|
|
|
||
|
|
files = {
|
||
|
|
'a.txt': b'AAA',
|
||
|
|
'dir/b.txt': b'BBB',
|
||
|
|
'c.txt': b'CCC',
|
||
|
|
}
|
||
|
|
blob = make_tar(files)
|
||
|
|
|
||
|
|
# -- _match is exact matching
|
||
|
|
assert tar._match('a.txt', ['a.txt', 'c.txt'])
|
||
|
|
assert not tar._match('a.txt', ['x.txt'])
|
||
|
|
assert not tar._match('a.txt', [])
|
||
|
|
|
||
|
|
# -- _filter_tar_file without a filter is a round trip
|
||
|
|
ret = tar._filter_tar_file(blob)
|
||
|
|
assert members(ret) == files
|
||
|
|
|
||
|
|
# -- With a filter, only the exact paths survive
|
||
|
|
matched: list[str] = []
|
||
|
|
ret = tar._filter_tar_file(blob, ['a.txt', 'c.txt'], matched = matched)
|
||
|
|
assert members(ret) == {'a.txt': b'AAA', 'c.txt': b'CCC'}
|
||
|
|
assert sorted(matched) == ['a.txt', 'c.txt']
|
||
|
|
|
||
|
|
# -- A filter without hits yields an empty tar
|
||
|
|
ret = tar._filter_tar_file(blob, ['nope.txt'])
|
||
|
|
assert members(ret) == {}
|
||
|
|
|
||
|
|
print('All TarIo tests passed')
|