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

chore(release): coordinate integrity and performance source updates

Browse Source 
Release v0.1.46. Coordinated integrity review: GovOPlaN/govoplan-core#298.
This commit is contained in:
zemion
2026-09-08 12:36:36 +02:00
parent 58d320d9b3
commit 845dcbafdb
5 changed files with 453 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/release/release-migration-audit.py
+190 -15
View File
@@ -8,7 +8,7 @@ from datetime import datetime, timezone
import json
import os
from pathlib import Path
from typing import Any
from typing import Any, NoReturn
META_ROOT = Path(__file__).resolve().parents[2]
@@ -168,12 +168,13 @@ def owner_for_versions_dir(versions_dir: Path) -> str:
def parse_migration_file(owner: str, path: Path) -> Migration | None:
if path.stat().st_size > 2 * 1024 * 1024:
raise ValueError(f"{path}: migration source exceeds the 2 MiB audit bound")
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
values: dict[str, Any] = {}
wrapped: Migration | None = None
release_peer = path.parent.parent / "versions" / path.name
if path.parent.name == "dev_versions" and release_peer.is_file():
wrapped = parse_migration_file(owner, release_peer)
values = _imported_release_metadata(owner, path, tree)
wrapped = _wrapped_release_metadata(owner, path, tree)
if values and wrapped:
raise ValueError(f"{path}: mixed migration metadata wrapper styles are ambiguous")
for statement in tree.body:
if isinstance(statement, ast.Assign):
for target in statement.targets:
@@ -196,6 +197,8 @@ def parse_migration_file(owner: str, path: Path) -> Migration | None:
)
revision = values.get("revision")
if not isinstance(revision, str):
if "revision" in values:
raise ValueError(f"{path}: migration revision must be an explicit string")
return None
return Migration(
owner=owner,
@@ -207,28 +210,200 @@ def parse_migration_file(owner: str, path: Path) -> Migration | None:
)
def _ast_binding_count(tree: ast.Module, name: str) -> int:
count = 0
for node in ast.walk(tree):
if isinstance(node, ast.Name) and node.id == name and isinstance(node.ctx, (ast.Store, ast.Del)):
count += 1
elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node.name == name:
count += 1
elif isinstance(node, (ast.Import, ast.ImportFrom)):
count += sum(
(alias.asname or (alias.name.split(".")[0] if isinstance(node, ast.Import) else alias.name)) == name
for alias in node.names
)
return count
def _release_wrapper_peer(owner: str, path: Path, filename: str) -> Migration:
versions = path.parent.parent / "versions"
peer = versions / filename
if (
path.parent.name != "dev_versions"
or Path(filename).name != filename or not filename.endswith(".py")
or versions.is_symlink() or peer.is_symlink() or not peer.is_file()
or peer.resolve().parent != versions.resolve()
or peer.stat().st_size > 2 * 1024 * 1024
):
raise ValueError(f"{path}: unsupported or ambiguous development migration wrapper")
migration = parse_migration_file(owner, peer)
if migration is None:
raise ValueError(f"{path}: release wrapper target has no migration metadata")
return migration
def _imported_release_metadata(owner: str, path: Path, tree: ast.Module) -> dict[str, Any]:
"""Recognize explicit metadata re-exports, never star/dynamic imports."""
names = {"revision", "down_revision", "depends_on", "branch_labels"}
prefix = f"{owner.replace('-', '_')}.backend.migrations.versions."
values: dict[str, Any] = {}
targets: set[str] = set()
for statement in tree.body:
if not isinstance(statement, ast.ImportFrom):
continue
selected = [alias for alias in statement.names if alias.name in names or (alias.asname or alias.name) in names]
if not selected:
if (statement.module or "").startswith(prefix) and any(alias.name == "*" for alias in statement.names):
raise ValueError(f"{path}: unsupported wildcard migration metadata")
continue
stem = (statement.module or "").removeprefix(prefix)
if (
statement.level or not (statement.module or "").startswith(prefix)
or not stem or any(character not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_" for character in stem)
):
raise ValueError(f"{path}: unsupported or ambiguous imported migration metadata")
targets.add(stem)
if len(targets) != 1:
raise ValueError(f"{path}: ambiguous imported migration metadata sources")
migration = _release_wrapper_peer(owner, path, stem + ".py")
projection = {
"revision": migration.revision, "down_revision": migration.down_revisions,
"depends_on": migration.depends_on, "branch_labels": migration.branch_labels,
}
for alias in selected:
name = alias.asname or alias.name
if name != alias.name or name in values or _ast_binding_count(tree, name) != 1:
raise ValueError(f"{path}: ambiguous imported migration metadata assignment")
values[name] = projection[name]
return values
def _wrapped_release_metadata(owner: str, path: Path, tree: ast.Module) -> dict[str, Migration]:
"""Resolve only known literal sibling wrappers, without importing any code."""
metadata_names = {"revision", "down_revision", "depends_on", "branch_labels"}
aliases: set[str] = set()
bindings: dict[str, list[ast.expr]] = {}
imported: dict[str, list[str]] = {}
for statement in tree.body:
if isinstance(statement, ast.ImportFrom) and not statement.level:
for alias in statement.names:
imported.setdefault(alias.asname or alias.name, []).append(f"{statement.module}.{alias.name}")
targets: list[ast.expr] = []
value: ast.expr | None = None
if isinstance(statement, ast.Assign):
targets, value = statement.targets, statement.value
elif isinstance(statement, ast.AnnAssign) and statement.value is not None:
targets, value = [statement.target], statement.value
for target in targets:
if isinstance(target, ast.Name) and value is not None:
bindings.setdefault(target.id, []).append(value)
if target.id in metadata_names and isinstance(value, ast.Attribute) and isinstance(value.value, ast.Name):
aliases.add(value.value.id)
if not aliases:
return {}
if path.parent.name != "dev_versions":
raise ValueError(f"{path}: non-literal release migration metadata is unsupported")
def reject() -> NoReturn:
raise ValueError(f"{path}: unsupported or ambiguous development migration wrapper")
def binding_count(name: str) -> int:
return _ast_binding_count(tree, name)
def assigned(name: str) -> ast.expr:
values = bindings.get(name, ())
if len(values) != 1 or binding_count(name) != 1 or name in imported:
reject()
return values[0]
def imported_as(node: ast.expr, qualified: str) -> bool:
return (
isinstance(node, ast.Name)
and imported.get(node.id) == [qualified]
and binding_count(node.id) == 1
)
def call(node: ast.expr, qualified: str, arguments: int) -> bool:
return isinstance(node, ast.Call) and imported_as(node.func, qualified) and len(node.args) == arguments and not node.keywords
def file_wrapper_target(node: ast.expr) -> str:
if binding_count("__file__"):
reject()
if isinstance(node, ast.Name):
node = assigned(node.id)
if not (
isinstance(node, ast.BinOp) and isinstance(node.op, ast.Div)
and isinstance(node.right, ast.Constant) and isinstance(node.right.value, str)
and isinstance(node.left, ast.BinOp) and isinstance(node.left.op, ast.Div)
and isinstance(node.left.right, ast.Constant) and node.left.right.value == "versions"
):
reject()
root = node.left.left
for name in imported:
if imported_as(ast.Name(id=name), "pathlib.Path"):
expected = ast.parse(f"{name}(__file__).resolve().parents[1]", mode="eval").body
if ast.dump(root) == ast.dump(expected):
return node.right.value
reject()
result: dict[str, Migration] = {}
resolved: set[Path] = set()
for alias in sorted(aliases):
value = assigned(alias)
if call(value, "importlib.import_module", 1):
argument = value.args[0]
if not isinstance(argument, ast.Constant) or not isinstance(argument.value, str):
reject()
prefix = f"{owner.replace('-', '_')}.backend.migrations.versions."
if not argument.value.startswith(prefix):
reject()
stem = argument.value.removeprefix(prefix)
if not stem or any(character not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_" for character in stem):
reject()
filename = stem + ".py"
elif call(value, "importlib.util.module_from_spec", 1):
argument = value.args[0]
if not isinstance(argument, ast.Name):
reject()
specification = assigned(argument.id)
if not call(specification, "importlib.util.spec_from_file_location", 2):
reject()
if not isinstance(specification.args[0], ast.Constant) or not isinstance(specification.args[0].value, str):
reject()
filename = file_wrapper_target(specification.args[1])
else:
reject()
migration = _release_wrapper_peer(owner, path, filename)
peer = migration.path
resolved.add(peer.resolve())
if len(resolved) > 1:
reject()
result[alias] = migration
return result
def _migration_assignment_value(
name: str,
value: ast.expr,
*,
wrapped: Migration | None,
wrapped: dict[str, Migration],
) -> Any:
try:
return ast.literal_eval(value)
except (ValueError, TypeError):
if (
wrapped is None
or not isinstance(value, ast.Attribute)
not isinstance(value, ast.Attribute)
or not isinstance(value.value, ast.Name)
or value.value.id != "_migration"
or value.value.id not in wrapped
or value.attr != name
):
return None
raise ValueError(f"Unsupported or ambiguous migration metadata: {name}")
migration = wrapped[value.value.id]
return {
"revision": wrapped.revision,
"down_revision": wrapped.down_revisions,
"depends_on": wrapped.depends_on,
"branch_labels": wrapped.branch_labels,
"revision": migration.revision,
"down_revision": migration.down_revisions,
"depends_on": migration.depends_on,
"branch_labels": migration.branch_labels,
}[name]