From 5965a7f7f1529722c320ec6ea7df49730df2fba8 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Fri, 17 Jul 2026 08:53:05 +0200 Subject: [PATCH 1/5] Replace tracking method string constants with a `TrackingMethod` enum We introduce `TrackingMethod(str, Enum)` with members VERSION, BRANCH, COMMIT, BUILTIN, and DIRECTORY. Using `str` as a mixin preserves JSON round-tripping through the manifest without a custom encoder. `PackageStatus.__init__` coerces plain strings to `TrackingMethod` so existing manifests load transparently. The if/elif dispatch chains in `upgrade()` and `_is_clone_outdated()` become `match` statements with an exhaustive `case _: raise NotImplementedError` arm. --- testing/test_manager.py | 44 +++++++++----------- zeekpkg/manager.py | 91 ++++++++++++++++++++--------------------- zeekpkg/package.py | 54 ++++++++++++++---------- zkg | 6 +-- 4 files changed, 101 insertions(+), 94 deletions(-) diff --git a/testing/test_manager.py b/testing/test_manager.py index a54129f..25294a6 100644 --- a/testing/test_manager.py +++ b/testing/test_manager.py @@ -18,16 +18,12 @@ _snapshot_from_git_repo, ) from zeekpkg.package import ( - TRACKING_METHOD_BRANCH, - TRACKING_METHOD_BUILTIN, - TRACKING_METHOD_COMMIT, - TRACKING_METHOD_DIRECTORY, - TRACKING_METHOD_VERSION, InstalledPackage, Package, PackageInfo, PackageSnapshot, PackageStatus, + TrackingMethod, ) @@ -59,7 +55,7 @@ def repo(tmp_path: pathlib.Path) -> git.Repo: def test_defaults_to_branch_when_no_tags(repo: git.Repo) -> None: resolution = _resolve_git_version(repo, "") assert isinstance(resolution, GitResolution) - assert resolution.tracking_method == TRACKING_METHOD_BRANCH + assert resolution.tracking_method == TrackingMethod.BRANCH assert resolution.version == "main" @@ -67,7 +63,7 @@ def test_defaults_to_latest_tag(repo: git.Repo) -> None: repo.create_tag("v1.0.0") repo.create_tag("v2.0.0") resolution = _resolve_git_version(repo, "") - assert resolution.tracking_method == TRACKING_METHOD_VERSION + assert resolution.tracking_method == TrackingMethod.VERSION assert resolution.version == "v2.0.0" @@ -75,7 +71,7 @@ def test_explicit_version_tag(repo: git.Repo) -> None: repo.create_tag("v1.0.0") repo.create_tag("v2.0.0") resolution = _resolve_git_version(repo, "v1.0.0") - assert resolution.tracking_method == TRACKING_METHOD_VERSION + assert resolution.tracking_method == TrackingMethod.VERSION assert resolution.version == "v1.0.0" @@ -85,14 +81,14 @@ def test_explicit_branch(repo: git.Repo) -> None: repo.remotes.origin.fetch() repo.git.checkout("feature") resolution = _resolve_git_version(repo, "feature") - assert resolution.tracking_method == TRACKING_METHOD_BRANCH + assert resolution.tracking_method == TrackingMethod.BRANCH assert resolution.version == "feature" def test_explicit_commit_hash(repo: git.Repo) -> None: hexsha = repo.head.object.hexsha resolution = _resolve_git_version(repo, hexsha) - assert resolution.tracking_method == TRACKING_METHOD_COMMIT + assert resolution.tracking_method == TrackingMethod.COMMIT assert resolution.current_hash == hexsha @@ -231,11 +227,11 @@ def _make_installed( @pytest.mark.parametrize( "method,expected", [ - (TRACKING_METHOD_VERSION, True), - (TRACKING_METHOD_BRANCH, True), - (TRACKING_METHOD_COMMIT, True), - (TRACKING_METHOD_BUILTIN, False), - (TRACKING_METHOD_DIRECTORY, False), + (TrackingMethod.VERSION, True), + (TrackingMethod.BRANCH, True), + (TrackingMethod.COMMIT, True), + (TrackingMethod.BUILTIN, False), + (TrackingMethod.DIRECTORY, False), (None, False), ], ) @@ -278,7 +274,7 @@ def test_prepare_snapshot_directory( package = Package(git_url=str(pkg_dir), canonical=True) snapshot = _prepare_snapshot(package, None, str(tmp_path / "dest")) assert snapshot.version == "1.0.0" - assert snapshot.tracking_method == TRACKING_METHOD_DIRECTORY + assert snapshot.tracking_method == TrackingMethod.DIRECTORY assert snapshot.current_hash is None assert snapshot.is_outdated is False @@ -298,7 +294,7 @@ def test_prepare_snapshot_git(repo: git.Repo, tmp_path: pathlib.Path) -> None: str(tmp_path / "dest"), existing_clone=repo, ) - assert snapshot.tracking_method == TRACKING_METHOD_BRANCH + assert snapshot.tracking_method == TrackingMethod.BRANCH assert snapshot.current_hash is not None @@ -307,7 +303,7 @@ def test_info_directory_backend(manager: Manager, pkg_dir: pathlib.Path) -> None info = manager.info(str(pkg_dir)) assert info.invalid_reason == "" assert info.metadata_version == "1.0.0" - assert info.version_type == TRACKING_METHOD_DIRECTORY + assert info.version_type == TrackingMethod.DIRECTORY def test_install_directory_backend(manager: Manager, pkg_dir: pathlib.Path) -> None: @@ -315,7 +311,7 @@ def test_install_directory_backend(manager: Manager, pkg_dir: pathlib.Path) -> N assert result == "" ipkg = manager.find_installed_package("mypkg") assert ipkg is not None - assert ipkg.status.tracking_method == TRACKING_METHOD_DIRECTORY + assert ipkg.status.tracking_method == TrackingMethod.DIRECTORY assert ipkg.status.current_version == "1.0.0" @@ -348,13 +344,13 @@ def test_info_from_snapshot(repo: git.Repo) -> None: status=None, versions=["v1.0.0"], default_branch="main", - version_type=TRACKING_METHOD_VERSION, + version_type=TrackingMethod.VERSION, ) assert isinstance(info, PackageInfo) assert info.metadata["description"] == "hello" assert info.versions == ["v1.0.0"] assert info.default_branch == "main" - assert info.version_type == TRACKING_METHOD_VERSION + assert info.version_type == TrackingMethod.VERSION assert info.invalid_reason == "" @@ -372,7 +368,7 @@ def test_info_installed_missing_metadata(manager: Manager) -> None: _make_installed( manager, pkg_name, - tracking_method=TRACKING_METHOD_BRANCH, + tracking_method=TrackingMethod.BRANCH, current_version="main", ) info = manager.info(f"https://example.com/{pkg_name}", prefer_installed=True) @@ -385,7 +381,7 @@ def test_refresh_skips_non_git_packages(manager: Manager) -> None: _make_installed( manager, "mypkg", - tracking_method=TRACKING_METHOD_DIRECTORY, + tracking_method=TrackingMethod.DIRECTORY, current_version="1.0.0", ) # Should complete without raising. @@ -484,7 +480,7 @@ def test_bundle_skips_non_git_existing_clone( _make_installed( manager, "mypkg", - tracking_method=TRACKING_METHOD_DIRECTORY, + tracking_method=TrackingMethod.DIRECTORY, current_version="1.0.0", ) bundle_file = str(tmp_path / "out.tar.gz") diff --git a/zeekpkg/manager.py b/zeekpkg/manager.py index 35f4481..86b6889 100644 --- a/zeekpkg/manager.py +++ b/zeekpkg/manager.py @@ -55,16 +55,13 @@ METADATA_FILENAME, PLUGIN_MAGIC_FILE, PLUGIN_MAGIC_FILE_DISABLED, - TRACKING_METHOD_BRANCH, - TRACKING_METHOD_COMMIT, - TRACKING_METHOD_DIRECTORY, - TRACKING_METHOD_VERSION, InstalledPackage, Package, PackageInfo, PackageSnapshot, PackageStatus, PackageVersion, + TrackingMethod, aliases, canonical_url, make_builtin_package, @@ -1354,21 +1351,20 @@ def upgrade(self, pkg_path: str) -> str: clone = self._open_package_clone(ipkg.package) - if ipkg.status.tracking_method == TRACKING_METHOD_VERSION: - version_tags = git_version_tags(clone) - return self._install(ipkg.package, version_tags[-1]) - - if ipkg.status.tracking_method == TRACKING_METHOD_BRANCH: - git_pull(clone) - assert ipkg.status.current_version - return self._install(ipkg.package, ipkg.status.current_version) - - if ipkg.status.tracking_method == TRACKING_METHOD_COMMIT: - # The above check for whether the installed package is outdated - # also should have already caught this situation. - return "package is not outdated" - - raise NotImplementedError + match ipkg.status.tracking_method: + case TrackingMethod.VERSION: + version_tags = git_version_tags(clone) + return self._install(ipkg.package, version_tags[-1]) + case TrackingMethod.BRANCH: + git_pull(clone) + assert ipkg.status.current_version + return self._install(ipkg.package, ipkg.status.current_version) + case TrackingMethod.COMMIT: + # The above check for whether the installed package is outdated + # also should have already caught this situation. + return "package is not outdated" + case _: + raise NotImplementedError def remove(self, pkg_path: str) -> bool: """Remove an installed package. @@ -2032,7 +2028,7 @@ def _info( status, versions=[], default_branch="", - version_type=TRACKING_METHOD_DIRECTORY, + version_type=TrackingMethod.DIRECTORY, ) def package_versions(self, installed_package: InstalledPackage) -> list[str]: @@ -2241,7 +2237,7 @@ def __str__(self) -> str: if zeek_version: node = Node("zeek") node.installed_version = PackageVersion( - TRACKING_METHOD_VERSION, + TrackingMethod.VERSION, zeek_version, ) graph["zeek"] = node @@ -2250,7 +2246,7 @@ def __str__(self) -> str: node = Node("zkg") node.installed_version = PackageVersion( - TRACKING_METHOD_VERSION, + TrackingMethod.VERSION, __version__, ) graph["zkg"] = node @@ -3227,7 +3223,7 @@ def _install( # `version` field, if present, matches the Git tag. meta_version = snapshot.meta.get("version") if ( - snapshot.tracking_method == TRACKING_METHOD_VERSION + snapshot.tracking_method == TrackingMethod.VERSION and meta_version and meta_version != snapshot.version ): @@ -3346,12 +3342,12 @@ class GitResolution: """The resolved Git state for a package after checkout.""" version: str - tracking_method: str + tracking_method: TrackingMethod current_hash: str is_outdated: bool -def _pick_version(clone: git.Repo, version: str | None) -> tuple[str, str]: +def _pick_version(clone: git.Repo, version: str | None) -> tuple[str, TrackingMethod]: """Return (resolved_version, tracking_method) for *version* in *clone*. When *version* is ``None`` or empty, the best available version is chosen @@ -3365,12 +3361,12 @@ def _pick_version(clone: git.Repo, version: str | None) -> tuple[str, str]: if version: if _is_commit_hash(clone, version): - return version, TRACKING_METHOD_COMMIT + return version, TrackingMethod.COMMIT if version in version_tags: - return version, TRACKING_METHOD_VERSION + return version, TrackingMethod.VERSION branches = _get_branch_names(clone) if version in branches: - return version, TRACKING_METHOD_BRANCH + return version, TrackingMethod.BRANCH LOG.info( 'branch "%s" not in available branches: %s', version, @@ -3379,8 +3375,8 @@ def _pick_version(clone: git.Repo, version: str | None) -> tuple[str, str]: raise ValueError(f'no such branch or version tag: "{version}"') if version_tags: - return version_tags[-1], TRACKING_METHOD_VERSION - return git_default_branch(clone), TRACKING_METHOD_BRANCH + return version_tags[-1], TrackingMethod.VERSION + return git_default_branch(clone), TrackingMethod.BRANCH def _resolve_git_version(clone: git.Repo, version: str | None) -> GitResolution: @@ -3484,7 +3480,7 @@ def _snapshot_from_directory(path: str) -> PackageSnapshot: working_dir=path, meta=meta, version=version, - tracking_method=TRACKING_METHOD_DIRECTORY, + tracking_method=TrackingMethod.DIRECTORY, ) @@ -3514,9 +3510,9 @@ def _is_git_package(status: PackageStatus) -> bool: tracking method; this predicate guards operations that require a git clone. """ return status.tracking_method in ( - TRACKING_METHOD_VERSION, - TRACKING_METHOD_BRANCH, - TRACKING_METHOD_COMMIT, + TrackingMethod.VERSION, + TrackingMethod.BRANCH, + TrackingMethod.COMMIT, ) @@ -3532,17 +3528,20 @@ def _is_branch_outdated(clone: git.Repo, branch: str) -> bool: return num_commits_behind > 0 -def _is_clone_outdated(clone: git.Repo, ref_name: str, tracking_method: str) -> bool: - if tracking_method == TRACKING_METHOD_VERSION: - return _is_version_outdated(clone, ref_name) - - if tracking_method == TRACKING_METHOD_BRANCH: - return _is_branch_outdated(clone, ref_name) - - if tracking_method == TRACKING_METHOD_COMMIT: - return False - - raise NotImplementedError +def _is_clone_outdated( + clone: git.Repo, + ref_name: str, + tracking_method: TrackingMethod, +) -> bool: + match tracking_method: + case TrackingMethod.VERSION: + return _is_version_outdated(clone, ref_name) + case TrackingMethod.BRANCH: + return _is_branch_outdated(clone, ref_name) + case TrackingMethod.COMMIT: + return False + case _: + raise NotImplementedError def _is_commit_hash(clone: git.Repo, text: str) -> bool: @@ -3695,7 +3694,7 @@ def _info_from_snapshot( status: PackageStatus | None, versions: list[str], default_branch: str, - version_type: str, + version_type: TrackingMethod | None, ) -> PackageInfo: """Build a :class:`.package.PackageInfo` from a :class:`.package.PackageSnapshot`. diff --git a/zeekpkg/package.py b/zeekpkg/package.py index b929080..8d52684 100644 --- a/zeekpkg/package.py +++ b/zeekpkg/package.py @@ -6,6 +6,7 @@ import os import re from dataclasses import dataclass, field +from enum import Enum from functools import total_ordering import semantic_version as semver @@ -17,11 +18,19 @@ METADATA_FILENAME = "zkg.meta" LEGACY_METADATA_FILENAME = "bro-pkg.meta" -TRACKING_METHOD_VERSION = "version" -TRACKING_METHOD_BRANCH = "branch" -TRACKING_METHOD_COMMIT = "commit" -TRACKING_METHOD_BUILTIN = "builtin" -TRACKING_METHOD_DIRECTORY = "directory" + +class TrackingMethod(str, Enum): + """How a package tracks upstream changes.""" + + VERSION = "version" + BRANCH = "branch" + COMMIT = "commit" + BUILTIN = "builtin" + DIRECTORY = "directory" + + def __str__(self) -> str: + return self.value + BUILTIN_SOURCE = "zeek-builtin" BUILTIN_SCHEME = "zeek-builtin://" @@ -170,8 +179,8 @@ class PackageSnapshot: working_dir: path to the package directory meta: parsed contents of the ``zkg.meta`` file version: resolved version string, or ``None`` if unavailable - tracking_method: how the package version is tracked (e.g. - ``TRACKING_METHOD_VERSION``); ``None`` if not yet determined + tracking_method: how the package version is tracked; ``None`` if not + yet determined current_hash: Git commit hash at the time of snapshot, or ``None`` for non-Git sources is_outdated: whether a newer version is available; always ``False`` @@ -181,7 +190,7 @@ class PackageSnapshot: working_dir: str meta: dict[str, str] = field(default_factory=dict) version: str | None = None - tracking_method: str | None = None + tracking_method: "TrackingMethod | None" = None current_hash: str | None = None is_outdated: bool = False @@ -191,7 +200,7 @@ class PackageVersion: Helper class to compare package versions with version specs. """ - def __init__(self, method: str | None, version: str | None) -> None: + def __init__(self, method: "TrackingMethod | None", version: str | None) -> None: self.method = method self.version = version self.req_semver = None @@ -206,13 +215,13 @@ def fullfills(self, version_spec: str) -> tuple[str, bool]: if version_spec == "*": # anything goes return "", True - if self.method == TRACKING_METHOD_COMMIT: + if self.method == TrackingMethod.COMMIT: return f'tracking method commit not compatible with "{version_spec}"', False - if self.method == TRACKING_METHOD_BRANCH: + if self.method == TrackingMethod.BRANCH: return "tracking method branch and commit", False - # TRACKING_METHOD_BRANCH / TRACKING_METHOD_BUILTIN + # TrackingMethod.BRANCH / TrackingMethod.BUILTIN if version_spec.startswith("branch="): branch = version_spec[len("branch=") :] return ( @@ -292,8 +301,7 @@ class PackageStatus: is_outdated (bool): whether a newer version of the package exists. - tracking_method (str): either "branch", "version", "commit", or - "builtin" to indicate (respectively) whether package upgrades + tracking_method (TrackingMethod): indicates whether package upgrades should stick to a git branch, use git version tags, do nothing because the package is to always use a specific git commit hash, or do nothing because the package is built into Zeek. @@ -310,14 +318,18 @@ def __init__( is_loaded: bool = False, is_pinned: bool = False, is_outdated: bool = False, - tracking_method: str | None = None, + tracking_method: "TrackingMethod | str | None" = None, current_version: str | None = None, current_hash: str | None = None, ) -> None: self.is_loaded = is_loaded self.is_pinned = is_pinned self.is_outdated = is_outdated - self.tracking_method = tracking_method + self.tracking_method = ( + TrackingMethod(tracking_method) + if isinstance(tracking_method, str) + else tracking_method + ) self.current_version = current_version self.current_hash = current_hash @@ -345,9 +357,9 @@ class PackageInfo: metadata_version: the package version that the metadata is from - version_type: either 'version', 'branch', or 'commit' to - indicate whether the package info/metadata was taken from a release - version tag, a branch, or a specific commit hash. + version_type: a :class:`TrackingMethod` indicating whether the package + info/metadata was taken from a release version tag, a branch, or a + specific commit hash. invalid_reason (str): this attribute is set when there is a problem with gathering package information and explains what went wrong. @@ -366,7 +378,7 @@ def __init__( versions: list[str] | None = None, metadata_version: str | None = "", invalid_reason: str = "", - version_type: str = "", + version_type: "TrackingMethod | None" = None, metadata_file: str | None = None, default_branch: str | None = None, ) -> None: @@ -648,7 +660,7 @@ def make_builtin_package( is_loaded=True, # May not hold in the future? is_outdated=False, is_pinned=True, - tracking_method=TRACKING_METHOD_BUILTIN, + tracking_method=TrackingMethod.BUILTIN, current_version=current_version, current_hash=current_hash, ) diff --git a/zkg b/zkg index e6fb50b..b9ab309 100755 --- a/zkg +++ b/zkg @@ -38,10 +38,10 @@ from zeekpkg._util import ( ) from zeekpkg.package import ( BUILTIN_SCHEME, - TRACKING_METHOD_VERSION, InstalledPackage, Package, PackageInfo, + TrackingMethod, ) from zeekpkg.template import ( InputError, @@ -1389,7 +1389,7 @@ def version_change_string( new_version = old_version version_change = "" - if installed_package.status.tracking_method == TRACKING_METHOD_VERSION: + if installed_package.status.tracking_method == TrackingMethod.VERSION: versions = manager.package_versions(installed_package) if len(versions): @@ -1443,7 +1443,7 @@ def cmd_upgrade( next_version = ipkg.status.current_version - if ipkg.status.tracking_method == TRACKING_METHOD_VERSION and info.versions: + if ipkg.status.tracking_method == TrackingMethod.VERSION and info.versions: next_version = info.versions[-1] assert next_version From 225a3cfffd007b78645ddd5e80433f41b6304c84 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Fri, 17 Jul 2026 09:01:52 +0200 Subject: [PATCH 2/5] Narrow tracking method types to enforce Git-only dispatch statically We introduce a `_GitTrackingMethod` type alias (`Literal[VERSION, BRANCH, COMMIT]`) and a `TypeGuard`-returning `_is_git_tracking_method` predicate. `GitResolution`, `_pick_version`, and `_is_clone_outdated` are updated to use this narrower type, so mypy can verify that BUILTIN and DIRECTORY never reach the Git-only `match` statements -- making `case _:` arms unnecessary. We add `typing-extensions` as a runtime dependency for `assert_never` and `TypeGuard`, both of which are available in stdlib only from Python 3.11 onward. --- pyproject.toml | 1 + testing/test_package.py | 23 +++++++++++++++++++++++ zeekpkg/manager.py | 40 +++++++++++++++++++++++++++------------- 3 files changed, 51 insertions(+), 13 deletions(-) create mode 100644 testing/test_package.py diff --git a/pyproject.toml b/pyproject.toml index 3e5dea4..ae7616a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,7 @@ dependencies = [ # Technically not a zkg dependency, but typically expected by users to be present. "btest>=1.1", "argcomplete>=3.6.3", + "typing-extensions>=4.0.0", ] [project.optional-dependencies] diff --git a/testing/test_package.py b/testing/test_package.py new file mode 100644 index 0000000..95178bb --- /dev/null +++ b/testing/test_package.py @@ -0,0 +1,23 @@ +"""Unit tests for zeekpkg.package data structures.""" + +import pytest + +from zeekpkg.package import PackageStatus, TrackingMethod + + +@pytest.mark.parametrize("value", [m.value for m in TrackingMethod]) +def test_package_status_coerces_string_tracking_method(value: str) -> None: + # Manifest JSON deserialises tracking_method as a plain string; verify + # PackageStatus converts it to the enum. + status = PackageStatus(tracking_method=value) + assert status.tracking_method is TrackingMethod(value) + + +def test_package_status_accepts_enum_tracking_method() -> None: + status = PackageStatus(tracking_method=TrackingMethod.BRANCH) + assert status.tracking_method is TrackingMethod.BRANCH + + +def test_package_status_rejects_invalid_tracking_method() -> None: + with pytest.raises(ValueError): + PackageStatus(tracking_method="bogus") diff --git a/zeekpkg/manager.py b/zeekpkg/manager.py index 86b6889..525eefc 100644 --- a/zeekpkg/manager.py +++ b/zeekpkg/manager.py @@ -17,6 +17,7 @@ import time from collections import deque from dataclasses import dataclass +from typing import Literal, TypeGuard from urllib.parse import urlparse import git @@ -1307,7 +1308,7 @@ def refresh_installed_packages(self) -> None: ) assert ipkg.status.current_version - assert ipkg.status.tracking_method + assert _is_git_tracking_method(ipkg.status.tracking_method) ipkg.status.is_outdated = _is_clone_outdated( clone, ipkg.status.current_version, @@ -1351,6 +1352,7 @@ def upgrade(self, pkg_path: str) -> str: clone = self._open_package_clone(ipkg.package) + assert _is_git_tracking_method(ipkg.status.tracking_method) match ipkg.status.tracking_method: case TrackingMethod.VERSION: version_tags = git_version_tags(clone) @@ -1363,8 +1365,6 @@ def upgrade(self, pkg_path: str) -> str: # The above check for whether the installed package is outdated # also should have already caught this situation. return "package is not outdated" - case _: - raise NotImplementedError def remove(self, pkg_path: str) -> bool: """Remove an installed package. @@ -3337,17 +3337,37 @@ def _clear_bin_dir(self, bin_dir: str) -> None: LOG.warning("failed to remove link %s", old) +_GitTrackingMethod = Literal[ + TrackingMethod.VERSION, + TrackingMethod.BRANCH, + TrackingMethod.COMMIT, +] + + +def _is_git_tracking_method( + method: TrackingMethod | None, +) -> TypeGuard[_GitTrackingMethod]: + return method in ( + TrackingMethod.VERSION, + TrackingMethod.BRANCH, + TrackingMethod.COMMIT, + ) + + @dataclass class GitResolution: """The resolved Git state for a package after checkout.""" version: str - tracking_method: TrackingMethod + tracking_method: _GitTrackingMethod current_hash: str is_outdated: bool -def _pick_version(clone: git.Repo, version: str | None) -> tuple[str, TrackingMethod]: +def _pick_version( + clone: git.Repo, + version: str | None, +) -> tuple[str, _GitTrackingMethod]: """Return (resolved_version, tracking_method) for *version* in *clone*. When *version* is ``None`` or empty, the best available version is chosen @@ -3509,11 +3529,7 @@ def _is_git_package(status: PackageStatus) -> bool: Non-git package sources (e.g. local directories) will use a different tracking method; this predicate guards operations that require a git clone. """ - return status.tracking_method in ( - TrackingMethod.VERSION, - TrackingMethod.BRANCH, - TrackingMethod.COMMIT, - ) + return _is_git_tracking_method(status.tracking_method) def _is_version_outdated(clone: git.Repo, version: str) -> bool: @@ -3531,7 +3547,7 @@ def _is_branch_outdated(clone: git.Repo, branch: str) -> bool: def _is_clone_outdated( clone: git.Repo, ref_name: str, - tracking_method: TrackingMethod, + tracking_method: _GitTrackingMethod, ) -> bool: match tracking_method: case TrackingMethod.VERSION: @@ -3540,8 +3556,6 @@ def _is_clone_outdated( return _is_branch_outdated(clone, ref_name) case TrackingMethod.COMMIT: return False - case _: - raise NotImplementedError def _is_commit_hash(clone: git.Repo, text: str) -> bool: From fdbcf9b3db5d2a3892c9fe92787cd48ac116f285 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Fri, 17 Jul 2026 09:38:24 +0200 Subject: [PATCH 3/5] Revert "Narrow tracking method types to enforce Git-only dispatch statically" This reverts commit 6cbbc43ce4599642910e8af6171b39018151e29b. --- pyproject.toml | 1 - testing/test_package.py | 23 ----------------------- zeekpkg/manager.py | 40 +++++++++++++--------------------------- 3 files changed, 13 insertions(+), 51 deletions(-) delete mode 100644 testing/test_package.py diff --git a/pyproject.toml b/pyproject.toml index ae7616a..3e5dea4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,7 +39,6 @@ dependencies = [ # Technically not a zkg dependency, but typically expected by users to be present. "btest>=1.1", "argcomplete>=3.6.3", - "typing-extensions>=4.0.0", ] [project.optional-dependencies] diff --git a/testing/test_package.py b/testing/test_package.py deleted file mode 100644 index 95178bb..0000000 --- a/testing/test_package.py +++ /dev/null @@ -1,23 +0,0 @@ -"""Unit tests for zeekpkg.package data structures.""" - -import pytest - -from zeekpkg.package import PackageStatus, TrackingMethod - - -@pytest.mark.parametrize("value", [m.value for m in TrackingMethod]) -def test_package_status_coerces_string_tracking_method(value: str) -> None: - # Manifest JSON deserialises tracking_method as a plain string; verify - # PackageStatus converts it to the enum. - status = PackageStatus(tracking_method=value) - assert status.tracking_method is TrackingMethod(value) - - -def test_package_status_accepts_enum_tracking_method() -> None: - status = PackageStatus(tracking_method=TrackingMethod.BRANCH) - assert status.tracking_method is TrackingMethod.BRANCH - - -def test_package_status_rejects_invalid_tracking_method() -> None: - with pytest.raises(ValueError): - PackageStatus(tracking_method="bogus") diff --git a/zeekpkg/manager.py b/zeekpkg/manager.py index 525eefc..86b6889 100644 --- a/zeekpkg/manager.py +++ b/zeekpkg/manager.py @@ -17,7 +17,6 @@ import time from collections import deque from dataclasses import dataclass -from typing import Literal, TypeGuard from urllib.parse import urlparse import git @@ -1308,7 +1307,7 @@ def refresh_installed_packages(self) -> None: ) assert ipkg.status.current_version - assert _is_git_tracking_method(ipkg.status.tracking_method) + assert ipkg.status.tracking_method ipkg.status.is_outdated = _is_clone_outdated( clone, ipkg.status.current_version, @@ -1352,7 +1351,6 @@ def upgrade(self, pkg_path: str) -> str: clone = self._open_package_clone(ipkg.package) - assert _is_git_tracking_method(ipkg.status.tracking_method) match ipkg.status.tracking_method: case TrackingMethod.VERSION: version_tags = git_version_tags(clone) @@ -1365,6 +1363,8 @@ def upgrade(self, pkg_path: str) -> str: # The above check for whether the installed package is outdated # also should have already caught this situation. return "package is not outdated" + case _: + raise NotImplementedError def remove(self, pkg_path: str) -> bool: """Remove an installed package. @@ -3337,37 +3337,17 @@ def _clear_bin_dir(self, bin_dir: str) -> None: LOG.warning("failed to remove link %s", old) -_GitTrackingMethod = Literal[ - TrackingMethod.VERSION, - TrackingMethod.BRANCH, - TrackingMethod.COMMIT, -] - - -def _is_git_tracking_method( - method: TrackingMethod | None, -) -> TypeGuard[_GitTrackingMethod]: - return method in ( - TrackingMethod.VERSION, - TrackingMethod.BRANCH, - TrackingMethod.COMMIT, - ) - - @dataclass class GitResolution: """The resolved Git state for a package after checkout.""" version: str - tracking_method: _GitTrackingMethod + tracking_method: TrackingMethod current_hash: str is_outdated: bool -def _pick_version( - clone: git.Repo, - version: str | None, -) -> tuple[str, _GitTrackingMethod]: +def _pick_version(clone: git.Repo, version: str | None) -> tuple[str, TrackingMethod]: """Return (resolved_version, tracking_method) for *version* in *clone*. When *version* is ``None`` or empty, the best available version is chosen @@ -3529,7 +3509,11 @@ def _is_git_package(status: PackageStatus) -> bool: Non-git package sources (e.g. local directories) will use a different tracking method; this predicate guards operations that require a git clone. """ - return _is_git_tracking_method(status.tracking_method) + return status.tracking_method in ( + TrackingMethod.VERSION, + TrackingMethod.BRANCH, + TrackingMethod.COMMIT, + ) def _is_version_outdated(clone: git.Repo, version: str) -> bool: @@ -3547,7 +3531,7 @@ def _is_branch_outdated(clone: git.Repo, branch: str) -> bool: def _is_clone_outdated( clone: git.Repo, ref_name: str, - tracking_method: _GitTrackingMethod, + tracking_method: TrackingMethod, ) -> bool: match tracking_method: case TrackingMethod.VERSION: @@ -3556,6 +3540,8 @@ def _is_clone_outdated( return _is_branch_outdated(clone, ref_name) case TrackingMethod.COMMIT: return False + case _: + raise NotImplementedError def _is_commit_hash(clone: git.Repo, text: str) -> bool: From 3f23882ad8c30491268fdbf96873102a05f4b122 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Fri, 17 Jul 2026 09:22:29 +0200 Subject: [PATCH 4/5] Restrict `TrackingMethod` to Git-backed tracking strategies BUILTIN and DIRECTORY were not tracking strategies -- they indicated the source type of a package. We remove them from the enum so `TrackingMethod` describes only how a Git clone is followed (VERSION, BRANCH, COMMIT), and non-Git packages express their state as `tracking_method=None`. Older manifests that contain "builtin" or "directory" as string values are handled in `PackageStatus.__init__`, which maps them to `None` for backward compatibility. --- testing/baselines/tests.builtin-info/out | 2 +- testing/test_manager.py | 12 ++++----- testing/test_package.py | 31 ++++++++++++++++++++++++ zeekpkg/manager.py | 15 +++++------- zeekpkg/package.py | 15 +++++++----- 5 files changed, 52 insertions(+), 23 deletions(-) create mode 100644 testing/test_package.py diff --git a/testing/baselines/tests.builtin-info/out b/testing/baselines/tests.builtin-info/out index efe9c5b..23d2ad5 100644 --- a/testing/baselines/tests.builtin-info/out +++ b/testing/baselines/tests.builtin-info/out @@ -8,7 +8,7 @@ is_loaded = True is_outdated = False is_pinned = True - tracking_method = builtin + tracking_method = None metadata (from version ""): diff --git a/testing/test_manager.py b/testing/test_manager.py index 25294a6..ae0f373 100644 --- a/testing/test_manager.py +++ b/testing/test_manager.py @@ -230,8 +230,6 @@ def _make_installed( (TrackingMethod.VERSION, True), (TrackingMethod.BRANCH, True), (TrackingMethod.COMMIT, True), - (TrackingMethod.BUILTIN, False), - (TrackingMethod.DIRECTORY, False), (None, False), ], ) @@ -274,7 +272,7 @@ def test_prepare_snapshot_directory( package = Package(git_url=str(pkg_dir), canonical=True) snapshot = _prepare_snapshot(package, None, str(tmp_path / "dest")) assert snapshot.version == "1.0.0" - assert snapshot.tracking_method == TrackingMethod.DIRECTORY + assert snapshot.tracking_method == None assert snapshot.current_hash is None assert snapshot.is_outdated is False @@ -303,7 +301,7 @@ def test_info_directory_backend(manager: Manager, pkg_dir: pathlib.Path) -> None info = manager.info(str(pkg_dir)) assert info.invalid_reason == "" assert info.metadata_version == "1.0.0" - assert info.version_type == TrackingMethod.DIRECTORY + assert info.version_type == None def test_install_directory_backend(manager: Manager, pkg_dir: pathlib.Path) -> None: @@ -311,7 +309,7 @@ def test_install_directory_backend(manager: Manager, pkg_dir: pathlib.Path) -> N assert result == "" ipkg = manager.find_installed_package("mypkg") assert ipkg is not None - assert ipkg.status.tracking_method == TrackingMethod.DIRECTORY + assert ipkg.status.tracking_method == None assert ipkg.status.current_version == "1.0.0" @@ -381,7 +379,7 @@ def test_refresh_skips_non_git_packages(manager: Manager) -> None: _make_installed( manager, "mypkg", - tracking_method=TrackingMethod.DIRECTORY, + tracking_method=None, current_version="1.0.0", ) # Should complete without raising. @@ -480,7 +478,7 @@ def test_bundle_skips_non_git_existing_clone( _make_installed( manager, "mypkg", - tracking_method=TrackingMethod.DIRECTORY, + tracking_method=None, current_version="1.0.0", ) bundle_file = str(tmp_path / "out.tar.gz") diff --git a/testing/test_package.py b/testing/test_package.py new file mode 100644 index 0000000..7110460 --- /dev/null +++ b/testing/test_package.py @@ -0,0 +1,31 @@ +"""Unit tests for zeekpkg.package data structures.""" + +import pytest + +from zeekpkg.package import PackageStatus, TrackingMethod + + +@pytest.mark.parametrize("value", [m.value for m in TrackingMethod]) +def test_package_status_coerces_string_tracking_method(value: str) -> None: + # Manifest JSON deserialises tracking_method as a plain string; verify + # PackageStatus converts it to the enum. + status = PackageStatus(tracking_method=value) + assert status.tracking_method is TrackingMethod(value) + + +@pytest.mark.parametrize("value", ["builtin", "directory"]) +def test_package_status_maps_legacy_non_git_methods_to_none(value: str) -> None: + # Non-Git packages have no tracking method; older manifests use string + # values for them that must round-trip to None. + status = PackageStatus(tracking_method=value) + assert status.tracking_method is None + + +def test_package_status_accepts_enum_tracking_method() -> None: + status = PackageStatus(tracking_method=TrackingMethod.BRANCH) + assert status.tracking_method is TrackingMethod.BRANCH + + +def test_package_status_rejects_invalid_tracking_method() -> None: + with pytest.raises(ValueError): + PackageStatus(tracking_method="bogus") diff --git a/zeekpkg/manager.py b/zeekpkg/manager.py index 86b6889..06f93a8 100644 --- a/zeekpkg/manager.py +++ b/zeekpkg/manager.py @@ -1351,6 +1351,7 @@ def upgrade(self, pkg_path: str) -> str: clone = self._open_package_clone(ipkg.package) + assert ipkg.status.tracking_method match ipkg.status.tracking_method: case TrackingMethod.VERSION: version_tags = git_version_tags(clone) @@ -2028,7 +2029,7 @@ def _info( status, versions=[], default_branch="", - version_type=TrackingMethod.DIRECTORY, + version_type=None, ) def package_versions(self, installed_package: InstalledPackage) -> list[str]: @@ -3480,7 +3481,7 @@ def _snapshot_from_directory(path: str) -> PackageSnapshot: working_dir=path, meta=meta, version=version, - tracking_method=TrackingMethod.DIRECTORY, + tracking_method=None, ) @@ -3506,14 +3507,10 @@ def _is_directory_package(path: str) -> bool: def _is_git_package(status: PackageStatus) -> bool: """Return True if *status* represents a git-backed package. - Non-git package sources (e.g. local directories) will use a different - tracking method; this predicate guards operations that require a git clone. + Non-git package sources (e.g. local directories) have no tracking method; + this predicate guards operations that require a git clone. """ - return status.tracking_method in ( - TrackingMethod.VERSION, - TrackingMethod.BRANCH, - TrackingMethod.COMMIT, - ) + return status.tracking_method is not None def _is_version_outdated(clone: git.Repo, version: str) -> bool: diff --git a/zeekpkg/package.py b/zeekpkg/package.py index 8d52684..139a401 100644 --- a/zeekpkg/package.py +++ b/zeekpkg/package.py @@ -20,13 +20,11 @@ class TrackingMethod(str, Enum): - """How a package tracks upstream changes.""" + """How a Git-backed package tracks upstream changes.""" VERSION = "version" BRANCH = "branch" COMMIT = "commit" - BUILTIN = "builtin" - DIRECTORY = "directory" def __str__(self) -> str: return self.value @@ -221,7 +219,7 @@ def fullfills(self, version_spec: str) -> tuple[str, bool]: if self.method == TrackingMethod.BRANCH: return "tracking method branch and commit", False - # TrackingMethod.BRANCH / TrackingMethod.BUILTIN + # TrackingMethod.VERSION / None (builtin or directory) if version_spec.startswith("branch="): branch = version_spec[len("branch=") :] return ( @@ -325,8 +323,13 @@ def __init__( self.is_loaded = is_loaded self.is_pinned = is_pinned self.is_outdated = is_outdated + # Non-Git sources have no tracking method; map their legacy string + # values to None for backward compatibility with older manifests. + _non_git = {"builtin", "directory"} self.tracking_method = ( - TrackingMethod(tracking_method) + None + if isinstance(tracking_method, str) and tracking_method in _non_git + else TrackingMethod(tracking_method) if isinstance(tracking_method, str) else tracking_method ) @@ -660,7 +663,7 @@ def make_builtin_package( is_loaded=True, # May not hold in the future? is_outdated=False, is_pinned=True, - tracking_method=TrackingMethod.BUILTIN, + tracking_method=None, current_version=current_version, current_hash=current_hash, ) From dcba69069ac11e66488e7a4d7e5075a53e28ed5a Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Fri, 17 Jul 2026 09:27:25 +0200 Subject: [PATCH 5/5] Inline and remove the trivial `_is_git_package` helper With `TrackingMethod` covering only Git-backed methods, non-Git packages are expressed as `tracking_method=None`. The helper reduces to a `None` check and adds no clarity over reading the condition directly. We inline it at both call sites and remove the function. --- testing/test_manager.py | 14 -------------- zeekpkg/manager.py | 15 ++++----------- 2 files changed, 4 insertions(+), 25 deletions(-) diff --git a/testing/test_manager.py b/testing/test_manager.py index ae0f373..2c8b86c 100644 --- a/testing/test_manager.py +++ b/testing/test_manager.py @@ -11,7 +11,6 @@ Manager, _info_from_snapshot, _is_directory_package, - _is_git_package, _prepare_snapshot, _resolve_git_version, _snapshot_from_directory, @@ -224,19 +223,6 @@ def _make_installed( manager.installed_pkgs[name] = InstalledPackage(pkg, status) -@pytest.mark.parametrize( - "method,expected", - [ - (TrackingMethod.VERSION, True), - (TrackingMethod.BRANCH, True), - (TrackingMethod.COMMIT, True), - (None, False), - ], -) -def test_is_git_package(method: str | None, expected: bool) -> None: - assert _is_git_package(PackageStatus(tracking_method=method)) is expected - - def test_snapshot_from_directory(pkg_dir: pathlib.Path) -> None: snapshot = _snapshot_from_directory(str(pkg_dir)) assert isinstance(snapshot, PackageSnapshot) diff --git a/zeekpkg/manager.py b/zeekpkg/manager.py index 06f93a8..4acee0d 100644 --- a/zeekpkg/manager.py +++ b/zeekpkg/manager.py @@ -1289,7 +1289,7 @@ def refresh_installed_packages(self) -> None: ) continue - if not _is_git_package(ipkg.status): + if ipkg.status.tracking_method is None: continue # Deliberate git entry point: fetch requires network access and @@ -2540,7 +2540,9 @@ def match_package_url_and_version( if prefer_existing_clones: ipkg = match_package_url_and_version(git_url, version) - if ipkg and _is_git_package(ipkg.status): + # Only reuse the existing clone if the installed package is + # git-backed; directory packages have no clone to copy. + if ipkg and ipkg.status.tracking_method is not None: src = os.path.join(self.package_clonedir, ipkg.package.name) shutil.copytree(src, clonepath, symlinks=True) clone = git.Repo(clonepath) @@ -3504,15 +3506,6 @@ def _is_directory_package(path: str) -> bool: return os.path.isdir(path) and not os.path.isdir(os.path.join(path, ".git")) -def _is_git_package(status: PackageStatus) -> bool: - """Return True if *status* represents a git-backed package. - - Non-git package sources (e.g. local directories) have no tracking method; - this predicate guards operations that require a git clone. - """ - return status.tracking_method is not None - - def _is_version_outdated(clone: git.Repo, version: str) -> bool: version_tags = git_version_tags(clone) latest = normalize_version_tag(version_tags[-1])