Files
govoplan/tools/release/govoplan_release/version_alignment.py

770 lines
29 KiB
Python

"""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
import tomllib
from .git_state import collect_versions, sanitized_git_environment
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 selected_release_webui_bundle_issues(
*,
repo_versions: dict[str, str],
workspace: Path,
) -> tuple[VersionAlignmentIssue, ...]:
"""Require selected WebUI modules to match Core's pinned release bundle.
Selective releases leave unselected module pins untouched. A selected module
that exposes a WebUI package, however, must be the version that Core's
immutable release package input and lockfile will actually install.
"""
core_webui = workspace / "govoplan-core" / "webui"
release_package_path = core_webui / "package.release.json"
release_lock_path = core_webui / "package-lock.release.json"
release_package = _json_object(release_package_path) if release_package_path.exists() else {}
release_dependencies = release_package.get("dependencies")
if not isinstance(release_dependencies, dict):
release_dependencies = {}
release_lock = _json_object(release_lock_path) if release_lock_path.exists() else {}
lock_packages = release_lock.get("packages")
if not isinstance(lock_packages, dict):
lock_packages = {}
lock_root = lock_packages.get("")
lock_root_dependencies = lock_root.get("dependencies") if isinstance(lock_root, dict) else None
if not isinstance(lock_root_dependencies, dict):
lock_root_dependencies = {}
issues: list[VersionAlignmentIssue] = []
for repo, requested_version in sorted(repo_versions.items()):
if repo == "govoplan-core":
continue
module_package_path = workspace / repo / "webui" / "package.json"
if not module_package_path.exists():
continue
module_package = _json_object(module_package_path)
package_name = module_package.get("name")
if not isinstance(package_name, str) or not package_name:
issues.append(
VersionAlignmentIssue(
repo=repo,
source="webui/package.json:name",
expected="WebUI package name",
actual="",
message="selected WebUI module must declare its package name",
)
)
continue
version = requested_version.removeprefix("v")
release_ref = release_dependencies.get(package_name)
release_match = _CATALOG_WEBUI_REF.search(release_ref) if isinstance(release_ref, str) else None
actual_release = (
f"{release_match.group('repo')}@v{release_match.group('version').removeprefix('v')}"
if release_match is not None
else str(release_ref or "")
)
expected_release = f"{repo}@v{version}"
if (
release_match is None
or release_match.group("repo") != repo
or release_match.group("version").removeprefix("v") != version
):
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"govoplan-core/webui/package.release.json:dependencies.{package_name}",
expected=expected_release,
actual=actual_release,
message="Core release WebUI dependency must match the selected module version",
)
)
locked_root_ref = lock_root_dependencies.get(package_name)
if locked_root_ref != release_ref:
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"govoplan-core/webui/package-lock.release.json:dependencies.{package_name}",
expected=str(release_ref or ""),
actual=str(locked_root_ref or ""),
message="Core release lock root dependency must match package.release.json",
)
)
locked_package = lock_packages.get(f"node_modules/{package_name}")
locked_version = locked_package.get("version") if isinstance(locked_package, dict) else None
if locked_version != version:
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"govoplan-core/webui/package-lock.release.json:{package_name}",
expected=version,
actual=str(locked_version or ""),
message="Core release lock must resolve the selected module version",
)
)
expected_commit = _git_tag_commit(workspace / repo, f"v{version}")
resolved = locked_package.get("resolved") if isinstance(locked_package, dict) else None
resolved_commit = resolved.rsplit("#", 1)[-1] if isinstance(resolved, str) and "#" in resolved else None
if expected_commit is not None and resolved_commit != expected_commit:
issues.append(
VersionAlignmentIssue(
repo=repo,
source=f"govoplan-core/webui/package-lock.release.json:{package_name}:resolved",
expected=expected_commit,
actual=resolved_commit or "",
message="Core release lock commit must match the selected module tag",
)
)
return tuple(issues)
def release_composition_issues(meta_root: Path, *, core_root: Path) -> tuple[VersionAlignmentIssue, ...]:
"""Require release references to agree and point at installable artifacts."""
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",
)
)
workspace = meta_root.parent
for repo, version in sorted(python_refs.items()):
repo_path = workspace / repo
if repo_path.exists():
issues.extend(
_release_artifact_issues(
repo=repo,
repo_path=repo_path,
version=version,
artifact="pyproject.toml",
)
)
for repo, version in sorted(webui_refs.items()):
repo_path = workspace / repo
if repo_path.exists():
issues.extend(
_release_artifact_issues(
repo=repo,
repo_path=repo_path,
version=version,
artifact="webui/package.json",
)
)
return tuple(issues)
def _release_artifact_issues(
*,
repo: str,
repo_path: Path,
version: str,
artifact: str,
) -> list[VersionAlignmentIssue]:
tag = f"v{version.removeprefix('v')}"
payload = _git_tag_file(repo_path, tag, artifact)
if payload is None:
return [
VersionAlignmentIssue(
repo=repo,
source=f"{tag}:{artifact}",
expected="installable package metadata",
actual="missing or unreadable",
message="release tag must contain its declared package artifact",
)
]
try:
if artifact == "pyproject.toml":
parsed = tomllib.loads(payload)
project = parsed.get("project")
artifact_version = project.get("version") if isinstance(project, dict) else None
else:
parsed = json.loads(payload)
artifact_version = parsed.get("version") if isinstance(parsed, dict) else None
except (json.JSONDecodeError, tomllib.TOMLDecodeError):
artifact_version = None
if artifact_version == version.removeprefix("v"):
return []
return [
VersionAlignmentIssue(
repo=repo,
source=f"{tag}:{artifact}",
expected=version.removeprefix("v"),
actual=str(artifact_version or ""),
message="release artifact version must match its immutable tag",
)
]
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(
["/usr/bin/git", "-C", str(repo_path), "rev-list", "-n", "1", tag],
check=False,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
timeout=10,
env=sanitized_git_environment(),
)
value = result.stdout.strip()
return value if result.returncode == 0 and value else None
def _git_tag_file(repo_path: Path, tag: str, relative_path: str) -> str | None:
if not (repo_path / ".git").exists():
return None
result = subprocess.run(
["/usr/bin/git", "-C", str(repo_path), "show", f"{tag}:{relative_path}"],
check=False,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
timeout=10,
env=sanitized_git_environment(),
)
return result.stdout if result.returncode == 0 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