feat(types): harden product version with validated ProductVersion type

Motivation:
Product version extraction used plain strings and empty strings for missing
values, causing ambiguity and potential errors. PR openSUSE#429 feedback requested
stricter definitions and better type safety.

Design Choices:
- Introduced ProductVersion string subclass in types.py with regex-based
  validation in __new__.
- Refactored gitea.py to return ProductVersion | None instead of "".
- Moved VERSION_EXTRACT_REGEX to types.py to centralize logic.
- Updated extraction logic to use fullmatch for stricter validation.
- Updated dependent code to use explicit is not None checks.

User Benefits:
- Improved type safety and runtime validation of product versions.
- Clearer API contracts with Optional types.
- Centralized version parsing logic improves maintainability.

Author: okurz

openqabot/loader/gitea.py

@@ -32,7 +32,7 @@
     from collections.abc import Iterator
 
     from openqabot.types.pullrequest import PullRequest
-    from openqabot.types.types import Repos
+from openqabot.types.types import VERSION_EXTRACT_REGEX, ProductVersion, Repos
 
 ARCHS = {"x86_64", "aarch64", "ppc64le", "s390x"}
 
@@ -54,7 +54,6 @@ class BuildResults:
 
 PROJECT_PRODUCT_REGEX = re.compile(r".*:PullRequest:\d+:(.*)")
 SCMSYNC_REGEX = re.compile(r".*/products/(.*)#([\d\.]{2,6})$")
-VERSION_EXTRACT_REGEX = re.compile(r"[.\d]+")
 OBS_PROJECT_SHOW_REGEX = re.compile(r".*/project/show/([^/\s\?\#\)]+)")
 # Regex to find all HTTPS URLs, excluding common trailing punctuation like dots or parentheses
 # that are likely part of the surrounding text (e.g. at the end of a sentence or in Markdown).
@@ -155,10 +154,10 @@ def get_product_name(obs_project: str) -> str:
     return product_match.group(1) if product_match else ""
 
 
-def get_product_name_and_version_from_scmsync(scmsync_url: str) -> tuple[str, str]:
+def get_product_name_and_version_from_scmsync(scmsync_url: str) -> tuple[str, ProductVersion | None]:
     """Extract product name and version from an scmsync URL."""
     m = SCMSYNC_REGEX.search(scmsync_url)
-    return (m.group(1), m.group(2)) if m else ("", "")
+    return (m.group(1), ProductVersion(m.group(2))) if m else ("", None)
 
 
 def compute_repo_url_for_job_setting(
@@ -284,14 +283,17 @@ def add_reviews(submission: dict[str, Any], reviews: list[Any]) -> int:
     return len(qam_states)
 
 
-def _extract_version(name: str, prefix: str) -> str:
+def _extract_version(name: str, prefix: str) -> ProductVersion | None:
     """Extract version number from a package name string."""
     remainder = name.removeprefix(prefix)
-    return next((part for part in remainder.split("-") if VERSION_EXTRACT_REGEX.search(part)), "")
+    return next(
+        (ProductVersion(part) for part in remainder.split("-") if VERSION_EXTRACT_REGEX.fullmatch(part)),
+        None,
+    )
 
 
 @lru_cache(maxsize=512)
-def get_product_version_from_repo_listing(project: str, product_name: str, repository: str) -> str:
+def get_product_version_from_repo_listing(project: str, product_name: str, repository: str) -> ProductVersion | None:
     """Determine the product version by inspecting an OBS repository listing."""
     project_path = project.replace(":", ":/")
     url = f"{config.settings.obs_download_url}/{project_path}/{repository}/repo?jsontable"
@@ -302,34 +304,34 @@ def get_product_version_from_repo_listing(project: str, product_name: str, repos
         data = r.json()["data"]
     except requests.exceptions.HTTPError as e:
         log.warning("Repo ignored: Could not query repository '%s' (%s->%s): %s", repository, product_name, project, e)
-        return ""
+        return None
     # Catching both because requests' JSONDecodeError might not inherit from json's
     except (json.JSONDecodeError, requests.exceptions.JSONDecodeError) as e:
         log.info("Invalid JSON document at '%s', ignoring: %s", url, e)
-        return ""
+        return None
     except requests.exceptions.RequestException as e:
         log.warning("Product version unresolved: Could not read from '%s': %s", url, e)
-        return ""
+        return None
     versions = (_extract_version(entry["name"], start) for entry in data if entry["name"].startswith(start))
-    return next((v for v in versions if len(v) > 0), "")
+    return next((v for v in versions if v is not None), None)
 
 
-def _get_product_version(res: etree._Element, project: str, product_name: str) -> str:
+def _get_product_version(res: etree._Element, project: str, product_name: str) -> ProductVersion | None:
     """Extract product version from scmsync element or repository listing."""
     # read product version from scmsync element if possible, e.g. 15.99
     product_version = next(
         (
             pv
-            for _, pv in (get_product_name_and_version_from_scmsync(e.text) for e in res.findall("scmsync"))
-            if len(pv) > 0
+            for _, pv in (get_product_name_and_version_from_scmsync(e.text) for e in res.findall("scmsync") if e.text)
+            if pv is not None
         ),
-        "",
+        None,
     )
 
     # read product version from directory listing if the project is for a concrete product
     if (
-        len(product_name) != 0
-        and len(product_version) == 0
+        product_name
+        and product_version is None
         and ("all" in config.settings.obs_products_set or product_name in config.settings.obs_products_set)
     ):
         product_version = get_product_version_from_repo_listing(project, product_name, res.get("repository"))
@@ -352,9 +354,9 @@ def add_channel_for_build_result(
     product_version = _get_product_version(res, project, product_name)
 
     # append product version to channel if known; otherwise skip channel if this is for a concrete product
-    if len(product_version) > 0:
+    if product_version is not None:
         channel = f"{channel}#{product_version}"
-    elif len(product_name) > 0:
+    elif product_name:
         log.debug("Channel skipped: Product version for build result %s:%s could not be determined", project, arch)
         return channel
 

openqabot/types/types.py

@@ -4,11 +4,27 @@
 
 from __future__ import annotations
 
+import re
 from enum import Enum, auto
 from typing import NamedTuple
 
 from openqabot.config import OBS_REPO_TYPE
 
+VERSION_EXTRACT_REGEX = re.compile(r"[.\d]+")
+
+
+class ProductVersion(str):  # noqa: FURB189
+    """A validated product version string."""
+
+    __slots__ = ()
+
+    def __new__(cls, value: str) -> ProductVersion:  # noqa: PYI034
+        """Create a new ProductVersion instance and validate its format."""
+        if not VERSION_EXTRACT_REGEX.fullmatch(value):
+            msg = f"Invalid product version format: '{value}'"
+            raise ValueError(msg)
+        return super().__new__(cls, value)
+
 
 class ChannelType(Enum):
     """Enumeration of channel types."""

tests/test_giteasync.py

@@ -247,7 +247,7 @@ def test_extracting_product_name_and_version() -> None:
 
     slfo_url = "https://src.suse.de/user1/SLFO.git?onlybuild=tree#f229f"
     prod_ver = get_product_name_and_version_from_scmsync(slfo_url)
-    assert prod_ver == ("", "")
+    assert prod_ver == ("", None)
     prod_url = "https://src.suse.de/products/SLES#15.99"
     prod_ver = get_product_name_and_version_from_scmsync(prod_url)
     assert prod_ver == ("SLES", "15.99")

tests/test_loader_gitea_helpers.py

@@ -74,7 +74,7 @@ def test_get_product_version_from_repo_listing_json_error(mocker: MockerFixture)
     mock_response.json.side_effect = json.JSONDecodeError("msg", "doc", 0)
     mocker.patch.object(gitea.retried_requests, "get", return_value=mock_response)
     res = gitea.get_product_version_from_repo_listing("project", "product", "repo")
-    assert not res
+    assert res is None
     assert mock_log.info.called
 
 
@@ -85,7 +85,7 @@ def test_get_product_version_from_repo_listing_http_error(mocker: MockerFixture)
     mock_response.raise_for_status.side_effect = requests.exceptions.HTTPError("error")
     mocker.patch.object(gitea.retried_requests, "get", return_value=mock_response)
     res = gitea.get_product_version_from_repo_listing("project", "product", "repo")
-    assert not res
+    assert res is None
     assert mock_log.warning.called
 
 
@@ -96,7 +96,7 @@ def test_get_product_version_from_repo_listing_request_exception(
     caplog.set_level(logging.WARNING, logger="bot.loader.gitea")
     mocker.patch("openqabot.loader.gitea.retried_requests.get", side_effect=requests.RequestException("error"))
     res = gitea.get_product_version_from_repo_listing("project", "product", "repo")
-    assert not res
+    assert res is None
     assert "Product version unresolved" in caplog.text
 
 
@@ -146,7 +146,7 @@ def test_get_product_version_from_repo_listing_requests_json_error(
     mock_response.json.side_effect = requests.exceptions.JSONDecodeError("msg", "doc", 0)
     mocker.patch("openqabot.loader.gitea.retried_requests.get", return_value=mock_response)
     res = gitea.get_product_version_from_repo_listing("project_json", "product_json", "repo_json")
-    assert not res
+    assert res is None
     assert "Invalid JSON document" in caplog.text
 
 

tests/test_types.py

@@ -4,7 +4,39 @@
 
 import pytest
 
-from openqabot.types.types import ChannelType, ProdVer, Repos, get_channel_type
+from openqabot.types.types import ChannelType, ProductVersion, ProdVer, Repos, get_channel_type
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        "15.4",
+        "15",
+        "0.1.2",
+        "1234.5678",
+    ],
+)
+def test_product_version_valid(value: str) -> None:
+    """Test valid product version strings."""
+    pv = ProductVersion(value)
+    assert pv == value
+    assert isinstance(pv, ProductVersion)
+    assert isinstance(pv, str)
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        "",
+        "a.b",
+        "15.4-SP1",
+        "15_4",
+    ],
+)
+def test_product_version_invalid(value: str) -> None:
+    """Test invalid product version strings."""
+    with pytest.raises(ValueError, match="Invalid product version format:"):
+        ProductVersion(value)
 
 
 @pytest.mark.parametrize(