446 lines
18 KiB
Python
446 lines
18 KiB
Python
"""Manifest contract extraction and validation for release planning.
|
|
|
|
The static scanner intentionally parses manifest source files with ``ast``
|
|
instead of importing module packages. That keeps contract checks usable before
|
|
``pip install`` has been rerun for a changed ``pyproject.toml`` or entry point.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from .model import (
|
|
CompatibilityIssue,
|
|
InterfaceProviderSnapshot,
|
|
InterfaceRequirementSnapshot,
|
|
ModuleContractSnapshot,
|
|
RepositorySnapshot,
|
|
)
|
|
|
|
|
|
def collect_contracts(repositories: tuple[RepositorySnapshot, ...]) -> tuple[ModuleContractSnapshot, ...]:
|
|
contracts: list[ModuleContractSnapshot] = []
|
|
for repo in repositories:
|
|
if not repo.exists or not repo.is_git:
|
|
continue
|
|
root = Path(repo.absolute_path)
|
|
for manifest in sorted(root.glob("src/**/backend/manifest.py")):
|
|
contract = parse_manifest_contract(manifest, repo_name=repo.spec.name)
|
|
if contract is not None:
|
|
contracts.append(contract)
|
|
return tuple(contracts)
|
|
|
|
|
|
def validate_contracts(contracts: tuple[ModuleContractSnapshot, ...]) -> tuple[CompatibilityIssue, ...]:
|
|
"""Validate the statically collected module interface graph.
|
|
|
|
This mirrors the runtime interface closure validation in
|
|
``govoplan_core.core.registry`` but reports all issues at once for CI and
|
|
release planning.
|
|
"""
|
|
|
|
issues: list[CompatibilityIssue] = []
|
|
providers: dict[str, list[tuple[ModuleContractSnapshot, InterfaceProviderSnapshot]]] = {}
|
|
module_ids: dict[str, list[ModuleContractSnapshot]] = {}
|
|
|
|
for contract in contracts:
|
|
module_ids.setdefault(contract.module_id, []).append(contract)
|
|
local_provider_names: set[str] = set()
|
|
local_requirement_names: set[str] = set()
|
|
|
|
for provider in contract.provides_interfaces:
|
|
if provider.name in local_provider_names:
|
|
issues.append(
|
|
CompatibilityIssue(
|
|
severity="blocker",
|
|
code="duplicate_provided_interface",
|
|
message=f"{contract.module_id} provides interface {provider.name!r} more than once.",
|
|
repo=contract.repo,
|
|
module_id=contract.module_id,
|
|
interface=provider.name,
|
|
)
|
|
)
|
|
local_provider_names.add(provider.name)
|
|
if not valid_interface_name(provider.name):
|
|
issues.append(
|
|
CompatibilityIssue(
|
|
severity="blocker",
|
|
code="invalid_provided_interface_name",
|
|
message=f"{contract.module_id} provides invalid interface name {provider.name!r}.",
|
|
repo=contract.repo,
|
|
module_id=contract.module_id,
|
|
interface=provider.name,
|
|
)
|
|
)
|
|
if not provider.version.strip():
|
|
issues.append(
|
|
CompatibilityIssue(
|
|
severity="blocker",
|
|
code="missing_provided_interface_version",
|
|
message=f"{contract.module_id} provides interface {provider.name!r} without a version.",
|
|
repo=contract.repo,
|
|
module_id=contract.module_id,
|
|
interface=provider.name,
|
|
)
|
|
)
|
|
providers.setdefault(provider.name, []).append((contract, provider))
|
|
|
|
for requirement in contract.requires_interfaces:
|
|
if requirement.name in local_requirement_names:
|
|
issues.append(
|
|
CompatibilityIssue(
|
|
severity="blocker",
|
|
code="duplicate_required_interface",
|
|
message=f"{contract.module_id} requires interface {requirement.name!r} more than once.",
|
|
repo=contract.repo,
|
|
module_id=contract.module_id,
|
|
interface=requirement.name,
|
|
)
|
|
)
|
|
local_requirement_names.add(requirement.name)
|
|
if not valid_interface_name(requirement.name):
|
|
issues.append(
|
|
CompatibilityIssue(
|
|
severity="blocker",
|
|
code="invalid_required_interface_name",
|
|
message=f"{contract.module_id} requires invalid interface name {requirement.name!r}.",
|
|
repo=contract.repo,
|
|
module_id=contract.module_id,
|
|
interface=requirement.name,
|
|
)
|
|
)
|
|
if requirement.version_min is not None and not requirement.version_min.strip():
|
|
issues.append(
|
|
CompatibilityIssue(
|
|
severity="blocker",
|
|
code="blank_required_interface_minimum",
|
|
message=f"{contract.module_id} has a blank minimum version for {requirement.name!r}.",
|
|
repo=contract.repo,
|
|
module_id=contract.module_id,
|
|
interface=requirement.name,
|
|
)
|
|
)
|
|
if requirement.version_max_exclusive is not None and not requirement.version_max_exclusive.strip():
|
|
issues.append(
|
|
CompatibilityIssue(
|
|
severity="blocker",
|
|
code="blank_required_interface_maximum",
|
|
message=f"{contract.module_id} has a blank maximum version for {requirement.name!r}.",
|
|
repo=contract.repo,
|
|
module_id=contract.module_id,
|
|
interface=requirement.name,
|
|
)
|
|
)
|
|
if not version_range_is_valid(
|
|
version_min=requirement.version_min,
|
|
version_max_exclusive=requirement.version_max_exclusive,
|
|
):
|
|
issues.append(
|
|
CompatibilityIssue(
|
|
severity="blocker",
|
|
code="invalid_required_interface_range",
|
|
message=(
|
|
f"{contract.module_id} requires {requirement.name!r} with invalid range "
|
|
f"{format_version_range(version_min=requirement.version_min, version_max_exclusive=requirement.version_max_exclusive)}."
|
|
),
|
|
repo=contract.repo,
|
|
module_id=contract.module_id,
|
|
interface=requirement.name,
|
|
)
|
|
)
|
|
|
|
for module_id, matches in sorted(module_ids.items()):
|
|
if len(matches) > 1:
|
|
repos = ", ".join(sorted(contract.repo for contract in matches))
|
|
for contract in matches:
|
|
issues.append(
|
|
CompatibilityIssue(
|
|
severity="blocker",
|
|
code="duplicate_module_id",
|
|
message=f"Module id {module_id!r} is declared by multiple repositories: {repos}.",
|
|
repo=contract.repo,
|
|
module_id=module_id,
|
|
)
|
|
)
|
|
|
|
for contract in contracts:
|
|
for requirement in contract.requires_interfaces:
|
|
available = providers.get(requirement.name, [])
|
|
matching = [
|
|
(provider_contract, provider)
|
|
for provider_contract, provider in available
|
|
if version_satisfies_range(
|
|
provider.version,
|
|
version_min=requirement.version_min,
|
|
version_max_exclusive=requirement.version_max_exclusive,
|
|
)
|
|
]
|
|
if matching:
|
|
continue
|
|
version_range = format_version_range(
|
|
version_min=requirement.version_min,
|
|
version_max_exclusive=requirement.version_max_exclusive,
|
|
)
|
|
if not available:
|
|
if requirement.optional:
|
|
continue
|
|
issues.append(
|
|
CompatibilityIssue(
|
|
severity="blocker",
|
|
code="required_interface_missing",
|
|
message=f"{contract.module_id} requires unavailable interface {requirement.name!r} ({version_range}).",
|
|
repo=contract.repo,
|
|
module_id=contract.module_id,
|
|
interface=requirement.name,
|
|
)
|
|
)
|
|
continue
|
|
available_versions = ", ".join(
|
|
f"{provider_contract.module_id}@{provider.version}" for provider_contract, provider in available
|
|
)
|
|
issues.append(
|
|
CompatibilityIssue(
|
|
severity="warning" if requirement.optional else "blocker",
|
|
code="interface_version_mismatch",
|
|
message=(
|
|
f"{contract.module_id} requires interface {requirement.name!r} ({version_range}) "
|
|
f"but available providers are {available_versions}."
|
|
),
|
|
repo=contract.repo,
|
|
module_id=contract.module_id,
|
|
interface=requirement.name,
|
|
)
|
|
)
|
|
|
|
return tuple(sorted(issues, key=issue_sort_key))
|
|
|
|
|
|
def interface_impact_map(
|
|
contracts: tuple[ModuleContractSnapshot, ...],
|
|
) -> dict[str, tuple[tuple[ModuleContractSnapshot, InterfaceRequirementSnapshot], ...]]:
|
|
consumers: dict[str, list[tuple[ModuleContractSnapshot, InterfaceRequirementSnapshot]]] = {}
|
|
for contract in contracts:
|
|
for requirement in contract.requires_interfaces:
|
|
consumers.setdefault(requirement.name, []).append((contract, requirement))
|
|
return {name: tuple(items) for name, items in sorted(consumers.items())}
|
|
|
|
|
|
def provided_interface_map(
|
|
contracts: tuple[ModuleContractSnapshot, ...],
|
|
) -> dict[str, tuple[tuple[ModuleContractSnapshot, InterfaceProviderSnapshot], ...]]:
|
|
providers: dict[str, list[tuple[ModuleContractSnapshot, InterfaceProviderSnapshot]]] = {}
|
|
for contract in contracts:
|
|
for provider in contract.provides_interfaces:
|
|
providers.setdefault(provider.name, []).append((contract, provider))
|
|
return {name: tuple(items) for name, items in sorted(providers.items())}
|
|
|
|
|
|
def parse_manifest_contract(path: Path, *, repo_name: str) -> ModuleContractSnapshot | None:
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
constants = imported_string_constants(tree, path)
|
|
constants.update(module_constants(tree, initial=constants))
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.Call):
|
|
continue
|
|
if call_name(node.func) != "ModuleManifest":
|
|
continue
|
|
keywords = {keyword.arg: keyword.value for keyword in node.keywords if keyword.arg}
|
|
module_id = value_as_string(keywords.get("id"), constants)
|
|
if module_id is None:
|
|
continue
|
|
return ModuleContractSnapshot(
|
|
repo=repo_name,
|
|
module_id=module_id,
|
|
module_version=value_as_string(keywords.get("version"), constants),
|
|
manifest_path=str(path),
|
|
provides_interfaces=parse_providers(keywords.get("provides_interfaces"), constants),
|
|
requires_interfaces=parse_requirements(keywords.get("requires_interfaces"), constants),
|
|
)
|
|
return None
|
|
|
|
|
|
def parse_providers(node: ast.AST | None, constants: dict[str, str]) -> tuple[InterfaceProviderSnapshot, ...]:
|
|
providers: list[InterfaceProviderSnapshot] = []
|
|
for call in interface_calls(node, "ModuleInterfaceProvider"):
|
|
keywords = {keyword.arg: keyword.value for keyword in call.keywords if keyword.arg}
|
|
name = value_as_string(keywords.get("name"), constants)
|
|
version = value_as_string(keywords.get("version"), constants)
|
|
if name and version:
|
|
providers.append(InterfaceProviderSnapshot(name=name, version=version))
|
|
return tuple(providers)
|
|
|
|
|
|
def parse_requirements(node: ast.AST | None, constants: dict[str, str]) -> tuple[InterfaceRequirementSnapshot, ...]:
|
|
requirements: list[InterfaceRequirementSnapshot] = []
|
|
for call in interface_calls(node, "ModuleInterfaceRequirement"):
|
|
keywords = {keyword.arg: keyword.value for keyword in call.keywords if keyword.arg}
|
|
name = value_as_string(keywords.get("name"), constants)
|
|
if not name:
|
|
continue
|
|
requirements.append(
|
|
InterfaceRequirementSnapshot(
|
|
name=name,
|
|
version_min=value_as_string(keywords.get("version_min"), constants),
|
|
version_max_exclusive=value_as_string(keywords.get("version_max_exclusive"), constants),
|
|
optional=value_as_bool(keywords.get("optional")),
|
|
)
|
|
)
|
|
return tuple(requirements)
|
|
|
|
|
|
def interface_calls(node: ast.AST | None, expected_name: str) -> tuple[ast.Call, ...]:
|
|
if node is None:
|
|
return ()
|
|
return tuple(
|
|
child
|
|
for child in ast.walk(node)
|
|
if isinstance(child, ast.Call) and call_name(child.func) == expected_name
|
|
)
|
|
|
|
|
|
def imported_string_constants(tree: ast.Module, path: Path) -> dict[str, str]:
|
|
constants: dict[str, str] = {}
|
|
for node in tree.body:
|
|
if not isinstance(node, ast.ImportFrom):
|
|
continue
|
|
source = import_source_path(path, module=node.module, level=node.level)
|
|
if source is None or not source.exists():
|
|
continue
|
|
try:
|
|
imported = module_constants(ast.parse(source.read_text(encoding="utf-8"), filename=str(source)))
|
|
except SyntaxError:
|
|
continue
|
|
for alias in node.names:
|
|
if alias.name in imported:
|
|
constants[alias.asname or alias.name] = imported[alias.name]
|
|
return constants
|
|
|
|
|
|
def import_source_path(path: Path, *, module: str | None, level: int) -> Path | None:
|
|
if level:
|
|
base = path.parent
|
|
for _ in range(level - 1):
|
|
base = base.parent
|
|
if module:
|
|
base = base.joinpath(*module.split("."))
|
|
candidate = base.with_suffix(".py")
|
|
else:
|
|
src_root = next((parent for parent in path.parents if parent.name == "src"), None)
|
|
if src_root is None or module is None:
|
|
return None
|
|
candidate = src_root.joinpath(*module.split(".")).with_suffix(".py")
|
|
if candidate.exists():
|
|
return candidate
|
|
init = candidate.with_suffix("") / "__init__.py"
|
|
return init if init.exists() else candidate
|
|
|
|
|
|
def module_constants(tree: ast.Module, *, initial: dict[str, str] | None = None) -> dict[str, str]:
|
|
constants: dict[str, str] = dict(initial or {})
|
|
for node in tree.body:
|
|
if isinstance(node, ast.Assign):
|
|
value = node.value
|
|
for target in node.targets:
|
|
if isinstance(target, ast.Name):
|
|
resolved = value_as_string(value, constants)
|
|
if resolved is not None:
|
|
constants[target.id] = resolved
|
|
elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
|
|
resolved = value_as_string(node.value, constants)
|
|
if resolved is not None:
|
|
constants[node.target.id] = resolved
|
|
return constants
|
|
|
|
|
|
def value_as_string(node: ast.AST | None, constants: dict[str, str]) -> str | None:
|
|
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
|
return node.value
|
|
if isinstance(node, ast.Name):
|
|
return constants.get(node.id)
|
|
return None
|
|
|
|
|
|
def value_as_bool(node: ast.AST | None) -> bool:
|
|
return bool(node.value) if isinstance(node, ast.Constant) and isinstance(node.value, bool) else False
|
|
|
|
|
|
def call_name(node: ast.AST) -> str:
|
|
if isinstance(node, ast.Name):
|
|
return node.id
|
|
if isinstance(node, ast.Attribute):
|
|
return node.attr
|
|
return ""
|
|
|
|
|
|
def issue_sort_key(issue: CompatibilityIssue) -> tuple[int, str, str, str, str]:
|
|
severity_order = {"blocker": 0, "warning": 1, "info": 2}
|
|
return (
|
|
severity_order.get(issue.severity, 9),
|
|
issue.repo or "",
|
|
issue.module_id or "",
|
|
issue.interface or "",
|
|
issue.code,
|
|
)
|
|
|
|
|
|
def valid_interface_name(value: str) -> bool:
|
|
return bool(re.match(r"^[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)+$", value))
|
|
|
|
|
|
def version_satisfies_range(
|
|
version: str,
|
|
*,
|
|
version_min: str | None = None,
|
|
version_max_exclusive: str | None = None,
|
|
) -> bool:
|
|
if version_min is not None and compare_versions(version, version_min) < 0:
|
|
return False
|
|
if version_max_exclusive is not None and compare_versions(version, version_max_exclusive) >= 0:
|
|
return False
|
|
return True
|
|
|
|
|
|
def version_range_is_valid(
|
|
*,
|
|
version_min: str | None = None,
|
|
version_max_exclusive: str | None = None,
|
|
) -> bool:
|
|
if version_min is None or version_max_exclusive is None:
|
|
return True
|
|
return compare_versions(version_min, version_max_exclusive) < 0
|
|
|
|
|
|
def format_version_range(
|
|
*,
|
|
version_min: str | None = None,
|
|
version_max_exclusive: str | None = None,
|
|
) -> str:
|
|
parts: list[str] = []
|
|
if version_min is not None:
|
|
parts.append(f">= {version_min}")
|
|
if version_max_exclusive is not None:
|
|
parts.append(f"< {version_max_exclusive}")
|
|
return ", ".join(parts) if parts else "any version"
|
|
|
|
|
|
def compare_versions(left: str, right: str) -> int:
|
|
left_parts = version_tuple(left)
|
|
right_parts = version_tuple(right)
|
|
max_length = max(len(left_parts), len(right_parts))
|
|
padded_left = left_parts + (0,) * (max_length - len(left_parts))
|
|
padded_right = right_parts + (0,) * (max_length - len(right_parts))
|
|
if padded_left < padded_right:
|
|
return -1
|
|
if padded_left > padded_right:
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def version_tuple(value: str) -> tuple[int, ...]:
|
|
match = re.match(r"^\s*v?(\d+(?:\.\d+)*)", value)
|
|
if not match:
|
|
return (0,)
|
|
return tuple(int(part) for part in match.group(1).split("."))
|