sandbox: B2 — get_translations runtime handler + string loader

Register a sandbox/get_translations handler in SandboxRuntime. It loads raw
translation strings for the requested domains from the sandbox's own
filesystem (built-in from the bundled package, custom from the fetched
<config>/custom_components/<domain>) by reusing core's
_async_get_component_strings against the sandbox-private hass — which also
pre-fills 'title' from integration.name. Main cannot run that fallback for a
custom domain because it holds no Integration, so the title must be injected
here. Replies with {language, strings: {domain: raw dict}}.

Tests cover built-in title pass-through, custom title injection, the empty
case, the Struct packing, and the no-flow-runner guard.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

sandbox/hass_client/hass_client/sandbox/__init__.py

@@ -39,13 +39,15 @@ from hass_client.entity_bridge import EntityBridge
 from hass_client.entry_runner import EntryRunner
 from hass_client.event_mirror import EventMirror
 from hass_client.flow_runner import FlowRunner
-from hass_client.protocol import MSG_READY, MSG_SHUTDOWN
+from hass_client.protocol import MSG_GET_TRANSLATIONS, MSG_READY, MSG_SHUTDOWN
 from hass_client.sandbox_bridge import ChannelSandboxBridge
 from hass_client.service_mirror import ServiceMirror
 from homeassistant.const import EVENT_HOMEASSISTANT_FINAL_WRITE
-from homeassistant.core import CoreState
+from homeassistant.core import CoreState, HomeAssistant
 from homeassistant.helpers import json as json_helper, restore_state
 from homeassistant.helpers.sandbox_context import current_sandbox
+from homeassistant.helpers.translation import _async_get_component_strings
+from homeassistant.loader import async_get_integrations
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -192,6 +194,9 @@ class SandboxRuntime:
             await _load_restore_state(hass)
             self._channel.register("sandbox/ping", _handle_ping)
             self._channel.register(MSG_SHUTDOWN, self._handle_shutdown)
+            self._channel.register(
+                MSG_GET_TRANSLATIONS, self._handle_get_translations
+            )
             self._flow_runner.register(self._channel)
             self._entry_runner.register(self._channel)
             self._entity_bridge.register(self._channel)
@@ -254,6 +259,29 @@ class SandboxRuntime:
             asyncio.get_running_loop().call_soon(self._shutdown.set)
         return summary
 
+    async def _handle_get_translations(
+        self, msg: pb.GetTranslations
+    ) -> pb.GetTranslationsResult:
+        """Serve a main-side ``sandbox/get_translations`` pull.
+
+        Main holds no ``Integration`` for a custom sandboxed domain, so it
+        cannot load the integration's ``translations/<lang>.json`` or run the
+        ``title``→``integration.name`` fallback. This sandbox does — it
+        fetched and imported the code — so it loads the raw strings here and
+        replies with the un-flattened nesting main's translation cache merges
+        as-is.
+        """
+        result = pb.GetTranslationsResult(language=msg.language)
+        flow_runner = self._flow_runner
+        if flow_runner is None:
+            return result
+        strings = await _collect_component_strings(
+            flow_runner.hass, msg.language, list(msg.domains)
+        )
+        if strings:
+            result.strings.update(strings)
+        return result
+
     async def _run_graceful_shutdown(self) -> pb.ShutdownResult:
         """Unload every loaded entry and snapshot RestoreEntity state.
 
@@ -325,6 +353,36 @@ class SandboxRuntime:
         return result
 
 
+async def _collect_component_strings(
+    hass: HomeAssistant, language: str, domains: list[str]
+) -> dict[str, Any]:
+    """Load raw translation strings for ``domains`` from this sandbox's disk.
+
+    Resolves each domain's ``Integration`` against the sandbox-private
+    ``hass`` (built-in from the bundled package, custom from the fetched
+    ``<config>/custom_components/<domain>``) and reuses core's
+    :func:`_async_get_component_strings`, which reads
+    ``translations/<language>.json`` and pre-fills ``title`` from
+    ``integration.name``. The return is ``{domain: <raw strings.json dict>}``
+    for the requested language — the exact shape main's translation cache
+    overlays. Domains the sandbox cannot resolve come back as ``{}`` (no
+    Integration ⇒ no file, no title), which is harmless on main.
+    """
+    if not domains:
+        return {}
+    components = set(domains)
+    ints_or_excs = await async_get_integrations(hass, components)
+    integrations = {
+        domain: result
+        for domain, result in ints_or_excs.items()
+        if not isinstance(result, Exception)
+    }
+    by_language = await _async_get_component_strings(
+        hass, [language], components, integrations
+    )
+    return by_language.get(language, {})
+
+
 async def _load_restore_state(hass: Any) -> None:
     """Warm-load this sandbox's ``core.restore_state`` cache.
 

sandbox/hass_client/tests/test_translation_provider.py

@@ -0,0 +1,144 @@
+"""Tests for the sandbox-side ``sandbox/get_translations`` handler.
+
+Covers :func:`hass_client.sandbox._collect_component_strings` (the loader that
+mirrors core's translation read) and :meth:`SandboxRuntime._handle_get_translations`
+(the channel handler that packs the result into a ``Struct``). The sandbox
+holds the ``Integration`` for a custom domain — main does not — so this is
+where the raw ``translations/<lang>.json`` and the ``title`` pre-fill come
+from.
+"""
+
+import json
+from pathlib import Path
+import tempfile
+from typing import Any
+
+from hass_client._proto import sandbox_pb2 as pb
+from hass_client.flow_runner import FlowRunner
+from hass_client.sandbox import SandboxRuntime, _collect_component_strings
+import pytest
+
+from homeassistant import loader as ha_loader
+from homeassistant.core import HomeAssistant
+
+
+@pytest.fixture(name="hass")
+async def _hass_fixture() -> HomeAssistant:
+    """A sandbox-private bare HA, as the runtime builds via FlowRunner."""
+    with tempfile.TemporaryDirectory(prefix="sandbox_translation_") as tmp:
+        flow_runner = await FlowRunner.create(config_dir=tmp)
+        try:
+            yield flow_runner.hass
+        finally:
+            await flow_runner.async_stop()
+
+
+def _install_custom_integration(
+    hass: HomeAssistant, tmp_path: Path, *, domain: str, strings: dict[str, Any]
+) -> ha_loader.Integration:
+    """Stand up a custom integration on disk + in the loader cache.
+
+    Writes ``<tmp_path>/<domain>/translations/en.json`` and injects a matching
+    custom :class:`Integration` into ``DATA_INTEGRATIONS`` so
+    ``async_get_integrations`` resolves it from cache.
+    """
+    root = tmp_path / domain
+    (root / "translations").mkdir(parents=True)
+    (root / "translations" / "en.json").write_text(json.dumps(strings))
+    integration = ha_loader.Integration(
+        hass,
+        f"custom_components.{domain}",
+        root,
+        {
+            "domain": domain,
+            "name": "My Custom",
+            "config_flow": True,
+            "documentation": "https://example.com",
+            "iot_class": "local_polling",
+            "requirements": [],
+            "dependencies": [],
+            "codeowners": [],
+        },
+        {"translations"},
+    )
+    assert not integration.is_built_in
+    assert integration.has_translations
+    cache = hass.data.setdefault(ha_loader.DATA_INTEGRATIONS, {})
+    cache[domain] = integration
+    return integration
+
+
+async def test_collect_strings_builtin_prefills_title(hass: HomeAssistant) -> None:
+    """A built-in domain loads its bundled strings with a ``title``."""
+    strings = await _collect_component_strings(hass, "en", ["counter"])
+
+    assert "counter" in strings
+    assert strings["counter"]
+    # Built-in en.json already ships a title; it is preserved verbatim.
+    assert strings["counter"]["title"] == "Counter"
+
+
+async def test_collect_strings_custom_injects_title(
+    hass: HomeAssistant, tmp_path: Path
+) -> None:
+    """A custom domain loads its on-disk strings; missing title ⇒ integration name."""
+    _install_custom_integration(
+        hass,
+        tmp_path,
+        domain="my_custom",
+        # No "title" — the helper must inject integration.name. Main cannot
+        # run this fallback (it holds no Integration for a custom domain).
+        strings={
+            "config": {"step": {"user": {"title": "Set up"}}},
+            "entity": {"sensor": {"widget": {"name": "Widget"}}},
+        },
+    )
+
+    strings = await _collect_component_strings(hass, "en", ["my_custom"])
+
+    assert strings["my_custom"]["title"] == "My Custom"
+    assert strings["my_custom"]["config"]["step"]["user"]["title"] == "Set up"
+    assert strings["my_custom"]["entity"]["sensor"]["widget"]["name"] == "Widget"
+
+
+async def test_collect_strings_empty_domains_returns_empty(
+    hass: HomeAssistant,
+) -> None:
+    """No domains requested ⇒ no work, empty result."""
+    assert await _collect_component_strings(hass, "en", []) == {}
+
+
+async def test_handle_get_translations_packs_struct(
+    hass: HomeAssistant, tmp_path: Path
+) -> None:
+    """The channel handler returns the raw dict packed into the result Struct."""
+    _install_custom_integration(
+        hass,
+        tmp_path,
+        domain="packed_custom",
+        strings={"entity": {"sensor": {"w": {"name": "W"}}}},
+    )
+    runtime = SandboxRuntime(url="ws://x", group="custom")
+    # The handler only reads ``flow_runner.hass``; wrap the fixture hass.
+    runtime._flow_runner = FlowRunner(hass)  # noqa: SLF001
+
+    result = await runtime._handle_get_translations(  # noqa: SLF001
+        pb.GetTranslations(language="en", domains=["packed_custom"])
+    )
+
+    assert result.language == "en"
+    packed = dict(result.strings)
+    assert packed["packed_custom"]["title"] == "My Custom"
+    assert packed["packed_custom"]["entity"]["sensor"]["w"]["name"] == "W"
+
+
+async def test_handle_get_translations_without_flow_runner_is_empty() -> None:
+    """No flow runner (channel never opened) ⇒ empty result, never raises."""
+    runtime = SandboxRuntime(url="ws://x", group="custom")
+
+    result = await runtime._handle_get_translations(  # noqa: SLF001
+        pb.GetTranslations(language="en", domains=["whatever"])
+    )
+
+    assert result.language == "en"
+    assert not dict(result.strings)