Добавить публикацию каталога прошивок

python/setprotocol/firmware_publish.py

@@ -0,0 +1,145 @@
+"""Reusable helpers for publishing the shared firmware release catalog."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from dataclasses import dataclass
+from pathlib import Path
+
+from .firmware_catalog import (
+    MAX_MANIFEST_BYTES,
+    SUPPORTED_TRANSPORTS,
+    parse_firmware_catalog,
+)
+
+MAX_FIRMWARE_BYTES = 128 * 1024 * 1024
+
+
+@dataclass(frozen=True)
+class FirmwarePublication:
+    """Metadata required to publish one firmware image."""
+
+    path: Path
+    product: str
+    version_name: str
+    version_code: int
+    transport: str
+    base_address: int | None = None
+    notes: str = ""
+
+    def validate(self) -> None:
+        if not self.path.is_file():
+            raise ValueError(f"Firmware file is missing: {self.path}")
+        if self.path.suffix.lower() not in {".bin", ".hex"}:
+            raise ValueError("Firmware file must have a .bin or .hex extension")
+        size = self.path.stat().st_size
+        if size <= 0:
+            raise ValueError("Firmware file is empty")
+        if size > MAX_FIRMWARE_BYTES:
+            raise ValueError("Firmware file exceeds the maximum size")
+        if not self.product.strip():
+            raise ValueError("Firmware product is empty")
+        if not self.version_name.strip():
+            raise ValueError("Firmware version name is empty")
+        if not 0 <= self.version_code <= 0x7FFFFFFF:
+            raise ValueError(
+                "Firmware version code must be between 0 and 2147483647"
+            )
+        if self.transport not in SUPPORTED_TRANSPORTS:
+            raise ValueError("Unsupported firmware transport")
+        if (
+            self.base_address is not None
+            and not 0 <= self.base_address <= 0xFFFFFFFF
+        ):
+            raise ValueError("Firmware base address is outside the uint32 range")
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def safe_release_tag_part(value: str) -> str:
+    source = value.strip()
+    part = re.sub(r"[^A-Za-z0-9._-]+", "-", source).strip("-.")
+    digest = hashlib.sha256(source.encode("utf-8")).hexdigest()[:8]
+    if not part:
+        return digest
+    return part if part == source else f"{part}-{digest}"
+
+
+def firmware_release_tag(publication: FirmwarePublication) -> str:
+    return "firmware-%s-v%s" % (
+        safe_release_tag_part(publication.product),
+        safe_release_tag_part(publication.version_name),
+    )
+
+
+def firmware_release_entry(
+    publication: FirmwarePublication, image_url: str, sha256: str
+) -> dict:
+    result = {
+        "product": publication.product.strip(),
+        "versionCode": publication.version_code,
+        "versionName": publication.version_name.strip(),
+        "imageUrl": image_url,
+        "fileName": publication.path.name,
+        "sha256": sha256,
+        "transport": publication.transport,
+        "notes": publication.notes.strip(),
+    }
+    if publication.base_address is not None:
+        result["baseAddress"] = f"0x{publication.base_address:08X}"
+    return result
+
+
+def firmware_entry_identity(entry: dict) -> tuple[str, int, str]:
+    return (
+        str(entry.get("product", entry.get("device", ""))).strip().casefold(),
+        int(entry.get("versionCode", 0)),
+        str(entry.get("transport", "rs485")).strip().lower(),
+    )
+
+
+def update_firmware_manifest(manifest: dict, entry: dict) -> dict:
+    """Insert or replace one release without disturbing other manifest data."""
+    result = dict(manifest)
+    existing = manifest.get("firmware")
+    firmware = dict(existing) if isinstance(existing, dict) else {}
+    rows = firmware.get("releases") if isinstance(existing, dict) else existing
+    releases = (
+        [dict(row) for row in rows if isinstance(row, dict)]
+        if isinstance(rows, list)
+        else []
+    )
+    identity = firmware_entry_identity(entry)
+    releases = [
+        row for row in releases if firmware_entry_identity(row) != identity
+    ]
+    releases.append(dict(entry))
+    releases.sort(
+        key=lambda row: (
+            str(row.get("product", row.get("device", ""))).casefold(),
+            -int(row.get("versionCode", 0)),
+            str(row.get("transport", "rs485")),
+        )
+    )
+    previous_rows = firmware.get("releases")
+    changed = releases != previous_rows
+    firmware["catalogVersion"] = (
+        int(firmware.get("catalogVersion", 0)) + int(changed)
+    )
+    firmware["releases"] = releases
+    result["firmware"] = firmware
+
+    encoded = (json.dumps(result, ensure_ascii=False) + "\n").encode("utf-8")
+    if len(encoded) > MAX_MANIFEST_BYTES:
+        raise ValueError("Updated update.json exceeds the maximum size")
+    parse_firmware_catalog(encoded, "https://catalog.invalid/update.json")
+    return result
+

python/tests/test_firmware_publish.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from setprotocol.firmware_publish import (
+    FirmwarePublication,
+    firmware_release_entry,
+    firmware_release_tag,
+    update_firmware_manifest,
+)
+
+
+class FirmwarePublishTests(unittest.TestCase):
+    def publication(self, path: Path, **overrides) -> FirmwarePublication:
+        fields = {
+            "path": path,
+            "product": "F103DS18",
+            "version_name": "1.1.0",
+            "version_code": 0x00010100,
+            "transport": "can",
+            "base_address": 0x08003000,
+            "notes": "Verified release",
+        }
+        fields.update(overrides)
+        return FirmwarePublication(**fields)
+
+    def test_publication_validates_file_and_metadata(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            image = Path(temporary) / "image.hex"
+            image.write_text(":00000001FF\n", encoding="ascii")
+            self.publication(image).validate()
+            with self.assertRaisesRegex(ValueError, "Unsupported"):
+                self.publication(image, transport="unknown").validate()
+
+    def test_entry_and_tag_are_deterministic(self) -> None:
+        publication = self.publication(Path("image.hex"))
+        entry = firmware_release_entry(
+            publication, "https://example.test/image.hex", "ab" * 32
+        )
+        self.assertEqual(
+            firmware_release_tag(publication), "firmware-F103DS18-v1.1.0"
+        )
+        self.assertEqual(entry["baseAddress"], "0x08003000")
+
+    def test_update_preserves_sections_and_replaces_same_release(self) -> None:
+        first = {
+            "product": "Device",
+            "versionCode": 7,
+            "versionName": "1.2.3",
+            "imageUrl": "https://example.test/old.bin",
+            "fileName": "old.bin",
+            "sha256": "11" * 32,
+            "transport": "rs485",
+        }
+        manifest = update_firmware_manifest(
+            {"windows": {"versionCode": 8}}, first
+        )
+        replacement = {
+            **first,
+            "imageUrl": "https://example.test/new.bin",
+            "fileName": "new.bin",
+            "sha256": "22" * 32,
+        }
+        updated = update_firmware_manifest(manifest, replacement)
+        self.assertEqual(updated["windows"], {"versionCode": 8})
+        self.assertEqual(len(updated["firmware"]["releases"]), 1)
+        self.assertEqual(
+            updated["firmware"]["releases"][0]["sha256"], "22" * 32
+        )
+        self.assertEqual(updated["firmware"]["catalogVersion"], 2)
+
+    def test_legacy_array_is_migrated_without_data_loss(self) -> None:
+        legacy = {
+            "product": "Legacy",
+            "versionCode": 1,
+            "versionName": "1.0",
+            "imageUrl": "https://example.test/legacy.bin",
+            "fileName": "legacy.bin",
+            "sha256": "33" * 32,
+            "transport": "rs485",
+        }
+        current = {**legacy, "product": "Current", "versionCode": 2}
+        updated = update_firmware_manifest({"firmware": [legacy]}, current)
+        self.assertCountEqual(
+            [row["product"] for row in updated["firmware"]["releases"]],
+            ["Legacy", "Current"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()