temporarily remove the intelligent mode select originally introduced in PR48 and expanded on in my follow-ups. It looks like my attempts to push it through the 'experimental' branch caused some serious regressions there and I/we need to look more thoroughly into a consistent solution across all branches. For now it's a small step backwards to ease fixing my recent breakage in 'experimental'. We'll enable additional unit tests again along the way.

git-svn-id: svn+ssh://svn.code.sf.net/p/migrid/code/trunk@6042 b75ad72c-e7d7-11dd-a971-7dbc132099af

Author: jonasbardino

mig/shared/fileio.py

@@ -129,7 +129,7 @@ def _write_chunk(path, chunk, offset, logger=None, mode='r+b',
                     for _ in xrange(offset - file_size):
                         filehandle.write('\0')
             # logger.debug("write %r chunk of size %d at position %d" %
-            #             (path, len(chunk), filehandle.tell()))
+            #              (path, len(chunk), filehandle.tell()))
             filehandle.write(chunk)
             # logger.debug("file %r chunk written at %d" % (path, offset))
             return True
@@ -146,7 +146,8 @@ def write_chunk(path, chunk, offset, logger, mode='r+b'):
     if not logger:
         logger = null_logger("dummy")
 
-    mode = _auto_adjust_mode(chunk, mode)
+    # TODO: enable this again once throuroughly tested and assured py2+3 safe
+    # mode = _auto_adjust_mode(chunk, mode)
 
     return _write_chunk(path, chunk, offset, logger, mode)
 
@@ -160,7 +161,8 @@ def write_file(content, path, logger, mode='w', make_parent=True, umask=None):
     if umask is not None:
         old_umask = os.umask(umask)
 
-    mode = _auto_adjust_mode(content, mode)
+    # TODO: enable this again once throuroughly tested and assured py2+3 safe
+    #mode = _auto_adjust_mode(content, mode)
 
     retval = _write_chunk(path, content, offset=0, logger=logger, mode=mode,
                           make_parent=make_parent, create_file=False)

tests/test_mig_shared_fileio.py

@@ -37,14 +37,14 @@
 try:
     import mig.shared.fileio as fileio
 except ImportError as ioe:
-    print("Failed to import mig core modules: %s" % ioe.message)
+    print("Failed to import mig core modules: %s" % ioe)
     exit(1)
 
 # NOTE: prevent autopep8 shuffling next imports up
 try:
     from support import MigTestCase, cleanpath, temppath, testmain
 except ImportError as ioe:
-    print("Failed to import mig test modules: %s" % ioe.message)
+    print("Failed to import mig test modules: %s" % ioe)
     exit(1)
 
 
@@ -106,32 +106,35 @@ def test_store_bytes_at_offset(self):
                              "expected a hole was left")
             self.assertEqual(content[3:], DUMMY_BYTES)
 
-    def test_store_bytes_in_text_mode(self):
-        fileio.write_chunk(self.tmp_path, DUMMY_BYTES, 0, self.logger,
-                           mode="r+")
-
-        with open(self.tmp_path, 'rb') as file:
-            content = file.read(1024)
-            self.assertEqual(len(content), DUMMY_BYTES_LENGTH)
-            self.assertEqual(content[:], DUMMY_BYTES)
-
-    def test_store_unicode(self):
-        fileio.write_chunk(self.tmp_path, DUMMY_UNICODE, 0, self.logger,
-                           mode='r+')
-
-        with open(self.tmp_path, 'r') as file:
-            content = file.read(1024)
-            self.assertEqual(len(content), DUMMY_UNICODE_LENGTH)
-            self.assertEqual(content[:], DUMMY_UNICODE)
-
-    def test_store_unicode_in_binary_mode(self):
-        fileio.write_chunk(self.tmp_path, DUMMY_UNICODE, 0, self.logger,
-                           mode='r+b')
-
-        with open(self.tmp_path, 'r') as file:
-            content = file.read(1024)
-            self.assertEqual(len(content), DUMMY_UNICODE_LENGTH)
-            self.assertEqual(content[:], DUMMY_UNICODE)
+    # 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+")
+    #
+    #    with open(self.tmp_path, 'rb') as file:
+    #        content = file.read(1024)
+    #        self.assertEqual(len(content), DUMMY_BYTES_LENGTH)
+    #        self.assertEqual(content[:], DUMMY_BYTES)
+
+    # TODO: enable again - requires the temporarily disabled auto mode select
+    # def test_store_unicode(self):
+    #    fileio.write_chunk(self.tmp_path, DUMMY_UNICODE, 0, self.logger,
+    #                       mode='r+')
+    #
+    #    with open(self.tmp_path, 'r') as file:
+    #        content = file.read(1024)
+    #        self.assertEqual(len(content), DUMMY_UNICODE_LENGTH)
+    #        self.assertEqual(content[:], DUMMY_UNICODE)
+
+    # TODO: enable again - requires the temporarily disabled auto mode select
+    # def test_store_unicode_in_binary_mode(self):
+    #    fileio.write_chunk(self.tmp_path, DUMMY_UNICODE, 0, self.logger,
+    #                       mode='r+b')
+    #
+    #    with open(self.tmp_path, 'r') as file:
+    #        content = file.read(1024)
+    #        self.assertEqual(len(content), DUMMY_UNICODE_LENGTH)
+    #        self.assertEqual(content[:], DUMMY_UNICODE)
 
 
 class MigSharedFileio__write_file(MigTestCase):
@@ -156,50 +159,59 @@ def test_return_false_on_missing_dir(self):
         self.assertFalse(did_succeed)
 
     def test_creates_directory(self):
-        did_succeed = fileio.write_file(DUMMY_BYTES, self.tmp_path, self.logger)
+        # TODO: temporarily use empty string to avoid any byte/unicode issues
+        #did_succeed = fileio.write_file(DUMMY_BYTES, self.tmp_path, self.logger)
+        did_succeed = fileio.write_file('', self.tmp_path, self.logger)
         self.assertTrue(did_succeed)
 
         path_kind = self.assertPathExists(DUMMY_FILE_WRITEFILE)
         self.assertEqual(path_kind, "file")
 
     def test_store_bytes(self):
-        did_succeed = fileio.write_file(DUMMY_BYTES, self.tmp_path, self.logger)
-        self.assertTrue(did_succeed)
-
-        with open(self.tmp_path, 'rb') as file:
-            content = file.read(1024)
-            self.assertEqual(len(content), DUMMY_BYTES_LENGTH)
-            self.assertEqual(content[:], DUMMY_BYTES)
-
-    def test_store_bytes_in_text_mode(self):
+        mode = 'w'
+        # TODO: remove next once we have auto adjust mode in write helper
+        mode = fileio._auto_adjust_mode(DUMMY_BYTES, mode)
         did_succeed = fileio.write_file(DUMMY_BYTES, self.tmp_path, self.logger,
-                                        mode="w")
+                                        mode=mode)
         self.assertTrue(did_succeed)
 
         with open(self.tmp_path, 'rb') as file:
             content = file.read(1024)
             self.assertEqual(len(content), DUMMY_BYTES_LENGTH)
             self.assertEqual(content[:], DUMMY_BYTES)
 
-    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:
-            content = file.read(1024)
-            self.assertEqual(len(content), DUMMY_UNICODE_LENGTH)
-            self.assertEqual(content[:], DUMMY_UNICODE)
-
-    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:
-            content = file.read(1024)
-            self.assertEqual(len(content), DUMMY_UNICODE_LENGTH)
-            self.assertEqual(content[:], DUMMY_UNICODE)
+    # 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")
+    #    self.assertTrue(did_succeed)
+    #
+    #    with open(self.tmp_path, 'rb') as file:
+    #        content = file.read(1024)
+    #        self.assertEqual(len(content), DUMMY_BYTES_LENGTH)
+    #        self.assertEqual(content[:], DUMMY_BYTES)
+
+    # 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:
+    #        content = file.read(1024)
+    #        self.assertEqual(len(content), DUMMY_UNICODE_LENGTH)
+    #        self.assertEqual(content[:], DUMMY_UNICODE)
+
+    # 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:
+    #        content = file.read(1024)
+    #        self.assertEqual(len(content), DUMMY_UNICODE_LENGTH)
+    #        self.assertEqual(content[:], DUMMY_UNICODE)
 
 
 if __name__ == '__main__':