lmxopcua/scripts/validation/opcua-lifecycle-memory/windows/prepare.py

#!/usr/bin/env python3
"""Publish and record a fresh immutable Windows x64 bundle of the canonical soak (lmxopcua#799) without running it.

The Windows counterpart of ../late-slope/prepare.py: same committed-source checks, same self-contained .NET 10.0.14
runtime, same exact SDK 2.0.1 / HistorianGateway 1.0.0 graph, same per-file SHA-256 manifest and managed MVIDs. Only
the RID (win-x64), the executable format check (PE32+ AMD64) and the absence of a Docker image differ.
"""

import argparse
from datetime import datetime, timezone
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile

sys.dont_write_bytecode = True
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE.parent / "late-slope"))

from analyze import APP, RUNTIME, digest, read_object, require  # noqa: E402
from bundle import CANONICAL_SOURCES, GATEWAY_PACKAGES, SDK_PACKAGES, check_modules, manifest  # noqa: E402
import prepare as linux_prepare  # noqa: E402

REPO = HERE.parents[3]
PROJECT = REPO / "tests/Server" / APP / (APP + ".csproj")
RID = "win-x64"
TOOLS = ("launch.ps1", "analyze.py", "README.md")
WINDOWS_SOURCES = (f"tests/Server/{APP}/Sdk2LifecycleWindowsDescriptors.cs", f"tests/Server/{APP}/ServerEventRig.cs")


def check_pe_amd64(path):
    data = path.read_bytes()[:4096]
    require(data[:2] == b"MZ", f"{path.name} is not a PE image")
    offset = int.from_bytes(data[0x3C:0x40], "little")
    require(data[offset:offset + 4] == b"PE\0\0" and data[offset + 4:offset + 6] == b"\x64\x86", f"{path.name} is not AMD64")


def check_application(app):
    check_pe_amd64(app / f"{APP}.exe")
    check_pe_amd64(app / "coreclr.dll")
    options = read_object(app / f"{APP}.runtimeconfig.json")["runtimeOptions"]
    frameworks = options["includedFrameworks"]
    require(options["tfm"] == "net10.0" and "framework" not in options and "frameworks" not in options
            and any(value["name"] == "Microsoft.NETCore.App" for value in frameworks)
            and all(value["version"] == RUNTIME for value in frameworks), "Wrong self-contained application runtime")
    dependencies = read_object(app / f"{APP}.deps.json")
    require(dependencies["runtimeTarget"]["name"] == f".NETCoreApp,Version=v10.0/{RID}", "Wrong application RID")
    libraries = dependencies["libraries"]
    sdk = {key for key in libraries if key.startswith("OPCFoundation.")}
    gateway = {key for key in libraries if key.startswith("ZB.MOM.WW.HistorianGateway.")}
    require(sdk == SDK_PACKAGES and gateway == GATEWAY_PACKAGES, "Expected exact SDK 2.0.1 and HistorianGateway 1.0.0 graph")
    return sorted(sdk), sorted(gateway)


def committed_sources(revision):
    sources = linux_prepare.check_source(revision)
    for name in (*WINDOWS_SOURCES, *(f"scripts/validation/opcua-lifecycle-memory/windows/{tool}" for tool in TOOLS),
                 "scripts/validation/opcua-lifecycle-memory/windows/prepare.py"):
        sources[name] = digest(REPO / name)
    return sources


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--source-revision", required=True)
    options = parser.parse_args()
    sources = committed_sources(options.source_revision)
    root = options.output.resolve()
    root.mkdir(mode=0o700, parents=True, exist_ok=False)
    with tempfile.TemporaryDirectory(prefix="opcua-windows-soak-publish-") as scratch:
        build = Path(scratch)
        command = ["dotnet", "publish", str(PROJECT), "-c", "Release", "-r", RID,
                   "--self-contained", "true", f"-p:RuntimeFrameworkVersion={RUNTIME}",
                   "-p:PublishSingleFile=false", "-p:PublishTrimmed=false", "-p:UseAppHost=true",
                   "--artifacts-path", str(build / "artifacts"), "-o", str(root / "app"), "-m:2", "--nologo"]
        with (root / "publish.log").open("w", encoding="utf-8") as log:
            subprocess.run(command, cwd=REPO, stdout=log, stderr=subprocess.STDOUT, check=True, timeout=900)
        sdk, gateway = check_application(root / "app")
        modules = linux_prepare.managed_modules(build, root)
    tools = root / "tools"
    tools.mkdir()
    for name in TOOLS:
        shutil.copy2(HERE / name, tools / name)
    require(committed_sources(options.source_revision) == sources, "Committed inputs changed during publication")
    files = manifest(root)
    check_modules(modules, files)
    receipt = {"kind": "windows-bundle", "protocol": "sdk2-late-slope-v1", "platform": "windows", "rid": RID,
               "runtimeVersion": RUNTIME, "selfContained": True, "mode": "late-slope", "requestedSeconds": 7200,
               "limits": {"fdGrowth": 32, "retainedManagedGrowthBytes": 64 << 20, "rssGrowthBytes": 128 << 20},
               "certificateLeakTracking": True, "sourceRevision": options.source_revision, "sourceFiles": sources,
               "preparedUtc": datetime.now(timezone.utc).isoformat(), "publishCommand": command,
               "sdkLibraries": sdk, "gatewayLibraries": gateway, "managedModules": modules,
               "coreClrSha256": files["app/coreclr.dll"], "files": files}
    path = root / "receipt.json"
    path.write_text(json.dumps(receipt, indent=2, allow_nan=False) + "\n", encoding="utf-8")
    # A flat manifest the Windows launcher re-verifies on the target before launch.
    (root / "manifest.sha256").write_text("".join(f"{sha}  {name}\n" for name, sha in files.items()), encoding="utf-8")
    print(json.dumps({"bundle": str(root), "sourceRevision": options.source_revision, "receiptSha256": digest(path),
                      "files": len(files), "workloadLaunched": False}, indent=2))


if __name__ == "__main__":
    main()