اصلاح ورک فلو های استقرار

deployment/serial/serial_command.py

@@ -0,0 +1,291 @@
+"""Check an existing U-Boot prompt or configure its volatile environment."""
+import argparse
+import ipaddress
+import fcntl
+import re
+import sys
+import time
+from pathlib import Path
+
+import serial
+
+# Match the complete final line, never a prompt embedded in command output.
+PROMPT = re.compile(r"(?:^|\n)([^\r\n]*=>) ?$")
+ERROR = re.compile(
+    r"Unknown command|## Error|^Error:|^Usage:|^.*(?:not defined|not found|failed|failure).*$",
+    re.IGNORECASE | re.MULTILINE
+)
+
+
+class UBoot:
+    def __init__(self, port):
+        self.port = port
+        self.prompt = None
+
+    def read_prompt(self, timeout):
+        output = ""
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            data = self.port.read(self.port.in_waiting or 1)
+            if data:
+                chunk = data.decode("utf-8", errors="replace").replace("\r", "")
+                print(chunk, end="", flush=True)
+                output += chunk
+            match = PROMPT.search(output)
+            if match and (self.prompt is None or match[1] == self.prompt):
+                # Wait for a quiet prompt rather than a partial output line.
+                time.sleep(0.15)
+                if self.port.in_waiting:
+                    continue
+                return output, match[1]
+        raise RuntimeError("Timed out waiting for the CLI prompt")
+
+    def detect(self):
+        # Observe before requesting a fresh prompt.
+        try:
+            _, candidate = self.read_prompt(3)
+        except RuntimeError:
+            candidate = None
+        if candidate is None:
+            # An idle prompt may have been printed before the port was opened.
+            # A newline only requests a fresh prompt; no U-Boot command is sent.
+            self.port.write(b"\r")
+            self.port.flush()
+            _, candidate = self.read_prompt(5)
+        if not candidate.endswith("=>"):
+            raise RuntimeError("Response is not a recognized U-Boot CLI prompt")
+        self.prompt = candidate
+        print("\nU-Boot-style CLI prompt detected: " + candidate, flush=True)
+
+    def command(self, command, expected=None, timeout=10, allow_missing=()):
+        print("\n>>> " + command, flush=True)
+        self.port.write((command + "\r").encode())
+        self.port.flush()
+        output, _ = self.read_prompt(timeout)
+        # Query immediately, before another command changes the return status.
+        self.port.write(b"echo __UBOOT_STATUS_ $?\r")
+        self.port.flush()
+        status, _ = self.read_prompt(5)
+        codes = re.findall(r"(?m)^__UBOOT_STATUS_ (\d+) *$", status)
+        # Only the initial environment inspection may report unset variables.
+        missing_pattern = r'^## Error: "(' + "|".join(map(re.escape, allow_missing)) + r')" not defined$'
+        missing = re.findall(missing_pattern, output, re.MULTILINE) if allow_missing else []
+        checked_output = re.sub(missing_pattern, "", output, flags=re.MULTILINE) if missing else output
+        if codes != ["0"] and not (codes == ["1"] and missing):
+            raise RuntimeError("Command failed or its exit status could not be verified: " + command)
+        if ERROR.search(checked_output):
+            raise RuntimeError("Error in command response: " + command)
+        if expected and not re.search(expected, output, re.MULTILINE):
+            raise RuntimeError("Expected response missing: " + command)
+        if missing:
+            print("\nInitially unset: " + ", ".join(missing), flush=True)
+        print("\nVerified: " + command, flush=True)
+        return output
+
+
+LAYOUTS = {
+    16: {"uboot": (0, 0x100000), "dtb": (0x100000, 0x10000),
+         "kernel": (0x110000, 0x500000), "rootfs": (0x610000, 0x900000)},
+    32: {"uboot": (0, 0x200000), "dtb": (0x200000, 0x110000),
+         "kernel": (0x310000, 0x500000), "rootfs": (0x810000, 0x17f0000)},
+}
+
+
+def detect_flash(cli):
+    output = cli.command("sf probe 0:0", r"^SF: Detected .+", timeout=20)
+    capacities = re.findall(r"^SF: Detected .*\btotal (\d+) MiB\s*$", output, re.MULTILINE)
+    if len(capacities) != 1 or int(capacities[0]) not in LAYOUTS:
+        raise RuntimeError("Unsupported or ambiguous flash capacity; expected 16 or 32 MiB")
+    capacity = int(capacities[0])
+    print(f"\nDetected {capacity} MiB ({capacity * 8} Mbit); selecting matching flash layout.")
+    return capacity
+
+
+def transfer(cli, firmware, *, dtb=False, kernel=False, rootfs=False, rootfs_size=None):
+    capacity = detect_flash(cli)
+    part = "rootfs" if rootfs else "kernel" if kernel else "dtb" if dtb else "uboot"
+    offset, limit = LAYOUTS[capacity][part]
+    address = 0x41000000 if (dtb or rootfs) else 0x42000000
+    dump_size = 64 if (kernel or rootfs) else 32
+    filename = {"uboot": "u-boot-sunxi-with-spl.bin", "dtb": "sun8i-v3s-licheepi-zero-dock.dtb",
+                "kernel": "zImage", "rootfs": "rootfs.squashfs"}[part]
+    if rootfs_size is not None and rootfs:
+        if not 64 <= rootfs_size <= limit:
+            raise RuntimeError("RootFS size override exceeds the selected partition or is invalid")
+        limit = rootfs_size
+    print(f"Selected {part}: flash offset {offset:#x}, maximum size {limit:#x}")
+    image = firmware.read_bytes()
+    if not dump_size <= len(image) <= limit:
+        raise RuntimeError(f"Firmware size must be between {dump_size} and {limit:#x} bytes")
+    if rootfs and len(image) > 0x1800000:
+        raise RuntimeError("RootFS exceeds the non-overlapping RAM buffer capacity (0x1800000)")
+    # Invalidate any previous transfer length before attempting a new transfer.
+    cli.command("setenv filesize 0")
+    cli.command("printenv filesize", r"^filesize=0$")
+    output = cli.command(
+        f"tftp {address:#x} {filename}",
+        r"^Bytes transferred = \d+", timeout=120,
+    )
+    transferred = re.findall(r"^Bytes transferred = (\d+)", output, re.MULTILINE)
+    if transferred != [str(len(image))]:
+        raise RuntimeError("TFTP byte count differs from the local firmware file")
+    output = cli.command("echo ${filesize}", r"^[0-9a-fA-F]+$")
+    sizes = re.findall(r"^([0-9a-fA-F]+)$", output, re.MULTILINE)
+    if len(sizes) != 1 or int(sizes[0], 16) != len(image):
+        raise RuntimeError("U-Boot filesize differs from the local firmware file")
+    output = cli.command(f"md.b {address:#x} {dump_size:#x}", rf"^{address:08x}:")
+    dumped = bytearray()
+    for row_offset in range(0, dump_size, 16):
+        row_address = f"{address + row_offset:08x}"
+        rows = re.findall(r"^" + row_address + r":((?: [0-9a-fA-F]{2}){16})(?:\s|$)", output, re.MULTILINE)
+        if len(rows) != 1:
+            raise RuntimeError("Missing or invalid memory dump at " + row_address)
+        dumped.extend(bytes.fromhex(rows[0]))
+    if dumped != image[:dump_size]:
+        raise RuntimeError(f"First {dump_size} bytes in RAM differ from the local firmware file")
+    print(f"\nTransfer verified: size and first {dump_size} bytes match.")
+    cli.command(
+        f"sf write {address:#x} {offset:#x} ${{filesize}}",
+        rf"^SF: {len(image)} bytes @ {offset:#x} Written: OK\s*$", timeout=120,
+    )
+    cli.command(
+        f"sf read 0x42800000 {offset:#x} ${{filesize}}",
+        rf"^SF: {len(image)} bytes @ {offset:#x} Read: OK\s*$", timeout=120,
+    )
+    output = cli.command(
+        f"cmp.b {address:#x} 0x42800000 ${{filesize}}",
+        r"^Total of \d+ byte\(s\) were the same", timeout=60,
+    )
+    compared = re.findall(r"^Total of (\d+) byte\(s\) were the same", output, re.MULTILINE)
+    if compared != [str(len(image))]:
+        raise RuntimeError("Flash comparison byte count differs from firmware size")
+    print("\nFirmware flash write and full read-back comparison verified.")
+
+
+def erase_flash(cli):
+    capacity = detect_flash(cli)
+    length = capacity * 1024 * 1024
+    cli.command(
+        f"sf erase 0x0 {length:#x}",
+        rf"^SF: {length} bytes @ 0x0 Erased: OK\s*$", timeout=600,
+    )
+    cli.command(
+        "sf read 0x42800000 0x0 0x100",
+        r"^SF: 256 bytes @ 0x0 Read: OK\s*$", timeout=20,
+    )
+    output = cli.command("md.b 0x42800000 0x100", r"^42800000:")
+    for offset in range(0, 0x100, 16):
+        address = f"{0x42800000 + offset:08x}"
+        rows = re.findall(
+            r"^" + address + r":((?: [0-9a-fA-F]{2}){16})(?:\s|$)",
+            output, re.MULTILINE,
+        )
+        if len(rows) != 1 or bytes.fromhex(rows[0]) != b"\xff" * 16:
+            raise RuntimeError("Erase verification failed at " + address)
+    print("\nFull-flash erase command succeeded; first 256 bytes verified as FF.")
+
+
+def set_bootcmd(cli):
+    capacity = detect_flash(cli)
+    dtb_offset, dtb_size = LAYOUTS[capacity]["dtb"]
+    kernel_offset, kernel_size = LAYOUTS[capacity]["kernel"]
+    bootcmd = (
+        f"sf probe 0; sf read 0x41800000 {dtb_offset:#x} {dtb_size:#x}; "
+        f"sf read 0x42000000 {kernel_offset:#x} {kernel_size:#x}; "
+        "bootz 0x42000000 - 0x41800000"
+    )
+    cli.command("setenv bootcmd '" + bootcmd + "'")
+    cli.command("printenv bootcmd", r"^bootcmd=" + re.escape(bootcmd) + r"$")
+    print("\nDynamic bootcmd verified in RAM; environment not saved and bootcmd not executed.")
+
+
+def boot_config(cli, confirmed=False):
+    cli.command(
+        "printenv bootcmd bootargs", allow_missing=("bootcmd", "bootargs"),
+    )
+    if not confirmed:
+        print("\nBoot configuration inspected; no changes requested.")
+        return
+    bootargs = "console=ttyS0,115200 loglevel=8 root=/dev/mtdblock3 rootfstype=squashfs ro rootwait"
+    cli.command("setenv bootargs '" + bootargs + "'")
+    cli.command("printenv bootargs", r"^bootargs=" + re.escape(bootargs) + r"$")
+    print("\nBootargs verified in RAM; environment not saved and board not rebooted.")
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("mode", choices=("check", "configure", "transfer", "transfer-dtb", "transfer-kernel", "transfer-rootfs", "erase", "boot-config", "preflight", "set-bootcmd", "saveenv", "reboot"), default="check", nargs="?")
+    parser.add_argument("--port", default="/dev/ttyUSB0")
+    parser.add_argument("--firmware", type=Path, default=None)
+    parser.add_argument("--rootfs-size", type=lambda value: int(value, 0), help="Actual RootFS partition size in bytes (decimal or 0x-prefixed hex)")
+    parser.add_argument("--confirm-squashfs-mtdblock3", action="store_true", help="Confirm actual RootFS is SquashFS on mtdblock3 and set bootargs")
+    parser.add_argument("--device-ip", type=ipaddress.IPv4Address, default="192.168.75.100")
+    parser.add_argument("--tftp-server-ip", type=ipaddress.IPv4Address, default="192.168.75.247")
+    args = parser.parse_args()
+    if args.firmware is None:
+        filename = "rootfs.squashfs" if args.mode == "transfer-rootfs" else "zImage" if args.mode == "transfer-kernel" else ("sun8i-v3s-licheepi-zero-dock.dtb" if args.mode == "transfer-dtb" else "u-boot-sunxi-with-spl.bin")
+        args.firmware = Path(__file__).resolve().parents[2] / "firmware" / filename
+    if args.mode in ("transfer", "transfer-dtb", "transfer-kernel", "transfer-rootfs") and not args.firmware.is_file():
+        raise RuntimeError("Firmware file not found: " + str(args.firmware))
+    # Shared by both workflows: fail rather than interleave serial commands.
+    with open("/tmp/uboot-serial.lock", "w") as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        with serial.Serial(args.port, 115200, timeout=0.1, write_timeout=2, exclusive=True) as port:
+            cli = UBoot(port)
+            cli.detect()
+            if args.mode == "check":
+                return
+            if args.mode == "saveenv":
+                cli.command("saveenv", timeout=120)
+                print("Current U-Boot environment saved successfully.", flush=True)
+                return
+            if args.mode == "reboot":
+                print("\n>>> reset", flush=True)
+                port.write(b"reset\r")
+                port.flush()
+                print("Reset command sent; reboot completion has not been verified.", flush=True)
+                return
+            if args.mode == "preflight":
+                capacity = detect_flash(cli)
+                files = {"uboot": "u-boot-sunxi-with-spl.bin", "dtb": "sun8i-v3s-licheepi-zero-dock.dtb",
+                         "kernel": "zImage", "rootfs": "rootfs.squashfs"}
+                for part, filename in files.items():
+                    path = Path(__file__).resolve().parents[2] / "firmware" / filename
+                    minimum = 64 if part in ("kernel", "rootfs") else 32
+                    limit = LAYOUTS[capacity][part][1]
+                    if not path.is_file() or not minimum <= path.stat().st_size <= limit:
+                        raise RuntimeError(f"Firmware missing or outside selected partition size: {path} (maximum {limit:#x})")
+                print("All firmware sizes fit the detected flash layout.")
+                return
+            if args.mode == "erase":
+                erase_flash(cli)
+                return
+            if args.mode == "boot-config":
+                boot_config(cli, confirmed=args.confirm_squashfs_mtdblock3)
+                return
+            if args.mode == "set-bootcmd":
+                set_bootcmd(cli)
+                return
+            if args.mode in ("transfer", "transfer-dtb", "transfer-kernel", "transfer-rootfs"):
+                transfer(cli, args.firmware, dtb=args.mode == "transfer-dtb", kernel=args.mode == "transfer-kernel", rootfs=args.mode == "transfer-rootfs", rootfs_size=args.rootfs_size)
+                return
+            cli.command("version", r"^U-Boot \d")
+            cli.command(
+                "printenv ipaddr serverip ethaddr bootcmd bootargs",
+                allow_missing=("ipaddr", "serverip", "ethaddr", "bootcmd", "bootargs"),
+            )
+            cli.command(f"setenv ipaddr {args.device_ip}")
+            cli.command("printenv ipaddr", r"^ipaddr=" + re.escape(str(args.device_ip)) + r"$")
+            cli.command(f"setenv serverip {args.tftp_server_ip}")
+            cli.command("printenv serverip", r"^serverip=" + re.escape(str(args.tftp_server_ip)) + r"$")
+            cli.command(f"ping {args.tftp_server_ip}", r"^host " + re.escape(str(args.tftp_server_ip)) + r" is alive$", timeout=30)
+            cli.command("sf probe 0:0", r"^SF: Detected .+", timeout=20)
+            print("\nAll commands verified; environment changes were not saved to flash.")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except (RuntimeError, OSError, serial.SerialException) as exc:
+        print("\nERROR: " + str(exc), file=sys.stderr)
+        sys.exit(1)

deployment/serial/software_test.py

@@ -0,0 +1,55 @@
+"""Wait up to three minutes for the flashed device to answer ICMP."""
+import argparse
+import ipaddress
+import json
+import subprocess
+import sys
+
+
+def test_device(interface, host_ip, device_ip):
+    host = str(ipaddress.IPv4Address(host_ip))
+    device = str(ipaddress.IPv4Address(device_ip))
+    if host == device:
+        raise ValueError('Device IP must differ from the test host IP')
+    if not interface or interface.startswith('-'):
+        raise ValueError('Invalid network interface')
+    result = subprocess.run(
+        ['ip', '-j', '-4', 'addr', 'show', 'dev', interface],
+        check=True, capture_output=True, text=True, timeout=5,
+    )
+    addresses = json.loads(result.stdout)
+    if not any(info.get('local') == host for entry in addresses for info in entry.get('addr_info', [])):
+        raise ValueError(f'Host IP {host} is not configured on {interface}')
+    result = subprocess.run(
+        ['ip', '-j', '-4', 'route', 'get', device, 'from', host],
+        check=True, capture_output=True, text=True, timeout=5,
+    )
+    routes = json.loads(result.stdout)
+    if not routes or any(route.get('dev') != interface for route in routes):
+        raise ValueError(f'Route to {device} does not use {interface}')
+    print(f'Waiting up to 180 seconds: {host} ({interface}) -> {device}', flush=True)
+    # -c 1 exits on the first reply; -w bounds the entire ping run.
+    result = subprocess.run(
+        ['ping', '-4', '-n', '-B', '-I', host, '-c', '1', '-i', '1', '-W', '1', '-w', '180', device],
+        timeout=180,
+    )
+    if result.returncode:
+        raise RuntimeError(f'FAIL: {device} did not pass the ping test (exit {result.returncode})')
+    print(f'PASS: device {device} answered ping', flush=True)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--interface', default='eno1')
+    parser.add_argument('--host-ip', default='192.168.75.15')
+    parser.add_argument('--device-ip', required=True)
+    args = parser.parse_args()
+    test_device(args.interface, args.host_ip, args.device_ip)
+
+
+if __name__ == '__main__':
+    try:
+        main()
+    except (ValueError, RuntimeError, OSError, subprocess.SubprocessError) as exc:
+        print(f'FAIL: {exc}', file=sys.stderr)
+        sys.exit(1)