Apply isort black

.pre-commit-config.yaml

@@ -0,0 +1,16 @@
+---
+repos:
+  - repo: https://github.com/python/black.git
+    rev: 22.3.0
+    hooks:
+      - id: black
+        name: black (python3)
+        language_version: python3
+        args: ["--check"]
+  - repo: https://github.com/pycqa/isort
+    rev: 5.10.1
+    hooks:
+      - id: isort
+        name: isort (python3)
+        language_version: python3
+        args: ["--check"]

drop_buffer_cache.py

@@ -4,9 +4,11 @@
 import os
 import sys
 
+
 class DropBufferCacheException(Exception):
     pass
 
+
 # Drop 'buffer' cache for the given range of the given file.
 
 POSIX_FADV_DONTNEED = 4
@@ -15,17 +17,19 @@ class DropBufferCacheException(Exception):
 
 # this function is used if we can't load the real libc function
 
+
 def noop_libc_function(*args):
     return 0
 
 
 # I have no idea what this code really does, but strace says it works.
 # does this code work under Cygwin?
 
+
 def load_libc_function(func_name):
     func = noop_libc_function
     try:
-        libc = ctypes.CDLL(ctypes.util.find_library('c'))
+        libc = ctypes.CDLL(ctypes.util.find_library("c"))
         func = getattr(libc, func_name)
     except AttributeError:
         # print("Unable to locate %s in libc.  Leaving as a no-op."% func_name)
@@ -35,28 +39,29 @@ def load_libc_function(func_name):
 
 # do this at module load time
 
-_posix_fadvise = load_libc_function('posix_fadvise64')
+_posix_fadvise = load_libc_function("posix_fadvise64")
 
 
 def drop_buffer_cache(fd, offset, length):
-    ret = _posix_fadvise(fd,
-                         ctypes.c_uint64(offset),
-                         ctypes.c_uint64(length),
-                         POSIX_FADV_DONTNEED)
+    ret = _posix_fadvise(
+        fd, ctypes.c_uint64(offset), ctypes.c_uint64(length), POSIX_FADV_DONTNEED
+    )
     if ret != OK:
-        raise DropBufferCacheException('posix_fadvise64(%s, %s, %s, 4) -> %s' %
-                        (fd, offset, length, ret))
+        raise DropBufferCacheException(
+            "posix_fadvise64(%s, %s, %s, 4) -> %s" % (fd, offset, length, ret)
+        )
+
 
 # unit test
 
-if __name__ == '__main__':
-    fd = os.open('/tmp/foo', os.O_WRONLY | os.O_CREAT)
-    if sys.version.startswith('3'):
-        ret = os.write(fd, bytes('hi there', 'UTF-8'))
-    elif sys.version.startswith('2'):
-        ret = os.write(fd, 'hi there')
+if __name__ == "__main__":
+    fd = os.open("/tmp/foo", os.O_WRONLY | os.O_CREAT)
+    if sys.version.startswith("3"):
+        ret = os.write(fd, bytes("hi there", "UTF-8"))
+    elif sys.version.startswith("2"):
+        ret = os.write(fd, "hi there")
     else:
-        raise DropBufferCacheException('unrecognized python version %s' % sys.version)
+        raise DropBufferCacheException("unrecognized python version %s" % sys.version)
     assert ret == 8
     drop_buffer_cache(fd, 0, 8)
     os.close(fd)

fallocate.py

@@ -18,17 +18,19 @@
 
 # this function is used if we can't load the real libc function
 
+
 def noop_libc_function(*args):
     return OK
 
 
 # I have no idea what this code really does, but strace says it works.
 # does this code work under Cygwin?
 
+
 def load_libc_function(func_name):
     func = noop_libc_function
     try:
-        libc = ctypes.CDLL(ctypes.util.find_library('c'))
+        libc = ctypes.CDLL(ctypes.util.find_library("c"))
         func = getattr(libc, func_name)
     except AttributeError:
         # print("Unable to locate %s in libc.  Leaving as a no-op."% func_name)
@@ -38,31 +40,30 @@ def load_libc_function(func_name):
 
 # do this at module load time
 
-_posix_fallocate = load_libc_function('fallocate64')
+_posix_fallocate = load_libc_function("fallocate64")
 
 
 # mode is one of FALLOC constants above
 
+
 def fallocate(fd, mode, offset, length):
-    return _posix_fallocate(fd,
-                            mode,
-                            ctypes.c_uint64(offset),
-                            ctypes.c_uint64(length))
+    return _posix_fallocate(fd, mode, ctypes.c_uint64(offset), ctypes.c_uint64(length))
+
 
 # unit test
 
-if __name__ == '__main__':
-    fd = os.open('/tmp/foo', os.O_WRONLY | os.O_CREAT)
+if __name__ == "__main__":
+    fd = os.open("/tmp/foo", os.O_WRONLY | os.O_CREAT)
     assert fd > 0x02
     ret = fallocate(fd, FALLOC_FL_KEEP_SIZE, 0, 8)
     assert ret == OK
-    if sys.version.startswith('3'):
-        ret = os.write(fd, bytes('hi there', 'UTF-8'))
-    elif sys.version.startswith('2'):
-        ret = os.write(fd, 'hi there')
+    if sys.version.startswith("3"):
+        ret = os.write(fd, bytes("hi there", "UTF-8"))
+    elif sys.version.startswith("2"):
+        ret = os.write(fd, "hi there")
     else:
-        print('unrecognized python version %s' % sys.version)
+        print("unrecognized python version %s" % sys.version)
         sys.exit(NOTOK)
     assert ret == 8
     os.close(fd)
-    print('SUCCESS')
+    print("SUCCESS")

invoke_process.py

@@ -1,30 +1,29 @@
 # -*- coding: utf-8 -*-
 
-'''
+"""
 invoke_process.py
 launch multiple subprocesses running SmallfileWorkload instance
 Copyright 2012 -- Ben England
 Licensed under the Apache License at http://www.apache.org/licenses/LICENSE-2.0
 See Appendix on this page for instructions pertaining to license.
-'''
+"""
 
 import multiprocessing
-import shutil
 import os
+import shutil
 import time
 
 import smallfile
-from smallfile import unittest_module, SMFRunException
+from smallfile import SMFRunException, unittest_module
 from sync_files import touch
 
-
 # this class launches multiple threads with SmallfileWorkload instances
 # we do this because we can use > 1 core this way, with python threading,
 # it doesn't really use > 1 core because of the GIL (global lock)
 # occasional status reports could be sent back using pipe as well
 
-class subprocess(multiprocessing.Process):
 
+class subprocess(multiprocessing.Process):
     def __init__(self, invocation):
         multiprocessing.Process.__init__(self)
         (conn1, conn2) = multiprocessing.Pipe(False)
@@ -38,11 +37,14 @@ def __init__(self, invocation):
     def run(self):
         try:
             self.invoke.do_workload()
-            self.invoke.log.debug('exiting subprocess and returning invoke '
-                                  + str(self.invoke))
+            self.invoke.log.debug(
+                "exiting subprocess and returning invoke " + str(self.invoke)
+            )
         except Exception as e:
-            print('Exception seen in thread %s host %s (tail %s) ' %
-                  (self.invoke.tid, self.invoke.onhost, self.invoke.log_fn()))
+            print(
+                "Exception seen in thread %s host %s (tail %s) "
+                % (self.invoke.tid, self.invoke.onhost, self.invoke.log_fn())
+            )
             self.invoke.log.error(str(e))
             self.status = self.invoke.NOTOK
         finally:
@@ -62,19 +64,19 @@ def run(self):
 # including multi-threaded test
 # to run, just do "python invoke_process.py"
 
-class Test(unittest_module.TestCase):
 
+class Test(unittest_module.TestCase):
     def setUp(self):
         self.invok = smallfile.SmallfileWorkload()
         self.invok.debug = True
         self.invok.verbose = True
-        self.invok.tid = 'regtest'
+        self.invok.tid = "regtest"
         self.invok.start_log()
         shutil.rmtree(self.invok.src_dirs[0], ignore_errors=True)
         os.makedirs(self.invok.src_dirs[0], 0o644)
 
     def test_multiproc_stonewall(self):
-        self.invok.log.info('starting stonewall test')
+        self.invok.log.info("starting stonewall test")
         thread_ready_timeout = 4
         thread_count = 4
         for tree in self.invok.top_dirs:
@@ -85,8 +87,7 @@ def test_multiproc_stonewall(self):
         for dir in self.invok.dest_dirs:
             os.mkdir(dir)
         os.mkdir(self.invok.network_dir)
-        self.invok.starting_gate = os.path.join(self.invok.network_dir,
-                                                'starting-gate')
+        self.invok.starting_gate = os.path.join(self.invok.network_dir, "starting-gate")
         sgate_file = self.invok.starting_gate
         invokeList = []
         for j in range(0, thread_count):
@@ -96,8 +97,8 @@ def test_multiproc_stonewall(self):
 
             s.verbose = True
             s.tid = str(j)
-            s.prefix = 'thr_'
-            s.suffix = 'foo'
+            s.prefix = "thr_"
+            s.suffix = "foo"
             s.iterations = 10
             s.stonewall = False
             s.starting_gate = sgate_file
@@ -118,8 +119,9 @@ def test_multiproc_stonewall(self):
                 break
             time.sleep(1)
         if not threads_ready:
-            raise SMFRunException('threads did not show up within %d seconds'
-                            % thread_ready_timeout)
+            raise SMFRunException(
+                "threads did not show up within %d seconds" % thread_ready_timeout
+            )
         time.sleep(1)
         touch(sgate_file)
         for t in threadList:
@@ -130,11 +132,13 @@ def test_multiproc_stonewall(self):
             assert rtnd_invok.rq_final is not None
             assert rtnd_invok.filenum_final is not None
             if rtnd_invok.status != rtnd_invok.OK:
-                raise SMFRunException('subprocess failure for %s invocation %s: '
-                                % (str(t), str(rtnd_invok)))
+                raise SMFRunException(
+                    "subprocess failure for %s invocation %s: "
+                    % (str(t), str(rtnd_invok))
+                )
 
 
 # so you can just do "python invoke_process.py" to test it
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     unittest_module.main()