Files
kaptan/firmware/libs/kaptan-err-reportable-macros/tests/final_link.py
T
Onur Özkan 84807dde4e fix(diagnostics): bound formatting and release ingress before callbacks
Bound UTF-8 formatting and release diagnostic ingress before invoking
callbacks. Check error-ID collisions at Rust/C++ release final link.

Validation: diagnostic tests, strict Clippy, Miri and final-link cases.
2026-11-07 18:12:11 +03:00

230 lines
7.5 KiB
Python

#!/usr/bin/env python3
"""Exercise the actual derive across crates at the host release-LTO final link.
Run with Python 3.11+. Build output is retained under the supplied directory.
This is a link experiment, not a trybuild or cargo-check substitute.
"""
import argparse
import hashlib
import json
from pathlib import Path
import subprocess
import tomllib
def run(arguments, cwd):
return subprocess.run(arguments, cwd=cwd, text=True, capture_output=True)
def require_success(result):
if result.returncode:
raise RuntimeError(result.stdout + result.stderr)
def registry_packages(lock):
return {
(package["source"], package["name"], package["version"], package["checksum"])
for package in tomllib.loads(lock.decode())["package"]
if package.get("source", "").startswith("registry+")
}
def verify_registry_packages(firmware_lock, fixture_lock):
packages = registry_packages(fixture_lock)
drift = packages - registry_packages(firmware_lock)
if drift:
raise RuntimeError(
f"fixture registry dependencies differ from firmware lock: {sorted(drift)!r}"
)
return sorted(packages)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--work-dir", type=Path, required=True)
args = parser.parse_args()
work = args.work_dir.resolve()
work.mkdir(parents=True, exist_ok=True)
runtime = Path(__file__).resolve().parents[2] / "kaptan-err-reportable"
version = run(["rustc", "+1.90.0", "-vV"], work)
require_success(version)
host = next(
line.removeprefix("host: ")
for line in version.stdout.splitlines()
if line.startswith("host: ")
)
print(version.stdout, end="")
for name in ["a", "b", "app"]:
(work / name / "src").mkdir(parents=True, exist_ok=True)
(work / "Cargo.toml").write_text("""
[workspace]
members = ["a", "b", "app"]
resolver = "2"
[profile.release]
opt-level = 3
codegen-units = 1
lto = "fat"
panic = "abort"
strip = "symbols"
""")
for name in ["a", "b"]:
(work / name / "Cargo.toml").write_text(f"""
[package]
name = "error-id-{name}"
version = "0.0.0"
edition = "2024"
[lib]
crate-type = ["rlib", "staticlib"]
[features]
duplicate = []
unused = []
[dependencies]
kaptan-err-reportable = {{ path = {json.dumps(str(runtime))} }}
""")
unique = 65401 if name == "a" else 65402
(work / name / "src/lib.rs").write_text(f"""
#[derive(Debug, kaptan_err_reportable::KaptanError)]
pub enum Error {{
#[cfg_attr(feature = "duplicate", kaptan_error(id = 65401))]
#[cfg_attr(not(feature = "duplicate"), kaptan_error(id = {unique}))]
Fault,
}}
#[unsafe(no_mangle)]
pub extern "C" fn fixture_{name}() -> u16 {{
#[cfg(not(feature = "unused"))]
{{ kaptan_err_reportable::ReportableError::error_id(&Error::Fault).get() }}
#[cfg(feature = "unused")]
{{ {unique} }}
}}
""")
(work / "app/Cargo.toml").write_text("""
[package]
name = "error-id-app"
version = "0.0.0"
edition = "2024"
[dependencies]
error-id-a = { path = "../a" }
error-id-b = { path = "../b" }
""")
(work / "app/src/main.rs").write_text("fn main() {}\n")
firmware_lock = (runtime.parents[1] / "Cargo.lock").read_bytes()
(work / "firmware.Cargo.lock").write_bytes(firmware_lock)
fixture_lock_path = work / "Cargo.lock"
fixture_lock_path.write_bytes(firmware_lock)
# Resolve the fixture's local workspace entries while retaining locked
# registry versions. generate-lockfile would instead select newer versions.
require_success(
run(
[
"cargo",
"+1.90.0",
"metadata",
"--offline",
"--all-features",
"--format-version",
"1",
],
work,
)
)
fixture_lock = fixture_lock_path.read_bytes()
packages = verify_registry_packages(firmware_lock, fixture_lock)
provenance = [
f"firmware Cargo.lock SHA-256: {hashlib.sha256(firmware_lock).hexdigest()}",
f"fixture Cargo.lock SHA-256: {hashlib.sha256(fixture_lock).hexdigest()}",
"Fixture registry dependencies verified as an exact subset of the firmware lock:",
*[
f" {name} {version} source={source} checksum={checksum}"
for source, name, version, checksum in packages
],
]
print("\n".join(provenance))
results = []
for mode in ["used", "unused", "unextracted"]:
use_b = mode != "unextracted"
rust_call = (
"assert!(std::hint::black_box(error_id_b::fixture_b()) >= 65401);"
if use_b
else ""
)
native_call = "|| fixture_b() < 65401" if use_b else ""
(work / "app/src/main.rs").write_text(f"""
extern crate error_id_b;
fn main() {{
assert_eq!(std::hint::black_box(error_id_a::fixture_a()), 65401);
{rust_call}
}}
""")
(work / "main.cpp").write_text(f"""
#include <cstdint>
extern "C" uint16_t fixture_a();
extern "C" uint16_t fixture_b();
int main() {{ return fixture_a() != 65401 {native_call}; }}
""")
for duplicate in [False, True]:
label = f"{mode}-{'duplicate' if duplicate else 'unique'}"
features = ["error-id-b/duplicate"] if duplicate else []
if mode != "used":
features += ["error-id-a/unused", "error-id-b/unused"]
target = work / "target" / label
cargo = [
"cargo",
"+1.90.0",
"build",
"--locked",
"--offline",
"--release",
"--target",
host,
"--target-dir",
str(target),
]
if features:
cargo += ["--features", ",".join(features)]
# Each archive must build independently before linking them together.
archives = run(cargo + ["-p", "error-id-a", "-p", "error-id-b"], work)
require_success(archives)
release = target / host / "release"
native = run(
[
"c++",
str(work / "main.cpp"),
str(release / "liberror_id_a.a"),
str(release / "liberror_id_b.a"),
"-ldl",
"-lpthread",
"-lm",
"-o",
str(release / "native-app"),
],
work,
)
rust = run(cargo + ["-p", "error-id-app"], work)
for kind, result, binary in [
("staticlib", native, release / "native-app"),
("rlib", rust, release / "error-id-app"),
]:
(work / f"{label}-{kind}.log").write_text(result.stdout + result.stderr)
if not duplicate:
require_success(result)
elif (use_b or kind == "rlib") and result.returncode == 0:
raise RuntimeError(f"{label} {kind}: duplicate unexpectedly linked")
if result.returncode == 0:
require_success(run([str(binary)], work))
outcome = "linked and ran"
else:
if "__kaptan_error_id_65401" not in result.stderr:
raise RuntimeError(result.stderr)
outcome = "duplicate ID rejected"
results.append(f"{label} {kind}: {outcome}")
print("\n".join(results))
(work / "results.txt").write_text(
version.stdout + "\n".join(provenance + results) + "\n"
)
if __name__ == "__main__":
main()