Miayui
199 lines
8.2 KiB
Python
199 lines
8.2 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import urllib.request
|
|
import zipfile
|
|
from http.server import ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from cognitive_agent.application.firmware_workbench import FirmwareWorkbenchService
|
|
from cognitive_agent.auth import AuthConfig, AuthRuntime
|
|
from cognitive_agent.server import SeatDemoHandler
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
class TestFirmwareWorkbenchApi:
|
|
@classmethod
|
|
def setup_class(cls) -> None:
|
|
cls.temporary_directory = tempfile.TemporaryDirectory()
|
|
root = Path(cls.temporary_directory.name)
|
|
image = root / "Seat_Controller_demo.hex"
|
|
image.write_bytes(b":020000040000FA\n" + b":100000000102030405060708090A0B0C0D0E0F1068\n" * 80)
|
|
|
|
def build_runner(task_id: str, log: object, cancel: threading.Event) -> Path:
|
|
log("info", f"IAR task {task_id}: 0 errors, 0 warnings")
|
|
return image
|
|
|
|
cls.workbench = FirmwareWorkbenchService(
|
|
ROOT,
|
|
artifact_root=root / "workbench",
|
|
ota_root=root / "ota",
|
|
build_runner=build_runner,
|
|
)
|
|
cls.original_workbench = SeatDemoHandler.firmware_workbench
|
|
cls.original_auth = SeatDemoHandler.auth
|
|
SeatDemoHandler.firmware_workbench = cls.workbench
|
|
SeatDemoHandler.auth = AuthRuntime(
|
|
AuthConfig(
|
|
enabled=False,
|
|
username="",
|
|
password_hash="",
|
|
session_secret=b"t" * 64,
|
|
session_ttl_seconds=3600,
|
|
login_attempts=5,
|
|
login_window_seconds=300,
|
|
requests_per_minute=120,
|
|
trust_proxy_headers=False,
|
|
)
|
|
)
|
|
cls.server = ThreadingHTTPServer(("127.0.0.1", 0), SeatDemoHandler)
|
|
cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
|
|
cls.thread.start()
|
|
cls.base_url = f"http://127.0.0.1:{cls.server.server_port}"
|
|
|
|
@classmethod
|
|
def teardown_class(cls) -> None:
|
|
cls.server.shutdown()
|
|
cls.server.server_close()
|
|
cls.thread.join(timeout=2)
|
|
cls.workbench.close()
|
|
SeatDemoHandler.firmware_workbench = cls.original_workbench
|
|
SeatDemoHandler.auth = cls.original_auth
|
|
cls.temporary_directory.cleanup()
|
|
|
|
@classmethod
|
|
def post(cls, path: str, payload: dict[str, object]) -> tuple[int, dict[str, object]]:
|
|
request = urllib.request.Request(
|
|
f"{cls.base_url}{path}",
|
|
method="POST",
|
|
headers={"Content-Type": "application/json"},
|
|
data=json.dumps(payload, ensure_ascii=False).encode(),
|
|
)
|
|
with urllib.request.urlopen(request, timeout=5) as response:
|
|
return response.status, json.loads(response.read().decode())
|
|
|
|
@pytest.mark.parametrize("instruction,capabilities", [
|
|
("增加位置控制、座椅通风和座椅加热功能", ["时间控制", "位置控制", "座椅通风", "座椅加热"]),
|
|
("增加动态木马功能", ["时间控制", "位置控制", "动态木马"]),
|
|
("增加动态木马、座椅通风和座椅加热功能", ["时间控制", "位置控制", "座椅通风", "座椅加热", "动态木马"]),
|
|
])
|
|
def test_real_http_task_publish_detail_and_download_flow(
|
|
self, instruction: str, capabilities: list[str], monkeypatch
|
|
) -> None:
|
|
status, created = self.post(
|
|
"/api/firmware-workbench/tasks", {"instruction": instruction}
|
|
)
|
|
assert status == 202
|
|
task_id = str(created["task"]["id"])
|
|
|
|
deadline = time.monotonic() + 5
|
|
while time.monotonic() < deadline:
|
|
with urllib.request.urlopen(
|
|
f"{self.base_url}/api/firmware-workbench/tasks/{task_id}", timeout=5
|
|
) as response:
|
|
task = json.loads(response.read().decode())
|
|
if task["status"] not in {"waiting", "running", "pending_confirmation"}:
|
|
break
|
|
time.sleep(0.03)
|
|
assert task["status"] == "success"
|
|
assert task["message"] == "已发布，待车机端下载升级"
|
|
with urllib.request.urlopen(
|
|
f"{self.base_url}/api/firmware-workbench/releases/latest", timeout=5
|
|
) as response:
|
|
latest = json.loads(response.read().decode())
|
|
assert latest["release_id"] == task["release_id"]
|
|
assert latest["download_url"].endswith(f"/{task['release_id']}/download")
|
|
|
|
repeated_status, repeated = self.post(
|
|
"/api/firmware-workbench/tasks", {"instruction": instruction}
|
|
)
|
|
assert repeated_status == 202
|
|
assert repeated["task"]["id"] != task_id
|
|
repeated_task_id = str(repeated["task"]["id"])
|
|
deadline = time.monotonic() + 5
|
|
while time.monotonic() < deadline:
|
|
with urllib.request.urlopen(
|
|
f"{self.base_url}/api/firmware-workbench/tasks/{repeated_task_id}", timeout=5
|
|
) as response:
|
|
repeated_task = json.loads(response.read().decode())
|
|
if repeated_task["status"] not in {"waiting", "running", "pending_confirmation"}:
|
|
break
|
|
time.sleep(0.03)
|
|
assert repeated_task["status"] == "success"
|
|
|
|
artifact_id = task["artifact_id"]
|
|
with urllib.request.urlopen(
|
|
f"{self.base_url}/api/firmware-workbench/artifacts/{artifact_id}", timeout=5
|
|
) as response:
|
|
artifact = json.loads(response.read().decode())
|
|
assert artifact["status"] == "published"
|
|
assert artifact["sha256"]
|
|
|
|
release_id = task["release_id"]
|
|
with urllib.request.urlopen(
|
|
f"{self.base_url}/api/firmware-workbench/releases/{release_id}/download",
|
|
timeout=5,
|
|
) as response:
|
|
package = response.read()
|
|
assert response.headers["Content-Type"] == "application/zip"
|
|
assert "attachment" in response.headers["Content-Disposition"]
|
|
assert len(package) == artifact["size_bytes"]
|
|
assert artifact["capabilities"] == capabilities
|
|
with zipfile.ZipFile(io.BytesIO(package)) as archive:
|
|
assert json.loads(archive.read("manifest.json"))["capabilities"] == capabilities
|
|
|
|
frames = []
|
|
|
|
def publish(frame):
|
|
frames.append(frame)
|
|
return {"status": "published"}
|
|
|
|
monkeypatch.setattr(
|
|
SeatDemoHandler, "service", SimpleNamespace(publish_firmware_notification=publish)
|
|
)
|
|
notify_status, notified = self.post(
|
|
f"/api/firmware-workbench/artifacts/{artifact_id}/notify", {}
|
|
)
|
|
assert notify_status == 200
|
|
assert notified["published_count"] == 3
|
|
assert [frame["sequence"] for frame in frames] == [1, 2, 3]
|
|
assert all(frame["firmware"]["release_id"] == release_id for frame in frames)
|
|
|
|
delete_request = urllib.request.Request(
|
|
f"{self.base_url}/api/firmware-workbench/artifacts/{artifact_id}",
|
|
method="DELETE",
|
|
)
|
|
with urllib.request.urlopen(delete_request, timeout=5) as response:
|
|
deleted = json.loads(response.read().decode())
|
|
assert deleted["deleted"] is True
|
|
repeated_artifact_id = str(repeated_task["artifact_id"])
|
|
repeated_delete_request = urllib.request.Request(
|
|
f"{self.base_url}/api/firmware-workbench/artifacts/{repeated_artifact_id}",
|
|
method="DELETE",
|
|
)
|
|
with urllib.request.urlopen(repeated_delete_request, timeout=5) as response:
|
|
assert json.loads(response.read().decode())["deleted"] is True
|
|
with urllib.request.urlopen(
|
|
f"{self.base_url}/api/firmware-workbench/state", timeout=5
|
|
) as response:
|
|
state = json.loads(response.read().decode())
|
|
assert [item["id"] for item in state["artifacts"]] == [
|
|
"fw-art-time-only-baseline"
|
|
]
|
|
|
|
with urllib.request.urlopen(
|
|
f"{self.base_url}/firmware-workbench.html", timeout=5
|
|
) as response:
|
|
page = response.read().decode()
|
|
assert response.headers["X-Frame-Options"] == "SAMEORIGIN"
|
|
assert "frame-ancestors 'self'" in response.headers["Content-Security-Policy"]
|
|
assert "座椅固件开发工作台" in page
|