GovOPlaN/govoplan
1
0
Fork 0
Code Issues 11 Pull Requests Actions 3 Packages Projects Releases Wiki Activity

feat(release): enforce aligned trusted publications

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 12:25:06 +02:00
parent c34892f6ea
commit a15a74c54c
18 changed files with 1359 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
tools/release/govoplan_release/git_state.py
View File

@@ -99,6 +99,7 @@ def collect_versions(path: Path) -> VersionSnapshot:
package=read_json_version(path / "package.json"),
webui_package=read_json_version(path / "webui" / "package.json"),
manifests=read_manifest_versions(path),
package_inits=read_package_init_versions(path),
)
@@ -137,6 +138,21 @@ def read_manifest_versions(path: Path) -> tuple[str, ...]:
return tuple(versions)
def read_package_init_versions(path: Path) -> tuple[str, ...]:
"""Read public runtime versions declared by top-level Python packages."""
src = path / "src"
if not src.exists():
return ()
versions: list[str] = []
for package_init in sorted(src.glob("*/__init__.py")):
text = package_init.read_text(encoding="utf-8")
match = re.search(r'(?m)^__version__\s*=\s*["\']([^"\']+)["\']', text)
if match is not None:
versions.append(match.group(1))
return tuple(versions)
def git_text(path: Path, *args: str, timeout: int = 10) -> str:
result = git(path, *args, timeout=timeout)
return result.stdout.strip() if result.returncode == 0 else ""

9
tools/release/govoplan_release/model.py
View File

@@ -26,10 +26,17 @@ class VersionSnapshot:
package: str | None = None
webui_package: str | None = None
manifests: tuple[str, ...] = ()
package_inits: tuple[str, ...] = ()
@property
def primary(self) -> str | None:
return self.pyproject or self.package or self.webui_package or (self.manifests[0] if self.manifests else None)
return (
self.pyproject
or self.package
or self.webui_package
or (self.manifests[0] if self.manifests else None)
or (self.package_inits[0] if self.package_inits else None)
)
@dataclass(frozen=True, slots=True)

55
tools/release/govoplan_release/publisher.py
View File

@@ -13,7 +13,9 @@ from govoplan_core.core.module_package_catalog import validate_module_package_ca
from .catalog import canonical_hash
from .model import CatalogPublishResult, CatalogPublishStep
from .module_directory import write_module_directory
from .selective_catalog import trusted_keys_from_keyring
from .version_alignment import candidate_catalog_version_issues
from .workspace import DEFAULT_WORKSPACE_ROOT, resolve_workspace_root, website_root
@@ -41,7 +43,6 @@ def publish_catalog_candidate(
resolved_web_root = Path(web_root).expanduser().resolve() if web_root is not None else website_root(workspace)
candidate_catalog = candidate_root / "channels" / f"{channel}.json"
candidate_keyring = candidate_root / "keyring.json"
candidate_modules = candidate_root / "modules"
target_catalog = resolved_web_root / "public" / "catalogs" / "v1" / "channels" / f"{channel}.json"
target_keyring = resolved_web_root / "public" / "catalogs" / "v1" / "keyring.json"
target_modules = resolved_web_root / "public" / "catalogs" / "v1" / "modules"
@@ -72,11 +73,36 @@ def publish_catalog_candidate(
validation_warnings: tuple[str, ...] = ()
if candidate_payload is not None and keyring_payload is not None:
version_issues = candidate_catalog_version_issues(candidate_payload)
blockers.extend(
f"candidate version alignment: {issue.source}: {issue.actual!r}; "
f"expected {issue.expected!r} ({issue.message})"
for issue in version_issues
)
candidate_keys = trusted_keys_from_keyring(keyring_payload if isinstance(keyring_payload, dict) else {})
target_keyring_payload = read_json(target_keyring) if target_keyring.exists() else None
publication_trust = trusted_keys_from_keyring(
target_keyring_payload if isinstance(target_keyring_payload, dict) else {}
)
if not publication_trust:
blockers.append(
"publication trust anchor is missing; bootstrap a trusted website keyring through a separate reviewed process"
)
for key_id in sorted(set(candidate_keys) & set(publication_trust)):
if candidate_keys[key_id] != publication_trust[key_id]:
blockers.append(f"candidate keyring changes the public key for trusted key id {key_id!r}")
if candidate_keyring_hash != target_keyring_hash_before:
release = candidate_payload.get("release") if isinstance(candidate_payload, dict) else None
signed_keyring_hash = release.get("keyring_sha256") if isinstance(release, dict) else None
if signed_keyring_hash != candidate_keyring_hash:
blockers.append(
"candidate keyring differs from the publication trust anchor without a matching signed keyring_sha256"
)
validation = validate_module_package_catalog(
candidate_catalog,
require_trusted=True,
approved_channels=(channel,),
trusted_keys=trusted_keys_from_keyring(keyring_payload if isinstance(keyring_payload, dict) else {}),
trusted_keys=publication_trust,
)
validation_valid = validation.get("valid") is True
validation_error = str(validation["error"]) if validation.get("error") else None
@@ -115,14 +141,12 @@ def publish_catalog_candidate(
steps.append(
CatalogPublishStep(
id="copy",
title="Copy candidate catalog, keyring, and module directory",
title="Copy candidate catalog/keyring and derive the module directory",
detail=f"{candidate_root} -> {resolved_web_root / 'public' / 'catalogs' / 'v1'}",
mutating=True,
status="planned" if not apply else "blocked" if blockers else "done",
)
)
if not candidate_modules.exists():
notes.append("Candidate has no module-directory tree; only catalog and keyring will be copied.")
if build_web:
steps.append(
CatalogPublishStep(
@@ -224,10 +248,15 @@ def publish_catalog_candidate(
target_catalog.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(candidate_catalog, target_catalog)
shutil.copy2(candidate_keyring, target_keyring)
if candidate_modules.exists():
if target_modules.exists():
shutil.rmtree(target_modules)
shutil.copytree(candidate_modules, target_modules)
if target_modules.exists():
shutil.rmtree(target_modules)
write_module_directory(
catalog_payload=candidate_payload,
keyring_payload=keyring_payload,
output_root=target_catalog.parent.parent,
channel=channel,
public_base_url=catalog_public_base_url(candidate_payload),
)
completed_steps = [replace_status(step, "done") if step.id == "copy" else step for step in steps]
if build_web:
@@ -282,6 +311,14 @@ def publish_catalog_candidate(
)
def catalog_public_base_url(payload: dict[str, object]) -> str:
release = payload.get("release")
catalog_url = release.get("catalog_url") if isinstance(release, dict) else None
if isinstance(catalog_url, str) and "/catalogs/" in catalog_url:
return catalog_url.split("/catalogs/", 1)[0]
return "https://govoplan.add-ideas.de"
def result(
*,
status: str,

30
tools/release/govoplan_release/selective_catalog.py
View File

@@ -19,6 +19,7 @@ from .contracts import collect_contracts
from .git_state import collect_repository_snapshot
from .model import CatalogEntryChange, SelectiveCatalogCandidate
from .module_directory import write_module_directory
from .version_alignment import selected_repository_version_issues
from .workspace import load_repository_specs, resolve_workspace_root, website_root
GITEA_BASE = "git+ssh://git@git.add-ideas.de/add-ideas"
@@ -46,6 +47,7 @@ def build_selective_catalog_candidate(
raise ValueError("At least one signing key is required.")
workspace = resolve_workspace_root(workspace_root)
enforce_selected_version_alignment(repo_versions=repo_versions, workspace=workspace)
web_root = website_root(workspace)
source_catalog = resolve_base_catalog(base_catalog=base_catalog, web_root=web_root, channel=channel)
payload = read_catalog(source_catalog)
@@ -75,22 +77,24 @@ def build_selective_catalog_candidate(
repository_base=repository_base.rstrip("/"),
)
candidate.pop("signature", None)
candidate.pop("signatures", None)
candidate["signatures"] = [signature(candidate, key_id=key_id, private_key=private_key) for key_id, private_key in parsed_keys]
output_root = resolve_output_dir(output_dir=output_dir, channel=channel, generated_at=generated_at)
catalog_path = output_root / "channels" / f"{channel}.json"
keyring_path = output_root / "keyring.json"
summary_path = output_root / "summary.json" if write_summary else None
catalog_path.parent.mkdir(parents=True, exist_ok=True)
catalog_path.write_text(json.dumps(candidate, indent=2, sort_keys=True) + "\n", encoding="utf-8")
keyring_payload = keyring_payload_for_candidate(
existing_keyring=web_root / "public" / "catalogs" / "v1" / "keyring.json",
signing_keys=parsed_keys,
generated_at=generated_at,
)
release["keyring_sha256"] = canonical_hash(keyring_payload)
candidate.pop("signature", None)
candidate.pop("signatures", None)
candidate["signatures"] = [signature(candidate, key_id=key_id, private_key=private_key) for key_id, private_key in parsed_keys]
catalog_path.parent.mkdir(parents=True, exist_ok=True)
catalog_path.write_text(json.dumps(candidate, indent=2, sort_keys=True) + "\n", encoding="utf-8")
keyring_path.write_text(json.dumps(keyring_payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
module_directory_files = write_module_directory(
catalog_payload=candidate,
@@ -159,6 +163,18 @@ def build_selective_catalog_candidate(
return result
def enforce_selected_version_alignment(*, repo_versions: dict[str, str], workspace: Path) -> None:
failures = [
f"{issue.repo}: {issue.source}={issue.actual!r}, expected {issue.expected!r} ({issue.message})"
for issue in selected_repository_version_issues(
repo_versions=repo_versions,
workspace=workspace,
)
]
if failures:
raise ValueError("Version alignment gate failed: " + "; ".join(failures))
def resolve_base_catalog(*, base_catalog: Path | str | None, web_root: Path, channel: str) -> Path | str:
if base_catalog is not None:
value = str(base_catalog)

13
tools/release/govoplan_release/selective_planner.py
View File

@@ -76,6 +76,19 @@ def build_unit(repo: RepositorySnapshot, *, target_version: str | None, contract
blockers.append(f"repository is behind {repo.upstream}")
if repo.exists and repo.is_git and not repo.has_head:
blockers.append("repository has no initial commit")
local_versions = tuple(
value
for value in (
repo.versions.pyproject,
repo.versions.package,
repo.versions.webui_package,
*repo.versions.manifests,
*repo.versions.package_inits,
)
if value
)
if len(set(local_versions)) > 1:
blockers.append("backend, manifest, and frontend version metadata is not aligned")
if repo.dirty:
warnings.append("uncommitted changes will need review before release")
if repo.ahead:

572
tools/release/govoplan_release/version_alignment.py Normal file
View File

@@ -0,0 +1,572 @@
"""Version-alignment checks for independently released GovOPlaN packages."""
from __future__ import annotations
from dataclasses import dataclass
import json
from pathlib import Path
import re
import subprocess
from .git_state import collect_versions
from .workspace import load_repository_specs, resolve_repo_path
_PYTHON_RELEASE_REF = re.compile(
r"^(?P<package>govoplan-[a-z0-9-]+)\s+@\s+git\+ssh://.+/"
r"(?P<repo>govoplan-[a-z0-9-]+)\.git@v(?P<version>[^\s;]+)$"
)
_WEBUI_RELEASE_REF = re.compile(r"/(?P<repo>govoplan-[a-z0-9-]+)\.git#v(?P<version>[^#\s]+)$")
_CATALOG_PYTHON_REF = re.compile(r"/(?P<repo>govoplan-[a-z0-9-]+)\.git@v(?P<version>[^\s;]+)$")
_CATALOG_WEBUI_REF = re.compile(r"/(?P<repo>govoplan-[a-z0-9-]+)\.git#v(?P<version>[^#\s]+)$")
@dataclass(frozen=True, slots=True)
class VersionAlignmentIssue:
repo: str
source: str
expected: str
actual: str
message: str
def repository_version_issues(
repo_path: Path,
*,
expected_version: str | None = None,
include_lockfiles: bool = True,
) -> tuple[VersionAlignmentIssue, ...]:
"""Compare every version-bearing file in one source repository."""
versions = collect_versions(repo_path)
declared = {
"pyproject.toml": versions.pyproject,
"package.json": versions.package,
"webui/package.json": versions.webui_package,
"webui/package.release.json": _json_version(repo_path / "webui" / "package.release.json")
if (repo_path / "webui" / "package.release.json").exists()
else None,
}
for index, version in enumerate(versions.manifests, start=1):
declared[f"module manifest {index}"] = version
for index, version in enumerate(versions.package_inits, start=1):
declared[f"package __init__.py {index}"] = version
canonical_source, canonical_version = next(
((source, version) for source, version in declared.items() if version),
(None, None),
)
repo = repo_path.name
if canonical_source is None or canonical_version is None:
if expected_version is None:
return ()
return (
VersionAlignmentIssue(
repo=repo,
source="repository",
expected=expected_version.removeprefix("v"),
actual="",
message="repository has no version metadata",
),
)
issues = [
VersionAlignmentIssue(
repo=repo,
source=source,
expected=canonical_version,
actual=version,
message=f"{source} must match {canonical_source}",
)
for source, version in declared.items()
if version is not None and version != canonical_version
]
if expected_version is not None and canonical_version.removeprefix("v") != expected_version.removeprefix("v"):
issues.append(
VersionAlignmentIssue(
repo=repo,
source=canonical_source,
expected=expected_version.removeprefix("v"),
actual=canonical_version,
message="source version must match the requested release version",
)
)
if include_lockfiles:
issues.extend(
_lockfile_issues(
repo=repo,
package_path=repo_path / "package.json",
lock_path=repo_path / "package-lock.json",
)
)
issues.extend(
_lockfile_issues(
repo=repo,
package_path=repo_path / "webui" / "package.json",
lock_path=repo_path / "webui" / "package-lock.json",
)
)
issues.extend(
_lockfile_issues(
repo=repo,
package_path=repo_path / "webui" / "package.release.json",
lock_path=repo_path / "webui" / "package-lock.release.json",
)
)
return tuple(issues)
def selected_repository_version_issues(
*,
repo_versions: dict[str, str],
workspace: Path,
include_lockfiles: bool = True,
) -> tuple[VersionAlignmentIssue, ...]:
"""Validate registered selected worktrees against explicit release versions."""
specs = {spec.name: spec for spec in load_repository_specs(include_website=False)}
issues: list[VersionAlignmentIssue] = []
for repo, expected_version in sorted(repo_versions.items()):
spec = specs.get(repo)
if spec is None:
issues.append(
VersionAlignmentIssue(
repo=repo,
source="repository",
expected=expected_version.removeprefix("v"),
actual="",
message="repository is not registered",
)
)
continue
repo_path = resolve_repo_path(spec, workspace)
if not repo_path.exists():
issues.append(
VersionAlignmentIssue(
repo=repo,
source="repository",
expected=expected_version.removeprefix("v"),
actual="",
message="repository path is missing",
)
)
continue
issues.extend(
repository_version_issues(
repo_path,
expected_version=expected_version,
include_lockfiles=include_lockfiles,
)
)
return tuple(issues)
def release_composition_issues(meta_root: Path, *, core_root: Path) -> tuple[VersionAlignmentIssue, ...]:
"""Require backend and WebUI release references for a module to agree."""
python_refs = _python_release_refs(meta_root / "requirements-release.txt")
webui_refs = _webui_release_refs(core_root / "webui" / "package.release.json")
issues: list[VersionAlignmentIssue] = []
for repo in sorted(set(python_refs) & set(webui_refs)):
python_version = python_refs[repo]
webui_version = webui_refs[repo]
if python_version == webui_version:
continue
issues.append(
VersionAlignmentIssue(
repo=repo,
source="webui/package.release.json",
expected=python_version,
actual=webui_version,
message="frontend release ref must match the backend release ref",
)
)
return tuple(issues)
def candidate_catalog_version_issues(payload: object) -> tuple[VersionAlignmentIssue, ...]:
"""Validate versions, refs, and selected-unit provenance inside a catalog."""
if not isinstance(payload, dict):
return (
VersionAlignmentIssue(
repo="catalog",
source="catalog",
expected="object",
actual=type(payload).__name__,
message="candidate catalog must be a JSON object",
),
)
issues: list[VersionAlignmentIssue] = []
represented: dict[str, str] = {}
core_release = payload.get("core_release")
if isinstance(core_release, dict):
issues.extend(_catalog_entry_issues(core_release, source="core_release", represented=represented))
else:
issues.append(_catalog_shape_issue("core_release", "candidate catalog has no core release"))
modules = payload.get("modules")
if not isinstance(modules, list):
issues.append(_catalog_shape_issue("modules", "candidate catalog has no modules list"))
else:
for index, entry in enumerate(modules):
if not isinstance(entry, dict):
issues.append(_catalog_shape_issue(f"modules[{index}]", "module entry must be an object"))
continue
issues.extend(
_catalog_entry_issues(
entry,
source=f"modules[{index}]",
represented=represented,
)
)
release = payload.get("release")
if isinstance(release, dict):
issues.extend(_catalog_release_issues(release, represented=represented))
return tuple(issues)
def _catalog_entry_issues(
entry: dict[str, object],
*,
source: str,
represented: dict[str, str],
) -> list[VersionAlignmentIssue]:
issues: list[VersionAlignmentIssue] = []
version = entry.get("version")
normalized_version = version.removeprefix("v") if isinstance(version, str) and version else None
if normalized_version is None:
issues.append(_catalog_shape_issue(f"{source}.version", "catalog entry has no version"))
python_ref = entry.get("python_ref")
python_match = _CATALOG_PYTHON_REF.search(python_ref) if isinstance(python_ref, str) else None
if python_match is None:
issues.append(_catalog_shape_issue(f"{source}.python_ref", "Python ref must end in a version tag"))
repo = str(entry.get("python_package") or entry.get("module_id") or source)
else:
repo = python_match.group("repo")
ref_version = python_match.group("version").removeprefix("v")
if normalized_version is not None and ref_version != normalized_version:
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"{source}.python_ref",
expected=normalized_version,
actual=ref_version,
message="Python ref tag must match the catalog entry version",
)
)
if normalized_version is not None:
previous = represented.setdefault(repo, normalized_version)
if previous != normalized_version:
issues.append(
VersionAlignmentIssue(
repo=repo,
source=source,
expected=previous,
actual=normalized_version,
message="repository appears with conflicting catalog versions",
)
)
webui_package = entry.get("webui_package")
webui_ref = entry.get("webui_ref")
if bool(webui_package) != bool(webui_ref):
issues.append(_catalog_shape_issue(f"{source}.webui_ref", "WebUI package and ref must be declared together"))
if isinstance(webui_ref, str):
webui_match = _CATALOG_WEBUI_REF.search(webui_ref)
if webui_match is None:
issues.append(_catalog_shape_issue(f"{source}.webui_ref", "WebUI ref must end in a version tag"))
else:
webui_repo = webui_match.group("repo")
webui_version = webui_match.group("version").removeprefix("v")
if python_match is not None and webui_repo != repo:
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"{source}.webui_ref",
expected=repo,
actual=webui_repo,
message="backend and WebUI refs must use the same repository",
)
)
if normalized_version is not None and webui_version != normalized_version:
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"{source}.webui_ref",
expected=normalized_version,
actual=webui_version,
message="WebUI ref tag must match the catalog entry version",
)
)
return issues
def _catalog_release_issues(
release: dict[str, object],
*,
represented: dict[str, str],
) -> list[VersionAlignmentIssue]:
issues: list[VersionAlignmentIssue] = []
version = release.get("version")
tag = release.get("tag")
if isinstance(version, str) and isinstance(tag, str) and tag != f"v{version.removeprefix('v')}":
issues.append(
VersionAlignmentIssue(
repo="govoplan-core",
source="release.tag",
expected=f"v{version.removeprefix('v')}",
actual=tag,
message="release tag must match release version",
)
)
selected_units = release.get("selected_units")
if selected_units is None:
return issues
if not isinstance(selected_units, list):
issues.append(_catalog_shape_issue("release.selected_units", "selected units must be a list"))
return issues
seen: set[str] = set()
for index, unit in enumerate(selected_units):
if not isinstance(unit, dict):
issues.append(_catalog_shape_issue(f"release.selected_units[{index}]", "selected unit must be an object"))
continue
repo = unit.get("repo")
unit_version = unit.get("version")
unit_tag = unit.get("tag")
if not isinstance(repo, str) or not isinstance(unit_version, str) or not isinstance(unit_tag, str):
issues.append(_catalog_shape_issue(f"release.selected_units[{index}]", "selected unit requires repo, version, and tag"))
continue
normalized_version = unit_version.removeprefix("v")
if repo in seen:
issues.append(_catalog_shape_issue(f"release.selected_units[{index}].repo", "selected repository is duplicated"))
seen.add(repo)
if unit_tag != f"v{normalized_version}":
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"release.selected_units[{index}].tag",
expected=f"v{normalized_version}",
actual=unit_tag,
message="selected-unit tag must match its version",
)
)
represented_version = represented.get(repo)
if represented_version is None:
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"release.selected_units[{index}]",
expected=normalized_version,
actual="",
message="selected repository is not represented in the catalog",
)
)
elif represented_version != normalized_version:
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"release.selected_units[{index}].version",
expected=represented_version,
actual=normalized_version,
message="selected-unit version must match the catalog entry",
)
)
return issues
def _catalog_shape_issue(source: str, message: str) -> VersionAlignmentIssue:
return VersionAlignmentIssue(
repo="catalog",
source=source,
expected="valid value",
actual="",
message=message,
)
def _lockfile_issues(*, repo: str, package_path: Path, lock_path: Path) -> list[VersionAlignmentIssue]:
if not package_path.exists() or not lock_path.exists():
return []
package_payload = _json_object(package_path)
lock_payload = _json_object(lock_path)
package_version = _payload_version(package_payload)
lock_version = _lockfile_root_version(lock_path)
issues: list[VersionAlignmentIssue] = []
if package_version is None or lock_version is None or package_version == lock_version:
pass
else:
issues.append(
VersionAlignmentIssue(
repo=repo,
source=str(lock_path.relative_to(package_path.parents[1] if package_path.parent.name == "webui" else package_path.parent)),
expected=package_version,
actual=lock_version,
message=f"lockfile root version must match {package_path.name}",
)
)
lock_packages = lock_payload.get("packages")
lock_root = lock_packages.get("") if isinstance(lock_packages, dict) else None
if isinstance(lock_root, dict):
for group in ("dependencies", "devDependencies", "optionalDependencies", "peerDependencies"):
package_dependencies = package_payload.get(group) or {}
lock_dependencies = lock_root.get(group) or {}
if package_dependencies != lock_dependencies:
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"{lock_path.name}:{group}",
expected=json.dumps(package_dependencies, sort_keys=True),
actual=json.dumps(lock_dependencies, sort_keys=True),
message=f"lockfile root {group} must match {package_path.name}",
)
)
issues.extend(
_git_lock_resolution_issues(
repo=repo,
package_payload=package_payload,
lock_packages=lock_packages,
workspace=package_path.parents[2] if package_path.parent.name == "webui" else package_path.parent.parent,
lock_name=lock_path.name,
)
)
return issues
def _git_lock_resolution_issues(
*,
repo: str,
package_payload: dict[str, object],
lock_packages: dict[str, object],
workspace: Path,
lock_name: str,
) -> list[VersionAlignmentIssue]:
dependencies = package_payload.get("dependencies")
if not isinstance(dependencies, dict):
return []
issues: list[VersionAlignmentIssue] = []
for package_name, spec in dependencies.items():
if not isinstance(package_name, str) or not isinstance(spec, str):
continue
match = _CATALOG_WEBUI_REF.search(spec)
if match is None:
continue
source_repo = match.group("repo")
version = match.group("version").removeprefix("v")
locked = lock_packages.get(f"node_modules/{package_name}")
if not isinstance(locked, dict):
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"{lock_name}:{package_name}",
expected=version,
actual="",
message="release lock has no resolved GovOPlaN package",
)
)
continue
locked_version = locked.get("version")
if locked_version != version:
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"{lock_name}:{package_name}",
expected=version,
actual=str(locked_version or ""),
message="resolved package version must match its release ref",
)
)
expected_commit = _git_tag_commit(workspace / source_repo, f"v{version}")
resolved = locked.get("resolved")
resolved_commit = resolved.rsplit("#", 1)[-1] if isinstance(resolved, str) and "#" in resolved else None
if expected_commit is None or resolved_commit != expected_commit:
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"{lock_name}:{package_name}:resolved",
expected=expected_commit or f"local tag v{version}",
actual=resolved_commit or "",
message="resolved package commit must match the local release tag",
)
)
return issues
def _git_tag_commit(repo_path: Path, tag: str) -> str | None:
if not (repo_path / ".git").exists():
return None
result = subprocess.run(
["git", "-C", str(repo_path), "rev-list", "-n", "1", tag],
check=False,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
timeout=10,
)
value = result.stdout.strip()
return value if result.returncode == 0 and value else None
def _json_version(path: Path) -> str | None:
value = _payload_version(_json_object(path))
return value
def _payload_version(payload: dict[str, object]) -> str | None:
value = payload.get("version")
return value if isinstance(value, str) else None
def _json_object(path: Path) -> dict[str, object]:
payload = json.loads(path.read_text(encoding="utf-8"))
return payload if isinstance(payload, dict) else {}
def _lockfile_root_version(path: Path) -> str | None:
payload = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(payload, dict):
return None
packages = payload.get("packages")
if isinstance(packages, dict):
root = packages.get("")
if isinstance(root, dict) and isinstance(root.get("version"), str):
return root["version"]
value = payload.get("version")
return value if isinstance(value, str) else None
def _python_release_refs(path: Path) -> dict[str, str]:
if not path.exists():
return {}
refs: dict[str, str] = {}
for line in path.read_text(encoding="utf-8").splitlines():
match = _PYTHON_RELEASE_REF.match(line.strip())
if match is None:
continue
repo = match.group("repo")
refs[repo] = match.group("version")
return refs
def _webui_release_refs(path: Path) -> dict[str, str]:
if not path.exists():
return {}
payload = json.loads(path.read_text(encoding="utf-8"))
dependencies = payload.get("dependencies") if isinstance(payload, dict) else None
if not isinstance(dependencies, dict):
return {}
refs: dict[str, str] = {}
for spec in dependencies.values():
if not isinstance(spec, str):
continue
match = _WEBUI_RELEASE_REF.search(spec)
if match is not None:
refs[match.group("repo")] = match.group("version")
return refs