Manipulate with tempfiles after they are closed (#139)

* Manipulate with tempfiles after they are closed

Fixes: #135

* CloseableTemporaryFile: close a file before unlink (if not closed yet)

cvise/passes/clang.py

@@ -1,9 +1,9 @@
 import logging
 import os
 import shutil
-import tempfile
 
 from cvise.passes.abstract import AbstractPass, PassResult
+from cvise.utils.misc import CloseableTemporaryFile
 
 
 class ClangPass(AbstractPass):
@@ -21,7 +21,7 @@ def advance_on_success(self, test_case, state):
 
     def transform(self, test_case, state, process_event_notifier):
         tmp = os.path.dirname(test_case)
-        with tempfile.NamedTemporaryFile(mode='w', delete=False, dir=tmp) as tmp_file:
+        with CloseableTemporaryFile(mode='w', dir=tmp) as tmp_file:
             args = [
                 self.external_programs['clang_delta'],
                 f'--transformation={self.arg}',
@@ -33,13 +33,13 @@ def transform(self, test_case, state, process_event_notifier):
 
             logging.debug(' '.join(cmd))
 
-            stdout, _stderr, returncode = process_event_notifier.run_process(cmd)
+            stdout, _, returncode = process_event_notifier.run_process(cmd)
             if returncode == 0:
                 tmp_file.write(stdout)
-                shutil.move(tmp_file.name, test_case)
+                tmp_file.close()
+                shutil.copy(tmp_file.name, test_case)
                 return (PassResult.OK, state)
             else:
-                os.unlink(tmp_file.name)
                 if returncode == 255 or returncode == 1:
                     return (PassResult.STOP, state)
                 else:

cvise/passes/clangbinarysearch.py

@@ -3,10 +3,10 @@
 import re
 import shutil
 import subprocess
-import tempfile
 import time
 
 from cvise.passes.abstract import AbstractPass, BinaryState, PassResult
+from cvise.utils.misc import CloseableTemporaryFile
 
 
 class ClangBinarySearchPass(AbstractPass):
@@ -97,7 +97,7 @@ def transform(self, test_case, state, process_event_notifier):
         logging.debug(f'TRANSFORM: {state}')
 
         tmp = os.path.dirname(test_case)
-        with tempfile.NamedTemporaryFile(mode='w', delete=False, dir=tmp) as tmp_file:
+        with CloseableTemporaryFile(mode='w', dir=tmp) as tmp_file:
             args = [
                 f'--transformation={self.arg}',
                 f'--counter={state.index + 1}',
@@ -115,11 +115,11 @@ def transform(self, test_case, state, process_event_notifier):
             stdout, stderr, returncode = process_event_notifier.run_process(cmd)
             self.parse_stderr(state, stderr)
             tmp_file.write(stdout)
+            tmp_file.close()
             if returncode == 0:
-                shutil.move(tmp_file.name, test_case)
+                shutil.copy(tmp_file.name, test_case)
                 return (PassResult.OK, state)
             else:
-                os.unlink(tmp_file.name)
                 return (
                     PassResult.STOP if returncode == 255 else PassResult.ERROR,
                     state,

cvise/passes/clex.py

@@ -1,8 +1,8 @@
 import os
 import shutil
-import tempfile
 
 from cvise.passes.abstract import AbstractPass, PassResult
+from cvise.utils.misc import CloseableTemporaryFile
 
 
 class ClexPass(AbstractPass):
@@ -20,15 +20,15 @@ def advance_on_success(self, test_case, state):
 
     def transform(self, test_case, state, process_event_notifier):
         tmp = os.path.dirname(test_case)
-        with tempfile.NamedTemporaryFile(mode='w', delete=False, dir=tmp) as tmp_file:
+        with CloseableTemporaryFile(mode='w', dir=tmp) as tmp_file:
             cmd = [self.external_programs['clex'], str(self.arg), str(state), test_case]
             stdout, _stderr, returncode = process_event_notifier.run_process(cmd)
             if returncode == 51:
                 tmp_file.write(stdout)
-                shutil.move(tmp_file.name, test_case)
+                tmp_file.close()
+                shutil.copy(tmp_file.name, test_case)
                 return (PassResult.OK, state)
             else:
-                os.unlink(tmp_file.name)
                 return (
                     PassResult.STOP if returncode == 71 else PassResult.ERROR,
                     state,

cvise/passes/lines.py

@@ -6,6 +6,7 @@
 
 from cvise.passes.abstract import AbstractPass, BinaryState, PassResult
 from cvise.utils.error import InsaneTestCaseError
+from cvise.utils.misc import CloseableTemporaryFile
 
 
 class LinesPass(AbstractPass):
@@ -15,9 +16,10 @@ def check_prerequisites(self):
     def __format(self, test_case, check_sanity):
         tmp = os.path.dirname(test_case)
 
-        with tempfile.NamedTemporaryFile(mode='w+', dir=tmp) as backup, tempfile.NamedTemporaryFile(
+        with CloseableTemporaryFile(mode='w+', dir=tmp) as backup, CloseableTemporaryFile(
             mode='w+', dir=tmp
         ) as tmp_file:
+            backup.close()
             with open(test_case) as in_file:
                 try:
                     cmd = [self.external_programs['topformflat'], self.arg]
@@ -28,7 +30,7 @@ def __format(self, test_case, check_sanity):
             for line in proc.stdout.splitlines(keepends=True):
                 if not line.isspace():
                     tmp_file.write(line)
-            tmp_file.flush()
+            tmp_file.close()
 
             # we need to check that sanity check is still fine
             if check_sanity:

cvise/passes/unifdef.py

@@ -36,6 +36,7 @@ def transform(self, test_case, state, process_event_notifier):
 
         tmp = os.path.dirname(test_case)
         with tempfile.NamedTemporaryFile(mode='w+', delete=False, dir=tmp) as tmp_file:
+            tmp_file.close()
             while True:
                 du = '-D' if state % 2 == 0 else '-U'
                 n_index = int(state / 2)

cvise/utils/misc.py

@@ -1,6 +1,24 @@
+import os
+import tempfile
+from contextlib import contextmanager
+
+
 def is_readable_file(filename):
     try:
         open(filename).read()
         return True
     except UnicodeDecodeError:
         return False
+
+
+# TODO: use tempfile.NamedTemporaryFile(delete_on_close=False) since Python 3.12 is the oldest supported release
+@contextmanager
+def CloseableTemporaryFile(mode='w+b', dir=None):
+    f = tempfile.NamedTemporaryFile(mode=mode, delete=False, dir=dir)
+    try:
+        yield f
+    finally:
+        # For Windows systems, be sure we always close the file before we remove it!
+        if not f.closed:
+            f.close()
+        os.remove(f.name)