scripts/ci_runtime.py
Rama main · réplica de lectura del repositorio original.
"""Source-bound PR runtime selection; never install or choose an ambient Python."""
from __future__ import annotations
import argparse
import errno
import hashlib
import importlib
import importlib.machinery
import importlib.metadata
import importlib.util
import json
import os
import platform
import re
import shutil
import subprocess
import sys
import tomllib
import zipfile
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def versions() -> dict[str, str]:
bootstrap = ROOT / "packages/apps/desktop/resources/bootstrap"
manifest = json.loads((bootstrap / "tool-manifest.json").read_text())
shell = (bootstrap / "flinttrade-bootstrap.sh").read_text()
python = re.findall(r"^export UV_PYTHON=(\d+\.\d+\.\d+)$", shell, re.MULTILINE)
if len(python) != 1:
raise ValueError("One exact managed bootstrap Python selector required")
package = json.loads((ROOT / "package.json").read_text())
pnpm = package["packageManager"].split("@", 1)[1].split("+", 1)[0]
if pnpm != manifest["pnpm"]["version"]:
raise ValueError("pnpm bootstrap/packageManager mismatch")
selected = {
"python-current": python[0],
"node-current": manifest["node"]["version"],
"uv-current": manifest["uv"]["version"],
"pnpm-current": pnpm,
}
if not all(re.fullmatch(r"\d+\.\d+\.\d+", value) for value in selected.values()):
raise ValueError("Exact stable runtime versions required")
return selected
def verify(tool: str, expected: str, output: Path | None) -> int:
if not re.fullmatch(r"\d+(?:\.\d+){0,2}", expected):
raise ValueError("A numeric version or support-floor prefix is required")
executable = sys.executable if tool == "python" else shutil.which(tool)
if executable is None:
raise RuntimeError("Selected tool is unavailable: " + tool)
command = (
[executable, "-I", "-S", "-c", "import platform;print(platform.python_version())"]
if tool == "python"
else [executable, "--version"]
)
observed = subprocess.check_output(command, text=True).strip()
uv_release = re.fullmatch(r"uv (\d+\.\d+\.\d+)(?: \([^()\r\n]+\))?", observed) if tool == "uv" else None
actual = uv_release.group(1) if uv_release else observed.removeprefix("v")
if not re.fullmatch(r"\d+\.\d+\.\d+", actual):
raise ValueError("Unrecognised stable tool version: " + observed)
matches = actual.split(".")[: len(expected.split("."))] == expected.split(".")
receipt = {"tool": tool, "executable": executable, "actual": actual, "reported": observed, "expected": expected, "matches": matches}
text = json.dumps(receipt, sort_keys=True)
if output is not None:
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(text + "\n")
print(text, flush=True)
return 0 if matches else 1
def native_inputs() -> dict:
crate = ROOT / "packages/core/ticks"
paths = [crate / name for name in ["Cargo.toml", "Cargo.lock", "pyproject.toml"]]
paths += sorted((crate / "src").rglob("*.rs"))
paths += sorted((crate / "python").rglob("*.py"))
return {"files": {p.relative_to(crate).as_posix(): hashlib.sha256(p.read_bytes()).hexdigest() for p in paths},
"uv_lock_sha256": hashlib.sha256((ROOT / "uv.lock").read_bytes()).hexdigest(),
"builder_constraint_sha256": hashlib.sha256((ROOT / "tools/acceptance/native-build.txt").read_bytes()).hexdigest()}
def native_admission(inputs: Path, wheel: Path) -> None:
"""Require the existing qualified boundary before loading extension code."""
if sys.flags.optimize:
raise RuntimeError("Python optimisation refused before native acceptance")
if os.environ.get("HOME") != "/run/test/home":
raise RuntimeError("Authorised isolated runner required")
paths = [ROOT / "scripts/ci_runtime.py",
*[ROOT / "packages/core/ticks" / name for name in native_inputs()["files"]],
ROOT / "uv.lock", ROOT / "tools/acceptance/native-build.txt",
ROOT / ".venv/pyvenv.cfg", inputs, wheel]
for path in paths:
if not path.resolve().is_relative_to(ROOT):
raise RuntimeError("Native inputs must belong to the source-only input")
# Do not truncate or write: permissions alone are not a read-only mount.
try:
descriptor = os.open(path, os.O_WRONLY)
except OSError as error:
if error.errno != errno.EROFS:
raise RuntimeError("Read-only native input required") from error
else:
os.close(descriptor)
raise RuntimeError("Read-only native input required")
spec = importlib.util.spec_from_file_location("native_acceptance_guard", ROOT / "tools/acceptance/guard.py")
if spec is None or spec.loader is None:
raise RuntimeError("Existing acceptance guard missing")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
module.guard()
if Path(sys.prefix).resolve() != (ROOT / ".venv").resolve():
raise RuntimeError("Native acceptance must use the prepared source environment")
def native(expected: str, inputs: Path, wheel: Path, output: Path | None) -> int:
if verify("python", expected, None):
return 1
if json.loads(inputs.read_text()) != native_inputs():
raise RuntimeError("Native source changed during build")
native_admission(inputs, wheel)
metadata = tomllib.loads((ROOT / "packages/core/ticks/pyproject.toml").read_text())
name = metadata["tool"]["maturin"]["module-name"]
package = importlib.import_module(name)
extension = importlib.import_module(name + "." + name)
if not isinstance(extension.__loader__, importlib.machinery.ExtensionFileLoader) or extension.__file__ is None or package.__file__ is None:
raise RuntimeError("An actual native extension loader is required")
origin = Path(extension.__file__).resolve(strict=True)
if not origin.is_relative_to(Path(sys.prefix).resolve()) or not any(str(origin).endswith(s) for s in importlib.machinery.EXTENSION_SUFFIXES):
raise RuntimeError("Native extension must load from the selected environment")
version = importlib.metadata.version(metadata["project"]["name"])
if version != metadata["project"]["version"]:
raise RuntimeError("Native distribution version mismatch")
with zipfile.ZipFile(wheel) as archive:
member = name + "/" + origin.name
if archive.read(member) != origin.read_bytes():
raise RuntimeError("Imported native bytes differ from the freshly built wheel")
if archive.read(name + "/__init__.py") != Path(package.__file__).read_bytes():
raise RuntimeError("Native Python wrapper differs from the freshly built wheel")
# A real extension call, not merely Cargo tests or an optional import/skip.
config = package.SpreadConfig(initial_capital=1000.0, fees=0.0)
config.add_leg(package.LegConfig(package.OptionType.Call, strike=100.0, quantity=-1, lot_size=1))
result = package.SpreadBacktest("CI", config).run([1, 2], [[10.0, 8.0]], [True, False], [False, True])
if result.total_pnl != 2.0:
raise RuntimeError("Native extension functional check failed")
receipt = {"python": platform.python_version(), "executable": sys.executable,
"module": name, "extension": str(origin), "version": version,
"extension_sha256": hashlib.sha256(origin.read_bytes()).hexdigest(),
"wheel_sha256": hashlib.sha256(wheel.read_bytes()).hexdigest(),
"inputs": native_inputs(), "functional_pnl": result.total_pnl}
text = json.dumps(receipt, sort_keys=True)
if output is not None:
output.write_text(text + "\n")
print(text)
return 0
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("mode", choices=["versions", "verify", "native-inputs", "native"])
parser.add_argument("--github-output", action="store_true")
parser.add_argument("--tool", choices=["python", "node", "pnpm", "uv"])
parser.add_argument("--expected")
parser.add_argument("--output", type=Path)
parser.add_argument("--inputs", type=Path)
parser.add_argument("--wheel", type=Path)
args = parser.parse_args()
if args.mode == "native-inputs":
if args.output is None:
parser.error("native-inputs requires --output")
args.output.write_text(json.dumps(native_inputs(), sort_keys=True) + "\n")
return 0
if args.mode == "native":
if args.expected is None or args.inputs is None or args.wheel is None:
parser.error("native requires --expected, --inputs and --wheel")
return native(args.expected, args.inputs, args.wheel, args.output)
if args.mode == "verify":
if args.tool is None or args.expected is None:
parser.error("verify requires --tool and --expected")
return verify(args.tool, args.expected, args.output)
selected = versions()
if args.github_output:
print("\n".join(f"{key}={value}" for key, value in selected.items()))
else:
print(json.dumps(selected, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())