diff --git a/kombu/utils/text.py b/kombu/utils/text.py index a6558cfc7..fd359f8bd 100644 --- a/kombu/utils/text.py +++ b/kombu/utils/text.py @@ -1,9 +1,11 @@ """Text Utilities.""" + # flake8: noqa from __future__ import annotations +import re from difflib import SequenceMatcher from typing import Iterable, Iterator @@ -14,12 +16,15 @@ def escape_regex(p, white=''): # type: (str, str) -> str """Escape string for use within a regular expression.""" # what's up with re.escape? that code must be neglected or something - return ''.join(c if c.isalnum() or c in white - else ('\\000' if c == '\000' else '\\' + c) - for c in p) + return ''.join( + c if c.isalnum() or c in white else ('\\000' if c == '\000' else '\\' + c) + for c in p + ) -def fmatch_iter(needle: str, haystack: Iterable[str], min_ratio: float = 0.6) -> Iterator[tuple[float, str]]: +def fmatch_iter( + needle: str, haystack: Iterable[str], min_ratio: float = 0.6 +) -> Iterator[tuple[float, str]]: """Fuzzy match: iteratively. Yields @@ -32,39 +37,76 @@ def fmatch_iter(needle: str, haystack: Iterable[str], min_ratio: float = 0.6) -> yield ratio, key -def fmatch_best(needle: str, haystack: Iterable[str], min_ratio: float = 0.6) -> str | None: +def fmatch_best( + needle: str, haystack: Iterable[str], min_ratio: float = 0.6 +) -> str | None: """Fuzzy match - Find best match (scalar).""" try: return sorted( - fmatch_iter(needle, haystack, min_ratio), reverse=True, - )[0][1] + fmatch_iter(needle, haystack, min_ratio), + reverse=True, + )[ + 0 + ][1] except IndexError: return None -def version_string_as_tuple(s: str) -> version_info_t: - """Convert version string to version info tuple.""" - v = _unpack_version(*s.split('.')) - # X.Y.3a1 -> (X, Y, 3, 'a1') - if isinstance(v.micro, str): - v = version_info_t(v.major, v.minor, *_splitmicro(*v[2:])) - # X.Y.3a1-40 -> (X, Y, 3, 'a1', '40') - if not v.serial and v.releaselevel and '-' in v.releaselevel: - v = version_info_t(*list(v[0:3]) + v.releaselevel.split('-')) - return v +def version_string_as_tuple(version: str) -> version_info_t: + """Parse a version string into its components and return a version_info_t tuple. + + The version string is expected to follow the pattern: + 'major.minor.micro[releaselevel][serial]'. Each component of the version + is extracted and returned as a tuple in the format (major, minor, micro, + releaselevel, serial). + + Args + ---- + version (str): The version string to parse. + + Returns + ------- + version_info_t: A tuple containing the parsed version components. + + Raises + ------ + ValueError: If the version string is invalid and does not match the expected pattern. + """ + pattern = r'^(\d+)' # catching the major version (mandatory) + pattern += r'(?:\.(\d+))?' # optionally catching the minor version + pattern += r'(?:\.(\d+))?' # optionally catching the micro version + pattern += r'(?:\.*([a-zA-Z+-][\da-zA-Z+-]*))?' # optionally catching the release level (starting with a letter, + or -) after a dot + pattern += r'(?:\.(.*))?' # optionally catching the serial number after a dot + + # applying the regex pattern to the input version string + match = re.match(pattern, version) + + if not match: + raise ValueError(f"Invalid version string: {version}") + + # extracting the matched groups + major = int(match.group(1)) + minor = int(match.group(2)) if match.group(2) else 0 + micro = int(match.group(3)) if match.group(3) else 0 + releaselevel = match.group(4) if match.group(4) else '' + serial = match.group(5) if match.group(5) else '' + + return _unpack_version(major, minor, micro, releaselevel, serial) def _unpack_version( - major: str, + major: str | int = 0, minor: str | int = 0, micro: str | int = 0, releaselevel: str = '', - serial: str = '' + serial: str = '', ) -> version_info_t: - return version_info_t(int(major), int(minor), micro, releaselevel, serial) + return version_info_t(int(major), int(minor), int(micro), releaselevel, serial) -def _splitmicro(micro: str, releaselevel: str = '', serial: str = '') -> tuple[int, str, str]: +def _splitmicro( + micro: str, releaselevel: str = '', serial: str = '' +) -> tuple[int, str, str]: for index, char in enumerate(micro): if not char.isdigit(): break diff --git a/t/unit/utils/test_utils.py b/t/unit/utils/test_utils.py index 08f950836..051928768 100644 --- a/t/unit/utils/test_utils.py +++ b/t/unit/utils/test_utils.py @@ -8,15 +8,25 @@ def test_dir(): import kombu + assert dir(kombu) -@pytest.mark.parametrize('version,expected', [ - ('3', version_info_t(3, 0, 0, '', '')), - ('3.3', version_info_t(3, 3, 0, '', '')), - ('3.3.1', version_info_t(3, 3, 1, '', '')), - ('3.3.1a3', version_info_t(3, 3, 1, 'a3', '')), - ('3.3.1.a3.40c32', version_info_t(3, 3, 1, 'a3', '40c32')), -]) +@pytest.mark.parametrize( + 'version,expected', + [ + ('3', version_info_t(3, 0, 0, '', '')), + ('3.3', version_info_t(3, 3, 0, '', '')), + ('3.3.1', version_info_t(3, 3, 1, '', '')), + ('3.3.1a3', version_info_t(3, 3, 1, 'a3', '')), + ('3.3.1.a3.40c32', version_info_t(3, 3, 1, 'a3', '40c32')), + ('4.0.0+beta.3.47.g4f1a05b', version_info_t( + 4, 0, 0, '+beta', '3.47.g4f1a05b')), + ('4.0.0-beta3.47.g4f1a05b', version_info_t( + 4, 0, 0, '-beta3', '47.g4f1a05b')), + ('4.0.1-alpha.3+40c32', version_info_t(4, 0, 1, '-alpha', '3+40c32')), + ('0+beta3.14159265', version_info_t(0, 0, 0, '+beta3', '14159265')), + ], +) def test_version_string_as_tuple(version, expected): assert version_string_as_tuple(version) == expected