cleanups: typing + whitespaces

rhodecode/api/tests/utils.py

@@ -1,123 +1,127 b''
 
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 import random
 import pytest
 
 from rhodecode.api.utils import get_origin
 from rhodecode.lib.ext_json import json
 
 
 def jsonify(obj):
     return json.loads(json.dumps(obj))
 
 
 API_URL = '/_admin/api'
 
 
 def assert_call_ok(id_, given):
     expected = jsonify({
         'id': id_,
         'error': None,
         'result': None
     })
     given = json.loads(given)
 
     assert expected['id'] == given['id']
     assert expected['error'] == given['error']
     return given['result']
 
 
 def assert_ok(id_, expected, given):
     given = json.loads(given)
     if given.get('error'):
         err = given['error']
         pytest.fail(f"Unexpected ERROR in expected success response: `{err}`")
 
     expected = jsonify({
         'id': id_,
         'error': None,
         'result': expected
     })
 
     assert expected == given
 
 
 def assert_error(id_, expected, given):
     expected = jsonify({
         'id': id_,
         'error': expected,
         'result': None
     })
     given = json.loads(given)
     assert expected == given
 
 
 def build_data(apikey, method, **kw):
     """
     Builds API data with given random ID
     """
     random_id = random.randrange(1, 9999)
     return random_id, json.dumps({
         "id": random_id,
         "api_key": apikey,
         "method": method,
         "args": kw
     })
 
 
-def api_call(app, params, status=None):
+def api_call(app, params, status=None, assert_no_error=False):
     response = app.post(
         API_URL, content_type='application/json', params=params, status=status,
         headers=[('Content-Type', 'application/json')])
+    if assert_no_error:
+        err_resp = response.json.get('error')
+        if err_resp:
+            raise AssertionError(f'ERROR in response: {err_resp}')
     return response
 
 
 def crash(*args, **kwargs):
     raise Exception('Total Crash !')
 
 
 def expected_permissions(object_with_permissions):
     """
     Returns the expected permissions structure for the given object.
 
     The object is expected to be a `Repository`, `RepositoryGroup`,
     or `UserGroup`. They all implement the same permission handling
     API.
     """
     permissions = []
     for _user in object_with_permissions.permissions():
         user_data = {
             'name': _user.username,
             'permission': _user.permission,
             'origin': get_origin(_user),
             'type': "user",
         }
         permissions.append(user_data)
 
     for _user_group in object_with_permissions.permission_user_groups():
         user_group_data = {
             'name': _user_group.users_group_name,
             'permission': _user_group.permission,
             'origin': get_origin(_user_group),
             'type': "user_group",
         }
         permissions.append(user_group_data)
     return permissions

rhodecode/lib/str_utils.py

@@ -1,169 +1,169 b''
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import typing
 import base64
 import logging
 from unidecode import unidecode
 
 import rhodecode
 from rhodecode.lib.type_utils import aslist
 
 
 log = logging.getLogger(__name__)
 
 
 def safe_int(val, default=None) -> int:
     """
     Returns int() of val if val is not convertable to int use default
     instead
 
     :param val:
     :param default:
     """
 
     try:
         val = int(val)
     except (ValueError, TypeError):
         val = default
 
     return val
 
 
 def safe_float(val, default=None) -> float:
     """
     Returns float() of val if val is not convertable to float use default
     instead
 
     :param val:
     :param default:
     """
 
     try:
         val = float(val)
     except (ValueError, TypeError):
         val = default
 
     return val
 
 
-def base64_to_str(text) -> str:
+def base64_to_str(text: str | bytes) -> str:
     return safe_str(base64.encodebytes(safe_bytes(text))).strip()
 
 
 def get_default_encodings() -> list[str]:
     return aslist(rhodecode.CONFIG.get('default_encoding', 'utf8'), sep=',')
 
 
 DEFAULT_ENCODINGS = get_default_encodings()
 
 
 def safe_str(str_, to_encoding=None) -> str:
     """
     safe str function. Does few trick to turn unicode_ into string
 
     :param str_: str to encode
     :param to_encoding: encode to this type UTF8 default
     """
     if isinstance(str_, str):
         return str_
 
     # if it's bytes cast to str
     if not isinstance(str_, bytes):
         return str(str_)
 
     to_encoding = to_encoding or DEFAULT_ENCODINGS
     if not isinstance(to_encoding, (list, tuple)):
         to_encoding = [to_encoding]
 
     for enc in to_encoding:
         try:
             return str(str_, enc)
         except UnicodeDecodeError:
             pass
 
     return str(str_, to_encoding[0], 'replace')
 
 
 def safe_bytes(str_, from_encoding=None) -> bytes:
     """
     safe bytes function. Does few trick to turn str_ into bytes string:
 
     :param str_: string to decode
     :param from_encoding: encode from this type UTF8 default
     """
     if isinstance(str_, bytes):
         return str_
 
     if not isinstance(str_, str):
         raise ValueError(f'safe_bytes cannot convert other types than str: got: {type(str_)}')
 
     from_encoding = from_encoding or get_default_encodings()
     if not isinstance(from_encoding, (list, tuple)):
         from_encoding = [from_encoding]
 
     for enc in from_encoding:
         try:
             return str_.encode(enc)
         except UnicodeDecodeError:
             pass
 
     return str_.encode(from_encoding[0], 'replace')
 
 
 def ascii_bytes(str_, allow_bytes=False) -> bytes:
     """
     Simple conversion from str to bytes, with assumption that str_ is pure ASCII.
     Fails with UnicodeError on invalid input.
     This should be used where encoding and "safe" ambiguity should be avoided.
     Where strings already have been encoded in other ways but still are unicode
     string - for example to hex, base64, json, urlencoding, or are known to be
     identifiers.
     """
     if allow_bytes and isinstance(str_, bytes):
         return str_
 
     if not isinstance(str_, str):
         raise ValueError(f'ascii_bytes cannot convert other types than str: got: {type(str_)}')
     return str_.encode('ascii')
 
 
 def ascii_str(str_) -> str:
     """
     Simple conversion from bytes to str, with assumption that str_ is pure ASCII.
     Fails with UnicodeError on invalid input.
     This should be used where encoding and "safe" ambiguity should be avoided.
     Where strings are encoded but also in other ways are known to be ASCII, and
     where a unicode string is wanted without caring about encoding. For example
     to hex, base64, urlencoding, or are known to be identifiers.
     """
 
     if not isinstance(str_, bytes):
         raise ValueError(f'ascii_str cannot convert other types than bytes: got: {type(str_)}')
     return str_.decode('ascii')
 
 
 def convert_special_chars(str_) -> str:
     """
         trie to replace non-ascii letters to their ascii representation eg::
 
             `żołw` converts into `zolw`
     """
     value = safe_str(str_)
     converted_value = unidecode(value)
     return converted_value

setup.py

@@ -1,203 +1,202 b''
-
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 # Import early to make sure things are patched up properly
 from setuptools import setup, find_packages, Extension
 
 import os
 import re
 import sys
 import pkgutil
 import platform
 import codecs
 
 import pip
 
 pip_major_version = int(pip.__version__.split(".")[0])
 if pip_major_version >= 20:
     from pip._internal.req import parse_requirements
     from pip._internal.network.session import PipSession
 elif pip_major_version >= 10:
     from pip._internal.req import parse_requirements
     from pip._internal.download import PipSession
 else:
     from pip.req import parse_requirements
     from pip.download import PipSession
 
 
 def get_package_name(req_object):
     package_name = None
     try:
         from pip._internal.req.constructors import install_req_from_parsed_requirement
     except ImportError:
         install_req_from_parsed_requirement = None
 
     # In 20.1 of pip, the requirements object changed
     if hasattr(req_object, 'req'):
         package_name = req_object.req.name
 
     if package_name is None:
         if install_req_from_parsed_requirement:
             package = install_req_from_parsed_requirement(req_object)
             package_name = package.req.name
 
     if package_name is None:
         # fallback for older pip
         package_name = re.split('===|<=|!=|==|>=|~=|<|>', req_object.requirement)[0]
 
     return package_name
 
 
 if sys.version_info < (3, 10):
     raise Exception('RhodeCode requires Python 3.10 or later')
 
 here = os.path.abspath(os.path.dirname(__file__))
 
 # defines current platform
 __platform__ = platform.system()
 __license__ = 'AGPLv3, and Commercial License'
 __author__ = 'RhodeCode GmbH'
 __url__ = 'https://code.rhodecode.com'
 is_windows = __platform__ in ('Windows',)
 
 
 def _get_requirements(req_filename, exclude=None, extras=None):
     extras = extras or []
     exclude = exclude or []
 
     try:
         parsed = parse_requirements(
             os.path.join(here, req_filename), session=PipSession())
     except TypeError:
         # try pip < 6.0.0, that doesn't support session
         parsed = parse_requirements(os.path.join(here, req_filename))
 
     requirements = []
     for int_req in parsed:
         req_name = get_package_name(int_req)
         if req_name not in exclude:
             requirements.append(req_name)
     return requirements + extras
 
 
 # requirements extract
 setup_requirements = ['PasteScript']
 install_requirements = _get_requirements(
     'requirements.txt', exclude=['setuptools', 'entrypoints'])
 test_requirements = _get_requirements(
     'requirements_test.txt')
 
 
 def get_version():
     version = pkgutil.get_data('rhodecode', 'VERSION')
     return version.decode().strip()
 
 
 # additional files that goes into package itself
 package_data = {
     '': ['*.txt', '*.rst'],
     'configs': ['*.ini'],
     'rhodecode': ['VERSION', 'i18n/*/LC_MESSAGES/*.mo', ],
 }
 
 description = 'Source Code Management Platform'
 keywords = ' '.join([
     'rhodecode', 'mercurial', 'git', 'svn',
     'code review',
     'repo groups', 'ldap', 'repository management', 'hgweb',
     'hgwebdir', 'gitweb', 'serving hgweb',
 ])
 
 
 # README/DESCRIPTION generation
 readme_file = 'README.rst'
 changelog_file = 'CHANGES.rst'
 try:
     long_description = codecs.open(readme_file).read() + '\n\n' + \
                        codecs.open(changelog_file).read()
 except IOError as err:
     sys.stderr.write(
         "[WARNING] Cannot find file specified as long_description (%s)\n "
         "or changelog (%s) skipping that file" % (readme_file, changelog_file))
     long_description = description
 
 
 setup(
     name='rhodecode-enterprise-ce',
     version=get_version(),
     description=description,
     long_description=long_description,
     keywords=keywords,
     license=__license__,
     author=__author__,
     author_email='[e-mail]',
     url=__url__,
     setup_requires=setup_requirements,
     install_requires=install_requirements,
     tests_require=test_requirements,
     zip_safe=False,
     packages=find_packages(exclude=["docs", "tests*"]),
     package_data=package_data,
     include_package_data=True,
     classifiers=[
         'Development Status :: 6 - Mature',
         'Environment :: Web Environment',
         'Intended Audience :: Developers',
         'Operating System :: OS Independent',
         'Topic :: Software Development :: Version Control',
         'License :: OSI Approved :: Affero GNU General Public License v3 or later (AGPLv3+)',
         'Programming Language :: Python :: 3.10',
     ],
     message_extractors={
         'rhodecode': [
             ('**.py', 'python', None),
             ('**.js', 'javascript', None),
             ('templates/**.mako', 'mako', {'input_encoding': 'utf-8'}),
             ('templates/**.html', 'mako', {'input_encoding': 'utf-8'}),
             ('public/**', 'ignore', None),
         ]
     },
     paster_plugins=['PasteScript'],
     entry_points={
         'paste.app_factory': [
             'main=rhodecode.config.middleware:make_pyramid_app',
         ],
         'paste.global_paster_command': [
             'ishell=rhodecode.lib.paster_commands.ishell:Command',
             'upgrade-db=rhodecode.lib.paster_commands.upgrade_db:UpgradeDb',
 
             'setup-rhodecode=rhodecode.lib.paster_commands.deprecated.setup_rhodecode:Command',
             'celeryd=rhodecode.lib.paster_commands.deprecated.celeryd:Command',
         ],
         'pyramid.pshell_runner': [
             'ipython = rhodecode.lib.pyramid_shell:ipython_shell_runner',
         ],
         'console_scripts': [
             'rc-setup-app=rhodecode.lib.rc_commands.setup_rc:main',
             'rc-upgrade-db=rhodecode.lib.rc_commands.upgrade_db:main',
             'rc-ishell=rhodecode.lib.rc_commands.ishell:main',
             'rc-add-artifact=rhodecode.lib.rc_commands.add_artifact:main',
             'rc-ssh-wrapper=rhodecode.apps.ssh_support.lib.ssh_wrapper:main',
         ],
         'beaker.backends': [
             'memorylru_base=rhodecode.lib.memory_lru_dict:MemoryLRUNamespaceManagerBase',
             'memorylru_debug=rhodecode.lib.memory_lru_dict:MemoryLRUNamespaceManagerDebug'
         ]
     },
 )