405 lines
14 KiB
Python
405 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi import APIRouter, Request
|
|
from fastapi.routing import APIRoute
|
|
from pydantic import ValidationError
|
|
|
|
from k1link.web.environment_api import (
|
|
EnvironmentBackground,
|
|
EnvironmentMediaDocument,
|
|
EnvironmentMediaItem,
|
|
EnvironmentPlaylistMediaDocument,
|
|
EnvironmentSettingsPut,
|
|
EnvironmentSettingsStore,
|
|
build_environment_router,
|
|
default_environment_settings,
|
|
)
|
|
|
|
|
|
def _endpoint(router: APIRouter, path: str, method: str) -> Callable[..., Any]:
|
|
for route in router.routes:
|
|
if (
|
|
isinstance(route, APIRoute)
|
|
and route.path == path
|
|
and route.methods is not None
|
|
and method in route.methods
|
|
):
|
|
return route.endpoint
|
|
raise AssertionError(f"{method} {path} route is missing")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("primary", "secondary", "expected"),
|
|
[
|
|
("spatial-scene", "cameras", ("cameras", None)),
|
|
("spatial-scene", None, ("cameras", None)),
|
|
("telemetry", "spatial-scene", ("telemetry", None)),
|
|
("objects", "world-map", ("objects", "world-map")),
|
|
],
|
|
)
|
|
def test_lab_scene_move_preserves_other_operator_settings(
|
|
tmp_path: Path, primary: str, secondary: str | None, expected: tuple[str, str | None]
|
|
) -> None:
|
|
store = EnvironmentSettingsStore(tmp_path)
|
|
original = default_environment_settings()
|
|
original.revision = 19
|
|
page = original.pages.observation
|
|
page.header_label = "Контроль"
|
|
page.primary_workspace_id = primary
|
|
page.secondary_workspace_id = secondary
|
|
store.settings_path.write_text(original.model_dump_json())
|
|
restored = store.read()
|
|
assert restored.revision == 19
|
|
assert restored.pages.observation.header_label == "Контроль"
|
|
assert restored.pages.observation.background == original.pages.observation.background
|
|
assert (restored.pages.observation.primary_workspace_id,
|
|
restored.pages.observation.secondary_workspace_id) == expected
|
|
assert restored.pages.home == original.pages.home
|
|
assert restored.pages.polygon == original.pages.polygon
|
|
|
|
|
|
def _streaming_request(payload: bytes, media_type: str) -> Request:
|
|
chunks = iter((payload[:8], payload[8:]))
|
|
|
|
async def receive() -> dict[str, Any]:
|
|
try:
|
|
chunk = next(chunks)
|
|
return {
|
|
"type": "http.request",
|
|
"body": chunk,
|
|
"more_body": True,
|
|
}
|
|
except StopIteration:
|
|
return {
|
|
"type": "http.request",
|
|
"body": b"",
|
|
"more_body": False,
|
|
}
|
|
|
|
return Request(
|
|
{
|
|
"type": "http",
|
|
"http_version": "1.1",
|
|
"method": "PUT",
|
|
"scheme": "http",
|
|
"path": "/api/v1/environment/media/home",
|
|
"raw_path": b"/api/v1/environment/media/home",
|
|
"query_string": b"",
|
|
"headers": [
|
|
(b"content-type", media_type.encode("ascii")),
|
|
(b"content-length", str(len(payload)).encode("ascii")),
|
|
],
|
|
"client": ("127.0.0.1", 1),
|
|
"server": ("127.0.0.1", 8000),
|
|
},
|
|
receive,
|
|
)
|
|
|
|
|
|
def test_environment_settings_are_versioned_and_persist_page_controls(tmp_path: Path) -> None:
|
|
store = EnvironmentSettingsStore(tmp_path / "operator-environment")
|
|
initial = store.read()
|
|
assert initial.revision == 0
|
|
assert initial.pages.polygon.header_label == "Тестировочный контур"
|
|
assert initial.pages.fleet.primary_workspace_id == "contour-health"
|
|
assert initial.pages.home.background.enabled is False
|
|
|
|
request = EnvironmentSettingsPut(
|
|
revision=initial.revision,
|
|
pages=initial.pages.model_copy(
|
|
update={
|
|
"observation": initial.pages.observation.model_copy(
|
|
update={
|
|
"header_label": "Контроль",
|
|
"title": "Контроль пространства",
|
|
}
|
|
)
|
|
}
|
|
),
|
|
)
|
|
saved = store.save(request)
|
|
restored = EnvironmentSettingsStore(store.root).read()
|
|
|
|
assert saved.revision == 1
|
|
assert restored == saved
|
|
assert restored.pages.observation.header_label == "Контроль"
|
|
assert restored.pages.observation.title == "Контроль пространства"
|
|
assert str(tmp_path) not in restored.model_dump_json()
|
|
|
|
|
|
def test_environment_settings_reject_stale_revision(tmp_path: Path) -> None:
|
|
store = EnvironmentSettingsStore(tmp_path / "operator-environment")
|
|
initial = store.read()
|
|
store.save(
|
|
EnvironmentSettingsPut(
|
|
revision=0,
|
|
pages=initial.pages,
|
|
)
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="revision changed"):
|
|
store.save(
|
|
EnvironmentSettingsPut(
|
|
revision=0,
|
|
pages=initial.pages,
|
|
)
|
|
)
|
|
|
|
|
|
def test_environment_background_rejects_untrusted_enabled_source() -> None:
|
|
with pytest.raises(ValidationError):
|
|
EnvironmentMediaItem(
|
|
id="media-invalid-url",
|
|
source="url",
|
|
url="file:///tmp/background.mp4",
|
|
media_kind="video",
|
|
)
|
|
|
|
with pytest.raises(ValidationError):
|
|
EnvironmentMediaItem(
|
|
id="media-invalid-file",
|
|
source="file",
|
|
url="/private/operator/background.mp4",
|
|
media_kind="video",
|
|
)
|
|
|
|
with pytest.raises(ValidationError):
|
|
EnvironmentBackground(enabled=True, items=[])
|
|
|
|
|
|
def test_environment_media_is_generation_bound_and_stored_outside_git(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
store = EnvironmentSettingsStore(tmp_path / "mission-data" / "ui-environment")
|
|
store.media_root.mkdir(parents=True)
|
|
payload = b"\x89PNG\r\n\x1a\nsynthetic-redacted"
|
|
temporary = store.media_root / ".home.upload"
|
|
temporary.write_bytes(payload)
|
|
digest = hashlib.sha256(payload).hexdigest()
|
|
|
|
document = store.finalize_media(
|
|
surface_id="home",
|
|
temporary_path=temporary,
|
|
file_name="mission-core.png",
|
|
media_type="image/png",
|
|
byte_length=len(payload),
|
|
sha256=digest,
|
|
)
|
|
path, restored = store.read_media("home", digest)
|
|
|
|
assert path == store.media_root / "home.png"
|
|
assert path.read_bytes() == payload
|
|
assert restored == document
|
|
assert document.url == f"/api/v1/environment/media/home?generation={digest}"
|
|
with pytest.raises(PermissionError):
|
|
store.read_media("home", "0" * 64)
|
|
|
|
|
|
def test_environment_media_upload_route_streams_and_publishes_safe_metadata(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
root = tmp_path / "mission-data" / "ui-environment"
|
|
router = build_environment_router(lambda: root)
|
|
payload = b"\x89PNG\r\n\x1a\nsynthetic-redacted"
|
|
upload = _endpoint(router, "/api/v1/environment/media/{surface_id}", "PUT")
|
|
|
|
document = asyncio.run(
|
|
upload(
|
|
surface_id="home",
|
|
request=_streaming_request(payload, "image/png"),
|
|
file_name="%D1%84%D0%BE%D0%BD.png",
|
|
)
|
|
)
|
|
|
|
assert isinstance(document, EnvironmentMediaDocument)
|
|
assert document.file_name == "фон.png"
|
|
assert document.media_kind == "image"
|
|
assert document.sha256 == hashlib.sha256(payload).hexdigest()
|
|
assert (root / "media" / "home.png").read_bytes() == payload
|
|
assert str(tmp_path) not in document.model_dump_json()
|
|
|
|
|
|
def test_playlist_media_is_item_scoped_and_generation_bound(tmp_path: Path) -> None:
|
|
root = tmp_path / "mission-data" / "ui-environment"
|
|
store = EnvironmentSettingsStore(root)
|
|
item_id = "media-11111111-1111-4111-8111-111111111111"
|
|
item_root = store.playlist_media_root("home")
|
|
item_root.mkdir(parents=True)
|
|
payload = b"\x89PNG\r\n\x1a\nplaylist-redacted"
|
|
temporary = item_root / ".playlist.upload"
|
|
temporary.write_bytes(payload)
|
|
digest = hashlib.sha256(payload).hexdigest()
|
|
|
|
document = store.finalize_playlist_media(
|
|
surface_id="home",
|
|
item_id=item_id,
|
|
temporary_path=temporary,
|
|
file_name="playlist.png",
|
|
media_type="image/png",
|
|
byte_length=len(payload),
|
|
sha256=digest,
|
|
)
|
|
path, restored = store.read_playlist_media("home", item_id, digest)
|
|
|
|
assert isinstance(document, EnvironmentPlaylistMediaDocument)
|
|
assert path == item_root / f"{item_id}.png"
|
|
assert path.read_bytes() == payload
|
|
assert restored == document
|
|
assert document.url == (
|
|
f"/api/v1/environment/media/home/{item_id}?generation={digest}"
|
|
)
|
|
with pytest.raises(PermissionError):
|
|
store.read_playlist_media("home", item_id, "0" * 64)
|
|
|
|
|
|
def test_playlist_media_upload_route_preserves_item_identity(tmp_path: Path) -> None:
|
|
root = tmp_path / "mission-data" / "ui-environment"
|
|
router = build_environment_router(lambda: root)
|
|
payload = b"\x89PNG\r\n\x1a\nplaylist-route-redacted"
|
|
item_id = "media-22222222-2222-4222-8222-222222222222"
|
|
upload = _endpoint(
|
|
router,
|
|
"/api/v1/environment/media/{surface_id}/{item_id}",
|
|
"PUT",
|
|
)
|
|
|
|
document = asyncio.run(
|
|
upload(
|
|
surface_id="home",
|
|
item_id=item_id,
|
|
request=_streaming_request(payload, "image/png"),
|
|
file_name="playlist.png",
|
|
)
|
|
)
|
|
|
|
assert isinstance(document, EnvironmentPlaylistMediaDocument)
|
|
assert document.item_id == item_id
|
|
assert document.media_kind == "image"
|
|
assert (root / "media" / "home" / f"{item_id}.png").read_bytes() == payload
|
|
|
|
|
|
def test_default_environment_has_every_product_surface() -> None:
|
|
document = default_environment_settings()
|
|
assert set(document.pages.model_dump()) == {
|
|
"home",
|
|
"fleet",
|
|
"observation",
|
|
"missions",
|
|
"data",
|
|
"system",
|
|
"polygon",
|
|
}
|
|
assert document.schema_version == "missioncore.operator-environment/v3"
|
|
assert document.pages.fleet.primary_workspace_id == "contour-health"
|
|
|
|
|
|
def test_legacy_v1_environment_is_migrated_without_losing_operator_media(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
store = EnvironmentSettingsStore(tmp_path / "operator-environment")
|
|
store.root.mkdir(parents=True)
|
|
empty_background = {
|
|
"enabled": False,
|
|
"source": "file",
|
|
"url": None,
|
|
"media_kind": None,
|
|
"file_name": None,
|
|
}
|
|
store.settings_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": "missioncore.operator-environment/v1",
|
|
"revision": 10,
|
|
"header_labels": {
|
|
"center": "Центр",
|
|
"fleet": "Парк",
|
|
"observation": "Контроль",
|
|
"missions": "Миссии",
|
|
"data": "Данные",
|
|
"system": "Система",
|
|
"polygon": "Тестировочный контур",
|
|
},
|
|
"backgrounds": {
|
|
"home": empty_background,
|
|
"center": empty_background,
|
|
"fleet": {
|
|
"enabled": True,
|
|
"source": "file",
|
|
"url": (
|
|
"/api/v1/environment/media/fleet?generation="
|
|
+ "a" * 64
|
|
),
|
|
"media_kind": "image",
|
|
"file_name": "park.png",
|
|
},
|
|
"observation": empty_background,
|
|
"missions": empty_background,
|
|
"data": empty_background,
|
|
"system": empty_background,
|
|
"polygon": empty_background,
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
migrated = store.read()
|
|
|
|
assert migrated.revision == 10
|
|
assert migrated.pages.observation.header_label == "Контроль"
|
|
assert migrated.pages.fleet.background.items[0].file_name == "park.png"
|
|
assert migrated.pages.fleet.background.image_duration_seconds == 10
|
|
assert migrated.pages.fleet.primary_workspace_id == "contour-health"
|
|
assert "center" not in migrated.pages.model_dump()
|
|
|
|
|
|
def test_v2_environment_is_migrated_to_ordered_playlist(tmp_path: Path) -> None:
|
|
store = EnvironmentSettingsStore(tmp_path / "operator-environment")
|
|
defaults = default_environment_settings()
|
|
store.root.mkdir(parents=True)
|
|
pages = defaults.pages.model_dump(mode="json")
|
|
for surface_id, page in pages.items():
|
|
page["background"] = {
|
|
"enabled": surface_id == "fleet",
|
|
"source": "file",
|
|
"url": (
|
|
f"/api/v1/environment/media/{surface_id}?generation="
|
|
+ "a" * 64
|
|
if surface_id == "fleet"
|
|
else None
|
|
),
|
|
"media_kind": "video" if surface_id == "fleet" else None,
|
|
"file_name": "legacy.mp4" if surface_id == "fleet" else None,
|
|
}
|
|
store.settings_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": "missioncore.operator-environment/v2",
|
|
"revision": 35,
|
|
"pages": pages,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
migrated = store.read()
|
|
|
|
assert migrated.schema_version == "missioncore.operator-environment/v3"
|
|
assert migrated.revision == 35
|
|
assert migrated.pages.fleet.background.enabled is True
|
|
assert [item.id for item in migrated.pages.fleet.background.items] == [
|
|
"media-legacy-fleet"
|
|
]
|
|
assert migrated.pages.fleet.background.items[0].media_kind == "video"
|