feat(ui): add environment media playlists

This commit is contained in:
DCCONSTRUCTIONS
2026-07-29 14:20:19 +03:00
parent c455301764
commit 8b1f109b09
11 changed files with 1076 additions and 138 deletions
+111 -6
View File
@@ -15,6 +15,8 @@ from pydantic import ValidationError
from k1link.web.environment_api import (
EnvironmentBackground,
EnvironmentMediaDocument,
EnvironmentMediaItem,
EnvironmentPlaylistMediaDocument,
EnvironmentSettingsPut,
EnvironmentSettingsStore,
build_environment_router,
@@ -124,21 +126,24 @@ def test_environment_settings_reject_stale_revision(tmp_path: Path) -> None:
def test_environment_background_rejects_untrusted_enabled_source() -> None:
with pytest.raises(ValidationError):
EnvironmentBackground(
enabled=True,
EnvironmentMediaItem(
id="media-invalid-url",
source="url",
url="file:///tmp/background.mp4",
media_kind="video",
)
with pytest.raises(ValidationError):
EnvironmentBackground(
enabled=True,
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,
@@ -192,6 +197,65 @@ def test_environment_media_upload_route_streams_and_publishes_safe_metadata(
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()) == {
@@ -203,7 +267,7 @@ def test_default_environment_has_every_product_surface() -> None:
"system",
"polygon",
}
assert document.schema_version == "missioncore.operator-environment/v2"
assert document.schema_version == "missioncore.operator-environment/v3"
assert document.pages.fleet.primary_workspace_id == "contour-health"
@@ -264,6 +328,47 @@ def test_legacy_v1_environment_is_migrated_without_losing_operator_media(
assert migrated.revision == 10
assert migrated.pages.observation.header_label == "Контроль"
assert migrated.pages.fleet.background.file_name == "park.png"
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"