feat(control-station): add configurable environment shell

tests/test_environment_api.py

@@ -0,0 +1,197 @@
+from __future__ import annotations
+
+import asyncio
+import hashlib
+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,
+    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")
+
+
+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_header_labels(tmp_path: Path) -> None:
+    store = EnvironmentSettingsStore(tmp_path / "operator-environment")
+    initial = store.read()
+    assert initial.revision == 0
+    assert initial.header_labels.polygon == "Тестировочный контур"
+    assert initial.backgrounds.home.enabled is False
+
+    request = EnvironmentSettingsPut(
+        revision=initial.revision,
+        header_labels=initial.header_labels.model_copy(update={"center": "Командный центр"}),
+        backgrounds=initial.backgrounds,
+    )
+    saved = store.save(request)
+    restored = EnvironmentSettingsStore(store.root).read()
+
+    assert saved.revision == 1
+    assert restored == saved
+    assert restored.header_labels.center == "Командный центр"
+    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,
+            header_labels=initial.header_labels,
+            backgrounds=initial.backgrounds,
+        )
+    )
+
+    with pytest.raises(RuntimeError, match="revision changed"):
+        store.save(
+            EnvironmentSettingsPut(
+                revision=0,
+                header_labels=initial.header_labels,
+                backgrounds=initial.backgrounds,
+            )
+        )
+
+
+def test_environment_background_rejects_untrusted_enabled_source() -> None:
+    with pytest.raises(ValidationError):
+        EnvironmentBackground(
+            enabled=True,
+            source="url",
+            url="file:///tmp/background.mp4",
+            media_kind="video",
+        )
+
+    with pytest.raises(ValidationError):
+        EnvironmentBackground(
+            enabled=True,
+            source="file",
+            url="/private/operator/background.mp4",
+            media_kind="video",
+        )
+
+
+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_default_environment_has_every_product_surface() -> None:
+    document = default_environment_settings()
+    assert set(document.backgrounds.model_dump()) == {
+        "home",
+        "center",
+        "fleet",
+        "observation",
+        "missions",
+        "data",
+        "system",
+        "polygon",
+    }