Skip to content

Commit

Permalink
fixup
Browse files Browse the repository at this point in the history
  • Loading branch information
albu-diku committed Jul 11, 2024
1 parent 1b3b154 commit 0a8f411
Show file tree
Hide file tree
Showing 2 changed files with 19 additions and 31 deletions.
15 changes: 0 additions & 15 deletions mig/shared/fileio.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,21 +74,6 @@
exit(1)


def _auto_adjust_mode(data, mode):
"""Select suitable file open mode based on string type of data. I.e. whether
to use binary or text mode depending on data in bytes or unicode format.
"""

# NOTE: detect byte/unicode writes and handle explicitly in a portable way
if isinstance(data, bytes):
if 'b' not in mode:
mode = "%sb" % mode # appended to avoid mode ordering error on PY2
else:
if 'b' in mode:
mode = mode.strip('b')
return mode


def _is_string(value):
return isinstance(value, unicode) if PY2 else isinstance(value, str)

Expand Down
35 changes: 19 additions & 16 deletions tests/test_mig_shared_fileio.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,12 +52,16 @@
assert isinstance(DUMMY_BYTES, bytes)


def as_unicode_string(value):
def _as_unicode_string(value):
assert isinstance(value, bytearray)
return unicode(codecs.decode(value, 'utf8')) if PY2 else str(value, 'utf8')


class TextFile:
class _ByteText:
"""File-like object that allows interacting with a text file as a bytearray.
Supports reading and directly usable as a context manager."""

def __init__(self, path, mode='r'):
self._file = None
self._path = path
Expand All @@ -79,7 +83,8 @@ def __exit__(self, *args):


class MigSharedFileio__write_chunk(MigTestCase):
# TODO: Add docstrings to this class and its methods
"""Coverage of the write_chunk() function."""

def setUp(self):
super(MigSharedFileio__write_chunk, self).setUp()
self.tmp_path = temppath(DUMMY_FILE_WRITECHUNK, self, skip_clean=True)
Expand Down Expand Up @@ -126,7 +131,6 @@ def test_store_bytes_at_offset(self):
"expected a hole was left")
self.assertEqual(content[3:], DUMMY_BYTES)

@unittest.skip("TODO: enable again - requires the temporarily disabled auto mode select")
def test_store_bytes_in_text_mode(self):
fileio.write_chunk(self.tmp_path, DUMMY_BYTES, 0, self.logger,
mode="r+")
Expand All @@ -141,7 +145,7 @@ def test_store_unicode(self):
mode='r+')
self.assertTrue(did_succeed)

with TextFile(self.tmp_path) as file:
with _ByteText(self.tmp_path) as file:
content = file.read(1024)
self.assertEqual(len(content), DUMMY_UNICODE_BYTES_LENGTH)
self.assertEqual(content[:], DUMMY_UNICODE_BYTES)
Expand All @@ -150,13 +154,15 @@ def test_store_unicode_in_binary_mode(self):
fileio.write_chunk(self.tmp_path, DUMMY_UNICODE, 0, self.logger,
mode='r+b')

with TextFile(self.tmp_path) as file:
with _ByteText(self.tmp_path) as file:
content = file.read(1024)
self.assertEqual(len(content), DUMMY_UNICODE_BYTES_LENGTH)
self.assertEqual(as_unicode_string(content[:]), DUMMY_UNICODE)
self.assertEqual(_as_unicode_string(content[:]), DUMMY_UNICODE)


class MigSharedFileio__write_file(MigTestCase):
"""Coverage of the write_file() function."""

def setUp(self):
super(MigSharedFileio__write_file, self).setUp()
self.tmp_path = temppath(DUMMY_FILE_WRITEFILE, self, skip_clean=True)
Expand Down Expand Up @@ -196,7 +202,6 @@ def test_store_bytes(self):
self.assertEqual(len(content), DUMMY_BYTES_LENGTH)
self.assertEqual(content[:], DUMMY_BYTES)

@unittest.skip("TODO: enable again - requires the temporarily disabled auto mode select")
def test_store_bytes_in_text_mode(self):
did_succeed = fileio.write_file(DUMMY_BYTES, self.tmp_path, self.logger,
mode="w")
Expand All @@ -207,27 +212,25 @@ def test_store_bytes_in_text_mode(self):
self.assertEqual(len(content), DUMMY_BYTES_LENGTH)
self.assertEqual(content[:], DUMMY_BYTES)

@unittest.skip("TODO: enable again - requires the temporarily disabled auto mode select")
def test_store_unicode(self):
did_succeed = fileio.write_file(DUMMY_UNICODE, self.tmp_path,
self.logger, mode='w')
self.assertTrue(did_succeed)

with open(self.tmp_path, 'r') as file:
with _ByteText(self.tmp_path, 'r') as file:
content = file.read(1024)
self.assertEqual(len(content), DUMMY_UNICODE_LENGTH)
self.assertEqual(content[:], DUMMY_UNICODE)
self.assertEqual(len(content), DUMMY_UNICODE_BYTES_LENGTH)
self.assertEqual(content[:], DUMMY_UNICODE_BYTES)

@unittest.skip("TODO: enable again - requires the temporarily disabled auto mode select")
def test_store_unicode_in_binary_mode(self):
did_succeed = fileio.write_file(DUMMY_UNICODE, self.tmp_path,
self.logger, mode='wb')
self.assertTrue(did_succeed)

with open(self.tmp_path, 'r') as file:
with _ByteText(self.tmp_path, 'r') as file:
content = file.read(1024)
self.assertEqual(len(content), DUMMY_UNICODE_LENGTH)
self.assertEqual(content[:], DUMMY_UNICODE)
self.assertEqual(len(content), DUMMY_UNICODE_BYTES_LENGTH)
self.assertEqual(content[:], DUMMY_UNICODE_BYTES)


if __name__ == '__main__':
Expand Down

0 comments on commit 0a8f411

Please sign in to comment.