feat: 发布固件底层开发 Skill 套件

This commit is contained in:
tangfahong committed 2026-09-11 15:06:55 +08:00
commit c7ae3b4ce1
72 files changed
+10212

No files matched your search

@@ -0,0 +1,186 @@
#!/usr/bin/env python3
"""Produce bounded BSP project and build preflight reports."""
from __future__ import annotations
import argparse
import json
import os
import re
import shlex
import sys
from pathlib import Path
import project_guard
VERSION_RE = re.compile(r"^\d{8}_[1-9]\d*$")
ASSIGNMENT_RE = re.compile(r"^\s*(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)=(.*)$")
def parse_assignments(path: Path) -> dict[str, str]:
values: dict[str, str] = {}
for line in path.read_text(encoding="utf-8").splitlines():
match = ASSIGNMENT_RE.match(line)
if not match:
continue
value = match.group(2).strip()
if value and value[0:1] in {'"', "'"} and value[-1:] == value[0]:
value = value[1:-1]
values[match.group(1)] = value
return values
def config_directories(project: Path) -> list[Path]:
configs = project / "configs"
if not configs.is_dir():
return []
return sorted(
path
for path in configs.iterdir()
if path.is_dir() and not path.is_symlink() and not path.name.startswith(".")
)
def detects_dual_os(config_dir: Path) -> bool:
for path in sorted(config_dir.glob("*_defconfig")):
if re.search(r"^CONFIG_DUAL_OS=y\s*$", path.read_text(encoding="utf-8"), re.MULTILINE):
return True
return False
def project_preflight(reference_name: str, target_name: str, root: Path) -> dict[str, object]:
errors: list[str] = []
warnings: list[str] = []
root = project_guard.normalize_root(root)
reference = project_guard.direct_child(root, reference_name, "reference-project", True)
target = project_guard.direct_child(root, target_name, "target-project", False)
if target.exists() or target.is_symlink():
errors.append(f"目标项目已存在: {target}")
build_env = reference / "buildEnvConfig.sh"
assignments: dict[str, str] = {}
if not build_env.is_file():
errors.append(f"缺少 buildEnvConfig.sh: {build_env}")
else:
assignments = parse_assignments(build_env)
config_dirs = config_directories(reference)
if len(config_dirs) != 1:
errors.append(f"参考项目必须唯一定位一个配置目录,实际为 {len(config_dirs)} 个")
config_dir = config_dirs[0] if len(config_dirs) == 1 else None
sync_script = reference / "configs" / "EPSYNCSDKConfig.sh"
if not sync_script.is_file():
warnings.append(f"未找到项目同步脚本: {sync_script}")
imgs = reference / "imgs"
imgs_target = os.readlink(imgs) if imgs.is_symlink() else None
if imgs_target != project_guard.expected_imgs_target(reference_name):
warnings.append("参考项目 imgs 不是约定的项目专属软链接")
return {
"schema_version": 1,
"kind": "project-preflight",
"passed": not errors,
"errors": errors,
"warnings": warnings,
"emxres_root": str(root),
"reference_project": reference_name,
"reference_path": str(reference),
"target_project": target_name,
"target_path": str(target),
"facts": {
"chip": assignments.get("EP_CHIP"),
"sdk": assignments.get("EP_SDK_VER"),
"architecture": assignments.get("EP_ARCH"),
"flash": assignments.get("EP_FLASH_TYPE"),
"nand_flash_id": assignments.get("NAND_FLASH_ID"),
"board_build_type": config_dir.name if config_dir else None,
"system_form": "dual-os" if config_dir and detects_dual_os(config_dir) else "single-os",
},
"bounded_paths": {
"build_env": str(build_env),
"config_directory": str(config_dir) if config_dir else None,
"sync_script": str(sync_script),
"reference_imgs_target": imgs_target,
"target_imgs_target": project_guard.expected_imgs_target(target_name),
},
}
def build_preflight(project_name: str, version: str, root: Path, output_root: Path) -> dict[str, object]:
errors: list[str] = []
warnings: list[str] = []
root = project_guard.normalize_root(root)
project = project_guard.direct_child(root, project_name, "project", True)
if not VERSION_RE.fullmatch(version):
errors.append("构建版本必须为 YYYYMMDD_N,且 N 为正整数")
completion_script = root / "tools" / "script_completion.sh"
if not completion_script.is_file():
errors.append(f"统一构建入口不存在: {completion_script}")
elif not os.access(completion_script, os.R_OK):
errors.append(f"统一构建入口不可读: {completion_script}")
imgs = project / "imgs"
expected_imgs = project_guard.expected_imgs_target(project_name)
actual_imgs = os.readlink(imgs) if imgs.is_symlink() else None
if actual_imgs != expected_imgs:
errors.append(f"项目 imgs 必须指向 {expected_imgs},实际为 {actual_imgs!r}")
output_root = output_root.expanduser().resolve(strict=False)
output_directory = output_root / project_name / "imgs" / version
if output_directory.exists() or output_directory.is_symlink():
errors.append(f"构建输出目录已存在,禁止覆盖: {output_directory}")
command = (
f"cd {shlex.quote(str(root))} && source tools/script_completion.sh && "
f"EMX2_BuildSDKTool {project_name} {version} all"
)
warnings.append("统一构建可能同步配置到项目外 SDK,并按工具实现执行 git fetch origin")
return {
"schema_version": 1,
"kind": "build-preflight",
"passed": not errors,
"errors": errors,
"warnings": warnings,
"project": project_name,
"project_path": str(project),
"build_version": version,
"working_directory": str(root),
"completion_script": str(completion_script),
"command": command,
"output_directory": str(output_directory),
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--emxres-root", type=Path, default=project_guard.production_root())
subparsers = parser.add_subparsers(dest="command", required=True)
project_parser = subparsers.add_parser("project")
project_parser.add_argument("--reference-project", required=True)
project_parser.add_argument("--target-project", required=True)
build_parser = subparsers.add_parser("build")
build_parser.add_argument("--project", required=True)
build_parser.add_argument("--version", required=True)
build_parser.add_argument(
"--output-root", type=Path, default=Path("/epbuild/EMX2.0/EmxRes")
)
args = parser.parse_args()
try:
if args.command == "project":
result = project_preflight(
args.reference_project, args.target_project, args.emxres_root
)
else:
result = build_preflight(
args.project, args.version, args.emxres_root, args.output_root
)
except (project_guard.GuardError, OSError, UnicodeDecodeError) as exc:
result = {"passed": False, "errors": [str(exc)]}
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0 if result.get("passed") is True else 2
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,100 @@
#!/usr/bin/env python3
"""Validate normalized pinmux data and write the final project CSV."""
import argparse
import csv
import json
import re
import sys
from pathlib import Path
HEADERS = ["引脚号", "功能名称", "引脚名称", "有效电平", "FMUX寄存器", "FMUX当前值", "FMUX默认值", "FMUX_list"]
INPUT_FIELDS = ["pin_number", "function_name", "pin_name", "is_gpio", "active_level", "fmux_register", "fmux_current", "fmux_default", "fmux_list"]
NUMBERED_MUX_RE = re.compile(r"(?:^|;\s*)([0-7])\s*:")
UNSAFE_FILENAME_RE = re.compile(r"[/\\:*?\"<>|]+")
def text(value):
return "null" if value is None else str(value).strip()
def safe_name(value):
value = UNSAFE_FILENAME_RE.sub("_", value.strip())
value = re.sub(r"\s+", "_", value)
value = re.sub(r"_+", "_", value).strip(" ._")
if not value:
raise ValueError("project_name 清理后为空")
return value
def normalize(raw, index, seen):
if not isinstance(raw, dict):
raise ValueError(f"rows[{index}] 必须是对象")
missing = [key for key in INPUT_FIELDS if key not in raw]
if missing:
raise ValueError(f"rows[{index}] 缺少字段: {', '.join(missing)}")
pin = text(raw["pin_number"])
if not pin or pin == "null":
raise ValueError(f"rows[{index}] pin_number 为空")
if pin in seen:
raise ValueError(f"rows[{index}] 引脚号重复: {pin}")
seen.add(pin)
if not isinstance(raw["is_gpio"], bool):
raise ValueError(f"rows[{index}] is_gpio 必须是布尔值")
active = text(raw["active_level"])
if raw["is_gpio"] and active not in {"高有效", "低有效", "待确认"}:
raise ValueError(f"rows[{index}] GPIO 有效电平无效")
if not raw["is_gpio"] and active != "null":
raise ValueError(f"rows[{index}] 非 GPIO 有效电平必须为 null")
mux_list = text(raw["fmux_list"])
mux_numbers = set(NUMBERED_MUX_RE.findall(mux_list))
current = text(raw["fmux_current"])
if current in set("01234567") and current not in mux_numbers:
raise ValueError(f"rows[{index}] 引脚 {pin}: FMUX当前值 {current} 未出现在 FMUX_list")
row = {
"引脚号": pin,
"功能名称": text(raw["function_name"]),
"引脚名称": text(raw["pin_name"]),
"有效电平": active,
"FMUX寄存器": text(raw["fmux_register"]),
"FMUX当前值": current,
"FMUX默认值": text(raw["fmux_default"]),
"FMUX_list": mux_list,
}
return row, len(mux_numbers)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--input", required=True, type=Path)
parser.add_argument("--project-name")
parser.add_argument("--output-dir", type=Path, default=Path.cwd())
parser.add_argument("--output", type=Path)
args = parser.parse_args()
try:
with args.input.open(encoding="utf-8-sig") as handle:
payload = json.load(handle)
if not isinstance(payload, dict) or not isinstance(payload.get("rows"), list):
raise ValueError("JSON 顶层必须是对象并包含 rows 数组")
project = args.project_name or text(payload.get("project_name", ""))
rows, seen = [], set()
for index, raw in enumerate(payload["rows"]):
row, mux_count = normalize(raw, index, seen)
if mux_count >= 2:
rows.append(row)
output = args.output or args.output_dir / f"{safe_name(project)}_pinmux.csv"
output.parent.mkdir(parents=True, exist_ok=True)
with output.open("w", encoding="utf-8-sig", newline="") as handle:
writer = csv.DictWriter(handle, fieldnames=HEADERS)
writer.writeheader()
writer.writerows(rows)
print(f"已输出 {len(rows)} 行(输入 {len(payload['rows'])} 行): {output}")
return 0
except (OSError, ValueError, json.JSONDecodeError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,601 @@
#!/usr/bin/env python3
"""Safely stage, verify, finalize, and audit an EmxRes BSP project clone."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
import shutil
import stat
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
ROOT_SUFFIX = Path("EapilBsp") / "EMX2.0" / "EmxRes"
PROJECT_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
EXCLUDED_DIR_NAMES = {
".git",
".cache",
".mypy_cache",
".pytest_cache",
".ruff_cache",
"__pycache__",
"build",
"logs",
"out",
"output",
}
EXCLUDED_FILE_SUFFIXES = {".log", ".pyc", ".tmp"}
CREDENTIAL_FILE_NAMES = {
".env",
"credentials.json",
"credentials.yaml",
"credentials.yml",
"id_ed25519",
"id_rsa",
}
CREDENTIAL_FILE_SUFFIXES = {".key", ".p12", ".pem", ".pfx"}
DOC_PREFIX = "doc/bsp_adaptation/"
DELTA_REPORT_PATH = "doc/bsp_adaptation/validation/project_delta.json"
FINAL_AUDIT_REPORT_PATH = "doc/bsp_adaptation/validation/final_audit.json"
class GuardError(RuntimeError):
"""A safety or validation gate failed."""
def now_utc() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def production_root() -> Path:
return Path.home() / ROOT_SUFFIX
def normalize_root(root: Path) -> Path:
root = root.expanduser()
if not root.is_dir():
raise GuardError(f"EmxRes 根目录不存在: {root}")
if root.is_symlink():
raise GuardError(f"EmxRes 根目录不能是软链接: {root}")
return root.resolve()
def validate_project_name(name: str, field: str) -> str:
if not PROJECT_NAME_RE.fullmatch(name or ""):
raise GuardError(
f"{field} 必须是 EmxRes 直接子目录名,只能包含字母、数字、点、下划线和连字符: {name!r}"
)
return name
def direct_child(
root: Path, name: str, field: str, must_exist: bool, allow_hidden: bool = False
) -> Path:
if allow_hidden and name.startswith("."):
validate_project_name(name[1:], field)
else:
name = validate_project_name(name, field)
path = root / name
if path.parent != root:
raise GuardError(f"{field} 发生路径逃逸: {name}")
if must_exist:
if not path.is_dir() or path.is_symlink():
raise GuardError(f"{field} 不是普通项目目录: {path}")
if path.resolve().parent != root:
raise GuardError(f"{field} 解析后不在 EmxRes 内: {path}")
return path
def is_excluded_name(path: Path, is_dir: bool) -> bool:
if is_dir and path.name in EXCLUDED_DIR_NAMES:
return True
return not is_dir and path.suffix.lower() in EXCLUDED_FILE_SUFFIXES
def copy_ignore(_directory: str, names: list[str]) -> set[str]:
ignored = set()
base = Path(_directory)
for name in names:
path = base / name
is_dir = path.is_dir()
if is_excluded_name(path, is_dir):
ignored.add(name)
return ignored
def scan_credentials(reference: Path) -> list[str]:
hits = []
for current, dirs, files in os.walk(reference, followlinks=False):
base = Path(current)
dirs[:] = [
name
for name in dirs
if not is_excluded_name(base / name, True) and not (base / name).is_symlink()
]
for name in files:
path = base / name
if path.is_symlink() or is_excluded_name(path, False):
continue
if name in CREDENTIAL_FILE_NAMES or path.suffix.lower() in CREDENTIAL_FILE_SUFFIXES:
hits.append(path.relative_to(reference).as_posix())
return sorted(hits)
def excluded_paths(reference: Path) -> list[str]:
hits = []
for current, dirs, files in os.walk(reference, followlinks=False):
base = Path(current)
kept_dirs = []
for name in dirs:
path = base / name
if is_excluded_name(path, True):
hits.append(path.relative_to(reference).as_posix() + "/")
elif path.is_symlink():
continue
else:
kept_dirs.append(name)
dirs[:] = kept_dirs
for name in files:
path = base / name
if not path.is_symlink() and is_excluded_name(path, False):
hits.append(path.relative_to(reference).as_posix())
return sorted(hits)
def within(path: Path, root: Path) -> bool:
try:
path.relative_to(root)
return True
except ValueError:
return False
def expected_imgs_target(project_name: str) -> str:
return f"/epbuild/EMX2.0/EmxRes/{project_name}/imgs"
def scan_symlinks(reference: Path, root: Path, reference_name: str) -> list[dict[str, str]]:
links = []
link_paths = []
for current, dirs, files in os.walk(reference, followlinks=False):
base = Path(current)
kept_dirs = []
for name in dirs:
path = base / name
if is_excluded_name(path, True):
continue
if path.is_symlink():
link_paths.append(path)
else:
kept_dirs.append(name)
dirs[:] = kept_dirs
for name in files:
path = base / name
if path.is_symlink() and not is_excluded_name(path, False):
link_paths.append(path)
for path in sorted(link_paths):
relative = path.relative_to(reference).as_posix()
raw_target = os.readlink(path)
if relative == "imgs" and raw_target == expected_imgs_target(reference_name):
kind = "imgs"
elif os.path.isabs(raw_target):
resolved = Path(raw_target).resolve(strict=False)
if not within(resolved, root):
raise GuardError(f"存在未经确认的外部绝对软链接: {relative} -> {raw_target}")
kind = "absolute-in-root"
else:
resolved = (path.parent / raw_target).resolve(strict=False)
if not within(resolved, root):
raise GuardError(f"存在未经确认的外部相对软链接: {relative} -> {raw_target}")
kind = "relative-in-root"
links.append({"path": relative, "target": raw_target, "kind": kind})
return links
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def inventory(
project: Path, omit_paths: set[str] | None = None
) -> dict[str, dict[str, object]]:
omit_paths = omit_paths or set()
entries: dict[str, dict[str, object]] = {}
for current, dirs, files in os.walk(project, followlinks=False):
base = Path(current)
kept_dirs = []
for name in sorted(dirs):
path = base / name
if is_excluded_name(path, True):
continue
if path.is_symlink():
rel = path.relative_to(project).as_posix()
entries[rel] = {
"kind": "symlink",
"mode": stat.S_IMODE(path.lstat().st_mode),
"target": os.readlink(path),
}
else:
kept_dirs.append(name)
dirs[:] = kept_dirs
for name in sorted(files):
path = base / name
if is_excluded_name(path, False):
continue
rel = path.relative_to(project).as_posix()
if rel in omit_paths:
continue
if path.is_symlink():
entries[rel] = {
"kind": "symlink",
"mode": stat.S_IMODE(path.lstat().st_mode),
"target": os.readlink(path),
}
elif path.is_file():
file_stat = path.stat()
entries[rel] = {
"kind": "file",
"mode": stat.S_IMODE(file_stat.st_mode),
"size": file_stat.st_size,
"sha256": sha256_file(path),
}
return dict(sorted(entries.items()))
def write_json(path: Path, payload: dict[str, object]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def prepare_project(reference_name: str, target_name: str, root: Path | None = None) -> dict[str, object]:
root = normalize_root(root or production_root())
reference = direct_child(root, reference_name, "reference-project", True)
target = direct_child(root, target_name, "target-project", False)
if target.exists() or target.is_symlink():
raise GuardError(f"目标项目已存在,禁止覆盖: {target}")
credentials = scan_credentials(reference)
if credentials:
raise GuardError("参考项目包含疑似凭据文件,未复制: " + ", ".join(credentials))
links = scan_symlinks(reference, root, reference_name)
source_before = inventory(reference)
staging_name = f".{target_name}.bsp-staging-{uuid.uuid4().hex[:12]}"
staging = root / staging_name
shutil.copytree(
reference,
staging,
symlinks=True,
copy_function=shutil.copy2,
ignore=copy_ignore,
)
imgs = staging / "imgs"
imgs_retargeted = False
if imgs.is_symlink() and os.readlink(imgs) == expected_imgs_target(reference_name):
imgs.unlink()
imgs.symlink_to(expected_imgs_target(target_name))
imgs_retargeted = True
doc_root = staging / "doc" / "bsp_adaptation"
(doc_root / "inputs").mkdir(parents=True, exist_ok=True)
validation = doc_root / "validation"
validation.mkdir(parents=True, exist_ok=True)
source_after = inventory(reference)
if source_before != source_after:
raise GuardError(f"复制期间参考项目发生变化,临时目录保留供检查: {staging}")
manifest: dict[str, object] = {
"schema_version": 1,
"created_at": now_utc(),
"emxres_root": str(root),
"reference_project": reference_name,
"target_project": target_name,
"staging_directory": staging_name,
"excluded_paths": excluded_paths(reference),
"symlinks": links,
"imgs_retargeted": imgs_retargeted,
"imgs_target": os.readlink(imgs) if imgs.is_symlink() else None,
"source_inventory": source_before,
"staging_inventory_before_edits": inventory(staging),
}
write_json(validation / "clone_manifest.json", manifest)
return {
"staging_directory": staging_name,
"staging_path": str(staging),
"manifest": str(validation / "clone_manifest.json"),
"imgs_retargeted": imgs_retargeted,
}
def normalize_allow_paths(values: list[str]) -> set[str]:
allowed = set()
for value in values:
pure = PurePosixPath(value)
if pure.is_absolute() or ".." in pure.parts or value in {"", "."}:
raise GuardError(f"allow-path 必须是项目内精确相对路径: {value!r}")
if any(char in value for char in "*?[]"):
raise GuardError(f"allow-path 不允许通配符,必须逐文件列出: {value!r}")
allowed.add(pure.as_posix())
return allowed
def compare_inventories(
source: dict[str, dict[str, object]], target: dict[str, dict[str, object]]
) -> dict[str, list[str]]:
source_paths = set(source)
target_paths = set(target)
return {
"added": sorted(target_paths - source_paths),
"removed": sorted(source_paths - target_paths),
"modified": sorted(path for path in source_paths & target_paths if source[path] != target[path]),
}
def load_clone_manifest(
staging: Path, root: Path, reference_name: str, target_name: str, staging_name: str
) -> dict[str, object]:
path = staging / "doc" / "bsp_adaptation" / "validation" / "clone_manifest.json"
if not path.is_file():
raise GuardError("缺少 clone_manifest.json;临时目录不是受控 prepare 结果")
manifest = json.loads(path.read_text(encoding="utf-8"))
expected = {
"emxres_root": str(root),
"reference_project": reference_name,
"target_project": target_name,
"staging_directory": staging_name,
}
mismatched = [key for key, value in expected.items() if manifest.get(key) != value]
if mismatched:
raise GuardError("clone_manifest 与当前操作不匹配: " + ", ".join(mismatched))
return manifest
def verify_project(
reference_name: str,
staging_name: str,
target_name: str,
allow_paths: list[str],
root: Path | None = None,
) -> dict[str, object]:
root = normalize_root(root or production_root())
reference = direct_child(root, reference_name, "reference-project", True)
validate_project_name(target_name, "target-project")
staging = direct_child(root, staging_name, "staging-dir", True, allow_hidden=True)
expected_prefix = f".{target_name}.bsp-staging-"
if not staging_name.startswith(expected_prefix):
raise GuardError(f"临时目录与目标项目不匹配: {staging_name}")
clone_manifest = load_clone_manifest(staging, root, reference_name, target_name, staging_name)
scan_symlinks(staging, root, target_name)
credentials = scan_credentials(staging)
if credentials:
raise GuardError("临时目录包含疑似凭据文件,禁止最终化: " + ", ".join(credentials))
unexpected_excluded = excluded_paths(staging)
if unexpected_excluded:
raise GuardError(
"临时目录新增了被排除的缓存、构建输出或日志,必须先处理: "
+ ", ".join(unexpected_excluded)
)
allowed = normalize_allow_paths(allow_paths)
source_inventory = clone_manifest.get("source_inventory")
if not isinstance(source_inventory, dict):
raise GuardError("clone_manifest 缺少 prepare 时的参考项目清单")
if inventory(reference) != source_inventory:
raise GuardError("prepare 后参考项目发生变化;必须重新确认并重新 prepare")
staging_inventory = inventory(staging, {DELTA_REPORT_PATH})
changes = compare_inventories(source_inventory, staging_inventory)
def permitted(path: str) -> bool:
return path == "imgs" or path.startswith(DOC_PREFIX) or path in allowed
violations = sorted(
{path for group in changes.values() for path in group if not permitted(path)}
)
report: dict[str, object] = {
"schema_version": 1,
"checked_at": now_utc(),
"reference_project": reference_name,
"target_project": target_name,
"staging_directory": staging_name,
"allowed_paths": sorted(allowed),
"changes": changes,
"violations": violations,
"passed": not violations,
"verified_staging_inventory": staging_inventory,
}
write_json(staging / "doc" / "bsp_adaptation" / "validation" / "project_delta.json", report)
return report
def finalize_project(
reference_name: str,
staging_name: str,
target_name: str,
allow_paths: list[str],
root: Path | None = None,
) -> dict[str, object]:
root = normalize_root(root or production_root())
target = direct_child(root, target_name, "target-project", False)
if target.exists() or target.is_symlink():
raise GuardError(f"目标项目已存在,禁止覆盖: {target}")
staging = direct_child(root, staging_name, "staging-dir", True, allow_hidden=True)
if not staging_name.startswith(f".{target_name}.bsp-staging-"):
raise GuardError(f"临时目录与目标项目不匹配: {staging_name}")
report_path = staging / "doc" / "bsp_adaptation" / "adaptation_report.md"
delta = verify_project(reference_name, staging_name, target_name, allow_paths, root=root)
if delta.get("passed") is not True:
raise GuardError("差异白名单未通过,禁止最终化")
verified_inventory = delta.get("verified_staging_inventory")
if not isinstance(verified_inventory, dict):
raise GuardError("差异报告缺少已验证目录清单;必须重新运行 verify")
current_inventory = inventory(staging, {DELTA_REPORT_PATH})
if current_inventory != verified_inventory:
raise GuardError("verify 后临时目录又发生变化;必须重新运行 verify")
if not report_path.is_file() or not report_path.read_text(encoding="utf-8").strip():
raise GuardError("缺少非空的 doc/bsp_adaptation/adaptation_report.md")
os.rename(staging, target)
finalized_at = now_utc()
clone_path = target / "doc" / "bsp_adaptation" / "validation" / "clone_manifest.json"
if clone_path.is_file():
clone = json.loads(clone_path.read_text(encoding="utf-8"))
clone["finalized_at"] = finalized_at
clone["final_path"] = str(target)
write_json(clone_path, clone)
delta_path = target / "doc" / "bsp_adaptation" / "validation" / "project_delta.json"
delta["finalized_at"] = finalized_at
delta["final_path"] = str(target)
write_json(delta_path, delta)
return {"target_project": target_name, "target_path": str(target), "finalized_at": finalized_at}
def audit_final_project(
target_name: str,
allow_paths: list[str] | None = None,
root: Path | None = None,
) -> dict[str, object]:
"""Audit a finalized project against its immutable prepare-time baseline."""
root = normalize_root(root or production_root())
target = direct_child(root, target_name, "target-project", True)
clone_path = target / "doc" / "bsp_adaptation" / "validation" / "clone_manifest.json"
if not clone_path.is_file():
raise GuardError("正式项目缺少 clone_manifest.json,无法恢复 prepare 时基线")
clone = json.loads(clone_path.read_text(encoding="utf-8"))
expected = {"emxres_root": str(root), "target_project": target_name}
mismatched = [key for key, value in expected.items() if clone.get(key) != value]
if mismatched:
raise GuardError("clone_manifest 与正式项目不匹配: " + ", ".join(mismatched))
source_inventory = clone.get("source_inventory")
if not isinstance(source_inventory, dict):
raise GuardError("clone_manifest 缺少 prepare 时的参考项目清单")
delta_path = target / DELTA_REPORT_PATH
delta: dict[str, object] = {}
if delta_path.is_file():
delta = json.loads(delta_path.read_text(encoding="utf-8"))
effective_allow_paths = allow_paths
if effective_allow_paths is None or not effective_allow_paths:
stored = delta.get("allowed_paths")
if not isinstance(stored, list) or not all(isinstance(item, str) for item in stored):
raise GuardError("未提供 allow-path,且 project_delta.json 中没有可复用白名单")
effective_allow_paths = stored
allowed = normalize_allow_paths(effective_allow_paths)
scan_symlinks(target, root, target_name)
credentials = scan_credentials(target)
if credentials:
raise GuardError("正式项目包含疑似凭据文件: " + ", ".join(credentials))
unexpected_excluded = excluded_paths(target)
if unexpected_excluded:
raise GuardError(
"正式项目包含被排除的缓存、构建输出或日志: " + ", ".join(unexpected_excluded)
)
target_inventory = inventory(target, {DELTA_REPORT_PATH, FINAL_AUDIT_REPORT_PATH})
changes = compare_inventories(source_inventory, target_inventory)
def permitted(path: str) -> bool:
return path == "imgs" or path.startswith(DOC_PREFIX) or path in allowed
violations = sorted(
{path for group in changes.values() for path in group if not permitted(path)}
)
reference_name = clone.get("reference_project")
reference_drifted: bool | None = None
if isinstance(reference_name, str):
reference = root / reference_name
if reference.is_dir() and not reference.is_symlink():
reference_drifted = inventory(reference) != source_inventory
report: dict[str, object] = {
"schema_version": 1,
"checked_at": now_utc(),
"reference_project": reference_name,
"reference_drifted_since_prepare": reference_drifted,
"target_project": target_name,
"target_path": str(target),
"baseline": "clone_manifest.source_inventory",
"allowed_paths": sorted(allowed),
"changes": changes,
"violations": violations,
"passed": not violations,
"audited_target_inventory": target_inventory,
}
write_json(target / FINAL_AUDIT_REPORT_PATH, report)
return report
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="command", required=True)
prepare = subparsers.add_parser("prepare", help="复制参考项目到唯一临时目录")
prepare.add_argument("--reference-project", required=True)
prepare.add_argument("--target-project", required=True)
verify = subparsers.add_parser("verify", help="按精确文件白名单检查差异")
verify.add_argument("--reference-project", required=True)
verify.add_argument("--staging-dir", required=True)
verify.add_argument("--target-project", required=True)
verify.add_argument("--allow-path", action="append", default=[])
finalize = subparsers.add_parser("finalize", help="将已验证临时目录原子落为正式项目")
finalize.add_argument("--reference-project", required=True)
finalize.add_argument("--staging-dir", required=True)
finalize.add_argument("--target-project", required=True)
finalize.add_argument("--allow-path", action="append", default=[])
audit = subparsers.add_parser("audit-final", help="复核已落盘正式项目的差异白名单")
audit.add_argument("--target-project", required=True)
audit.add_argument("--allow-path", action="append", default=[])
return parser
def main() -> int:
args = build_parser().parse_args()
try:
if args.command == "prepare":
result = prepare_project(args.reference_project, args.target_project)
elif args.command == "verify":
result = verify_project(
args.reference_project,
args.staging_dir,
args.target_project,
args.allow_path,
)
if not result["passed"]:
print(json.dumps(result, ensure_ascii=False, indent=2))
return 2
elif args.command == "finalize":
result = finalize_project(
args.reference_project,
args.staging_dir,
args.target_project,
args.allow_path,
)
else:
result = audit_final_project(args.target_project, args.allow_path)
if not result["passed"]:
print(json.dumps(result, ensure_ascii=False, indent=2))
return 2
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0
except (GuardError, OSError, json.JSONDecodeError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,453 @@
#!/usr/bin/env python3
"""Select a bundled CV18 pinout workbook and query its pin/FMUX information."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import posixpath
import re
import shlex
import tempfile
from pathlib import Path, PurePosixPath
from typing import Any, Iterator
from xml.etree import ElementTree as ET
from zipfile import BadZipFile, ZipFile
MAIN_NS = "http://schemas.openxmlformats.org/spreadsheetml/2006/main"
DOC_REL_NS = "http://schemas.openxmlformats.org/officeDocument/2006/relationships"
PKG_REL_NS = "http://schemas.openxmlformats.org/package/2006/relationships"
NS = {"m": MAIN_NS}
SKILL_ROOT = Path(__file__).resolve().parents[2]
CATALOG_ROOT = SKILL_ROOT / "references" / "gpio-pinout"
MANIFEST_PATH = CATALOG_ROOT / "manifest.json"
EP_CHIP_RE = re.compile(r"^\s*(?:export\s+)?EP_CHIP\s*=\s*(.*?)\s*$")
CELL_RE = re.compile(r"^([A-Z]+)([0-9]+)$")
MUX_RE = re.compile(r"^\s*([0-7])\s*:\s*(.+?)\s*$")
MAX_XLSX_SIZE = 16 * 1024 * 1024
MAX_UNCOMPRESSED_SIZE = 64 * 1024 * 1024
class PinoutError(RuntimeError):
"""A bundled pinout cannot be selected or queried safely."""
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def write_json_atomic(path: Path, payload: dict[str, object]) -> None:
path = path.expanduser().resolve(strict=False)
path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(
"w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False
) as handle:
json.dump(payload, handle, ensure_ascii=False, indent=2)
handle.write("\n")
temporary = Path(handle.name)
os.replace(temporary, path)
def normalize_chip(value: str) -> str:
normalized = re.sub(r"[^A-Za-z0-9]", "", value).upper()
if not normalized:
raise PinoutError("芯片名称规范化后为空")
return normalized
def project_chip(project: Path) -> tuple[str, dict[str, object]]:
project = project.expanduser().resolve(strict=True)
if not project.is_dir() or project.is_symlink():
raise PinoutError(f"项目必须是普通目录: {project}")
config = project / "buildEnvConfig.sh"
if not config.is_file() or config.is_symlink():
raise PinoutError(f"项目缺少普通文件 buildEnvConfig.sh: {config}")
values: list[str] = []
for line_number, line in enumerate(
config.read_text(encoding="utf-8", errors="replace").splitlines(), 1
):
match = EP_CHIP_RE.match(line)
if match is None:
continue
try:
tokens = shlex.split(match.group(1), comments=True, posix=True)
except ValueError as exc:
raise PinoutError(f"buildEnvConfig.sh:{line_number} EP_CHIP 无法解析: {exc}") from exc
if len(tokens) != 1 or not tokens[0]:
raise PinoutError(
f"buildEnvConfig.sh:{line_number} EP_CHIP 必须是一个静态芯片名称"
)
if re.fullmatch(r"[A-Za-z0-9_-]+", tokens[0]) is None:
raise PinoutError(
f"buildEnvConfig.sh:{line_number} EP_CHIP 包含展开或非法字符,必须是静态芯片名称"
)
values.append(normalize_chip(tokens[0]))
unique = sorted(set(values))
if len(unique) != 1:
raise PinoutError(f"buildEnvConfig.sh 必须唯一确定 EP_CHIP,实际 {unique}")
return unique[0], {
"source": "project-buildEnvConfig",
"project": str(project),
"build_env_config": str(config),
"build_env_config_sha256": sha256_file(config),
}
def safe_relative_file(value: object, field: str) -> str:
if not isinstance(value, str) or not value:
raise PinoutError(f"manifest {field} 必须是非空字符串")
pure = PurePosixPath(value)
if pure.is_absolute() or ".." in pure.parts or len(pure.parts) != 1:
raise PinoutError(f"manifest {field} 必须是当前目录内文件名: {value!r}")
return value
def load_manifest() -> list[dict[str, Any]]:
if not MANIFEST_PATH.is_file() or MANIFEST_PATH.is_symlink():
raise PinoutError(f"内置 GPIO manifest 缺失或不是普通文件: {MANIFEST_PATH}")
path = MANIFEST_PATH.resolve(strict=True)
payload = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(payload, dict) or payload.get("schema_version") != 1:
raise PinoutError("gpio-pinout/manifest.json schema_version 必须为 1")
documents = payload.get("documents")
if not isinstance(documents, list) or not documents:
raise PinoutError("gpio-pinout/manifest.json.documents 必须是非空数组")
checked: list[dict[str, Any]] = []
ids: set[str] = set()
for index, item in enumerate(documents):
if not isinstance(item, dict):
raise PinoutError(f"manifest documents[{index}] 必须是对象")
identifier = item.get("id")
if not isinstance(identifier, str) or not identifier or identifier in ids:
raise PinoutError(f"manifest documents[{index}].id 非法或重复")
ids.add(identifier)
filename = safe_relative_file(item.get("file"), f"documents[{index}].file")
digest = item.get("sha256")
if not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest):
raise PinoutError(f"manifest documents[{index}].sha256 非法")
patterns = item.get("chip_patterns")
aliases = item.get("chip_aliases")
if not isinstance(patterns, list) or not patterns or not all(
isinstance(pattern, str) and pattern for pattern in patterns
):
raise PinoutError(f"manifest documents[{index}].chip_patterns 非法")
if not isinstance(aliases, list) or not all(
isinstance(alias, str) and alias for alias in aliases
):
raise PinoutError(f"manifest documents[{index}].chip_aliases 非法")
try:
compiled = [re.compile(pattern) for pattern in patterns]
except re.error as exc:
raise PinoutError(f"manifest documents[{index}] 正则非法: {exc}") from exc
workbook_candidate = CATALOG_ROOT / filename
if not workbook_candidate.is_file() or workbook_candidate.is_symlink():
raise PinoutError(f"内置 GPIO 表格缺失或不是普通文件: {workbook_candidate}")
workbook = workbook_candidate.resolve(strict=True)
if workbook.parent != CATALOG_ROOT.resolve():
raise PinoutError(f"内置 GPIO 表格越出目录: {workbook}")
actual_digest = sha256_file(workbook)
if actual_digest != digest:
raise PinoutError(
f"内置 GPIO 表格 hash 与 manifest 不一致: {filename}: {actual_digest}"
)
checked.append(
{
**item,
"path": workbook,
"compiled_patterns": compiled,
"normalized_aliases": {normalize_chip(alias) for alias in aliases},
}
)
return checked
def select_document(chip: str) -> dict[str, Any]:
chip = normalize_chip(chip)
matched = [
item
for item in load_manifest()
if chip in item["normalized_aliases"]
or any(pattern.fullmatch(chip) for pattern in item["compiled_patterns"])
]
if len(matched) != 1:
identifiers = [item["id"] for item in matched]
if not matched:
raise PinoutError(f"芯片 {chip} 没有对应的内置 GPIO 设计表")
raise PinoutError(f"芯片 {chip} 同时匹配多个内置 GPIO 设计表: {identifiers}")
return matched[0]
def xml_text(node: ET.Element) -> str:
return "".join(text.text or "" for text in node.iter(f"{{{MAIN_NS}}}t"))
def workbook_parts(archive: ZipFile) -> tuple[list[str], list[tuple[str, str]]]:
names = archive.namelist()
total_size = sum(item.file_size for item in archive.infolist())
if total_size > MAX_UNCOMPRESSED_SIZE:
raise PinoutError(f"XLSX 解压后超过 {MAX_UNCOMPRESSED_SIZE} 字节限制")
shared: list[str] = []
if "xl/sharedStrings.xml" in names:
root = ET.fromstring(archive.read("xl/sharedStrings.xml"))
shared = [xml_text(item) for item in root.findall(f"{{{MAIN_NS}}}si")]
workbook = ET.fromstring(archive.read("xl/workbook.xml"))
relationships = ET.fromstring(archive.read("xl/_rels/workbook.xml.rels"))
targets = {
item.attrib["Id"]: item.attrib["Target"]
for item in relationships.findall(f"{{{PKG_REL_NS}}}Relationship")
}
sheets: list[tuple[str, str]] = []
for sheet in workbook.findall(f".//{{{MAIN_NS}}}sheet"):
rel_id = sheet.attrib.get(f"{{{DOC_REL_NS}}}id")
if rel_id not in targets:
raise PinoutError(f"工作表 {sheet.attrib.get('name')} 缺少 relationship")
target = targets[rel_id].lstrip("/")
if not target.startswith("xl/"):
target = posixpath.normpath(posixpath.join("xl", target))
if target.startswith("../") or target not in names:
raise PinoutError(f"工作表目标路径非法或缺失: {target}")
sheets.append((sheet.attrib.get("name", ""), target))
return shared, sheets
def column_number(cell_reference: str) -> int:
match = CELL_RE.fullmatch(cell_reference)
if match is None:
raise PinoutError(f"XLSX cell reference 非法: {cell_reference!r}")
value = 0
for character in match.group(1):
value = value * 26 + ord(character) - ord("A") + 1
return value
def cell_value(cell: ET.Element, shared: list[str]) -> str:
cell_type = cell.attrib.get("t")
if cell_type == "inlineStr":
return xml_text(cell)
value = cell.find(f"{{{MAIN_NS}}}v")
raw = "" if value is None or value.text is None else value.text
if cell_type == "s" and raw:
try:
return shared[int(raw)]
except (IndexError, ValueError) as exc:
raise PinoutError(f"XLSX shared string index 非法: {raw!r}") from exc
if cell_type == "b":
return "TRUE" if raw == "1" else "FALSE"
return raw
def sheet_rows(archive: ZipFile, target: str, shared: list[str]) -> Iterator[tuple[int, dict[int, str]]]:
root = ET.fromstring(archive.read(target))
for row in root.findall(f".//{{{MAIN_NS}}}sheetData/{{{MAIN_NS}}}row"):
values: dict[int, str] = {}
for cell in row.findall(f"{{{MAIN_NS}}}c"):
reference = cell.attrib.get("r")
if reference is None:
continue
value = cell_value(cell, shared).strip()
if value:
values[column_number(reference)] = value
if values:
yield int(row.attrib.get("r", "0")), values
def normalized_header(value: str) -> str:
return re.sub(r"[^a-z0-9]", "", value.casefold())
def header_columns(row: dict[int, str]) -> dict[str, int]:
aliases = {
"pin_number": {"pinnum", "pinnumber"},
"pin_name": {"pinname"},
"io_type": {"iotype"},
"io_group": {"iogroup"},
"power_domain": {"powerdomain"},
"io_cfg_register": {"iocfgregister"},
"fmux_register": {"functionselectregister"},
"fmux_default": {"fmuxdefault"},
"description": {"description"},
"note": {"note"},
}
result: dict[str, int] = {}
for column, value in row.items():
normalized = normalized_header(value)
for field, candidates in aliases.items():
if normalized in candidates and field not in result:
result[field] = column
return result
def parse_muxes(description: str) -> list[dict[str, object]]:
muxes: list[dict[str, object]] = []
for line in description.replace("\r", "\n").splitlines():
match = MUX_RE.fullmatch(line)
if match is None:
continue
raw_function = match.group(2).strip()
default = bool(re.search(r"\(\s*default\s*\)\s*$", raw_function, re.IGNORECASE))
function = re.sub(
r"\s*\(\s*default\s*\)\s*$", "", raw_function, flags=re.IGNORECASE
).strip()
muxes.append({"number": int(match.group(1)), "function": function, "default": default})
return muxes
def load_pin_rows(path: Path) -> tuple[str, list[dict[str, object]]]:
path = path.resolve(strict=True)
if path.stat().st_size > MAX_XLSX_SIZE:
raise PinoutError(f"内置 XLSX 超过 {MAX_XLSX_SIZE} 字节限制: {path}")
try:
with ZipFile(path) as archive:
shared, sheets = workbook_parts(archive)
candidates = [item for item in sheets if "管腳信息表" in item[0] or "管脚信息表" in item[0]]
if len(candidates) != 1:
raise PinoutError(f"必须唯一定位管脚信息表,实际 {[item[0] for item in candidates]}")
sheet_name, target = candidates[0]
rows = iter(sheet_rows(archive, target, shared))
try:
_, header = next(rows)
except StopIteration as exc:
raise PinoutError(f"管脚信息表为空: {sheet_name}") from exc
columns = header_columns(header)
required = {"pin_number", "pin_name", "fmux_register", "fmux_default", "description"}
missing = sorted(required - set(columns))
if missing:
raise PinoutError(f"管脚信息表缺少列: {missing}")
parsed: list[dict[str, object]] = []
for excel_row, values in rows:
pin_number = values.get(columns["pin_number"], "").strip()
pin_name = values.get(columns["pin_name"], "").strip()
if not pin_number and not pin_name:
continue
item: dict[str, object] = {"excel_row": excel_row}
for field, column in columns.items():
item[field] = values.get(column, "").strip()
item["muxes"] = parse_muxes(str(item.get("description", "")))
parsed.append(item)
if not parsed:
raise PinoutError(f"管脚信息表未解析出数据行: {sheet_name}")
return sheet_name, parsed
except (BadZipFile, KeyError, ET.ParseError) as exc:
raise PinoutError(f"内置 XLSX 结构无效: {path}: {exc}") from exc
def lookup_aliases(value: object) -> set[str]:
normalized = re.sub(r"[^A-Za-z0-9]", "", str(value)).upper()
if not normalized:
return set()
aliases = {normalized}
if normalized.startswith("XGPIO"):
aliases.add(normalized[1:])
elif normalized.startswith("GPIO"):
aliases.add("X" + normalized)
return aliases
def query_rows(
rows: list[dict[str, object]], queries: list[str], match_mode: str
) -> tuple[list[dict[str, object]], list[str]]:
results: list[dict[str, object]] = []
unmatched: list[str] = []
for query in queries:
needles = lookup_aliases(query)
query_matches: list[dict[str, object]] = []
for row in rows:
fields: list[tuple[str, object]] = [
("pin_number", row.get("pin_number", "")),
("pin_name", row.get("pin_name", "")),
]
fields.extend(
("mux_function", mux.get("function", ""))
for mux in row.get("muxes", [])
if isinstance(mux, dict)
)
matched_by: list[dict[str, str]] = []
for field, raw in fields:
haystacks = lookup_aliases(raw)
matched = bool(needles & haystacks)
if match_mode == "contains":
matched = any(needle in haystack for needle in needles for haystack in haystacks)
if matched:
matched_by.append({"field": field, "value": str(raw)})
if matched_by:
query_matches.append({**row, "matched_by": matched_by})
if query_matches:
results.append({"query": query, "matches": query_matches})
else:
unmatched.append(query)
return results, unmatched
def run(project: Path | None, chip_arg: str | None, queries: list[str], match_mode: str) -> dict[str, object]:
if project is not None:
chip, chip_evidence = project_chip(project)
elif chip_arg is not None:
chip = normalize_chip(chip_arg)
chip_evidence = {"source": "explicit-chip"}
else:
raise PinoutError("必须提供 --project 或 --chip")
document = select_document(chip)
sheet, rows = load_pin_rows(document["path"])
query_results, unmatched = query_rows(rows, queries, match_mode)
errors = [f"内置 GPIO 表中未找到查询项: {item}" for item in unmatched]
return {
"schema_version": 1,
"passed": not errors,
"chip": chip,
"chip_evidence": chip_evidence,
"pinout": {
"id": document["id"],
"revision": document.get("revision"),
"path": str(document["path"]),
"sha256": document["sha256"],
"worksheet": sheet,
"row_count": len(rows),
},
"match_mode": match_mode,
"queries": query_results,
"unmatched": unmatched,
"errors": errors,
}
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
source = parser.add_mutually_exclusive_group(required=True)
source.add_argument("--project", type=Path, help="从项目 buildEnvConfig.sh 的 EP_CHIP 选表")
source.add_argument("--chip", help="显式芯片名称;不从项目目录名推断")
parser.add_argument(
"--query",
action="append",
default=[],
help="查询 Pin Num、Pin Name、FMUX function;可重复,不提供时只解析并报告选表",
)
parser.add_argument("--match", choices=("exact", "contains"), default="exact")
parser.add_argument("--output", type=Path)
return parser
def main() -> int:
args = build_parser().parse_args()
try:
result = run(args.project, args.chip, args.query, args.match)
if args.output is not None:
write_json_atomic(args.output, result)
except (PinoutError, OSError, json.JSONDecodeError) as exc:
result = {
"schema_version": 1,
"passed": False,
"errors": [str(exc)],
}
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0 if result.get("passed") is True else 2
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,417 @@
#!/usr/bin/env python3
"""Record a verified BSP imgs path and produce the BSP-to-PSP handoff manifest."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import shlex
import subprocess
import tempfile
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from pathlib import Path
import bsp_preflight
import project_guard
REQUIRED_IMAGE_MARGIN_BYTES = 512 * 1024
REQUIRED_PARTITION_IMAGES = ("kernel.bin", "rootfs.bin")
def now_utc() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def write_text_atomic(path: Path, content: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(
"w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False
) as handle:
handle.write(content)
temporary = Path(handle.name)
os.replace(temporary, path)
def read_reference_project(project: Path) -> str | None:
path = project / "doc" / "bsp_adaptation" / "validation" / "clone_manifest.json"
if not path.is_file():
return None
payload = json.loads(path.read_text(encoding="utf-8"))
value = payload.get("reference_project")
return value if isinstance(value, str) else None
def git_state(project: Path) -> dict[str, object]:
top = subprocess.run(
["git", "-C", str(project), "rev-parse", "--show-toplevel"],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=False,
)
if top.returncode != 0:
return {"tracked": False, "repository_root": None, "commit": None, "dirty": None}
commit = subprocess.run(
["git", "-C", str(project), "rev-parse", "HEAD"],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=False,
)
status = subprocess.run(
["git", "-C", str(project), "status", "--porcelain", "--untracked-files=all", "--", "."],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=False,
)
if commit.returncode != 0 or status.returncode != 0:
return {"tracked": True, "repository_root": top.stdout.strip(), "commit": None, "dirty": None}
lines = [line for line in status.stdout.splitlines() if line.strip()]
return {
"tracked": True,
"repository_root": str(Path(top.stdout.strip()).resolve()),
"commit": commit.stdout.strip(),
"dirty": bool(lines),
"status": lines,
}
def artifact_inventory(output_directory: Path) -> tuple[list[dict[str, object]], list[str]]:
artifacts: list[dict[str, object]] = []
warnings: list[str] = []
for path in sorted(output_directory.rglob("*")):
relative = path.relative_to(output_directory).as_posix()
if path.is_symlink():
warnings.append(f"未将软链接作为固件产物计入 hash: {relative}")
elif path.is_file():
size = path.stat().st_size
if size <= 0:
warnings.append(f"零长度文件未计入固件产物: {relative}")
continue
artifacts.append(
{
"name": path.name,
"relative_path": relative,
"absolute_path": str(path),
"size": size,
"sha256": sha256_file(path),
}
)
return artifacts, warnings
def resolve_partition_table(project: Path, requested: Path | None) -> Path:
configs = project / "configs"
if not configs.is_dir() or configs.is_symlink():
raise project_guard.GuardError(f"项目配置目录不存在或不是普通目录: {configs}")
if requested is not None:
candidate = requested.expanduser()
if not candidate.is_absolute():
candidate = project / candidate
if not candidate.is_file() or candidate.is_symlink():
raise project_guard.GuardError(f"指定分区表不存在或不是普通文件: {candidate}")
candidate = candidate.resolve()
try:
candidate.relative_to(configs.resolve())
except ValueError as exc:
raise project_guard.GuardError("指定分区表必须位于当前项目 configs/ 内") from exc
return candidate
candidates = sorted(
path.resolve()
for path in configs.rglob("partition*.xml")
if path.is_file() and not path.is_symlink()
)
if len(candidates) != 1:
rendered = ", ".join(str(path) for path in candidates) or "无"
raise project_guard.GuardError(
f"无法唯一定位生效分区表:找到 {len(candidates)} 个候选({rendered});"
"必须用 --partition-table 指定当前构建实际使用的 configs/ 内文件"
)
return candidates[0]
def parse_size_in_kb(value: str, partition_table: Path, image_name: str) -> int:
try:
size_in_kb = int(value, 10)
except ValueError as exc:
raise project_guard.GuardError(
f"分区表 {partition_table} 中 {image_name} 的 size_in_kb 不是十进制整数: {value!r}"
) from exc
if size_in_kb <= 0:
raise project_guard.GuardError(
f"分区表 {partition_table} 中 {image_name} 的 size_in_kb 必须大于零"
)
return size_in_kb
def validate_image_partition_margin(
output_directory: Path,
artifacts: list[dict[str, object]],
partition_table: Path,
) -> dict[str, object]:
required_margin_bytes = REQUIRED_IMAGE_MARGIN_BYTES
try:
root = ET.parse(partition_table).getroot()
except ET.ParseError as exc:
raise project_guard.GuardError(f"分区表 XML 解析失败 {partition_table}: {exc}") from exc
checks: list[dict[str, object]] = []
failures: list[str] = []
for image_name in REQUIRED_PARTITION_IMAGES:
image_artifacts = [item for item in artifacts if item.get("name") == image_name]
if len(image_artifacts) != 1:
failures.append(
f"最终 imgs 中 {image_name} 必须唯一存在,实际找到 {len(image_artifacts)} 个"
)
continue
partitions: list[dict[str, object]] = []
for element in root.iter("partition"):
if Path(element.attrib.get("file", "")).name != image_name:
continue
raw_size = element.attrib.get("size_in_kb")
if raw_size is None:
raise project_guard.GuardError(
f"分区表 {partition_table} 中 {image_name} 对应项缺少 size_in_kb"
)
size_in_kb = parse_size_in_kb(raw_size.strip(), partition_table, image_name)
partitions.append(
{
"label": element.attrib.get("label"),
"alias": element.attrib.get("alias"),
"file": element.attrib.get("file"),
"size_in_kb": size_in_kb,
"size_bytes": size_in_kb * 1024,
}
)
if not partitions:
failures.append(f"分区表未找到 file={image_name!r} 的对应分区")
continue
image = image_artifacts[0]
image_size = int(image["size"])
effective_partition_size = min(int(item["size_bytes"]) for item in partitions)
margin = effective_partition_size - image_size
passed = margin >= required_margin_bytes
check = {
"image": image_name,
"image_path": str(output_directory / str(image["relative_path"])),
"image_size_bytes": image_size,
"partitions": partitions,
"effective_partition_size_bytes": effective_partition_size,
"required_margin_bytes": required_margin_bytes,
"actual_margin_bytes": margin,
"passed": passed,
}
checks.append(check)
if not passed:
failures.append(
f"{image_name} 分区余量不足:镜像 {image_size} 字节,最小对应分区 "
f"{effective_partition_size} 字节,实际余量 {margin} 字节,"
f"要求至少 {required_margin_bytes} 字节"
)
if failures:
raise project_guard.GuardError(";".join(failures))
return {
"schema_version": 1,
"passed": True,
"checked_at": now_utc(),
"partition_table": str(partition_table),
"partition_table_sha256": sha256_file(partition_table),
"required_margin_bytes": required_margin_bytes,
"checks": checks,
}
def record_handoff(
project_name: str,
build_version: str,
exit_code: int,
completion_marker_seen: bool,
root: Path,
output_root: Path,
build_command: str | None,
risks: list[str],
partition_table: Path | None = None,
) -> dict[str, object]:
if exit_code != 0:
raise project_guard.GuardError("构建退出码非零,禁止生成 build_passed 交接")
if not completion_marker_seen:
raise project_guard.GuardError("未确认统一构建完成标记,禁止生成交接")
if not bsp_preflight.VERSION_RE.fullmatch(build_version):
raise project_guard.GuardError("构建版本必须为 YYYYMMDD_N")
root = project_guard.normalize_root(root)
project = project_guard.direct_child(root, project_name, "project", True)
output_root = output_root.expanduser().resolve(strict=False)
output_directory = output_root / project_name / "imgs" / build_version
if not output_directory.is_dir() or output_directory.is_symlink():
raise project_guard.GuardError(f"最终 imgs 版本目录不存在或不是普通目录: {output_directory}")
artifacts, warnings = artifact_inventory(output_directory)
if not artifacts:
raise project_guard.GuardError("最终 imgs 版本目录没有非空普通文件")
partition_table = resolve_partition_table(project, partition_table)
image_partition_check = validate_image_partition_margin(
output_directory, artifacts, partition_table
)
build_env = project / "buildEnvConfig.sh"
assignments = bsp_preflight.parse_assignments(build_env) if build_env.is_file() else {}
config_dirs = bsp_preflight.config_directories(project)
config_dir = config_dirs[0] if len(config_dirs) == 1 else None
if len(config_dirs) != 1:
warnings.append(f"无法唯一确定 board build type:找到 {len(config_dirs)} 个配置目录")
system_form = (
"dual-os" if config_dir is not None and bsp_preflight.detects_dual_os(config_dir) else "single-os"
)
command = build_command or (
f"cd {shlex.quote(str(root))} && source tools/script_completion.sh && "
f"EMX2_BuildSDKTool {project_name} {build_version} all"
)
validation_dir = project / "doc" / "bsp_adaptation" / "validation"
handoff_dir = project / "doc" / "bsp_adaptation" / "handoff"
hash_path = validation_dir / f"artifact_manifest_{build_version}.sha256"
margin_path = validation_dir / f"image_partition_margin_{build_version}.json"
hash_lines = "".join(
f"{item['sha256']} {item['relative_path']}\n" for item in artifacts
)
write_text_atomic(hash_path, hash_lines)
write_text_atomic(
margin_path, json.dumps(image_partition_check, ensure_ascii=False, indent=2) + "\n"
)
manifest: dict[str, object] = {
"schema_version": 1,
"handoff_type": "bsp-to-psp",
"generated_at": now_utc(),
"producer": {
"module": "BSP",
"project": project_name,
"project_path": str(project),
"reference_project": read_reference_project(project),
"git": git_state(project),
},
"baseline": {
"chip": assignments.get("EP_CHIP"),
"sdk": assignments.get("EP_SDK_VER"),
"architecture": assignments.get("EP_ARCH"),
"flash": assignments.get("EP_FLASH_TYPE"),
"board_build_type": config_dir.name if config_dir else None,
"system_form": system_form,
},
"build": {
"version": build_version,
"status": "build_passed",
"exit_code": exit_code,
"completion_marker_seen": True,
"working_directory": str(root),
"command": command,
"imgs_root": project_guard.expected_imgs_target(project_name),
"imgs_version_path": str(output_directory),
"artifact_manifest_path": str(hash_path),
"image_partition_margin_path": str(margin_path),
},
"artifacts": artifacts,
"validation": {
"local_build": "passed",
"image_partition_margin": image_partition_check,
"hardware": "pending",
"owner_acceptance": "pending",
"release": "not-authorized",
},
"risks": risks,
"warnings": warnings,
"consumer_contract": {
"module": "PSP",
"immutable_artifacts": True,
"verify_sha256_before_use": True,
"select_latest_directory": False,
"baseline_change_rule": "Any project, build version, artifact path/size/hash, partition table/hash, or image-margin change invalidates downstream PSP validation.",
},
}
handoff_path = handoff_dir / "psp_handoff.json"
write_text_atomic(
handoff_path, json.dumps(manifest, ensure_ascii=False, indent=2) + "\n"
)
status = {
"schema_version": 1,
"updated_at": now_utc(),
"project": project_name,
"state": "build_passed",
"imgs_version_path": str(output_directory),
"psp_handoff_path": str(handoff_path),
"image_partition_margin_path": str(margin_path),
"image_partition_margin": image_partition_check,
"hardware_validation": "pending",
"owner_acceptance": "pending",
"release": "not-authorized",
}
status_path = project / "doc" / "bsp_adaptation" / "status.json"
write_text_atomic(status_path, json.dumps(status, ensure_ascii=False, indent=2) + "\n")
return {
"passed": True,
"imgs_version_path": str(output_directory),
"artifact_count": len(artifacts),
"artifact_manifest_path": str(hash_path),
"image_partition_margin_path": str(margin_path),
"image_partition_margin": image_partition_check,
"psp_handoff_path": str(handoff_path),
"status_path": str(status_path),
"warnings": warnings,
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--project", required=True)
parser.add_argument("--build-version", required=True)
parser.add_argument("--build-exit-code", type=int, required=True)
parser.add_argument("--completion-marker-seen", action="store_true")
parser.add_argument("--build-command")
parser.add_argument("--risk", action="append", default=[])
parser.add_argument(
"--partition-table",
type=Path,
help="自动定位不唯一时,指定当前项目 configs/ 内实际生效的分区表",
)
parser.add_argument("--emxres-root", type=Path, default=project_guard.production_root())
parser.add_argument(
"--output-root", type=Path, default=Path("/epbuild/EMX2.0/EmxRes")
)
args = parser.parse_args()
try:
result = record_handoff(
args.project,
args.build_version,
args.build_exit_code,
args.completion_marker_seen,
args.emxres_root,
args.output_root,
args.build_command,
args.risk,
args.partition_table,
)
except (project_guard.GuardError, OSError, json.JSONDecodeError) as exc:
result = {"passed": False, "error": str(exc)}
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0 if result.get("passed") is True else 2
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,187 @@
#!/usr/bin/env python3
"""Behavioral tests for bounded preflight and BSP-to-PSP handoff recording."""
import hashlib
import json
import tempfile
from pathlib import Path
import bsp_preflight
import project_guard
import record_build_handoff
def make_project(root: Path, name: str) -> Path:
project = root / name
config = project / "configs" / "cv1841cp_wevb_0015a_spinand"
config.mkdir(parents=True)
(project / "buildEnvConfig.sh").write_text(
"EP_CHIP=CV1841CP\nEP_SDK_VER=cv184x_v6.3.3\n"
"EP_ARCH=MUSL\nEP_FLASH_TYPE=nand\nNAND_FLASH_ID=0xa1d4\n",
encoding="utf-8",
)
(project / "configs" / "EPSYNCSDKConfig.sh").write_text("#!/bin/bash\n", encoding="utf-8")
(config / "cv1841cp_wevb_0015a_spinand_defconfig").write_text(
"CONFIG_DUAL_OS=y\n", encoding="utf-8"
)
partition = config / "partition" / "partition_spinand.xml"
partition.parent.mkdir()
partition.write_text(
"""<physical_partition type="spinand">
<partition label="BOOT" alias="kernel.A" size_in_kb="1024" file="kernel.bin"/>
<partition label="BOOT2" alias="kernel.B" size_in_kb="1536" file="kernel.bin"/>
<partition label="ROOTFS" alias="rootfs.A" size_in_kb="1024" file="rootfs.bin"/>
</physical_partition>
""",
encoding="utf-8",
)
(project / "imgs").symlink_to(project_guard.expected_imgs_target(name))
return project
def main() -> int:
with tempfile.TemporaryDirectory() as tmp:
base = Path(tmp)
root = base / "EmxRes"
root.mkdir()
tools = root / "tools"
tools.mkdir()
(tools / "script_completion.sh").write_text("# test\n", encoding="utf-8")
reference = make_project(root, "CV1841CP-REF_dualos")
project_report = bsp_preflight.project_preflight(
reference.name, "CV1841CP-NEW_dualos", root
)
assert project_report["passed"] is True, project_report
assert project_report["facts"]["board_build_type"] == "cv1841cp_wevb_0015a_spinand"
assert project_report["facts"]["system_form"] == "dual-os"
output_root = base / "epbuild"
build_report = bsp_preflight.build_preflight(
reference.name, "20260902_1", root, output_root
)
assert build_report["passed"] is True, build_report
assert build_report["working_directory"] == str(root.resolve())
assert build_report["command"].startswith(f"cd {root.resolve()} &&")
output = output_root / reference.name / "imgs" / "20260902_1"
output.mkdir(parents=True)
boot = output / "boot.bin"
boot.write_bytes(b"verified BSP artifact")
kernel = output / "kernel.bin"
kernel.write_bytes(b"k" * (512 * 1024))
rootfs = output / "rootfs.bin"
rootfs.write_bytes(b"r" * (256 * 1024))
conflict = bsp_preflight.build_preflight(
reference.name, "20260902_1", root, output_root
)
assert conflict["passed"] is False
clone = reference / "doc" / "bsp_adaptation" / "validation" / "clone_manifest.json"
clone.parent.mkdir(parents=True)
clone.write_text(
json.dumps({"reference_project": "CV1841CP-BASE_dualos"}), encoding="utf-8"
)
recorded = record_build_handoff.record_handoff(
reference.name,
"20260902_1",
0,
True,
root,
output_root,
None,
["NAND part number pending"],
)
assert recorded["passed"] is True, recorded
assert recorded["imgs_version_path"] == str(output.resolve())
assert recorded["image_partition_margin"]["passed"] is True
handoff_path = Path(recorded["psp_handoff_path"])
handoff = json.loads(handoff_path.read_text(encoding="utf-8"))
assert handoff["handoff_type"] == "bsp-to-psp"
assert handoff["producer"]["git"]["tracked"] is False
assert handoff["build"]["imgs_version_path"] == str(output.resolve())
assert handoff["build"]["status"] == "build_passed"
margin = handoff["validation"]["image_partition_margin"]
assert margin["passed"] is True
assert margin["required_margin_bytes"] == 512 * 1024
assert margin["partition_table_sha256"]
kernel_check = next(item for item in margin["checks"] if item["image"] == "kernel.bin")
assert kernel_check["actual_margin_bytes"] == 512 * 1024
assert kernel_check["effective_partition_size_bytes"] == 1024 * 1024
assert Path(recorded["image_partition_margin_path"]).is_file()
assert handoff["consumer_contract"]["select_latest_directory"] is False
assert handoff["artifacts"][0]["sha256"] == hashlib.sha256(boot.read_bytes()).hexdigest()
assert handoff["validation"]["hardware"] == "pending"
overflow_output = output_root / reference.name / "imgs" / "20260902_2"
overflow_output.mkdir(parents=True)
(overflow_output / "boot.bin").write_bytes(b"boot")
(overflow_output / "kernel.bin").write_bytes(b"k" * (512 * 1024))
(overflow_output / "rootfs.bin").write_bytes(b"r" * (512 * 1024 + 1))
try:
record_build_handoff.record_handoff(
reference.name,
"20260902_2",
0,
True,
root,
output_root,
None,
[],
)
except project_guard.GuardError as exc:
assert "rootfs.bin 分区余量不足" in str(exc)
else:
raise AssertionError("image with less than 512 KiB margin must fail")
assert not (
reference
/ "doc"
/ "bsp_adaptation"
/ "validation"
/ "image_partition_margin_20260902_2.json"
).exists()
original_partition = (
reference
/ "configs"
/ "cv1841cp_wevb_0015a_spinand"
/ "partition"
/ "partition_spinand.xml"
)
duplicate_partition = original_partition.with_name("partition_backup.xml")
duplicate_partition.write_text(original_partition.read_text(encoding="utf-8"), encoding="utf-8")
try:
record_build_handoff.resolve_partition_table(reference, None)
except project_guard.GuardError as exc:
assert "无法唯一定位" in str(exc)
else:
raise AssertionError("ambiguous partition tables must fail")
assert (
record_build_handoff.resolve_partition_table(
reference,
original_partition.relative_to(reference),
)
== original_partition.resolve()
)
try:
record_build_handoff.record_handoff(
reference.name,
"20260902_1",
1,
True,
root,
output_root,
None,
[],
)
except project_guard.GuardError as exc:
assert "退出码" in str(exc)
else:
raise AssertionError("failed build must not produce a PSP handoff")
print("BSP preflight and handoff behavioral tests passed")
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,76 @@
#!/usr/bin/env python3
"""Behavioral smoke tests for build_pinmux_csv.py."""
import csv
import json
import subprocess
import sys
import tempfile
from pathlib import Path
SCRIPT = Path(__file__).with_name("build_pinmux_csv.py")
HEADERS = ["引脚号", "功能名称", "引脚名称", "有效电平", "FMUX寄存器", "FMUX当前值", "FMUX默认值", "FMUX_list"]
def run(payload):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source = root / "input.json"
output = root / "out.csv"
source.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
result = subprocess.run(
[sys.executable, str(SCRIPT), "--input", str(source), "--output", str(output)],
text=True,
capture_output=True,
check=False,
)
data = output.read_bytes() if output.exists() else b""
return result, data
def valid_row():
return {
"pin_number": "A12",
"function_name": "CAM_RST0",
"pin_name": "PAD_MIPI_TXM4",
"is_gpio": True,
"active_level": "低有效",
"fmux_register": "0x030010A4",
"fmux_current": "3",
"fmux_default": "0",
"fmux_list": "0:XGPIOA_20; 1:CAM_MCLK0; 3:GPIOA_20",
}
def main():
result, data = run({"project_name": "测试 项目", "rows": [valid_row()]})
assert result.returncode == 0, result.stderr
assert data.startswith(b"\xef\xbb\xbf")
rows = list(csv.reader(data.decode("utf-8-sig").splitlines()))
assert rows[0] == HEADERS
assert len(rows) == 2 and rows[1][0] == "A12"
bad = valid_row()
bad["active_level"] = None
result, data = run({"project_name": "bad", "rows": [bad]})
assert result.returncode == 2 and not data
assert "GPIO 有效电平无效" in result.stderr
duplicate = valid_row()
result, data = run({"project_name": "dup", "rows": [valid_row(), duplicate]})
assert result.returncode == 2 and not data
assert "引脚号重复" in result.stderr
mismatched_mux = valid_row()
mismatched_mux["fmux_current"] = "7"
result, data = run({"project_name": "mux", "rows": [mismatched_mux]})
assert result.returncode == 2 and not data
assert "FMUX当前值 7 未出现在 FMUX_list" in result.stderr
print("Pinmux CSV behavioral tests passed")
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,303 @@
#!/usr/bin/env python3
"""Behavioral tests for project_guard.py using isolated temporary projects."""
import json
import os
import sys
import tempfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
import project_guard # noqa: E402
def make_reference(root: Path, name: str = "CV1841CP-REF_dualos") -> Path:
reference = root / name
uboot = reference / "configs" / "uboot"
uboot.mkdir(parents=True)
(reference / "buildEnvConfig.sh").write_text(
"EP_CHIP=CV1841CP\nEP_SDK_VER=v6.3.2\nEP_FLASH_TYPE=nand\n",
encoding="utf-8",
)
(uboot / "cvi_board_init.c").write_text("int cvi_board_init(void) { return 0; }\n", encoding="utf-8")
generated = reference / "build"
generated.mkdir()
(generated / "artifact.bin").write_bytes(b"generated")
(generated / "external-cache-link").symlink_to("/outside/ignored-build-cache")
(reference / "debug.log").write_text("ignored\n", encoding="utf-8")
(reference / "imgs").symlink_to(project_guard.expected_imgs_target(name))
return reference
def test_prepare_verify_finalize(root: Path) -> None:
reference_name = "CV1841CP-REF_dualos"
target_name = "CV1841CP-NEW_dualos"
reference = make_reference(root, reference_name)
prepared = project_guard.prepare_project(reference_name, target_name, root=root)
staging = root / prepared["staging_directory"]
assert staging.is_dir()
assert not (staging / "build").exists()
assert not (staging / "debug.log").exists()
assert (staging / "imgs").is_symlink()
assert os.readlink(staging / "imgs") == project_guard.expected_imgs_target(target_name)
manifest_path = staging / "doc" / "bsp_adaptation" / "validation" / "clone_manifest.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
assert manifest["reference_project"] == reference_name
assert manifest["imgs_retargeted"] is True
relative_config = "configs/uboot/cvi_board_init.c"
(staging / relative_config).write_text("int cvi_board_init(void) { return 1; }\n", encoding="utf-8")
report = staging / "doc" / "bsp_adaptation" / "adaptation_report.md"
report.write_text("# BSP adaptation\n\nStatic validation passed.\n", encoding="utf-8")
verified = project_guard.verify_project(
reference_name,
staging.name,
target_name,
[relative_config],
root=root,
)
assert verified["passed"] is True, verified
assert relative_config in verified["changes"]["modified"]
assert "imgs" in verified["changes"]["modified"]
finalized = project_guard.finalize_project(
reference_name,
staging.name,
target_name,
[relative_config],
root=root,
)
target = root / target_name
assert finalized["target_path"] == str(target)
assert target.is_dir() and not staging.exists()
assert reference.is_dir()
final_manifest = json.loads(
(target / "doc" / "bsp_adaptation" / "validation" / "clone_manifest.json").read_text(
encoding="utf-8"
)
)
assert final_manifest["final_path"] == str(target)
handoff = target / "doc" / "bsp_adaptation" / "handoff" / "psp_handoff.json"
handoff.parent.mkdir(parents=True)
handoff.write_text('{"schema_version": 1}\n', encoding="utf-8")
audited = project_guard.audit_final_project(target_name, root=root)
assert audited["passed"] is True, audited
assert audited["reference_drifted_since_prepare"] is False
(target / "buildEnvConfig.sh").write_text("EP_CHIP=CV1842HP\n", encoding="utf-8")
violated = project_guard.audit_final_project(target_name, root=root)
assert violated["passed"] is False
assert violated["violations"] == ["buildEnvConfig.sh"]
def test_unapproved_delta_is_rejected(root: Path) -> None:
reference_name = "CV1841CP-REF2_dualos"
target_name = "CV1841CP-NEW2_dualos"
make_reference(root, reference_name)
prepared = project_guard.prepare_project(reference_name, target_name, root=root)
staging = root / prepared["staging_directory"]
(staging / "buildEnvConfig.sh").write_text("EP_CHIP=CV1842HP\n", encoding="utf-8")
(staging / "doc" / "bsp_adaptation" / "adaptation_report.md").write_text(
"not final\n", encoding="utf-8"
)
verified = project_guard.verify_project(
reference_name,
staging.name,
target_name,
["configs/uboot/cvi_board_init.c"],
root=root,
)
assert verified["passed"] is False
assert verified["violations"] == ["buildEnvConfig.sh"]
try:
project_guard.finalize_project(
reference_name,
staging.name,
target_name,
["configs/uboot/cvi_board_init.c"],
root=root,
)
except project_guard.GuardError as exc:
assert "未通过" in str(exc)
else:
raise AssertionError("finalize should reject a failed delta report")
def test_external_link_and_credentials_are_rejected(root: Path) -> None:
external = make_reference(root, "CV1815JA-EXTERNAL")
(external / "imgs").unlink()
(external / "other-link").symlink_to("/outside/emxres/shared")
try:
project_guard.prepare_project(external.name, "CV1815JA-NEW", root=root)
except project_guard.GuardError as exc:
assert "外部绝对软链接" in str(exc)
else:
raise AssertionError("unexpected external symlink should be rejected")
credential = make_reference(root, "CV1815JA-CREDENTIAL")
(credential / ".env").write_text("TOKEN=secret\n", encoding="utf-8")
try:
project_guard.prepare_project(credential.name, "CV1815JA-CLEAN", root=root)
except project_guard.GuardError as exc:
assert "疑似凭据" in str(exc)
else:
raise AssertionError("credential-like files should be rejected")
def test_changes_after_verify_are_rejected(root: Path) -> None:
reference_name = "CV1841CP-REF3_dualos"
target_name = "CV1841CP-NEW3_dualos"
make_reference(root, reference_name)
prepared = project_guard.prepare_project(reference_name, target_name, root=root)
staging = root / prepared["staging_directory"]
report = staging / "doc" / "bsp_adaptation" / "adaptation_report.md"
report.write_text("# verified\n", encoding="utf-8")
verified = project_guard.verify_project(
reference_name,
staging.name,
target_name,
[],
root=root,
)
assert verified["passed"] is True
(staging / "buildEnvConfig.sh").write_text("EP_CHIP=CV1842HP\n", encoding="utf-8")
try:
project_guard.finalize_project(reference_name, staging.name, target_name, [], root=root)
except project_guard.GuardError as exc:
assert "未通过" in str(exc)
else:
raise AssertionError("finalize should reject post-verification changes")
def test_reference_drift_is_rejected(root: Path) -> None:
reference_name = "CV1841CP-REF4_dualos"
target_name = "CV1841CP-NEW4_dualos"
reference = make_reference(root, reference_name)
prepared = project_guard.prepare_project(reference_name, target_name, root=root)
staging = root / prepared["staging_directory"]
(reference / "buildEnvConfig.sh").write_text("EP_CHIP=CV1842HP\n", encoding="utf-8")
try:
project_guard.verify_project(reference_name, staging.name, target_name, [], root=root)
except project_guard.GuardError as exc:
assert "参考项目发生变化" in str(exc)
else:
raise AssertionError("verify should reject reference drift after prepare")
def test_staging_credentials_and_excluded_outputs_are_rejected(root: Path) -> None:
reference_name = "CV1815JA-REF5"
target_name = "CV1815JA-NEW5"
make_reference(root, reference_name)
prepared = project_guard.prepare_project(reference_name, target_name, root=root)
staging = root / prepared["staging_directory"]
(staging / ".env").write_text("TOKEN=secret\n", encoding="utf-8")
try:
project_guard.verify_project(reference_name, staging.name, target_name, [], root=root)
except project_guard.GuardError as exc:
assert "疑似凭据" in str(exc)
else:
raise AssertionError("verify should reject credentials added during editing")
(staging / ".env").unlink()
(staging / "logs").mkdir()
(staging / "logs" / "build.txt").write_text("generated\n", encoding="utf-8")
try:
project_guard.verify_project(reference_name, staging.name, target_name, [], root=root)
except project_guard.GuardError as exc:
assert "被排除" in str(exc)
else:
raise AssertionError("verify should reject excluded outputs added during editing")
def test_excluded_output_added_after_verify_is_rejected(root: Path) -> None:
reference_name = "CV1815JA-REF6"
target_name = "CV1815JA-NEW6"
make_reference(root, reference_name)
prepared = project_guard.prepare_project(reference_name, target_name, root=root)
staging = root / prepared["staging_directory"]
(staging / "doc" / "bsp_adaptation" / "adaptation_report.md").write_text(
"# verified\n", encoding="utf-8"
)
verified = project_guard.verify_project(
reference_name, staging.name, target_name, [], root=root
)
assert verified["passed"] is True
(staging / "logs").mkdir()
(staging / "logs" / "late.txt").write_text("late\n", encoding="utf-8")
try:
project_guard.finalize_project(reference_name, staging.name, target_name, [], root=root)
except project_guard.GuardError as exc:
assert "被排除" in str(exc)
else:
raise AssertionError("finalize should reject excluded output added after verify")
def test_tampered_delta_report_is_not_trusted(root: Path) -> None:
reference_name = "CV1815JA-REF7"
target_name = "CV1815JA-NEW7"
make_reference(root, reference_name)
prepared = project_guard.prepare_project(reference_name, target_name, root=root)
staging = root / prepared["staging_directory"]
(staging / "doc" / "bsp_adaptation" / "adaptation_report.md").write_text(
"# report\n", encoding="utf-8"
)
(staging / "buildEnvConfig.sh").write_text("EP_CHIP=CV1842HP\n", encoding="utf-8")
verified = project_guard.verify_project(
reference_name, staging.name, target_name, [], root=root
)
assert verified["passed"] is False
delta_path = staging / "doc" / "bsp_adaptation" / "validation" / "project_delta.json"
tampered = json.loads(delta_path.read_text(encoding="utf-8"))
tampered["passed"] = True
tampered["violations"] = []
project_guard.write_json(delta_path, tampered)
try:
project_guard.finalize_project(reference_name, staging.name, target_name, [], root=root)
except project_guard.GuardError as exc:
assert "未通过" in str(exc)
else:
raise AssertionError("finalize should recompute instead of trusting a tampered delta report")
def test_allow_paths_must_be_exact() -> None:
try:
project_guard.normalize_allow_paths(["configs/**"])
except project_guard.GuardError as exc:
assert "不允许通配符" in str(exc)
else:
raise AssertionError("wildcard allow paths should be rejected")
def test_audit_requires_frozen_baseline(root: Path) -> None:
target = root / "CV1841CP-UNMANAGED_dualos"
target.mkdir()
try:
project_guard.audit_final_project(target.name, root=root)
except project_guard.GuardError as exc:
assert "clone_manifest" in str(exc)
else:
raise AssertionError("audit should reject an unmanaged project")
def main() -> int:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp) / "EmxRes"
root.mkdir()
test_prepare_verify_finalize(root)
test_unapproved_delta_is_rejected(root)
test_external_link_and_credentials_are_rejected(root)
test_changes_after_verify_are_rejected(root)
test_reference_drift_is_rejected(root)
test_staging_credentials_and_excluded_outputs_are_rejected(root)
test_excluded_output_added_after_verify_is_rejected(root)
test_tampered_delta_report_is_not_trusted(root)
test_audit_requires_frozen_baseline(root)
test_allow_paths_must_be_exact()
print("Project guard behavioral tests passed")
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,96 @@
#!/usr/bin/env python3
"""Behavioral tests for the bundled CV18 GPIO pinout catalog."""
from __future__ import annotations
import importlib.util
import tempfile
from pathlib import Path
SCRIPT = Path(__file__).with_name("query_builtin_pinout.py")
SPEC = importlib.util.spec_from_file_location("query_builtin_pinout", SCRIPT)
assert SPEC is not None and SPEC.loader is not None
MODULE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MODULE)
def selected_id(chip: str) -> str:
return str(MODULE.select_document(chip)["id"])
def main() -> int:
expected = {
"CV1810": "cv181xc",
"CV1811C": "cv181xc",
"CV1812CPA": "cv181xc",
"CV1812HP": "cv181xh",
"CV1815JA": "cv1815j",
"CV1841CP": "cv184xc",
"CV1842CP": "cv184xc",
"CV1842HP": "cv184xh",
"CV1843HP": "cv184xh",
}
for chip, identifier in expected.items():
assert selected_id(chip) == identifier, chip
try:
MODULE.select_document("CV1801B")
except MODULE.PinoutError as exc:
assert "没有对应" in str(exc)
else:
raise AssertionError("unsupported chips must fail closed")
manifest = MODULE.load_manifest()
assert len(manifest) == 5
for document in manifest:
sheet, rows = MODULE.load_pin_rows(document["path"])
assert "管腳信息表" in sheet or "管脚信息表" in sheet
assert rows
results, unmatched = MODULE.query_rows(rows, ["SD0_CLK", "GPIOA_7"], "exact")
assert not unmatched, document["id"]
assert [item["query"] for item in results] == ["SD0_CLK", "GPIOA_7"]
gpio_matches = results[1]["matches"]
assert any(
mux.get("function") == "XGPIOA[7]"
for row in gpio_matches
for mux in row["muxes"]
)
with tempfile.TemporaryDirectory() as temporary:
project = Path(temporary) / "CV1815JA-misleading-name"
project.mkdir()
config = project / "buildEnvConfig.sh"
config.write_text("export EP_CHIP='CV1842HP'\n", encoding="utf-8")
result = MODULE.run(project, None, ["SD0_CLK"], "exact")
assert result["passed"] is True
assert result["chip"] == "CV1842HP"
assert result["pinout"]["id"] == "cv184xh"
assert result["chip_evidence"]["source"] == "project-buildEnvConfig"
config.write_text("EP_CHIP=CV1811C\nEP_CHIP=CV1841CP\n", encoding="utf-8")
try:
MODULE.project_chip(project)
except MODULE.PinoutError as exc:
assert "唯一确定" in str(exc)
else:
raise AssertionError("ambiguous EP_CHIP must fail closed")
config.write_text("EP_CHIP=$CV1841CP\n", encoding="utf-8")
try:
MODULE.project_chip(project)
except MODULE.PinoutError as exc:
assert "静态芯片名称" in str(exc)
else:
raise AssertionError("dynamic EP_CHIP must fail closed")
result = MODULE.run(None, "CV1841CP", ["NOT_A_REAL_PIN"], "exact")
assert result["passed"] is False
assert result["unmatched"] == ["NOT_A_REAL_PIN"]
print("Bundled CV18 GPIO pinout behavioral tests passed")
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,83 @@
#!/usr/bin/env python3
"""Behavioral tests for validate_gpio_plan.py."""
import copy
import validate_gpio_plan
def valid_plan() -> dict[str, object]:
return {
"schema_version": 1,
"project": "CV1841CP-BOARD_dualos",
"validation_mode": "emxres-config",
"sources": ["schematic.pdf page 3", "GPIO List row 12"],
"confirmation": {
"confirmed": True,
"confirmed_at": "2026-09-02T10:00:00+08:00",
"summary": "用户确认整张 GPIO 实施表",
},
"signals": [
{
"function": "CAM_RST0",
"pad": "PAD_MIPI_TXP2",
"mux": "XGPIOC_17",
"gpio": "C17",
"direction": "output",
"active_level": "low",
"boot_value": 0,
"runtime_owner": "camera",
"dts_consumers": ["mipi_rx.snsr-reset"],
"source_evidence": ["schematic.pdf page 3"],
"decision": "confirmed",
"note": "early reset",
},
{
"function": "CAM_MCLK0",
"pad": "PAD_MIPI_TXP0",
"mux": "CAM_MCLK0",
"gpio": None,
"direction": "peripheral",
"active_level": None,
"boot_value": None,
"runtime_owner": "camera",
"dts_consumers": [],
"source_evidence": ["schematic.pdf page 3"],
"decision": "inherited",
"note": "camera clock",
},
],
}
def main() -> int:
result = validate_gpio_plan.validate_plan(valid_plan())
assert result["passed"] is True, result
candidate = valid_plan()
candidate["confirmation"] = {"confirmed": False}
candidate["signals"][0]["decision"] = "provisional"
candidate["signals"][0]["active_level"] = "unknown"
candidate_result = validate_gpio_plan.validate_plan(candidate, allow_unconfirmed=True)
assert candidate_result["passed"] is True, candidate_result
assert candidate_result["warnings"]
assert validate_gpio_plan.validate_plan(candidate)["passed"] is False
duplicate = copy.deepcopy(valid_plan())
duplicate["signals"].append(copy.deepcopy(duplicate["signals"][0]))
duplicate["signals"][-1]["function"] = "WIFI_PWREN"
duplicate_result = validate_gpio_plan.validate_plan(duplicate)
assert duplicate_result["passed"] is False
assert any("重复 Pad" in error for error in duplicate_result["errors"])
assert any("重复 GPIO" in error for error in duplicate_result["errors"])
invalid_input = copy.deepcopy(valid_plan())
invalid_input["signals"][0]["direction"] = "input"
invalid_input["signals"][0]["boot_value"] = 1
assert validate_gpio_plan.validate_plan(invalid_input)["passed"] is False
print("GPIO plan behavioral tests passed")
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,184 @@
#!/usr/bin/env python3
"""Validate a machine-readable BSP GPIO implementation plan."""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
DIRECTIONS = {"input", "output", "peripheral", "none"}
ACTIVE_LEVELS = {"high", "low", "paired", "unknown", None}
DECISIONS = {"confirmed", "inherited", "no-gpio", "provisional", "blocked"}
IMPLEMENTABLE = {"confirmed", "inherited", "no-gpio"}
REQUIRED_SIGNAL_FIELDS = {
"function",
"pad",
"mux",
"gpio",
"direction",
"active_level",
"boot_value",
"runtime_owner",
"dts_consumers",
"source_evidence",
"decision",
"note",
}
def validate_plan(payload: object, allow_unconfirmed: bool = False) -> dict[str, object]:
errors: list[str] = []
warnings: list[str] = []
if not isinstance(payload, dict):
return {"passed": False, "errors": ["顶层必须是 JSON 对象"], "warnings": []}
if payload.get("schema_version") != 1:
errors.append("schema_version 必须为 1")
if not isinstance(payload.get("project"), str) or not payload.get("project", "").strip():
errors.append("project 必须是非空字符串")
if payload.get("validation_mode") not in {"emxres-config", "full-dts-audit"}:
errors.append("validation_mode 必须为 emxres-config 或 full-dts-audit")
sources = payload.get("sources")
if not isinstance(sources, list) or not sources or not all(
isinstance(item, str) and item.strip() for item in sources
):
errors.append("sources 必须是非空字符串数组")
confirmation = payload.get("confirmation")
confirmed = isinstance(confirmation, dict) and confirmation.get("confirmed") is True
if not confirmed and not allow_unconfirmed:
errors.append("confirmation.confirmed 必须为 true")
if confirmed:
if not isinstance(confirmation.get("confirmed_at"), str) or not confirmation.get(
"confirmed_at", ""
).strip():
errors.append("确认后必须填写 confirmation.confirmed_at")
if not isinstance(confirmation.get("summary"), str) or not confirmation.get(
"summary", ""
).strip():
errors.append("确认后必须填写 confirmation.summary")
signals = payload.get("signals")
if not isinstance(signals, list) or not signals:
errors.append("signals 必须是非空数组")
signals = []
functions: set[str] = set()
pads: set[str] = set()
gpios: set[str] = set()
for index, signal in enumerate(signals):
prefix = f"signals[{index}]"
if not isinstance(signal, dict):
errors.append(f"{prefix} 必须是对象")
continue
missing = sorted(REQUIRED_SIGNAL_FIELDS - set(signal))
if missing:
errors.append(f"{prefix} 缺少字段: {', '.join(missing)}")
continue
function = signal.get("function")
if not isinstance(function, str) or not function.strip():
errors.append(f"{prefix}.function 必须是非空字符串")
elif function in functions:
errors.append(f"重复功能名: {function}")
else:
functions.add(function)
direction = signal.get("direction")
decision = signal.get("decision")
active_level = signal.get("active_level")
boot_value = signal.get("boot_value")
pad = signal.get("pad")
gpio = signal.get("gpio")
if direction not in DIRECTIONS:
errors.append(f"{prefix}.direction 非法: {direction!r}")
if active_level not in ACTIVE_LEVELS:
errors.append(f"{prefix}.active_level 非法: {active_level!r}")
if decision not in DECISIONS:
errors.append(f"{prefix}.decision 非法: {decision!r}")
elif decision not in IMPLEMENTABLE and not allow_unconfirmed:
errors.append(f"{prefix} 仍为 {decision},不能进入实现")
if not isinstance(signal.get("runtime_owner"), str):
errors.append(f"{prefix}.runtime_owner 必须是字符串")
if not isinstance(signal.get("note"), str):
errors.append(f"{prefix}.note 必须是字符串")
if not isinstance(signal.get("dts_consumers"), list) or not all(
isinstance(item, str) for item in signal.get("dts_consumers", [])
):
errors.append(f"{prefix}.dts_consumers 必须是字符串数组")
evidence = signal.get("source_evidence")
if not isinstance(evidence, list) or not evidence or not all(
isinstance(item, str) and item.strip() for item in evidence
):
errors.append(f"{prefix}.source_evidence 必须是非空字符串数组")
if decision == "no-gpio":
if direction != "none" or any(
value is not None
for value in (pad, signal.get("mux"), gpio, active_level, boot_value)
):
errors.append(
f"{prefix} 的 no-gpio 行必须使用 direction=none 且 pad/mux/gpio/active_level/boot_value=null"
)
elif direction in {"input", "output"}:
if not isinstance(pad, str) or not pad.strip():
errors.append(f"{prefix} 的 GPIO 行必须填写 pad")
if not isinstance(gpio, str) or not gpio.strip():
errors.append(f"{prefix} 的 GPIO 行必须填写 gpio")
if active_level is None:
errors.append(f"{prefix} 的 GPIO 行不能使用 active_level=null")
if active_level == "unknown" and not allow_unconfirmed:
errors.append(f"{prefix} 的 active_level 仍为 unknown")
if direction == "output" and boot_value not in {0, 1}:
errors.append(f"{prefix} 的输出必须填写 boot_value=0 或 1")
if direction == "input" and boot_value is not None:
errors.append(f"{prefix} 的输入必须使用 boot_value=null")
elif direction == "peripheral":
if not isinstance(pad, str) or not pad.strip():
errors.append(f"{prefix} 的外设复用必须填写 pad")
if gpio is not None or active_level is not None or boot_value is not None:
errors.append(f"{prefix} 的外设复用必须使用 gpio/active_level/boot_value=null")
elif direction == "none" and decision != "no-gpio":
errors.append(f"{prefix} 只有 no-gpio 决策可以使用 direction=none")
if isinstance(pad, str) and pad:
if pad in pads:
errors.append(f"重复 Pad: {pad}")
pads.add(pad)
if isinstance(gpio, str) and gpio:
if gpio in gpios:
errors.append(f"重复 GPIO: {gpio}")
gpios.add(gpio)
if allow_unconfirmed and not confirmed:
warnings.append("计划尚未获得用户确认,仅通过候选结构检查")
return {
"passed": not errors,
"signal_count": len(signals),
"errors": errors,
"warnings": warnings,
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("plan", type=Path)
parser.add_argument("--allow-unconfirmed", action="store_true")
args = parser.parse_args()
try:
payload = json.loads(args.plan.read_text(encoding="utf-8"))
result = validate_plan(payload, allow_unconfirmed=args.allow_unconfirmed)
except (OSError, json.JSONDecodeError) as exc:
print(json.dumps({"passed": False, "errors": [str(exc)]}, ensure_ascii=False, indent=2))
return 2
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0 if result["passed"] else 2
if __name__ == "__main__":
raise SystemExit(main())