some bugfixes
This commit is contained in:
parent
852762ddfc
commit
4870acbb29
|
@ -78,7 +78,8 @@ class BunkeredFile(io.RawIOBase):
|
||||||
"""
|
"""
|
||||||
size = None
|
size = None
|
||||||
if(file_.seekable()):
|
if(file_.seekable()):
|
||||||
size = file_.seek(0, 2)
|
file_.seek(0, 2)
|
||||||
|
size = file_.tell()
|
||||||
file_.seek(0, 0)
|
file_.seek(0, 0)
|
||||||
return cls(file_, name, size=size, isvirtual=False, ismem=False)
|
return cls(file_, name, size=size, isvirtual=False, ismem=False)
|
||||||
|
|
||||||
|
@ -96,9 +97,10 @@ class BunkeredFile(io.RawIOBase):
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_BytesIO(cls, bytes_, name):
|
def from_BytesIO(cls, bytes_, name):
|
||||||
size = None
|
size = None
|
||||||
if(file_.seekable()):
|
if(bytes_.seekable()):
|
||||||
size = file_.seek(0, 2)
|
bytes_.seek(0, 2)
|
||||||
file_.seek(0, 0)
|
size = bytes_.tell()
|
||||||
|
bytes_.seek(0, 0)
|
||||||
return cls(bytes_, name, size=size, isvirtual=True, ismem=True)
|
return cls(bytes_, name, size=size, isvirtual=True, ismem=True)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
|
|
|
@ -73,7 +73,7 @@ class RewriteableTarFile(object):
|
||||||
for member in handle.getmembers():
|
for member in handle.getmembers():
|
||||||
if(member.name == membername):
|
if(member.name == membername):
|
||||||
continue
|
continue
|
||||||
new_handle.add_file(member, handle.extractfile(member))
|
new_handle.addfile(member, handle.extractfile(member))
|
||||||
new_handle.close()
|
new_handle.close()
|
||||||
|
|
||||||
def close(self):
|
def close(self):
|
||||||
|
|
|
@ -1,5 +1,6 @@
|
||||||
import os
|
import os
|
||||||
import tarfile
|
import tarfile
|
||||||
|
from io import BytesIO
|
||||||
from bunker.files.bunkerfile import BunkeredFile
|
from bunker.files.bunkerfile import BunkeredFile
|
||||||
|
|
||||||
def test_load_from_tar(tmpdir):
|
def test_load_from_tar(tmpdir):
|
||||||
|
@ -27,3 +28,12 @@ def test_load_from_tar(tmpdir):
|
||||||
|
|
||||||
assert a.read() == b"abcdefg"
|
assert a.read() == b"abcdefg"
|
||||||
assert b.read() == b"foobar"
|
assert b.read() == b"foobar"
|
||||||
|
|
||||||
|
def test_from_BytesIO():
|
||||||
|
b = BytesIO(b"foobar")
|
||||||
|
f = BunkeredFile.from_BytesIO(b, "test")
|
||||||
|
|
||||||
|
assert len(f) == len(b"foobar")
|
||||||
|
assert f.read() == b"foobar"
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -1,5 +1,6 @@
|
||||||
import os
|
import os
|
||||||
import tarfile
|
import tarfile
|
||||||
|
import pytest
|
||||||
|
|
||||||
from bunker.files.tarfile import RewriteableTarFile
|
from bunker.files.tarfile import RewriteableTarFile
|
||||||
from bunker.files.bunkerfile import BunkeredFile
|
from bunker.files.bunkerfile import BunkeredFile
|
||||||
|
@ -21,3 +22,75 @@ def test_create(tmpdir):
|
||||||
f = tf.get_file("__bunker_main__")
|
f = tf.get_file("__bunker_main__")
|
||||||
|
|
||||||
assert f.read() == b"foobar"
|
assert f.read() == b"foobar"
|
||||||
|
|
||||||
|
def test_rewrite(tmpdir):
|
||||||
|
tmpdname = str(tmpdir)
|
||||||
|
|
||||||
|
tf = RewriteableTarFile.open(os.path.join(tmpdname, "test.bunker"))
|
||||||
|
f = BunkeredFile.empty("__bunker_main__")
|
||||||
|
tf.add_file(f)
|
||||||
|
tf.add_file(BunkeredFile.empty("test.tx"))
|
||||||
|
|
||||||
|
f = tf.get_file("__bunker_main__")
|
||||||
|
f.write(b"foobar")
|
||||||
|
|
||||||
|
tf.close()
|
||||||
|
|
||||||
|
assert f.closed == True
|
||||||
|
|
||||||
|
tf = RewriteableTarFile.open(os.path.join(tmpdname, "test.bunker"))
|
||||||
|
f = tf.get_file("__bunker_main__")
|
||||||
|
f.write(b"foobaz")
|
||||||
|
tf.close()
|
||||||
|
|
||||||
|
tf = RewriteableTarFile.open(os.path.join(tmpdname, "test.bunker"))
|
||||||
|
f = tf.get_file("__bunker_main__")
|
||||||
|
assert f.read() == b"foobaz"
|
||||||
|
|
||||||
|
def test_writeback(tmpdir):
|
||||||
|
tmpdname = str(tmpdir)
|
||||||
|
|
||||||
|
tf = RewriteableTarFile.open(os.path.join(tmpdname, "test.bunker"))
|
||||||
|
f = BunkeredFile.empty("__bunker_main__")
|
||||||
|
tf.add_file(f)
|
||||||
|
tf.add_file(BunkeredFile.empty("test.tx"))
|
||||||
|
|
||||||
|
f = tf.get_file("__bunker_main__")
|
||||||
|
f.write(b"foobar")
|
||||||
|
|
||||||
|
tf.close()
|
||||||
|
|
||||||
|
tf = RewriteableTarFile.open(os.path.join(tmpdname, "test.bunker"))
|
||||||
|
f = tf.get_file("__bunker_main__")
|
||||||
|
f.write(b"foobaz")
|
||||||
|
f.close()
|
||||||
|
tf.close()
|
||||||
|
|
||||||
|
tf = RewriteableTarFile.open(os.path.join(tmpdname, "test.bunker"))
|
||||||
|
f = tf.get_file("__bunker_main__")
|
||||||
|
assert f.read() == b"foobaz"
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
def test_delete(tmpdir):
|
||||||
|
tmpdname = str(tmpdir)
|
||||||
|
|
||||||
|
tf = RewriteableTarFile.open(os.path.join(tmpdname, "test.bunker"))
|
||||||
|
f = BunkeredFile.empty("__bunker_main__")
|
||||||
|
tf.add_file(f)
|
||||||
|
tf.add_file(BunkeredFile.empty("test.tx"))
|
||||||
|
|
||||||
|
f = tf.get_file("__bunker_main__")
|
||||||
|
f.write(b"foobar")
|
||||||
|
|
||||||
|
tf.close()
|
||||||
|
|
||||||
|
tf = RewriteableTarFile.open(os.path.join(tmpdname, "test.bunker"))
|
||||||
|
tf.delete_file("__bunker_main__")
|
||||||
|
tf.close()
|
||||||
|
|
||||||
|
tf = RewriteableTarFile.open(os.path.join(tmpdname, "test.bunker"))
|
||||||
|
|
||||||
|
with pytest.raises(KeyError):
|
||||||
|
f = tf.get_file("__bunker_main__")
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue
Block a user