ITWorx_MCP_Hub-Public/scripts/generate_moby_v1_golden.py

"""Regenerate test-only V1 wire goldens with Go and source-pinned Moby structs.

No Hub normalizer code is imported. This installs nothing, builds no image and
changes only the fixed test fixture. Go 1.27.2 is a local regeneration tool.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import subprocess
import tempfile
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
FIXTURE = ROOT / "tests/fixtures/moby_v1_golden.json"
REFERENCE = ROOT / "tests/fixtures/moby_v1_reference/main.go"
PINS = (
    (
        "Old",
        "onbuild_null",
        "89c5e8fd66634b6128fc4c0e6f1236e2540e46e0",
        "image/image.go",
        "5ce9ea9e6daeb5cecf03dbfafe7c895516da782010e2797b1fc0f979161cf5be",
        "08df9dce6147eb2a4edb9859d81fedbdb13f2af2e249db38bf79361f67db4bcf",
    ),
    (
        "New",
        "onbuild_omitempty",
        "5834c866f83c0bae2d0be28f38b92f32cd07ca38",
        "daemon/internal/image/image.go",
        "c186304737e0343a04e0f9a4e88f8f81b867e86159cd7bf68dd0d35a8392d3a8",
        "06f5e71501399a33906ef7b006d49d71a700e6c84d7442a90e9d240cb52b81ee",
    ),
)


def source(commit: str, path: str, expected: str) -> str:
    url = f"https://raw.githubusercontent.com/moby/moby/{commit}/{path}"
    with urllib.request.build_opener(urllib.request.ProxyHandler({})).open(
        url, timeout=20
    ) as response:
        data = response.read(200_001)
    if (
        not isinstance(data, bytes)
        or len(data) > 200_000
        or hashlib.sha256(data).hexdigest() != expected
    ):
        raise ValueError("pinned reference source digest mismatch")
    return data.decode("utf-8")


def declaration(text: str, original: str, replacement: str) -> str:
    match = re.search(rf"(?ms)^type {original} struct \{{\n.*?^\}}", text)
    if not match:
        raise ValueError("pinned reference struct is missing")
    matched_source = match.group()
    if not isinstance(matched_source, str):
        raise ValueError("pinned reference declaration is not text")
    return matched_source.replace(f"type {original} struct", f"type {replacement} struct", 1)


def generate(go: Path) -> None:
    if not go.is_absolute() or not go.is_file():
        raise ValueError("an existing absolute Go executable is required")
    env = dict(os.environ)
    env.update(
        GOTOOLCHAIN="local",
        GOPROXY="off",
        GOSUMDB="off",
        GO111MODULE="off",
        GOMAXPROCS="2",
        GOFLAGS="-p=2",
        CGO_ENABLED="0",
    )
    version = subprocess.run(  # noqa: S603 - explicit existing absolute regeneration compiler
        [str(go), "version"], check=True, capture_output=True, text=True, env=env
    )
    if version.stdout.split()[2] != "go1.27.2":
        raise ValueError("reference regeneration requires Go 1.27.2")
    # These are public synthetic inputs: no runtime config, credentials or image
    # bytes are sent to the reference compiler or persisted in this fixture.
    inputs = [
        {
            "architecture": "amd64",
            "os": "linux",
            "created": "2026-10-14T00:01:02.123456789Z",
            "container_config": {
                "User": "constructor",
                "Cmd": ["µ<&>\u2028\u2029"],
                "Labels": {"z": "last", "a": "first"},
                "StopTimeout": 0,
            },
            "config": {
                "User": "10001:10001",
                "Cmd": ["reference"],
                "Labels": {"z": "µ<&>\u2028\u2029", "a": "first"},
                "StopTimeout": 0,
                "Healthcheck": {"Test": ["CMD", "true"], "Interval": 30000000000, "Retries": 2},
            },
            "rootfs": {"type": "layers", "diff_ids": ["sha256:" + "a" * 64, "sha256:" + "b" * 64]},
        },
        {
            "architecture": "amd64",
            "os": "linux",
            "config": {"User": "10001:10001"},
            "rootfs": {"type": "layers", "diff_ids": ["sha256:" + "c" * 64]},
        },
    ]
    declarations = [
        'package main\nimport "time"\n',
        """type HealthConfig struct {
Test []string `json:",omitempty"`
Interval time.Duration `json:",omitempty"`
Timeout time.Duration `json:",omitempty"`
StartPeriod time.Duration `json:",omitempty"`
StartInterval time.Duration `json:",omitempty"`
Retries int `json:",omitempty"`
}\n""",
    ]
    for label, _dialect, commit, image_path, image_hash, config_hash in PINS:
        config = declaration(
            source(commit, "api/types/container/config.go", config_hash), "Config", label + "Config"
        )
        config = (
            config.replace("strslice.StrSlice", "[]string")
            .replace("nat.PortSet", "map[string]struct{}")
            .replace("network.PortSet", "map[string]struct{}")
        )
        image = declaration(source(commit, image_path, image_hash), "V1Image", label + "V1")
        declarations.extend((config, image.replace("container.Config", label + "Config")))
    local = ROOT / ".luna"
    local.mkdir(exist_ok=True)
    if local.is_symlink() or not local.resolve().is_relative_to(ROOT.resolve()):
        raise ValueError("reference scratch directory must stay in this checkout")
    with tempfile.TemporaryDirectory(prefix="moby-v1-reference-", dir=local) as name:
        temp = Path(name)
        if not temp.resolve().is_relative_to(local.resolve()):
            raise ValueError("reference temporary directory escaped its owner")
        env["GOCACHE"] = str(temp / "cache")
        (temp / "main.go").write_bytes(REFERENCE.read_bytes())
        (temp / "types.go").write_text("\n".join(declarations), encoding="utf-8")
        (temp / "inputs.json").write_text(json.dumps(inputs, ensure_ascii=False), encoding="utf-8")
        completed = subprocess.run(  # noqa: S603 - fixed standard-library-only reference program
            [str(go), "run", ".", str(temp / "inputs.json")],
            cwd=temp,
            env=env,
            stdin=subprocess.DEVNULL,
            check=True,
            capture_output=True,
            timeout=180,
        )
        if len(completed.stdout) > 100_000:
            raise ValueError("reference output exceeded its fixed bound")
        cases = json.loads(completed.stdout)
    result = {
        "schema": "itworx.moby-v1-wire-golden/v1",
        "goVersion": "go1.27.2",
        "referenceMainSha256": hashlib.sha256(REFERENCE.read_bytes()).hexdigest(),
        "sourceCommits": [pin[2] for pin in PINS],
        "cases": cases,
    }
    FIXTURE.write_text(json.dumps(result, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    print(f"Generated {len(cases)} independent Go reference cases")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--go", type=Path, required=True)
    generate(parser.parse_args().go)