vee
113 lines
4.7 KiB
Python
113 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import threading
|
|
import urllib.request
|
|
from http.server import ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
from cognitive_agent.firmware_ota_server import FirmwareOtaHandler, FirmwareOtaRepository
|
|
|
|
|
|
class TestFirmwareOtaServer:
|
|
def test_latest_download_head_and_resume_are_read_only(self, tmp_path: Path) -> None:
|
|
release_id = "fw-rel-123456789abc"
|
|
filename = "PassengerSeat_test.zip"
|
|
package = (b"vehicle-firmware-package" * 128) + b"end"
|
|
release_root = tmp_path / "releases" / release_id
|
|
release_root.mkdir(parents=True)
|
|
package_path = release_root / filename
|
|
package_path.write_bytes(package)
|
|
manifest = {
|
|
"schema": "n7-ota-release/v1",
|
|
"release_id": release_id,
|
|
"firmware_filename": filename,
|
|
"size_bytes": len(package),
|
|
"sha256": hashlib.sha256(package).hexdigest(),
|
|
"capabilities": ["时间控制", "座椅加热"],
|
|
"published_at": "2026-09-16T13:17:17+08:00",
|
|
"status": "published",
|
|
"download_url": "http://stale-address.invalid/download",
|
|
}
|
|
(release_root / "release.json").write_text(
|
|
json.dumps(manifest, ensure_ascii=False), encoding="utf-8"
|
|
)
|
|
(tmp_path / "latest.json").write_text(
|
|
json.dumps(manifest, ensure_ascii=False), encoding="utf-8"
|
|
)
|
|
older_release_id = "fw-rel-000000000001"
|
|
older_root = tmp_path / "releases" / older_release_id
|
|
older_root.mkdir(parents=True)
|
|
older_filename = "PassengerSeat_older.zip"
|
|
older_package = b"older-firmware-package"
|
|
(older_root / older_filename).write_bytes(older_package)
|
|
older_manifest = {
|
|
**manifest,
|
|
"release_id": older_release_id,
|
|
"firmware_filename": older_filename,
|
|
"size_bytes": len(older_package),
|
|
"sha256": hashlib.sha256(older_package).hexdigest(),
|
|
"published_at": "2026-09-15T13:17:17+08:00",
|
|
}
|
|
(older_root / "release.json").write_text(
|
|
json.dumps(older_manifest, ensure_ascii=False), encoding="utf-8"
|
|
)
|
|
|
|
class Handler(FirmwareOtaHandler):
|
|
repository = FirmwareOtaRepository(
|
|
tmp_path, "https://n7-ota.example.ts.net/api/firmware-workbench/releases"
|
|
)
|
|
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
base_url = f"http://127.0.0.1:{server.server_port}"
|
|
try:
|
|
with urllib.request.urlopen(
|
|
f"{base_url}/api/firmware-workbench/releases/latest", timeout=3
|
|
) as response:
|
|
latest = json.loads(response.read().decode("utf-8"))
|
|
assert latest["download_url"] == (
|
|
f"https://n7-ota.example.ts.net/api/firmware-workbench/releases/"
|
|
f"{release_id}/download"
|
|
)
|
|
|
|
with urllib.request.urlopen(
|
|
f"{base_url}/api/firmware-workbench/releases", timeout=3
|
|
) as response:
|
|
releases = json.loads(response.read().decode("utf-8"))
|
|
assert releases["schema"] == "n7-ota-release-list/v1"
|
|
assert releases["count"] == 2
|
|
assert releases["latest_release_id"] == release_id
|
|
assert [item["release_id"] for item in releases["releases"]] == [
|
|
release_id,
|
|
older_release_id,
|
|
]
|
|
assert releases["releases"][1]["download_url"].endswith(
|
|
f"/{older_release_id}/download"
|
|
)
|
|
|
|
request = urllib.request.Request(
|
|
f"{base_url}/api/firmware-workbench/releases/{release_id}/download",
|
|
headers={"Range": "bytes=8-31"},
|
|
)
|
|
with urllib.request.urlopen(request, timeout=3) as response:
|
|
assert response.status == 206
|
|
assert response.headers["Content-Range"] == f"bytes 8-31/{len(package)}"
|
|
assert response.headers["X-Firmware-SHA256"] == manifest["sha256"]
|
|
assert response.read() == package[8:32]
|
|
|
|
head = urllib.request.Request(
|
|
f"{base_url}/api/firmware-workbench/releases/{release_id}/download",
|
|
method="HEAD",
|
|
)
|
|
with urllib.request.urlopen(head, timeout=3) as response:
|
|
assert response.status == 200
|
|
assert int(response.headers["Content-Length"]) == len(package)
|
|
assert response.read() == b""
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=3)
|