refactor(cargo_provider): cleanup and get rid of potential type errors

Author: bearomorphism

commitizen/providers/cargo_provider.py

@@ -3,19 +3,15 @@
 import fnmatch
 import glob
 from pathlib import Path
+from typing import TYPE_CHECKING
 
-import tomlkit
+from tomlkit import TOMLDocument, dumps, parse
+from tomlkit.exceptions import NonExistentKey
+from tomlkit.items import AoT
 
 from commitizen.providers.base_provider import TomlProvider
 
 
-def matches_exclude(path: str, exclude_patterns: list[str]) -> bool:
-    for pattern in exclude_patterns:
-        if fnmatch.fnmatch(path, pattern):
-            return True
-    return False
-
-
 class CargoProvider(TomlProvider):
     """
     Cargo version management
@@ -30,65 +26,75 @@ class CargoProvider(TomlProvider):
     def lock_file(self) -> Path:
         return Path() / self.lock_filename
 
-    def get(self, document: tomlkit.TOMLDocument) -> str:
-        # If there is a root package, change its version (but not the workspace version)
-        try:
-            return document["package"]["version"]  # type: ignore[index,return-value]
-        # Else, bump the workspace version
-        except tomlkit.exceptions.NonExistentKey:
-            ...
-        return document["workspace"]["package"]["version"]  # type: ignore[index,return-value]
+    def get(self, document: TOMLDocument) -> str:
+        out = _try_get_workspace(document)["package"]["version"]
+        assert isinstance(out, str)
+        return out
 
-    def set(self, document: tomlkit.TOMLDocument, version: str) -> None:
-        try:
-            document["workspace"]["package"]["version"] = version  # type: ignore[index]
-            return
-        except tomlkit.exceptions.NonExistentKey:
-            ...
-        document["package"]["version"] = version  # type: ignore[index]
+    def set(self, document: TOMLDocument, version: str) -> None:
+        _try_get_workspace(document)["package"]["version"] = version
 
     def set_version(self, version: str) -> None:
         super().set_version(version)
         if self.lock_file.exists():
             self.set_lock_version(version)
 
     def set_lock_version(self, version: str) -> None:
-        cargo_toml_content = tomlkit.parse(self.file.read_text())
-        cargo_lock_content = tomlkit.parse(self.lock_file.read_text())
-        packages: tomlkit.items.AoT = cargo_lock_content["package"]  # type: ignore[assignment]
+        cargo_toml_content = parse(self.file.read_text())
+        cargo_lock_content = parse(self.lock_file.read_text())
+        packages = cargo_lock_content["package"]
+
+        assert isinstance(packages, AoT)
+
         try:
-            package_name = cargo_toml_content["package"]["name"]  # type: ignore[index]
+            cargo_package_name = cargo_toml_content["package"]["name"]  # type: ignore[index]
+            if TYPE_CHECKING:
+                assert isinstance(cargo_package_name, str)
             for i, package in enumerate(packages):
-                if package["name"] == package_name:
+                if package["name"] == cargo_package_name:
                     cargo_lock_content["package"][i]["version"] = version  # type: ignore[index]
                     break
-        except tomlkit.exceptions.NonExistentKey:
-            workspace_members = cargo_toml_content.get("workspace", {}).get(
-                "members", []
-            )
-            excluded_workspace_members = cargo_toml_content.get("workspace", {}).get(
-                "exclude", []
-            )
-            members_inheriting = []
+        except NonExistentKey:
+            workspace = cargo_toml_content.get("workspace", {})
+            assert isinstance(workspace, dict)
+            workspace_members = workspace.get("members", [])
+            excluded_workspace_members = workspace.get("exclude", [])
+            members_inheriting: list[str] = []
 
             for member in workspace_members:
                 for path in glob.glob(member, recursive=True):
-                    if matches_exclude(path, excluded_workspace_members):
+                    if any(
+                        fnmatch.fnmatch(path, pattern)
+                        for pattern in excluded_workspace_members
+                    ):
                         continue
+
                     cargo_file = Path(path) / "Cargo.toml"
-                    cargo_toml_content = tomlkit.parse(cargo_file.read_text())
+                    package_content = parse(cargo_file.read_text()).get("package", {})
+                    if TYPE_CHECKING:
+                        assert isinstance(package_content, dict)
                     try:
-                        version_workspace = cargo_toml_content["package"]["version"][  # type: ignore[index]
-                            "workspace"
-                        ]
+                        version_workspace = package_content["version"]["workspace"]
                         if version_workspace is True:
-                            package_name = cargo_toml_content["package"]["name"]  # type: ignore[index]
+                            package_name = package_content["name"]
+                            if TYPE_CHECKING:
+                                assert isinstance(package_name, str)
                             members_inheriting.append(package_name)
-                    except tomlkit.exceptions.NonExistentKey:
-                        continue
+                    except NonExistentKey:
+                        pass
 
             for i, package in enumerate(packages):
                 if package["name"] in members_inheriting:
                     cargo_lock_content["package"][i]["version"] = version  # type: ignore[index]
 
-        self.lock_file.write_text(tomlkit.dumps(cargo_lock_content))
+        self.lock_file.write_text(dumps(cargo_lock_content))
+
+
+def _try_get_workspace(document: TOMLDocument) -> dict:
+    try:
+        workspace = document["workspace"]
+        if TYPE_CHECKING:
+            assert isinstance(workspace, dict)
+        return workspace
+    except NonExistentKey:
+        return document

tests/providers/test_cargo_provider.py

@@ -301,3 +301,150 @@ def test_cargo_provider_with_lock(
     provider.set_version("42.1")
     assert file.read_text() == dedent(toml_expected)
     assert lock_file.read_text() == dedent(lock_expected)
+
+
+def test_cargo_provider_workspace_member_without_version_key(
+    config: BaseConfig,
+    chdir: Path,
+):
+    """Test workspace member that has no version key at all (should not crash)."""
+    workspace_toml = """\
+[workspace]
+members = ["member_without_version"]
+
+[workspace.package]
+version = "0.1.0"
+"""
+
+    # Create a member that has no version key at all
+    member_content = """\
+[package]
+name = "member_without_version"
+# No version key - this should trigger NonExistentKey exception
+"""
+
+    lock_content = """\
+[[package]]
+name = "member_without_version"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "123abc"
+"""
+
+    expected_workspace_toml = """\
+[workspace]
+members = ["member_without_version"]
+
+[workspace.package]
+version = "42.1"
+"""
+
+    expected_lock_content = """\
+[[package]]
+name = "member_without_version"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "123abc"
+"""
+
+    # Create the workspace file
+    filename = CargoProvider.filename
+    file = chdir / filename
+    file.write_text(dedent(workspace_toml))
+
+    # Create the member directory and file
+    os.mkdir(chdir / "member_without_version")
+    member_file = chdir / "member_without_version" / "Cargo.toml"
+    member_file.write_text(dedent(member_content))
+
+    # Create the lock file
+    lock_filename = CargoProvider.lock_filename
+    lock_file = chdir / lock_filename
+    lock_file.write_text(dedent(lock_content))
+
+    config.settings["version_provider"] = "cargo"
+
+    provider = get_provider(config)
+    assert isinstance(provider, CargoProvider)
+    assert provider.get_version() == "0.1.0"
+
+    # This should not crash even though the member has no version key
+    provider.set_version("42.1")
+    assert file.read_text() == dedent(expected_workspace_toml)
+    # The lock file should remain unchanged since the member doesn't inherit workspace version
+    assert lock_file.read_text() == dedent(expected_lock_content)
+
+
+def test_cargo_provider_workspace_member_without_workspace_key(
+    config: BaseConfig,
+    chdir: Path,
+):
+    """Test workspace member that has version key but no workspace subkey."""
+    workspace_toml = """\
+[workspace]
+members = ["member_without_workspace"]
+
+[workspace.package]
+version = "0.1.0"
+"""
+
+    # Create a member that has version as a table but no workspace subkey
+    # This should trigger NonExistentKey when trying to access version["workspace"]
+    member_content = """\
+[package]
+name = "member_without_workspace"
+
+[package.version]
+# Has version table but no workspace key - should trigger NonExistentKey
+"""
+
+    lock_content = """\
+[[package]]
+name = "member_without_workspace"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "123abc"
+"""
+
+    expected_workspace_toml = """\
+[workspace]
+members = ["member_without_workspace"]
+
+[workspace.package]
+version = "42.1"
+"""
+
+    expected_lock_content = """\
+[[package]]
+name = "member_without_workspace"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "123abc"
+"""
+
+    # Create the workspace file
+    filename = CargoProvider.filename
+    file = chdir / filename
+    file.write_text(dedent(workspace_toml))
+
+    # Create the member directory and file
+    os.mkdir(chdir / "member_without_workspace")
+    member_file = chdir / "member_without_workspace" / "Cargo.toml"
+    member_file.write_text(dedent(member_content))
+
+    # Create the lock file
+    lock_filename = CargoProvider.lock_filename
+    lock_file = chdir / lock_filename
+    lock_file.write_text(dedent(lock_content))
+
+    config.settings["version_provider"] = "cargo"
+
+    provider = get_provider(config)
+    assert isinstance(provider, CargoProvider)
+    assert provider.get_version() == "0.1.0"
+
+    # This should not crash even though the member has no version.workspace key
+    provider.set_version("42.1")
+    assert file.read_text() == dedent(expected_workspace_toml)
+    # The lock file should remain unchanged since the member doesn't inherit workspace version
+    assert lock_file.read_text() == dedent(expected_lock_content)