"""Publish reviewed release catalog candidates into the website repository.""" from __future__ import annotations from datetime import UTC, datetime import json import os from pathlib import Path import re import secrets import shlex import shutil import stat import subprocess import tempfile from govoplan_core.core.module_package_catalog import validate_module_package_catalog from .catalog import canonical_hash from .artifact_identity import selected_artifact_identity_issues from .candidate_artifact import validate_release_channel from .model import CatalogPublishResult, CatalogPublishStep from .module_directory import module_directory_payloads from .selective_catalog import trusted_keys_from_keyring from .source_provenance import catalog_source_selection, source_tag_provenance_issues from .version_alignment import candidate_catalog_version_issues from .workspace import DEFAULT_WORKSPACE_ROOT, resolve_workspace_root, website_root def publish_catalog_candidate( *, candidate_dir: Path | str, workspace_root: Path | str | None = None, web_root: Path | str | None = None, channel: str = "stable", apply: bool = False, # noqa: A002 - CLI uses --apply. build_web: bool = False, commit: bool = False, tag: bool = False, push: bool = False, remote: str = "origin", source_remote: str = "origin", branch: str | None = None, npm: str = "npm", tag_name: str | None = None, allow_dirty_website: bool = False, ) -> CatalogPublishResult: channel = validate_release_channel(channel) workspace = resolve_workspace_root(workspace_root or DEFAULT_WORKSPACE_ROOT) candidate_root = Path(candidate_dir).expanduser() if not candidate_root.is_absolute(): candidate_root = Path.cwd() / candidate_root 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" 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" steps: list[CatalogPublishStep] = [] notes: list[str] = [] blockers: list[str] = [] if not candidate_catalog.exists(): blockers.append(f"candidate catalog is missing: {candidate_catalog}") if not candidate_keyring.exists(): blockers.append(f"candidate keyring is missing: {candidate_keyring}") if not resolved_web_root.exists(): blockers.append(f"website root is missing: {resolved_web_root}") if push: commit = True if tag: commit = True candidate_payload = read_json(candidate_catalog) if candidate_catalog.exists() else None keyring_payload = read_json(candidate_keyring) if candidate_keyring.exists() else None candidate_catalog_hash = canonical_hash(candidate_payload) if candidate_payload is not None else None candidate_keyring_hash = canonical_hash(keyring_payload) if keyring_payload is not None else None target_catalog_hash_before = file_json_hash(target_catalog) target_keyring_hash_before = file_json_hash(target_keyring) validation_valid = False validation_error: str | None = None validation_warnings: tuple[str, ...] = () if candidate_payload is not None and keyring_payload is not None: if not isinstance(candidate_payload, dict) or candidate_payload.get("channel") != channel: blockers.append("candidate catalog channel does not match the publication channel") 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 ) source_selection = catalog_source_selection(candidate_payload) blockers.extend( f"source provenance: {issue.describe()}" for issue in source_selection.issues ) artifact_issues = selected_artifact_identity_issues(candidate_payload) if apply or commit or tag or push: blockers.extend( f"release artifact identity: {issue}" for issue in artifact_issues ) else: notes.extend( f"Source-only preview; publication would block: {issue}" for issue in artifact_issues ) source_issues = source_tag_provenance_issues( repo_versions=source_selection.all_versions, workspace=workspace, remote=source_remote, require_head_repos=source_selection.selected_versions, expected_commits=source_selection.selected_commits, expected_tag_objects=source_selection.selected_tag_objects, ) blockers.extend( f"source provenance: {issue.describe()}" for issue in source_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_catalog_payload( candidate_payload, require_trusted=True, approved_channels=(channel,), trusted_keys=publication_trust, ) validation_valid = validation.get("valid") is True validation_error = str(validation["error"]) if validation.get("error") else None validation_warnings = tuple(str(item) for item in validation.get("warnings") or ()) if not validation_valid: blockers.append(validation_error or "candidate catalog validation failed") catalog_changed = candidate_catalog_hash != target_catalog_hash_before keyring_changed = candidate_keyring_hash != target_keyring_hash_before if not catalog_changed and not keyring_changed: notes.append("Candidate catalog and keyring already match the website repository.") if apply or commit or tag or push: if not (resolved_web_root / ".git").exists(): blockers.append(f"website root is not a Git repository: {resolved_web_root}") if not allow_dirty_website and website_dirty(resolved_web_root): blockers.append("website repository has uncommitted changes") effective_branch = branch or git_text(resolved_web_root, "branch", "--show-current") or None effective_tag_name = tag_name or default_tag_name(candidate_payload, channel=channel) if (tag or push) and not effective_tag_name: blockers.append("could not infer catalog tag name; pass --tag-name") if (commit or push) and not effective_branch: blockers.append("could not infer website branch; pass --branch") if tag and effective_tag_name and git_success(resolved_web_root, "rev-parse", "--quiet", "--verify", f"refs/tags/{effective_tag_name}"): blockers.append(f"website tag already exists: {effective_tag_name}") steps.append( CatalogPublishStep( id="validate", title="Validate candidate catalog", detail="Validate signature, approved channel, freshness, interface closure, and keyring trust.", status="done" if validation_valid else "blocked", ) ) steps.append( CatalogPublishStep( id="copy", 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 build_web: steps.append( CatalogPublishStep( id="build-web", title="Build website", detail="Run the website build after copying catalog files.", command=shlex.join([npm, "--prefix", str(resolved_web_root), "run", "build"]), mutating=True, status="planned" if not apply else "blocked" if blockers else "pending", ) ) if commit: steps.append( CatalogPublishStep( id="commit", title="Commit website catalog update", detail="Commit generated catalog/keyring files in the website repository.", command=shlex.join(["git", "-C", str(resolved_web_root), "commit", "-m", commit_message(candidate_payload, channel=channel)]), mutating=True, status="planned" if not apply else "blocked" if blockers else "pending", ) ) if tag: steps.append( CatalogPublishStep( id="tag", title=f"Tag website catalog publication {effective_tag_name}", detail="Create an annotated website publication tag.", command=shlex.join(["git", "-C", str(resolved_web_root), "tag", "-a", effective_tag_name, "-m", commit_message(candidate_payload, channel=channel)]), mutating=True, status="planned" if not apply else "blocked" if blockers else "pending", ) ) if push: steps.append( CatalogPublishStep( id="push", title="Push website catalog publication", detail="Push website branch and tag if a tag was requested.", command=push_command(resolved_web_root, remote=remote, branch=effective_branch, tag_name=effective_tag_name if tag else None), mutating=True, status="planned" if not apply else "blocked" if blockers else "pending", ) ) if blockers: return result( status="blocked", applied=False, candidate_root=candidate_root, candidate_catalog=candidate_catalog, candidate_keyring=candidate_keyring, resolved_web_root=resolved_web_root, target_catalog=target_catalog, target_keyring=target_keyring, channel=channel, branch=effective_branch, tag_name=effective_tag_name, validation_valid=validation_valid, validation_error=validation_error, validation_warnings=validation_warnings, candidate_catalog_hash=candidate_catalog_hash, candidate_keyring_hash=candidate_keyring_hash, target_catalog_hash_before=target_catalog_hash_before, target_keyring_hash_before=target_keyring_hash_before, catalog_changed=catalog_changed, keyring_changed=keyring_changed, steps=tuple(steps), notes=tuple([*notes, *blockers]), ) if not apply: notes.append("Dry run only. Pass --apply to copy files; pass --commit/--tag/--push for Git publication steps.") return result( status="ready", applied=False, candidate_root=candidate_root, candidate_catalog=candidate_catalog, candidate_keyring=candidate_keyring, resolved_web_root=resolved_web_root, target_catalog=target_catalog, target_keyring=target_keyring, channel=channel, branch=effective_branch, tag_name=effective_tag_name, validation_valid=validation_valid, validation_error=validation_error, validation_warnings=validation_warnings, candidate_catalog_hash=candidate_catalog_hash, candidate_keyring_hash=candidate_keyring_hash, target_catalog_hash_before=target_catalog_hash_before, target_keyring_hash_before=target_keyring_hash_before, catalog_changed=catalog_changed, keyring_changed=keyring_changed, steps=tuple(steps), notes=tuple(notes), ) directory_payloads = module_directory_payloads( catalog_payload=candidate_payload, keyring_payload=keyring_payload, channel=channel, public_base_url=catalog_public_base_url(candidate_payload), ) _atomic_write_public_json( target_catalog, candidate_payload, trusted_root=resolved_web_root ) _atomic_write_public_json( target_keyring, keyring_payload, trusted_root=resolved_web_root ) if target_modules.exists(): _remove_public_tree(target_modules, trusted_root=resolved_web_root) for relative_path, payload in directory_payloads: _atomic_write_public_json( target_catalog.parent.parent / relative_path, payload, trusted_root=resolved_web_root, ) expected_blobs = _expected_publication_blobs( web_root=resolved_web_root, target_catalog=target_catalog, candidate_payload=candidate_payload, target_keyring=target_keyring, keyring_payload=keyring_payload, module_root=target_catalog.parent.parent, directory_payloads=directory_payloads, ) completed_steps = [replace_status(step, "done") if step.id == "copy" else step for step in steps] if build_web: run_checked([npm, "--prefix", str(resolved_web_root), "run", "build"], cwd=resolved_web_root) completed_steps = [replace_status(step, "done") if step.id == "build-web" else step for step in completed_steps] publication_commit: str | None = None if commit: add_paths = ["git", "-C", str(resolved_web_root), "add", str(target_catalog), str(target_keyring)] if target_modules.exists(): add_paths.append(str(target_modules)) run_checked(add_paths, cwd=resolved_web_root) if git_success(resolved_web_root, "diff", "--cached", "--quiet"): notes.append("No website catalog changes were staged for commit.") completed_steps = [replace_status(step, "skipped") if step.id == "commit" else step for step in completed_steps] else: run_checked(["git", "-C", str(resolved_web_root), "commit", "-m", commit_message(candidate_payload, channel=channel)], cwd=resolved_web_root) completed_steps = [replace_status(step, "done") if step.id == "commit" else step for step in completed_steps] publication_commit = committed_publication_sha(resolved_web_root) verify_committed_publication( web_root=resolved_web_root, commit_sha=publication_commit, expected_blobs=expected_blobs, module_root=target_modules, ) publication_tag_object: str | None = None if tag and effective_tag_name and publication_commit: run_checked(["git", "-C", str(resolved_web_root), "tag", "-a", effective_tag_name, publication_commit, "-m", commit_message(candidate_payload, channel=channel)], cwd=resolved_web_root) publication_tag_object = git_text( resolved_web_root, "rev-parse", "--verify", f"refs/tags/{effective_tag_name}" ) tagged_commit = git_text( resolved_web_root, "rev-parse", "--verify", f"refs/tags/{effective_tag_name}^{{commit}}", ) if ( re.fullmatch(r"[0-9a-f]{40,64}", publication_tag_object) is None or tagged_commit != publication_commit ): raise RuntimeError("website publication tag does not bind the verified commit") completed_steps = [replace_status(step, "done") if step.id == "tag" else step for step in completed_steps] if push and effective_branch and publication_commit: push_args = ["git", "-C", str(resolved_web_root), "push"] if tag and effective_tag_name and publication_tag_object: push_args.extend(["--atomic", remote, f"{publication_commit}:refs/heads/{effective_branch}", f"{publication_tag_object}:refs/tags/{effective_tag_name}"]) else: push_args.extend([remote, f"{publication_commit}:refs/heads/{effective_branch}"]) run_checked(push_args, cwd=resolved_web_root) completed_steps = [replace_status(step, "done") if step.id == "push" else step for step in completed_steps] return result( status="published" if push else "applied", applied=True, candidate_root=candidate_root, candidate_catalog=candidate_catalog, candidate_keyring=candidate_keyring, resolved_web_root=resolved_web_root, target_catalog=target_catalog, target_keyring=target_keyring, channel=channel, branch=effective_branch, tag_name=effective_tag_name, validation_valid=validation_valid, validation_error=validation_error, validation_warnings=validation_warnings, candidate_catalog_hash=candidate_catalog_hash, candidate_keyring_hash=candidate_keyring_hash, target_catalog_hash_before=target_catalog_hash_before, target_keyring_hash_before=target_keyring_hash_before, catalog_changed=catalog_changed, keyring_changed=keyring_changed, steps=tuple(completed_steps), notes=tuple(notes), ) 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, applied: bool, candidate_root: Path, candidate_catalog: Path, candidate_keyring: Path, resolved_web_root: Path, target_catalog: Path, target_keyring: Path, channel: str, branch: str | None, tag_name: str | None, validation_valid: bool, validation_error: str | None, validation_warnings: tuple[str, ...], candidate_catalog_hash: str | None, candidate_keyring_hash: str | None, target_catalog_hash_before: str | None, target_keyring_hash_before: str | None, catalog_changed: bool, keyring_changed: bool, steps: tuple[CatalogPublishStep, ...], notes: tuple[str, ...], ) -> CatalogPublishResult: return CatalogPublishResult( generated_at=datetime.now(tz=UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z"), channel=channel, status=status, applied=applied, candidate_dir=str(candidate_root), candidate_catalog_path=str(candidate_catalog), candidate_keyring_path=str(candidate_keyring), web_root=str(resolved_web_root), target_catalog_path=str(target_catalog), target_keyring_path=str(target_keyring), branch=branch, tag_name=tag_name, validation_valid=validation_valid, validation_error=validation_error, validation_warnings=validation_warnings, candidate_catalog_hash=candidate_catalog_hash, candidate_keyring_hash=candidate_keyring_hash, target_catalog_hash_before=target_catalog_hash_before, target_keyring_hash_before=target_keyring_hash_before, catalog_changed=catalog_changed, keyring_changed=keyring_changed, steps=steps, notes=notes, ) def read_json(path: Path) -> object: return json.loads(path.read_text(encoding="utf-8")) def validate_catalog_payload( payload: object, *, require_trusted: bool, approved_channels: tuple[str, ...], trusted_keys: dict[str, bytes], ) -> dict[str, object]: """Validate exactly the in-memory object that publication will consume.""" encoded = _public_json_bytes(payload) descriptor, temporary_name = tempfile.mkstemp( prefix="govoplan-catalog-validation-", suffix=".json" ) temporary = Path(temporary_name) try: os.fchmod(descriptor, 0o600) with os.fdopen(descriptor, "wb", closefd=True) as handle: handle.write(encoded) handle.flush() os.fsync(handle.fileno()) return validate_module_package_catalog( temporary, require_trusted=require_trusted, approved_channels=approved_channels, trusted_keys=trusted_keys, ) finally: try: os.close(descriptor) except OSError: pass temporary.unlink(missing_ok=True) def _public_json_bytes(payload: object) -> bytes: encoded = ( json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=True) + "\n" ).encode("utf-8") if len(encoded) > 16 * 1024 * 1024: raise ValueError("published catalog object exceeds its size limit") return encoded def _atomic_write_public_json( path: Path, payload: object, *, trusted_root: Path ) -> None: """Write through pinned directories from a trusted worktree root.""" encoded = _public_json_bytes(payload) relative = _trusted_relative_path(path, trusted_root=trusted_root) if len(relative.parts) < 2: raise ValueError("published catalog target requires a parent directory") parent_descriptor = _open_relative_directory( trusted_root, relative.parts[:-1], create=True ) target_name = relative.parts[-1] try: try: observed = os.stat( target_name, dir_fd=parent_descriptor, follow_symlinks=False ) except FileNotFoundError: observed = None if observed is not None and not stat.S_ISREG(observed.st_mode): raise ValueError("published catalog target must be a regular file") temporary_name = f".{target_name}.{secrets.token_hex(16)}.tmp" flags = ( os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) ) descriptor = os.open( temporary_name, flags, 0o600, dir_fd=parent_descriptor ) replaced = False try: with os.fdopen(descriptor, "wb", closefd=True) as handle: handle.write(encoded) handle.flush() os.fsync(handle.fileno()) os.replace( temporary_name, target_name, src_dir_fd=parent_descriptor, dst_dir_fd=parent_descriptor, ) replaced = True published = os.open( target_name, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), dir_fd=parent_descriptor, ) try: if not stat.S_ISREG(os.fstat(published).st_mode): raise ValueError("published catalog target must remain regular") os.fchmod(published, 0o644) finally: os.close(published) os.fsync(parent_descriptor) finally: if not replaced: try: os.close(descriptor) except OSError: pass try: os.unlink(temporary_name, dir_fd=parent_descriptor) except FileNotFoundError: pass finally: os.close(parent_descriptor) def _trusted_relative_path(path: Path, *, trusted_root: Path) -> Path: try: relative = path.absolute().relative_to(trusted_root.absolute()) except ValueError as exc: raise ValueError("published catalog target leaves the trusted worktree") from exc if not relative.parts or any(part in {"", ".", ".."} for part in relative.parts): raise ValueError("published catalog target is not canonical") return relative def _open_relative_directory( trusted_root: Path, parts: tuple[str, ...], *, create: bool ) -> int: root_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr( os, "O_NOFOLLOW", 0 ) descriptor = os.open(trusted_root, root_flags) try: if not stat.S_ISDIR(os.fstat(descriptor).st_mode): raise ValueError("publication worktree root must be a directory") for part in parts: if part in {"", ".", ".."} or "/" in part: raise ValueError("published catalog path is not canonical") try: child = os.open(part, root_flags, dir_fd=descriptor) except FileNotFoundError: if not create: raise os.mkdir(part, mode=0o755, dir_fd=descriptor) child = os.open(part, root_flags, dir_fd=descriptor) os.close(descriptor) descriptor = child return descriptor except BaseException: os.close(descriptor) raise def _remove_public_tree(path: Path, *, trusted_root: Path) -> None: _trusted_relative_path(path, trusted_root=trusted_root) if not shutil.rmtree.avoids_symlink_attacks: raise RuntimeError("safe directory removal is unavailable on this platform") try: mode = path.lstat().st_mode except FileNotFoundError: return if stat.S_ISLNK(mode) or not stat.S_ISDIR(mode): raise ValueError("published module target must be a real directory") shutil.rmtree(path) def _expected_publication_blobs( *, web_root: Path, target_catalog: Path, candidate_payload: object, target_keyring: Path, keyring_payload: object, module_root: Path, directory_payloads: tuple[tuple[Path, dict[str, object]], ...], ) -> dict[str, bytes]: expected = { _trusted_relative_path( target_catalog, trusted_root=web_root ).as_posix(): _public_json_bytes(candidate_payload), _trusted_relative_path( target_keyring, trusted_root=web_root ).as_posix(): _public_json_bytes(keyring_payload), } for relative_path, payload in directory_payloads: target = module_root / relative_path repository_path = _trusted_relative_path( target, trusted_root=web_root ).as_posix() if repository_path in expected: raise ValueError("publication output paths are ambiguous") expected[repository_path] = _public_json_bytes(payload) return expected def committed_publication_sha(web_root: Path) -> str: commit_sha = git_text(web_root, "rev-parse", "--verify", "HEAD^{commit}") if re.fullmatch(r"[0-9a-f]{40,64}", commit_sha) is None: raise RuntimeError("could not resolve the publication commit") return commit_sha def verify_committed_publication( *, web_root: Path, commit_sha: str, expected_blobs: dict[str, bytes], module_root: Path, ) -> None: """Require the exact validated objects in one immutable commit tree.""" if re.fullmatch(r"[0-9a-f]{40,64}", commit_sha) is None: raise RuntimeError("publication commit identity is malformed") module_prefix = _trusted_relative_path( module_root, trusted_root=web_root ).as_posix() expected_modules = { path for path in expected_blobs if path.startswith(f"{module_prefix}/") } tree = subprocess.run( [ "git", "-C", str(web_root), "ls-tree", "-r", "--name-only", "-z", commit_sha, "--", module_prefix, ], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ).stdout observed_modules = { item.decode("utf-8") for item in tree.split(b"\0") if item } if observed_modules != expected_modules: raise RuntimeError( "committed module directory differs from validated publication outputs" ) for repository_path, expected in expected_blobs.items(): entry = subprocess.run( [ "git", "-C", str(web_root), "ls-tree", "-z", commit_sha, "--", repository_path, ], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ).stdout metadata, separator, encoded_path = entry.rstrip(b"\0").partition(b"\t") fields = metadata.split() if ( not separator or encoded_path.decode("utf-8") != repository_path or len(fields) != 3 or fields[0] != b"100644" or fields[1] != b"blob" or re.fullmatch(rb"[0-9a-f]{40,64}", fields[2]) is None ): raise RuntimeError( f"committed publication path is not one regular blob: {repository_path}" ) observed = subprocess.run( ["git", "-C", str(web_root), "cat-file", "blob", fields[2].decode()], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ).stdout if observed != expected: raise RuntimeError( f"committed publication blob differs from validated output: {repository_path}" ) def _reject_symlink_components(path: Path) -> None: absolute = path.absolute() for component in reversed((absolute, *absolute.parents)): try: mode = component.lstat().st_mode except FileNotFoundError: continue if stat.S_ISLNK(mode): raise ValueError("published catalog path must not traverse symlinks") def file_json_hash(path: Path) -> str | None: if not path.exists(): return None return canonical_hash(read_json(path)) def website_dirty(path: Path) -> bool: if not (path / ".git").exists(): return False return bool(git_text(path, "status", "--porcelain")) def default_tag_name(payload: object, *, channel: str) -> str | None: if not isinstance(payload, dict): return None sequence = payload.get("sequence") release = payload.get("release") selected_units = release.get("selected_units") if isinstance(release, dict) else None if isinstance(selected_units, list) and len(selected_units) == 1 and isinstance(selected_units[0], dict): repo = str(selected_units[0].get("repo") or "catalog") version = str(selected_units[0].get("version") or "").removeprefix("v") if version: return f"catalog-{channel}-{repo}-v{version}" return f"catalog-{channel}-{sequence}" if sequence is not None else None def commit_message(payload: object, *, channel: str) -> str: if isinstance(payload, dict): release = payload.get("release") selected_units = release.get("selected_units") if isinstance(release, dict) else None if isinstance(selected_units, list) and selected_units: units = ", ".join( f"{item.get('repo')} v{str(item.get('version') or '').removeprefix('v')}" for item in selected_units if isinstance(item, dict) ) if units: return f"Publish {channel} catalog for {units}" return f"Publish {channel} catalog" def push_command(web_root: Path, *, remote: str, branch: str | None, tag_name: str | None) -> str: if tag_name: return shlex.join(["git", "-C", str(web_root), "push", "--atomic", remote, f"HEAD:refs/heads/{branch or ''}", f"refs/tags/{tag_name}"]) return shlex.join(["git", "-C", str(web_root), "push", remote, f"HEAD:refs/heads/{branch or ''}"]) def replace_status(step: CatalogPublishStep, status: str) -> CatalogPublishStep: return CatalogPublishStep( id=step.id, title=step.title, detail=step.detail, command=step.command, mutating=step.mutating, status=status, ) def git_text(path: Path, *args: str) -> str: result = subprocess.run(["git", *args], cwd=path, check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) return result.stdout.strip() if result.returncode == 0 else "" def git_success(path: Path, *args: str) -> bool: return subprocess.run(["git", *args], cwd=path, check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0 def run_checked(args: list[str], *, cwd: Path) -> None: subprocess.run(args, cwd=cwd, check=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)