lint: more autofixes

vcsserver/git_lfs/lib.py

@@ -1,175 +1,175 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
 import os
 import shutil
 import logging
 from collections import OrderedDict
 
 log = logging.getLogger(__name__)
 
 
 class OidHandler:
 
     def __init__(self, store, repo_name, auth, oid, obj_size, obj_data, obj_href,
                  obj_verify_href=None):
         self.current_store = store
         self.repo_name = repo_name
         self.auth = auth
         self.oid = oid
         self.obj_size = obj_size
         self.obj_data = obj_data
         self.obj_href = obj_href
         self.obj_verify_href = obj_verify_href
 
     def get_store(self, mode=None):
         return self.current_store
 
     def get_auth(self):
         """returns auth header for re-use in upload/download"""
         return " ".join(self.auth)
 
     def download(self):
 
         store = self.get_store()
         response = None
         has_errors = None
 
         if not store.has_oid():
             # error reply back to client that something is wrong with dl
             err_msg = f'object: {store.oid} does not exist in store'
             has_errors = OrderedDict(
                 error=OrderedDict(
                     code=404,
                     message=err_msg
                 )
             )
 
         download_action = OrderedDict(
             href=self.obj_href,
             header=OrderedDict([("Authorization", self.get_auth())])
         )
         if not has_errors:
             response = OrderedDict(download=download_action)
         return response, has_errors
 
     def upload(self, skip_existing=True):
         """
         Write upload action for git-lfs server
         """
 
         store = self.get_store()
         response = None
         has_errors = None
 
         # verify if we have the OID before, if we do, reply with empty
         if store.has_oid():
             log.debug('LFS: store already has oid %s', store.oid)
 
             # validate size
             store_size = store.size_oid()
             size_match = store_size == self.obj_size
             if not size_match:
                 log.warning(
                     'LFS: size mismatch for oid:%s, in store:%s expected: %s',
                     self.oid, store_size, self.obj_size)
             elif skip_existing:
                 log.debug('LFS: skipping further action as oid is existing')
                 return response, has_errors
 
         chunked = ("Transfer-Encoding", "chunked")
         upload_action = OrderedDict(
             href=self.obj_href,
             header=OrderedDict([("Authorization", self.get_auth()), chunked])
         )
         if not has_errors:
             response = OrderedDict(upload=upload_action)
             # if specified in handler, return the verification endpoint
             if self.obj_verify_href:
                 verify_action = OrderedDict(
                     href=self.obj_verify_href,
                     header=OrderedDict([("Authorization", self.get_auth())])
                 )
                 response['verify'] = verify_action
         return response, has_errors
 
     def exec_operation(self, operation, *args, **kwargs):
         handler = getattr(self, operation)
         log.debug('LFS: handling request using %s handler', handler)
         return handler(*args, **kwargs)
 
 
 class LFSOidStore:
 
     def __init__(self, oid, repo, store_location=None):
         self.oid = oid
         self.repo = repo
         self.store_path = store_location or self.get_default_store()
         self.tmp_oid_path = os.path.join(self.store_path, oid + '.tmp')
         self.oid_path = os.path.join(self.store_path, oid)
         self.fd = None
 
     def get_engine(self, mode):
         """
         engine = .get_engine(mode='wb')
         with engine as f:
             f.write('...')
         """
 
-        class StoreEngine(object):
+        class StoreEngine:
             def __init__(self, mode, store_path, oid_path, tmp_oid_path):
                 self.mode = mode
                 self.store_path = store_path
                 self.oid_path = oid_path
                 self.tmp_oid_path = tmp_oid_path
 
             def __enter__(self):
                 if not os.path.isdir(self.store_path):
                     os.makedirs(self.store_path)
 
                 # TODO(marcink): maybe write metadata here with size/oid ?
                 fd = open(self.tmp_oid_path, self.mode)
                 self.fd = fd
                 return fd
 
             def __exit__(self, exc_type, exc_value, traceback):
                 # close tmp file, and rename to final destination
                 self.fd.close()
                 shutil.move(self.tmp_oid_path, self.oid_path)
 
         return StoreEngine(
             mode, self.store_path, self.oid_path, self.tmp_oid_path)
 
     def get_default_store(self):
         """
         Default store, consistent with defaults of Mercurial large files store
         which is /home/username/.cache/largefiles
         """
         user_home = os.path.expanduser("~")
         return os.path.join(user_home, '.cache', 'lfs-store')
 
     def has_oid(self):
         return os.path.exists(os.path.join(self.store_path, self.oid))
 
     def size_oid(self):
         size = -1
 
         if self.has_oid():
             oid = os.path.join(self.store_path, self.oid)
             size = os.stat(oid).st_size
 
         return size

vcsserver/lib/_vendor/statsd/base.py

@@ -1,154 +1,154 b''
 import re
 import random
 from collections import deque
 from datetime import timedelta
 from repoze.lru import lru_cache
 
 from .timer import Timer
 
 TAG_INVALID_CHARS_RE = re.compile(
     r"[^\w\d_\-:/\.]",
     #re.UNICODE
 )
 TAG_INVALID_CHARS_SUBS = "_"
 
 # we save and expose methods called by statsd for discovery
 buckets_dict = {
 
 }
 
 
 @lru_cache(maxsize=500)
 def _normalize_tags_with_cache(tag_list):
     return [TAG_INVALID_CHARS_RE.sub(TAG_INVALID_CHARS_SUBS, tag) for tag in tag_list]
 
 
 def normalize_tags(tag_list):
     # We have to turn our input tag list into a non-mutable tuple for it to
     # be hashable (and thus usable) by the @lru_cache decorator.
     return _normalize_tags_with_cache(tuple(tag_list))
 
 
 class StatsClientBase:
     """A Base class for various statsd clients."""
 
     def close(self):
         """Used to close and clean up any underlying resources."""
         raise NotImplementedError()
 
     def _send(self):
         raise NotImplementedError()
 
     def pipeline(self):
         raise NotImplementedError()
 
     def timer(self, stat, rate=1, tags=None, auto_send=True):
         """
         statsd = StatsdClient.statsd
         with statsd.timer('bucket_name', auto_send=True) as tmr:
             # This block will be timed.
             for i in range(0, 100000):
                 i ** 2
         # you can access time here...
         elapsed_ms = tmr.ms
         """
         return Timer(self, stat, rate, tags, auto_send=auto_send)
 
     def timing(self, stat, delta, rate=1, tags=None, use_decimals=True):
         """
         Send new timing information.
 
         `delta` can be either a number of milliseconds or a timedelta.
         """
         if isinstance(delta, timedelta):
             # Convert timedelta to number of milliseconds.
             delta = delta.total_seconds() * 1000.
         if use_decimals:
             fmt = '%0.6f|ms'
         else:
             fmt = '%s|ms'
         self._send_stat(stat, fmt % delta, rate, tags)
 
     def incr(self, stat, count=1, rate=1, tags=None):
         """Increment a stat by `count`."""
-        self._send_stat(stat, '%s|c' % count, rate, tags)
+        self._send_stat(stat, f'{count}|c', rate, tags)
 
     def decr(self, stat, count=1, rate=1, tags=None):
         """Decrement a stat by `count`."""
         self.incr(stat, -count, rate, tags)
 
     def gauge(self, stat, value, rate=1, delta=False, tags=None):
         """Set a gauge value."""
         if value < 0 and not delta:
             if rate < 1:
                 if random.random() > rate:
                     return
             with self.pipeline() as pipe:
                 pipe._send_stat(stat, '0|g', 1)
-                pipe._send_stat(stat, '%s|g' % value, 1)
+                pipe._send_stat(stat, f'{value}|g', 1)
         else:
             prefix = '+' if delta and value >= 0 else ''
-            self._send_stat(stat, '%s%s|g' % (prefix, value), rate, tags)
+            self._send_stat(stat, f'{prefix}{value}|g', rate, tags)
 
     def set(self, stat, value, rate=1):
         """Set a set value."""
-        self._send_stat(stat, '%s|s' % value, rate)
+        self._send_stat(stat, f'{value}|s', rate)
 
     def histogram(self, stat, value, rate=1, tags=None):
         """Set a histogram"""
-        self._send_stat(stat, '%s|h' % value, rate, tags)
+        self._send_stat(stat, f'{value}|h', rate, tags)
 
     def _send_stat(self, stat, value, rate, tags=None):
         self._after(self._prepare(stat, value, rate, tags))
 
     def _prepare(self, stat, value, rate, tags=None):
         global buckets_dict
         buckets_dict[stat] = 1
 
         if rate < 1:
             if random.random() > rate:
                 return
-            value = '%s|@%s' % (value, rate)
+            value = f'{value}|@{rate}'
 
         if self._prefix:
-            stat = '%s.%s' % (self._prefix, stat)
+            stat = f'{self._prefix}.{stat}'
 
         res = '%s:%s%s' % (
             stat,
             value,
             ("|#" + ",".join(normalize_tags(tags))) if tags else "",
         )
         return res
 
     def _after(self, data):
         if data:
             self._send(data)
 
 
 class PipelineBase(StatsClientBase):
 
     def __init__(self, client):
         self._client = client
         self._prefix = client._prefix
         self._stats = deque()
 
     def _send(self):
         raise NotImplementedError()
 
     def _after(self, data):
         if data is not None:
             self._stats.append(data)
 
     def __enter__(self):
         return self
 
     def __exit__(self, typ, value, tb):
         self.send()
 
     def send(self):
         if not self._stats:
             return
         self._send()
 
     def pipeline(self):
         return self.__class__(self)

vcsserver/pygrack.py

@@ -1,417 +1,417 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
 """Handles the Git smart protocol."""
 
 import os
 import socket
 import logging
 
 import dulwich.protocol
 from dulwich.protocol import CAPABILITY_SIDE_BAND, CAPABILITY_SIDE_BAND_64K
 from webob import Request, Response, exc
 
 from vcsserver.lib.rc_json import json
 from vcsserver import hooks, subprocessio
 from vcsserver.str_utils import ascii_bytes
 
 
 log = logging.getLogger(__name__)
 
 
 class FileWrapper:
     """File wrapper that ensures how much data is read from it."""
 
     def __init__(self, fd, content_length):
         self.fd = fd
         self.content_length = content_length
         self.remain = content_length
 
     def read(self, size):
         if size <= self.remain:
             try:
                 data = self.fd.read(size)
             except socket.error:
                 raise IOError(self)
             self.remain -= size
         elif self.remain:
             data = self.fd.read(self.remain)
             self.remain = 0
         else:
             data = None
         return data
 
     def __repr__(self):
         return '<FileWrapper {} len: {}, read: {}>'.format(
             self.fd, self.content_length, self.content_length - self.remain
         )
 
 
 class GitRepository:
     """WSGI app for handling Git smart protocol endpoints."""
 
     git_folder_signature = frozenset(('config', 'head', 'info', 'objects', 'refs'))
     commands = frozenset(('git-upload-pack', 'git-receive-pack'))
     valid_accepts = frozenset(f'application/x-{c}-result' for c in commands)
 
     # The last bytes are the SHA1 of the first 12 bytes.
     EMPTY_PACK = (
         b'PACK\x00\x00\x00\x02\x00\x00\x00\x00\x02\x9d\x08' +
         b'\x82;\xd8\xa8\xea\xb5\x10\xadj\xc7\\\x82<\xfd>\xd3\x1e'
     )
     FLUSH_PACKET = b"0000"
 
     SIDE_BAND_CAPS = frozenset((CAPABILITY_SIDE_BAND, CAPABILITY_SIDE_BAND_64K))
 
     def __init__(self, repo_name, content_path, git_path, update_server_info, extras):
         files = frozenset(f.lower() for f in os.listdir(content_path))
         valid_dir_signature = self.git_folder_signature.issubset(files)
 
         if not valid_dir_signature:
             raise OSError(f'{content_path} missing git signature')
 
         self.content_path = content_path
         self.repo_name = repo_name
         self.extras = extras
         self.git_path = git_path
         self.update_server_info = update_server_info
 
     def _get_fixedpath(self, path):
         """
         Small fix for repo_path
 
         :param path:
         """
         path = path.split(self.repo_name, 1)[-1]
         if path.startswith('.git'):
             # for bare repos we still get the .git prefix inside, we skip it
             # here, and remove from the service command
             path = path[4:]
 
         return path.strip('/')
 
     def inforefs(self, request, unused_environ):
         """
         WSGI Response producer for HTTP GET Git Smart
         HTTP /info/refs request.
         """
 
         git_command = request.GET.get('service')
         if git_command not in self.commands:
             log.debug('command %s not allowed', git_command)
             return exc.HTTPForbidden()
 
         # please, resist the urge to add '\n' to git capture and increment
         # line count by 1.
         # by git docs: Documentation/technical/http-protocol.txt#L214 \n is
         # a part of protocol.
         # The code in Git client not only does NOT need '\n', but actually
         # blows up if you sprinkle "flush" (0000) as "0001\n".
         # It reads binary, per number of bytes specified.
         # if you do add '\n' as part of data, count it.
-        server_advert = '# service=%s\n' % git_command
+        server_advert = f'# service={git_command}\n'
         packet_len = hex(len(server_advert) + 4)[2:].rjust(4, '0').lower()
         try:
             gitenv = dict(os.environ)
             # forget all configs
             gitenv['RC_SCM_DATA'] = json.dumps(self.extras)
             command = [self.git_path, git_command[4:], '--stateless-rpc',
                        '--advertise-refs', self.content_path]
             out = subprocessio.SubprocessIOChunker(
                 command,
                 env=gitenv,
                 starting_values=[ascii_bytes(packet_len + server_advert) + self.FLUSH_PACKET],
                 shell=False
             )
         except OSError:
             log.exception('Error processing command')
             raise exc.HTTPExpectationFailed()
 
         resp = Response()
         resp.content_type = f'application/x-{git_command}-advertisement'
         resp.charset = None
         resp.app_iter = out
 
         return resp
 
     def _get_want_capabilities(self, request):
         """Read the capabilities found in the first want line of the request."""
         pos = request.body_file_seekable.tell()
         first_line = request.body_file_seekable.readline()
         request.body_file_seekable.seek(pos)
 
         return frozenset(
             dulwich.protocol.extract_want_line_capabilities(first_line)[1])
 
     def _build_failed_pre_pull_response(self, capabilities, pre_pull_messages):
         """
         Construct a response with an empty PACK file.
 
         We use an empty PACK file, as that would trigger the failure of the pull
         or clone command.
 
         We also print in the error output a message explaining why the command
         was aborted.
 
         If additionally, the user is accepting messages we send them the output
         of the pre-pull hook.
 
         Note that for clients not supporting side-band we just send them the
         emtpy PACK file.
         """
 
         if self.SIDE_BAND_CAPS.intersection(capabilities):
             response = []
             proto = dulwich.protocol.Protocol(None, response.append)
             proto.write_pkt_line(dulwich.protocol.NAK_LINE)
 
             self._write_sideband_to_proto(proto, ascii_bytes(pre_pull_messages, allow_bytes=True), capabilities)
             # N.B.(skreft): Do not change the sideband channel to 3, as that
             # produces a fatal error in the client:
             #   fatal: error in sideband demultiplexer
             proto.write_sideband(
                 dulwich.protocol.SIDE_BAND_CHANNEL_PROGRESS,
                 ascii_bytes('Pre pull hook failed: aborting\n', allow_bytes=True))
             proto.write_sideband(
                 dulwich.protocol.SIDE_BAND_CHANNEL_DATA,
                 ascii_bytes(self.EMPTY_PACK, allow_bytes=True))
 
             # writes b"0000" as default
             proto.write_pkt_line(None)
 
             return response
         else:
             return [ascii_bytes(self.EMPTY_PACK, allow_bytes=True)]
 
     def _build_post_pull_response(self, response, capabilities, start_message, end_message):
         """
         Given a list response we inject the post-pull messages.
 
         We only inject the messages if the client supports sideband, and the
         response has the format:
             0008NAK\n...0000
 
         Note that we do not check the no-progress capability as by default, git
         sends it, which effectively would block all messages.
         """
 
         if not self.SIDE_BAND_CAPS.intersection(capabilities):
             return response
 
         if not start_message and not end_message:
             return response
 
         try:
             iter(response)
             # iterator probably will work, we continue
         except TypeError:
             raise TypeError(f'response must be an iterator: got {type(response)}')
         if isinstance(response, (list, tuple)):
             raise TypeError(f'response must be an iterator: got {type(response)}')
 
         def injected_response():
 
             do_loop = 1
             header_injected = 0
             next_item = None
             has_item = False
             item = b''
 
             while do_loop:
 
                 try:
                     next_item = next(response)
                 except StopIteration:
                     do_loop = 0
 
                 if has_item:
                     # last item ! alter it now
                     if do_loop == 0 and item.endswith(self.FLUSH_PACKET):
                         new_response = [item[:-4]]
                         new_response.extend(self._get_messages(end_message, capabilities))
                         new_response.append(self.FLUSH_PACKET)
                         item = b''.join(new_response)
 
                     yield item
 
                 has_item = True
                 item = next_item
 
                 # alter item if it's the initial chunk
                 if not header_injected and item.startswith(b'0008NAK\n'):
                     new_response = [b'0008NAK\n']
                     new_response.extend(self._get_messages(start_message, capabilities))
                     new_response.append(item[8:])
                     item = b''.join(new_response)
                     header_injected = 1
 
         return injected_response()
 
     def _write_sideband_to_proto(self, proto, data, capabilities):
         """
         Write the data to the proto's sideband number 2 == SIDE_BAND_CHANNEL_PROGRESS
 
         We do not use dulwich's write_sideband directly as it only supports
         side-band-64k.
         """
         if not data:
             return
 
         # N.B.(skreft): The values below are explained in the pack protocol
         # documentation, section Packfile Data.
         # https://github.com/git/git/blob/master/Documentation/technical/pack-protocol.txt
         if CAPABILITY_SIDE_BAND_64K in capabilities:
             chunk_size = 65515
         elif CAPABILITY_SIDE_BAND in capabilities:
             chunk_size = 995
         else:
             return
 
         chunker = (data[i:i + chunk_size] for i in range(0, len(data), chunk_size))
 
         for chunk in chunker:
             proto.write_sideband(dulwich.protocol.SIDE_BAND_CHANNEL_PROGRESS, ascii_bytes(chunk, allow_bytes=True))
 
     def _get_messages(self, data, capabilities):
         """Return a list with packets for sending data in sideband number 2."""
         response = []
         proto = dulwich.protocol.Protocol(None, response.append)
 
         self._write_sideband_to_proto(proto, data, capabilities)
 
         return response
 
     def backend(self, request, environ):
         """
         WSGI Response producer for HTTP POST Git Smart HTTP requests.
         Reads commands and data from HTTP POST's body.
         returns an iterator obj with contents of git command's
         response to stdout
         """
         # TODO(skreft): think how we could detect an HTTPLockedException, as
         # we probably want to have the same mechanism used by mercurial and
         # simplevcs.
         # For that we would need to parse the output of the command looking for
         # some signs of the HTTPLockedError, parse the data and reraise it in
         # pygrack. However, that would interfere with the streaming.
         #
         # Now the output of a blocked push is:
         # Pushing to http://test_regular:test12@127.0.0.1:5001/vcs_test_git
         # POST git-receive-pack (1047 bytes)
         # remote: ERROR: Repository `vcs_test_git` locked by user `test_admin`. Reason:`lock_auto`
         # To http://test_regular:test12@127.0.0.1:5001/vcs_test_git
         # ! [remote rejected] master -> master (pre-receive hook declined)
         # error: failed to push some refs to 'http://test_regular:test12@127.0.0.1:5001/vcs_test_git'
 
         git_command = self._get_fixedpath(request.path_info)
         if git_command not in self.commands:
             log.debug('command %s not allowed', git_command)
             return exc.HTTPForbidden()
 
         capabilities = None
         if git_command == 'git-upload-pack':
             capabilities = self._get_want_capabilities(request)
 
         if 'CONTENT_LENGTH' in environ:
             inputstream = FileWrapper(request.body_file_seekable,
                                       request.content_length)
         else:
             inputstream = request.body_file_seekable
 
         resp = Response()
         resp.content_type = f'application/x-{git_command}-result'
         resp.charset = None
 
         pre_pull_messages = ''
         # Upload-pack == clone
         if git_command == 'git-upload-pack':
             hook_response = hooks.git_pre_pull(self.extras)
             if hook_response.status != 0:
                 pre_pull_messages = hook_response.output
                 resp.app_iter = self._build_failed_pre_pull_response(
                     capabilities, pre_pull_messages)
                 return resp
 
         gitenv = dict(os.environ)
         # forget all configs
         gitenv['GIT_CONFIG_NOGLOBAL'] = '1'
         gitenv['RC_SCM_DATA'] = json.dumps(self.extras)
         cmd = [self.git_path, git_command[4:], '--stateless-rpc',
                self.content_path]
         log.debug('handling cmd %s', cmd)
 
         out = subprocessio.SubprocessIOChunker(
             cmd,
             input_stream=inputstream,
             env=gitenv,
             cwd=self.content_path,
             shell=False,
             fail_on_stderr=False,
             fail_on_return_code=False
         )
 
         if self.update_server_info and git_command == 'git-receive-pack':
             # We need to fully consume the iterator here, as the
             # update-server-info command needs to be run after the push.
             out = list(out)
 
             # Updating refs manually after each push.
             # This is required as some clients are exposing Git repos internally
             # with the dumb protocol.
             cmd = [self.git_path, 'update-server-info']
             log.debug('handling cmd %s', cmd)
             output = subprocessio.SubprocessIOChunker(
                 cmd,
                 input_stream=inputstream,
                 env=gitenv,
                 cwd=self.content_path,
                 shell=False,
                 fail_on_stderr=False,
                 fail_on_return_code=False
             )
             # Consume all the output so the subprocess finishes
             for _ in output:
                 pass
 
         # Upload-pack == clone
         if git_command == 'git-upload-pack':
             hook_response = hooks.git_post_pull(self.extras)
             post_pull_messages = hook_response.output
             resp.app_iter = self._build_post_pull_response(out, capabilities, pre_pull_messages, post_pull_messages)
         else:
             resp.app_iter = out
 
         return resp
 
     def __call__(self, environ, start_response):
         request = Request(environ)
         _path = self._get_fixedpath(request.path_info)
         if _path.startswith('info/refs'):
             app = self.inforefs
         else:
             app = self.backend
 
         try:
             resp = app(request, environ)
         except exc.HTTPException as error:
             log.exception('HTTP Error')
             resp = error
         except Exception:
             log.exception('Unknown error')
             resp = exc.HTTPInternalServerError()
 
         return resp(environ, start_response)

vcsserver/tests/test_install_hooks.py

@@ -1,206 +1,206 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
 import os
 import sys
 import stat
 import pytest
 import vcsserver
 import tempfile
 from vcsserver import hook_utils
 from vcsserver.tests.fixture import no_newline_id_generator
 from vcsserver.str_utils import safe_bytes, safe_str
 from vcsserver.utils import AttributeDict
 
 
 class TestCheckRhodecodeHook:
 
     def test_returns_false_when_hook_file_is_wrong_found(self, tmpdir):
         hook = os.path.join(str(tmpdir), 'fake_hook_file.py')
         with open(hook, 'wb') as f:
             f.write(b'dummy test')
             result = hook_utils.check_rhodecode_hook(hook)
             assert result is False
 
     def test_returns_true_when_no_hook_file_found(self, tmpdir):
         hook = os.path.join(str(tmpdir), 'fake_hook_file_not_existing.py')
         result = hook_utils.check_rhodecode_hook(hook)
         assert result
 
     @pytest.mark.parametrize("file_content, expected_result", [
         ("RC_HOOK_VER = '3.3.3'\n", True),
         ("RC_HOOK = '3.3.3'\n", False),
     ], ids=no_newline_id_generator)
     def test_signatures(self, file_content, expected_result, tmpdir):
         hook = os.path.join(str(tmpdir), 'fake_hook_file_1.py')
         with open(hook, 'wb') as f:
             f.write(safe_bytes(file_content))
 
         result = hook_utils.check_rhodecode_hook(hook)
 
         assert result is expected_result
 
 
 class BaseInstallHooks:
     HOOK_FILES = ()
 
     def _check_hook_file_mode(self, file_path):
-        assert os.path.exists(file_path), 'path %s missing' % file_path
+        assert os.path.exists(file_path), f'path {file_path} missing'
         stat_info = os.stat(file_path)
 
         file_mode = stat.S_IMODE(stat_info.st_mode)
         expected_mode = int('755', 8)
         assert expected_mode == file_mode
 
     def _check_hook_file_content(self, file_path, executable):
         executable = executable or sys.executable
         with open(file_path, 'rt') as hook_file:
             content = hook_file.read()
 
         expected_env = '#!{}'.format(executable)
         expected_rc_version = "\nRC_HOOK_VER = '{}'\n".format(vcsserver.__version__)
         assert content.strip().startswith(expected_env)
         assert expected_rc_version in content
 
     def _create_fake_hook(self, file_path, content):
         with open(file_path, 'w') as hook_file:
             hook_file.write(content)
 
     def create_dummy_repo(self, repo_type):
         tmpdir = tempfile.mkdtemp()
         repo = AttributeDict()
         if repo_type == 'git':
             repo.path = os.path.join(tmpdir, 'test_git_hooks_installation_repo')
             os.makedirs(repo.path)
             os.makedirs(os.path.join(repo.path, 'hooks'))
             repo.bare = True
 
         elif repo_type == 'svn':
             repo.path = os.path.join(tmpdir, 'test_svn_hooks_installation_repo')
             os.makedirs(repo.path)
             os.makedirs(os.path.join(repo.path, 'hooks'))
 
         return repo
 
     def check_hooks(self, repo_path, repo_bare=True):
         for file_name in self.HOOK_FILES:
             if repo_bare:
                 file_path = os.path.join(repo_path, 'hooks', file_name)
             else:
                 file_path = os.path.join(repo_path, '.git', 'hooks', file_name)
             self._check_hook_file_mode(file_path)
             self._check_hook_file_content(file_path, sys.executable)
 
 
 class TestInstallGitHooks(BaseInstallHooks):
     HOOK_FILES = ('pre-receive', 'post-receive')
 
     def test_hooks_are_installed(self):
         repo = self.create_dummy_repo('git')
         result = hook_utils.install_git_hooks(repo.path, repo.bare)
         assert result
         self.check_hooks(repo.path, repo.bare)
 
     def test_hooks_are_replaced(self):
         repo = self.create_dummy_repo('git')
         hooks_path = os.path.join(repo.path, 'hooks')
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             self._create_fake_hook(
                 file_path, content="RC_HOOK_VER = 'abcde'\n")
 
         result = hook_utils.install_git_hooks(repo.path, repo.bare)
         assert result
         self.check_hooks(repo.path, repo.bare)
 
     def test_non_rc_hooks_are_not_replaced(self):
         repo = self.create_dummy_repo('git')
         hooks_path = os.path.join(repo.path, 'hooks')
         non_rc_content = 'echo "non rc hook"\n'
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             self._create_fake_hook(
                 file_path, content=non_rc_content)
 
         result = hook_utils.install_git_hooks(repo.path, repo.bare)
         assert result
 
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             with open(file_path, 'rt') as hook_file:
                 content = hook_file.read()
             assert content == non_rc_content
 
     def test_non_rc_hooks_are_replaced_with_force_flag(self):
         repo = self.create_dummy_repo('git')
         hooks_path = os.path.join(repo.path, 'hooks')
         non_rc_content = 'echo "non rc hook"\n'
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             self._create_fake_hook(
                 file_path, content=non_rc_content)
 
         result = hook_utils.install_git_hooks(
             repo.path, repo.bare, force_create=True)
         assert result
         self.check_hooks(repo.path, repo.bare)
 
 
 class TestInstallSvnHooks(BaseInstallHooks):
     HOOK_FILES = ('pre-commit', 'post-commit')
 
     def test_hooks_are_installed(self):
         repo = self.create_dummy_repo('svn')
         result = hook_utils.install_svn_hooks(repo.path)
         assert result
         self.check_hooks(repo.path)
 
     def test_hooks_are_replaced(self):
         repo = self.create_dummy_repo('svn')
         hooks_path = os.path.join(repo.path, 'hooks')
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             self._create_fake_hook(
                 file_path, content="RC_HOOK_VER = 'abcde'\n")
 
         result = hook_utils.install_svn_hooks(repo.path)
         assert result
         self.check_hooks(repo.path)
 
     def test_non_rc_hooks_are_not_replaced(self):
         repo = self.create_dummy_repo('svn')
         hooks_path = os.path.join(repo.path, 'hooks')
         non_rc_content = 'echo "non rc hook"\n'
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             self._create_fake_hook(
                 file_path, content=non_rc_content)
 
         result = hook_utils.install_svn_hooks(repo.path)
         assert result
 
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             with open(file_path, 'rt') as hook_file:
                 content = hook_file.read()
             assert content == non_rc_content
 
     def test_non_rc_hooks_are_replaced_with_force_flag(self):
         repo = self.create_dummy_repo('svn')
         hooks_path = os.path.join(repo.path, 'hooks')
         non_rc_content = 'echo "non rc hook"\n'
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             self._create_fake_hook(
                 file_path, content=non_rc_content)
 
         result = hook_utils.install_svn_hooks(
             repo.path, force_create=True)
         assert result
         self.check_hooks(repo.path, )