feat(release): synthesize initial catalog entries
This commit is contained in:
378
tools/release/govoplan_release/catalog_entry_synthesis.py
Normal file
378
tools/release/govoplan_release/catalog_entry_synthesis.py
Normal file
@@ -0,0 +1,378 @@
|
||||
"""Synthesize initial catalog entries from a repository's package and manifest."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
from contextlib import contextmanager
|
||||
import importlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
import tomllib
|
||||
from types import ModuleType
|
||||
from typing import Iterator
|
||||
|
||||
from govoplan_core.core.modules import ModuleManifest
|
||||
from govoplan_core.core.versioning import version_satisfies_range
|
||||
|
||||
from .workspace import load_repository_specs, resolve_repo_path
|
||||
|
||||
|
||||
_INSPECTION_CHILD = "GOVOPLAN_CATALOG_ENTRY_INSPECTION_CHILD"
|
||||
_INSPECTION_MARKER = "GOVOPLAN_CATALOG_ENTRY_JSON="
|
||||
|
||||
|
||||
def synthesize_repository_catalog_entries(
|
||||
*,
|
||||
repo: str,
|
||||
version: str,
|
||||
workspace: Path,
|
||||
repository_base: str,
|
||||
) -> tuple[dict[str, object], ...]:
|
||||
"""Build install entries from tagged, preflighted local source metadata.
|
||||
|
||||
The caller owns source-tag and worktree provenance checks. This function
|
||||
accepts no hand-maintained module catalog registry: distribution metadata
|
||||
identifies the runtime entry point and the runtime ``ModuleManifest`` is
|
||||
the authoritative dependency/interface/frontend description.
|
||||
"""
|
||||
|
||||
if os.getenv(_INSPECTION_CHILD) == "1":
|
||||
return synthesize_repository_catalog_entries_in_process(
|
||||
repo=repo,
|
||||
version=version,
|
||||
workspace=workspace,
|
||||
repository_base=repository_base,
|
||||
)
|
||||
command = (
|
||||
sys.executable,
|
||||
"-m",
|
||||
"govoplan_release.catalog_entry_synthesis",
|
||||
"--inspect-repo",
|
||||
repo,
|
||||
"--version",
|
||||
version,
|
||||
"--workspace",
|
||||
str(workspace),
|
||||
"--repository-base",
|
||||
repository_base,
|
||||
)
|
||||
environment = os.environ.copy()
|
||||
environment[_INSPECTION_CHILD] = "1"
|
||||
release_root = str(Path(__file__).resolve().parents[1])
|
||||
environment["PYTHONPATH"] = os.pathsep.join(
|
||||
item for item in (release_root, environment.get("PYTHONPATH", "")) if item
|
||||
)
|
||||
try:
|
||||
result = subprocess.run(
|
||||
command,
|
||||
check=False,
|
||||
text=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
timeout=30,
|
||||
env=environment,
|
||||
)
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
raise ValueError(f"Cannot synthesize {repo}: isolated manifest inspection timed out.") from exc
|
||||
if result.returncode != 0:
|
||||
detail = result.stderr.strip() or result.stdout.strip() or "unknown inspection error"
|
||||
raise ValueError(f"Cannot synthesize {repo}: isolated manifest inspection failed: {detail}")
|
||||
payload_line = next(
|
||||
(line.removeprefix(_INSPECTION_MARKER) for line in reversed(result.stdout.splitlines()) if line.startswith(_INSPECTION_MARKER)),
|
||||
None,
|
||||
)
|
||||
if payload_line is None:
|
||||
raise ValueError(f"Cannot synthesize {repo}: isolated manifest inspection returned no metadata.")
|
||||
payload = json.loads(payload_line)
|
||||
if not isinstance(payload, list) or not all(isinstance(item, dict) for item in payload):
|
||||
raise ValueError(f"Cannot synthesize {repo}: isolated manifest inspection returned invalid metadata.")
|
||||
return tuple(payload)
|
||||
|
||||
|
||||
def synthesize_repository_catalog_entries_in_process(
|
||||
*,
|
||||
repo: str,
|
||||
version: str,
|
||||
workspace: Path,
|
||||
repository_base: str,
|
||||
) -> tuple[dict[str, object], ...]:
|
||||
specs = {item.name: item for item in load_repository_specs(include_website=False)}
|
||||
spec = specs.get(repo)
|
||||
if spec is None:
|
||||
raise ValueError(f"Cannot synthesize {repo}: repository is not registered.")
|
||||
root = resolve_repo_path(spec, workspace)
|
||||
project = read_project_metadata(root / "pyproject.toml")
|
||||
package = required_text(project, "name", source=f"{repo}/pyproject.toml")
|
||||
project_version = required_text(project, "version", source=f"{repo}/pyproject.toml").removeprefix("v")
|
||||
expected_version = version.removeprefix("v")
|
||||
if project_version != expected_version:
|
||||
raise ValueError(
|
||||
f"Cannot synthesize {repo}: project version {project_version!r} does not match selected version {expected_version!r}."
|
||||
)
|
||||
description = optional_text(project.get("description"))
|
||||
entry_points = module_entry_points(project, repo=repo)
|
||||
entries: list[dict[str, object]] = []
|
||||
for declared_module_id, target in sorted(entry_points.items()):
|
||||
manifest = load_manifest(root=root, target=target, repo=repo)
|
||||
if manifest.id != declared_module_id:
|
||||
raise ValueError(
|
||||
f"Cannot synthesize {repo}: entry point {declared_module_id!r} returns manifest {manifest.id!r}."
|
||||
)
|
||||
if manifest.version.removeprefix("v") != expected_version:
|
||||
raise ValueError(
|
||||
f"Cannot synthesize {repo}/{manifest.id}: manifest version {manifest.version!r} does not match {expected_version!r}."
|
||||
)
|
||||
entry = manifest_catalog_entry(
|
||||
manifest=manifest,
|
||||
repo=repo,
|
||||
package=package,
|
||||
version=expected_version,
|
||||
description=description,
|
||||
root=root,
|
||||
repository_base=repository_base.rstrip("/"),
|
||||
)
|
||||
entries.append(entry)
|
||||
return tuple(entries)
|
||||
|
||||
|
||||
def manifest_catalog_entry(
|
||||
*,
|
||||
manifest: ModuleManifest,
|
||||
repo: str,
|
||||
package: str,
|
||||
version: str,
|
||||
description: str | None,
|
||||
root: Path,
|
||||
repository_base: str,
|
||||
) -> dict[str, object]:
|
||||
tag = f"v{version}"
|
||||
entry: dict[str, object] = {
|
||||
"module_id": manifest.id,
|
||||
"name": manifest.name,
|
||||
"version": version,
|
||||
"action": "install",
|
||||
"python_package": package,
|
||||
"python_ref": f"{package} @ {repository_base}/{repo}.git@{tag}",
|
||||
"license_features": [f"module.{manifest.id}"],
|
||||
"tags": ["official"],
|
||||
}
|
||||
if description:
|
||||
entry["description"] = description
|
||||
if manifest.dependencies:
|
||||
entry["dependencies"] = list(manifest.dependencies)
|
||||
if manifest.optional_dependencies:
|
||||
entry["optional_dependencies"] = list(manifest.optional_dependencies)
|
||||
if manifest.provides_interfaces:
|
||||
entry["provides_interfaces"] = [
|
||||
{"name": item.name, "version": item.version}
|
||||
for item in manifest.provides_interfaces
|
||||
]
|
||||
if manifest.requires_interfaces:
|
||||
entry["requires_interfaces"] = [
|
||||
{
|
||||
"name": item.name,
|
||||
**({"version_min": item.version_min} if item.version_min else {}),
|
||||
**({"version_max_exclusive": item.version_max_exclusive} if item.version_max_exclusive else {}),
|
||||
"optional": item.optional,
|
||||
}
|
||||
for item in manifest.requires_interfaces
|
||||
]
|
||||
if manifest.frontend is not None and manifest.frontend.package_name:
|
||||
validate_webui_package(root=root, package=manifest.frontend.package_name, version=version, repo=repo)
|
||||
entry["webui_package"] = manifest.frontend.package_name
|
||||
entry["webui_ref"] = f"{repository_base}/{repo}.git#{tag}"
|
||||
if manifest.migration_spec is not None:
|
||||
migration = manifest.migration_spec
|
||||
entry["migration_safety"] = "requires_review"
|
||||
entry["migration_notes"] = (
|
||||
"Module owns database migrations; review release notes and migration output before activation."
|
||||
)
|
||||
if migration.migration_after:
|
||||
entry["migration_after"] = list(migration.migration_after)
|
||||
if migration.migration_before:
|
||||
entry["migration_before"] = list(migration.migration_before)
|
||||
if migration.migration_tasks:
|
||||
entry["migration_tasks"] = [
|
||||
{
|
||||
"task_id": task.task_id,
|
||||
"phase": task.phase,
|
||||
"summary": task.summary,
|
||||
"task_version": task.task_version,
|
||||
"safety": task.safety,
|
||||
"idempotent": task.idempotent,
|
||||
**({"timeout_seconds": task.timeout_seconds} if task.timeout_seconds is not None else {}),
|
||||
}
|
||||
for task in migration.migration_tasks
|
||||
]
|
||||
return entry
|
||||
|
||||
|
||||
def validate_initial_entry_closure(
|
||||
*,
|
||||
catalog_modules: list[object],
|
||||
initial_module_ids: set[str],
|
||||
) -> None:
|
||||
entries = [item for item in catalog_modules if isinstance(item, dict)]
|
||||
by_id = {str(item.get("module_id") or ""): item for item in entries}
|
||||
providers: dict[str, list[str]] = {}
|
||||
for entry in entries:
|
||||
for raw in entry.get("provides_interfaces") or ():
|
||||
if isinstance(raw, dict) and isinstance(raw.get("name"), str) and isinstance(raw.get("version"), str):
|
||||
providers.setdefault(raw["name"], []).append(raw["version"])
|
||||
|
||||
issues: list[str] = []
|
||||
for module_id in sorted(initial_module_ids):
|
||||
entry = by_id[module_id]
|
||||
for dependency in entry.get("dependencies") or ():
|
||||
if isinstance(dependency, str) and dependency not in by_id:
|
||||
issues.append(f"{module_id} requires catalog module {dependency!r}")
|
||||
for raw in entry.get("requires_interfaces") or ():
|
||||
if not isinstance(raw, dict) or raw.get("optional") is True:
|
||||
continue
|
||||
name = raw.get("name")
|
||||
if not isinstance(name, str):
|
||||
continue
|
||||
available = providers.get(name, [])
|
||||
if not any(
|
||||
version_satisfies_range(
|
||||
provider_version,
|
||||
version_min=raw.get("version_min") if isinstance(raw.get("version_min"), str) else None,
|
||||
version_max_exclusive=raw.get("version_max_exclusive")
|
||||
if isinstance(raw.get("version_max_exclusive"), str)
|
||||
else None,
|
||||
)
|
||||
for provider_version in available
|
||||
):
|
||||
issues.append(f"{module_id} requires unavailable compatible interface {name!r}")
|
||||
if issues:
|
||||
raise ValueError("Initial catalog entry dependency closure failed: " + "; ".join(issues))
|
||||
|
||||
|
||||
def read_project_metadata(path: Path) -> dict[str, object]:
|
||||
if not path.exists():
|
||||
raise ValueError(f"Cannot synthesize catalog entry: missing {path}.")
|
||||
with path.open("rb") as handle:
|
||||
payload = tomllib.load(handle)
|
||||
project = payload.get("project")
|
||||
if not isinstance(project, dict):
|
||||
raise ValueError(f"Cannot synthesize catalog entry: {path} has no [project] table.")
|
||||
return project
|
||||
|
||||
|
||||
def module_entry_points(project: dict[str, object], *, repo: str) -> dict[str, str]:
|
||||
groups = project.get("entry-points")
|
||||
raw = groups.get("govoplan.modules") if isinstance(groups, dict) else None
|
||||
if not isinstance(raw, dict) or not raw:
|
||||
raise ValueError(f"Cannot synthesize {repo}: pyproject has no govoplan.modules entry point.")
|
||||
result = {
|
||||
str(module_id).strip(): str(target).strip()
|
||||
for module_id, target in raw.items()
|
||||
if str(module_id).strip() and str(target).strip()
|
||||
}
|
||||
if len(result) != len(raw):
|
||||
raise ValueError(f"Cannot synthesize {repo}: govoplan.modules contains an empty id or target.")
|
||||
return result
|
||||
|
||||
|
||||
def load_manifest(*, root: Path, target: str, repo: str) -> ModuleManifest:
|
||||
module_name, separator, attribute = target.partition(":")
|
||||
if not separator or not module_name or not attribute or "[" in attribute:
|
||||
raise ValueError(f"Cannot synthesize {repo}: unsupported module entry point {target!r}.")
|
||||
src = (root / "src").resolve()
|
||||
expected_module = (src / Path(*module_name.split("."))).with_suffix(".py")
|
||||
expected_package = src / Path(*module_name.split(".")) / "__init__.py"
|
||||
expected = expected_module if expected_module.exists() else expected_package
|
||||
if not expected.exists():
|
||||
raise ValueError(f"Cannot synthesize {repo}: entry point source {expected_module} is missing.")
|
||||
|
||||
top_level = module_name.split(".", 1)[0]
|
||||
with isolated_source_import(src=src, top_level=top_level):
|
||||
try:
|
||||
module = importlib.import_module(module_name)
|
||||
except Exception as exc: # noqa: BLE001 - exact load failure is release evidence.
|
||||
raise ValueError(f"Cannot synthesize {repo}: loading {target!r} failed: {exc}") from exc
|
||||
origin = Path(str(getattr(module, "__file__", ""))).resolve()
|
||||
if origin != expected.resolve():
|
||||
raise ValueError(f"Cannot synthesize {repo}: entry point resolved outside selected source ({origin}).")
|
||||
factory = getattr(module, attribute, None)
|
||||
if not callable(factory):
|
||||
raise ValueError(f"Cannot synthesize {repo}: entry point target {target!r} is not callable.")
|
||||
manifest = factory()
|
||||
if not isinstance(manifest, ModuleManifest):
|
||||
raise ValueError(f"Cannot synthesize {repo}: entry point {target!r} did not return ModuleManifest.")
|
||||
return manifest
|
||||
|
||||
|
||||
@contextmanager
|
||||
def isolated_source_import(*, src: Path, top_level: str) -> Iterator[None]:
|
||||
saved: dict[str, ModuleType] = {
|
||||
name: module
|
||||
for name, module in tuple(sys.modules.items())
|
||||
if name == top_level or name.startswith(f"{top_level}.")
|
||||
}
|
||||
for name in saved:
|
||||
sys.modules.pop(name, None)
|
||||
sys.path.insert(0, str(src))
|
||||
importlib.invalidate_caches()
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
sys.path.remove(str(src))
|
||||
for name in tuple(sys.modules):
|
||||
if name == top_level or name.startswith(f"{top_level}."):
|
||||
sys.modules.pop(name, None)
|
||||
sys.modules.update(saved)
|
||||
importlib.invalidate_caches()
|
||||
|
||||
|
||||
def validate_webui_package(*, root: Path, package: str, version: str, repo: str) -> None:
|
||||
declarations: list[tuple[str, str]] = []
|
||||
for path in (root / "package.json", root / "webui" / "package.json"):
|
||||
if not path.exists():
|
||||
continue
|
||||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||||
if isinstance(payload, dict) and isinstance(payload.get("name"), str) and isinstance(payload.get("version"), str):
|
||||
declarations.append((payload["name"], payload["version"]))
|
||||
if (package, version) not in declarations:
|
||||
raise ValueError(
|
||||
f"Cannot synthesize {repo}: frontend manifest declares {package}@{version}, but package metadata does not."
|
||||
)
|
||||
|
||||
|
||||
def required_text(payload: dict[str, object], key: str, *, source: str) -> str:
|
||||
value = optional_text(payload.get(key))
|
||||
if value is None:
|
||||
raise ValueError(f"Cannot synthesize catalog entry: {source} has no {key!r}.")
|
||||
return value
|
||||
|
||||
|
||||
def optional_text(value: object) -> str | None:
|
||||
return value.strip() if isinstance(value, str) and value.strip() else None
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(add_help=False)
|
||||
parser.add_argument("--inspect-repo", required=True)
|
||||
parser.add_argument("--version", required=True)
|
||||
parser.add_argument("--workspace", type=Path, required=True)
|
||||
parser.add_argument("--repository-base", required=True)
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
entries = synthesize_repository_catalog_entries(
|
||||
repo=args.inspect_repo,
|
||||
version=args.version,
|
||||
workspace=args.workspace.resolve(),
|
||||
repository_base=args.repository_base,
|
||||
)
|
||||
except ValueError as exc:
|
||||
print(str(exc), file=sys.stderr)
|
||||
return 1
|
||||
print(_INSPECTION_MARKER + json.dumps(entries, sort_keys=True))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user