Apply ruff formatting.

.pre-commit-config.yaml

@@ -9,7 +9,7 @@ repos:
     rev: v0.5.0
     hooks:
       - id: ruff
-        args: [--fix, --exit-non-zero-on-fix, --show-fixes]
+        args: [--fix, --exit-non-zero-on-fix, --show-fixes, --unsafe-fixes]
       - id: ruff-format
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v4.6.0

src/manhole/__init__.py

@@ -186,7 +186,7 @@ def start(self):
         self.should_run = True
         super().start()
         if not self.serious.wait(self.start_timeout):
-            _LOG("WARNING: Waited %s seconds but Manhole thread didn't start yet :(" % self.start_timeout)
+            _LOG(f"WARNING: Waited {self.start_timeout} seconds but Manhole thread didn't start yet :(")
 
     def run(self):
         """
@@ -201,12 +201,12 @@ def run(self):
         pthread_setname_np(self.ident, self.psname)
 
         if self.bind_delay:
-            _LOG('Delaying UDS binding %s seconds ...' % self.bind_delay)
+            _LOG(f'Delaying UDS binding {self.bind_delay} seconds ...')
             _ORIGINAL_SLEEP(self.bind_delay)
 
         sock = self.get_socket()
         while self.should_run:
-            _LOG('Waiting for new connection (in pid:%s) ...' % os.getpid())
+            _LOG(f'Waiting for new connection (in pid:{os.getpid()}) ...')
             try:
                 client = ManholeConnectionThread(sock.accept()[0], self.connection_handler, self.daemon_connection)
                 client.start()
@@ -243,7 +243,7 @@ def run(self):
         try:
             self.connection_handler(self.client)
         except BaseException as exc:
-            _LOG('ManholeConnectionThread failure: %r' % exc)
+            _LOG(f'ManholeConnectionThread failure: {exc!r}')
 
 
 def check_credentials(client):
@@ -280,7 +280,7 @@ def exit():
             try:
                 payload = fh.readline()
                 while payload:
-                    _LOG('Running: %r.' % payload)
+                    _LOG(f'Running: {payload!r}.')
                     eval(compile(payload, '<manhole>', 'exec'), {'exit': exit}, _MANHOLE.locals)
                     payload = fh.readline()
             except ExitExecLoop:
@@ -313,7 +313,7 @@ def handle_connection_repl(client: socket.socket):
         except BrokenPipeError:
             _LOG('REPL client disconnected')
         except Exception as exc:
-            _LOG('REPL failed with %r.' % exc)
+            _LOG(f'REPL failed with {exc!r}.')
         _LOG('DONE.')
     finally:
         try:
@@ -482,11 +482,11 @@ def configure(
                 self.patch_os_fork_functions()
             else:
                 if activate_on:
-                    _LOG('Not patching os.fork and os.forkpty. Activation is done by signal %s' % activate_on)
+                    _LOG(f'Not patching os.fork and os.forkpty. Activation is done by signal {activate_on}')
                 elif oneshot_on:
-                    _LOG('Not patching os.fork and os.forkpty. Oneshot activation is done by signal %s' % oneshot_on)
+                    _LOG(f'Not patching os.fork and os.forkpty. Oneshot activation is done by signal {oneshot_on}')
                 elif socket_path:
-                    _LOG('Not patching os.fork and os.forkpty. Using user socket path %s' % socket_path)
+                    _LOG(f'Not patching os.fork and os.forkpty. Using user socket path {socket_path}')
 
     def release(self):
         if self._thread:
@@ -532,15 +532,15 @@ def handle_oneshot(self, _signum=None, _frame=None):
         try:
             try:
                 sock = self.get_socket()
-                _LOG('Waiting for new connection (in pid:%s) ...' % os.getpid())
+                _LOG(f'Waiting for new connection (in pid:{os.getpid()}) ...')
                 client = force_original_socket(sock.accept()[0])
                 check_credentials(client)
                 self.connection_handler(client)
             finally:
                 self.remove_manhole_uds()
         except BaseException as exc:  # pylint: disable=W0702
             # we don't want to let any exception out, it might make the application misbehave
-            _LOG('Oneshot failure: %r' % exc)
+            _LOG(f'Oneshot failure: {exc!r}')
 
     def remove_manhole_uds(self):
         name = self.uds_name
@@ -551,7 +551,7 @@ def remove_manhole_uds(self):
     @property
     def uds_name(self):
         if self.socket_path is None:
-            return '/tmp/manhole-%s' % os.getpid()
+            return f'/tmp/manhole-{os.getpid()}'
         return self.socket_path
 
     def patched_fork(self):
@@ -649,7 +649,7 @@ def dump_stacktraces():
         for filename, lineno, name, line in traceback.extract_stack(stack):
             lines.append('File: "%s", line %d, in %s' % (filename, lineno, name))
             if line:
-                lines.append('  %s' % (line.strip()))
+                lines.append(f'  {line.strip()}')
     lines.append('#############################################\n\n')
 
     print('\n'.join(lines), file=sys.stderr if _MANHOLE.redirect_stderr else sys.stdout)

src/manhole/cli.py

@@ -49,12 +49,15 @@ def parse_signal(value):
     if value in SIG_NAMES:
         return SIG_NAMES[value]
     else:
-        raise argparse.ArgumentTypeError('Invalid signal name %r.' % value)
+        raise argparse.ArgumentTypeError(f'Invalid signal name {value!r}.')
 
 
 parser = argparse.ArgumentParser(description='Connect to a manhole.')
 parser.add_argument(
-    'pid', metavar='PID', type=parse_pid, help='A numerical process id, or a path in the form: /tmp/manhole-1234'  # nargs='?',
+    'pid',
+    metavar='PID',
+    type=parse_pid,
+    help='A numerical process id, or a path in the form: /tmp/manhole-1234',  # nargs='?',
 )
 parser.add_argument('-t', '--timeout', dest='timeout', default=1, type=float, help='Timeout to use. Default: %(default)s seconds.')
 group = parser.add_mutually_exclusive_group()
@@ -119,7 +122,7 @@ def main():
         os.kill(args.pid, args.signal)
 
     start = time.time()
-    uds_path = '/tmp/manhole-%s' % args.pid
+    uds_path = f'/tmp/manhole-{args.pid}'
     sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
     sock.settimeout(args.timeout)
     while time.time() - start < args.timeout:
@@ -131,7 +134,7 @@ def main():
         else:
             break
     else:
-        print('Failed to connect to %r: Timeout' % uds_path, file=sys.stderr)
+        print(f'Failed to connect to {uds_path!r}: Timeout', file=sys.stderr)
         sys.exit(5)
 
     is_closing = threading.Event()

tests/helper.py

@@ -195,7 +195,7 @@ def handle_usr2(_sig, _frame):
 
             @partial(signal.signal, signal.SIGUSR1)
             def signal_handler(sig, _):
-                print('Received signal %s' % sig)
+                print(f'Received signal {sig}')
                 global signalled
                 signalled = True
 
@@ -212,7 +212,7 @@ def signal_handler(sig, _):
             sys.setcheckinterval(1)
             for _ in range(100000):
                 os.kill(os.getpid(), signal.SIGUSR1)
-            print('signalled=%s' % signalled)
+            print(f'signalled={signalled}')
             time.sleep(TIMEOUT * 10)
         elif test_name == 'test_auth_fail':
             manhole.get_peercred = lambda _: (-1, -1, -1)
@@ -282,7 +282,7 @@ def cleanup():
             else:
                 raise RuntimeError('Invalid test spec.')
     except:  # noqa
-        print('Died with %s.' % sys.exc_info()[0].__name__, file=OUTPUT)
+        print(f'Died with {sys.exc_info()[0].__name__}.', file=OUTPUT)
         import traceback
 
         traceback.print_exc(file=OUTPUT)

tests/test_manhole.py

@@ -54,7 +54,7 @@ def assert_manhole_running(proc, uds_path, oneshot=False, extra=None):
             wait_for_strings(client.read, TIMEOUT, 'ProcessID', 'ThreadID', '>>>')
             sock.send(b"print('FOOBAR')\n")
             wait_for_strings(client.read, TIMEOUT, 'FOOBAR')
-            wait_for_strings(proc.read, TIMEOUT, 'UID:%s' % os.getuid())
+            wait_for_strings(proc.read, TIMEOUT, f'UID:{os.getuid()}')
             if extra:
                 extra(client)
     wait_for_strings(proc.read, TIMEOUT, 'Cleaned up.', *[] if oneshot else ['Waiting for new connection'])
@@ -102,7 +102,7 @@ def test_connection_handler_exec(variant):
                 wait_for_strings(
                     proc.read,
                     TIMEOUT,
-                    'UID:%s' % os.getuid(),
+                    f'UID:{os.getuid()}',
                 )
                 with TestSocket(sock) as client:
                     with dump_on_error(client.read):
@@ -238,7 +238,7 @@ def test_redirect_stderr_default():
                 with dump_on_error(client.read):
                     wait_for_strings(client.read, 1, '>>>')
                     client.reset()
-                    sock.send(b"import sys\n" b"sys.stderr.write('OK')\n")
+                    sock.send(b'import sys\n' b"sys.stderr.write('OK')\n")
                     wait_for_strings(client.read, 1, 'OK')
 
 
@@ -265,7 +265,7 @@ def test_redirect_stderr_disabled():
                 with dump_on_error(client.read):
                     wait_for_strings(client.read, 1, '>>>')
                     client.reset()
-                    sock.send(b"import sys\n" b"sys.stderr.write('STDERR')\n" b"sys.stdout.write('STDOUT')\n")
+                    sock.send(b'import sys\n' b"sys.stderr.write('STDERR')\n" b"sys.stdout.write('STDOUT')\n")
                     wait_for_strings(client.read, 1, 'STDOUT')
                     assert 'STDERR' not in client.read()
 
@@ -321,18 +321,18 @@ def test_exit_with_grace():
                     sock.send(b"print('FOOBAR')\n")
                     wait_for_strings(client.read, TIMEOUT, 'FOOBAR')
 
-                    wait_for_strings(proc.read, TIMEOUT, 'UID:%s' % os.getuid())
+                    wait_for_strings(proc.read, TIMEOUT, f'UID:{os.getuid()}')
                     sock.shutdown(socket.SHUT_WR)
                     select.select([sock], [], [], 5)
                     sock.recv(1024)
                     try:
                         sock.shutdown(socket.SHUT_RD)
                     except Exception as exc:
-                        print('Failed to SHUT_RD: %s' % exc)
+                        print(f'Failed to SHUT_RD: {exc}')
                     try:
                         sock.close()
                     except Exception as exc:
-                        print('Failed to close socket: %s' % exc)
+                        print(f'Failed to close socket: {exc}')
             wait_for_strings(proc.read, TIMEOUT, 'DONE.', 'Cleaned up.', 'Waiting for new connection')
 
 
@@ -396,7 +396,7 @@ def test_auth_fail():
                 wait_for_strings(
                     proc.read,
                     TIMEOUT,
-                    "SuspiciousClient: Can't accept client with PID:-1 UID:-1 GID:-1. It doesn't match the current " "EUID:",
+                    "SuspiciousClient: Can't accept client with PID:-1 UID:-1 GID:-1. It doesn't match the current " 'EUID:',
                     'Waiting for new connection',
                 )
                 proc.proc.send_signal(signal.SIGINT)
@@ -551,7 +551,7 @@ def test_uwsgi():
         with dump_on_error(proc.read):
             wait_for_strings(proc.read, TIMEOUT, 'uWSGI http bound')
             port = re.findall(r'uWSGI http bound on :(\d+) fd', proc.read())[0]
-            assert requests.get('http://127.0.0.1:%s/' % port, timeout=TIMEOUT).text == 'OK'
+            assert requests.get(f'http://127.0.0.1:{port}/', timeout=TIMEOUT).text == 'OK'
 
             wait_for_strings(proc.read, TIMEOUT, 'spawned uWSGI worker 1')
             pid = re.findall(r'spawned uWSGI worker 1 \(pid: (\d+), ', proc.read())[0]
@@ -560,7 +560,7 @@ def test_uwsgi():
                 with open('/tmp/manhole-pid', 'w') as fh:
                     fh.write(pid)
                 try:
-                    assert_manhole_running(proc, '/tmp/manhole-%s' % pid, oneshot=True)
+                    assert_manhole_running(proc, f'/tmp/manhole-{pid}', oneshot=True)
                 except Exception:
                     if not retry:
                         raise

tests/test_manhole_cli.py

@@ -85,7 +85,7 @@ def test_path():
     with TestProcess(sys.executable, HELPER, 'test_simple') as service:
         with dump_on_error(service.read):
             wait_for_strings(service.read, TIMEOUT, '/tmp/manhole-')
-            with TestProcess('manhole-cli', '/tmp/manhole-%s' % service.proc.pid, bufsize=0, stdin=subprocess.PIPE) as client:
+            with TestProcess('manhole-cli', f'/tmp/manhole-{service.proc.pid}', bufsize=0, stdin=subprocess.PIPE) as client:
                 with dump_on_error(client.read):
                     wait_for_strings(client.read, TIMEOUT, '(ManholeConsole)', '>>>')
                     client.proc.stdin.write('1234+2345\n')