rf: readonly fixes from #105

Author: mgxd

heudiconv/bids.py

@@ -15,7 +15,7 @@
 
 from .parser import find_files
 from .utils import (load_json, save_json, create_file_if_missing,
-                    json_dumps_pretty, set_readonly)
+                    json_dumps_pretty, set_readonly, is_readonly)
 
 lgr = logging.getLogger(__name__)
 
@@ -142,11 +142,14 @@ def tuneup_bids_json_files(json_files):
                                           '_magnitude%d.json' % i)['EchoTime'])
                 except IOError as exc:
                     lgr.error("Failed to open magnitude file: %s", exc)
-            # might have been made R/O already
-            set_readonly(json_phasediffname, False)
-            #json.dump(json_, open(json_phasediffname, 'w'), indent=2)
+            # might have been made R/O already, but if not -- it will be set
+            # only later in the pipeline, so we must not make it read-only yet
+            was_readonly = is_readonly(json_phasediffname)
+            if was_readonly:
+                set_readonly(json_phasediffname, False)
             save_json(json_phasediffname, json_, indent=2)
-            set_readonly(json_phasediffname)
+            if was_readonly:
+                set_readonly(json_phasediffname)
 
 
 def add_participant_record(studydir, subject, age, sex):

heudiconv/utils.py

@@ -276,6 +276,11 @@ def safe_copyfile(src, dest):
     shutil.copyfile(src, dest)
 
 
+# Globals to check filewriting permissions
+ALL_CAN_WRITE = (stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH)
+ALL_CAN_READ = (stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
+assert ALL_CAN_READ >> 1 == ALL_CAN_WRITE  # Assumption in the code
+
 def set_readonly(path, read_only=True):
     """Make file read only or writeable while preserving "access levels"
 
@@ -290,9 +295,6 @@ def set_readonly(path, read_only=True):
         writeable for levels where it is readable
 
     """
-    ALL_CAN_WRITE = (stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH)
-    ALL_CAN_READ = (stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
-    assert ALL_CAN_READ >> 1 == ALL_CAN_WRITE  # Assumption in the code
 
     # get current permissions
     perms = stat.S_IMODE(os.lstat(path).st_mode)
@@ -305,6 +307,15 @@ def set_readonly(path, read_only=True):
         whocanread = perms & ALL_CAN_READ
         thosecanwrite = whocanread >> 1
         new_perms = perms | thosecanwrite
-        # apply and return those target permissions
-        os.chmod(path, new_perms)
-        return new_perms
+    # apply and return those target permissions
+    os.chmod(path, new_perms)
+    return new_perms
+
+
+def is_readonly(path):
+    """Return True if it is a fully read-only file (dereferences the symlink)
+    """
+    # get current permissions
+    perms = stat.S_IMODE(os.lstat(os.path.realpath(path)).st_mode)
+    # should be true if anyone is allowed to write
+    return not bool(perms & ALL_CAN_WRITE)

tests/test_main.py

@@ -2,7 +2,10 @@
 
 from heudiconv.cli.run import main as runner
 from heudiconv import __version__
-from heudiconv.utils import create_file_if_missing, json_dumps_pretty
+from heudiconv.utils import (create_file_if_missing,
+                             json_dumps_pretty,
+                             set_readonly,
+                             is_readonly)
 from heudiconv.bids import (populate_bids_templates,
                             add_participant_record,
                             get_formatted_scans_key_row,
@@ -18,9 +21,7 @@
 from mock import patch
 from os.path import join as opj
 from six.moves import StringIO
-
-
-import heudiconv
+import stat
 
 
 @patch('sys.stdout', new_callable=StringIO)
@@ -215,3 +216,27 @@ def test__find_subj_ses():
                                                                        's1')
     assert find_subj_ses(
         'fmap/sub-01-fmap_acq-3mm_acq-3mm_phasediff.nii.gz') == ('01', None)
+
+
+def test_make_readonly(tmpdir):
+    # we could test it all without torturing a poor file, but for going all
+    # the way, let's do it on a file
+    path = tmpdir.join('f')
+    pathname = str(path)
+    with open(pathname, 'w'):
+        pass
+    symname = pathname + 'link'
+    os.symlink(pathname, symname)
+    for orig, ro, rw in [
+        (0o600, 0o400, 0o600),  # fully returned
+        (0o624, 0o404, 0o606),  # it will not get write bit where it is not readable
+        (0o1777, 0o1555, 0o1777),  # and other bits should be preserved
+    ]:
+        os.chmod(pathname, orig)
+        assert not is_readonly(pathname)
+        assert set_readonly(pathname) == ro
+        assert is_readonly(pathname)
+        assert stat.S_IMODE(os.lstat(pathname).st_mode) == ro
+        # and it should go back if we set it back to non-read_only
+        assert set_readonly(pathname, read_only=False) == rw
+        assert not is_readonly(pathname)