feat: require pre-declaring preview url ports on create

.gitignore

@@ -7,6 +7,7 @@ build/
 .pytest_cache/
 .ruff_cache/
 .coverage
+coverage.xml
 htmlcov/
 .env
 .env.*

README.md

@@ -73,6 +73,7 @@ sandbox = await Sandbox.create(
     name="my-sandbox",
     type="cpu:default",
     max_lifetime=3600,
+    ports=[3000, 8080],
     envs={"API_KEY": "your-api-key"},
 )
 ```
@@ -149,10 +150,19 @@ await sandbox.destroy()
 
 ### Preview URLs
 
-Use preview URLs to get access to servers or web services running in a sandbox on a particular port: 
+Ports that should be publicly accessible must be declared at creation time via the `ports` list.
 
 ```python
-url = await sandbox.get_url(port=3000)
+sandbox = await Sandbox.create(
+    name="web-sandbox",
+    ports=[3000, 8080],
+)
+
+# Start a server inside the sandbox on the declared port
+await sandbox.exec("python -m http.server 3000 &", timeout=5_000)
+
+# Retrieve the pre-provisioned preview URL — synchronous, no network call
+url = sandbox.get_url(3000)
 print("preview:", url)
 # https://sb-abc123-va6-0-xK3mPq2nAeB-3000.sandbox-adobeioruntime.net
 ```

src/aio_lib_sandbox/__init__.py

@@ -10,7 +10,9 @@
 from .errors import (
     SandboxClientError,
     SandboxInitializationError,
+    SandboxInvalidPortError,
     SandboxNotFoundError,
+    SandboxPortNotProvisionedError,
     SandboxSDKError,
     SandboxTimeoutError,
     SandboxUnauthorizedError,
@@ -43,7 +45,9 @@
     "SandboxSDKError",
     "SandboxInitializationError",
     "SandboxClientError",
+    "SandboxInvalidPortError",
     "SandboxNotFoundError",
+    "SandboxPortNotProvisionedError",
     "SandboxUnauthorizedError",
     "SandboxTimeoutError",
     "SandboxWebSocketError",

src/aio_lib_sandbox/errors.py

@@ -32,3 +32,11 @@ class SandboxTimeoutError(SandboxSDKError):
 
 class SandboxWebSocketError(SandboxSDKError):
     """WebSocket transport error."""
+
+
+class SandboxPortNotProvisionedError(SandboxClientError):
+    """Port was not declared in ``create(ports=[...])`` and cannot be retrieved."""
+
+
+class SandboxInvalidPortError(SandboxClientError):
+    """Port value is not a valid integer in the range 1–65535."""

src/aio_lib_sandbox/sandbox.py

@@ -6,7 +6,6 @@
 import asyncio
 import base64
 import os
-import re
 import secrets
 from typing import Any, Callable
 
@@ -24,6 +23,8 @@
 from .errors import (
     SandboxClientError,
     SandboxInitializationError,
+    SandboxInvalidPortError,
+    SandboxPortNotProvisionedError,
     SandboxWebSocketError,
 )
 from .types import (
@@ -57,7 +58,7 @@ def __init__(
         api_host: str,
         api_key: str,
         token: str | None = None,
-        public_url_template: str | None = None,
+        preview_urls: dict[int, str] | None = None,
         management_endpoint: str | None = None,
         verify_ssl: bool = True,
     ) -> None:
@@ -72,7 +73,7 @@ def __init__(
         self.api_host = api_host
         self.api_key = api_key
         self.token = token
-        self.public_url_template = public_url_template
+        self.preview_urls: dict[int, str] = preview_urls or {}
         self.management_endpoint = management_endpoint
         self.verify_ssl = verify_ssl
 
@@ -93,6 +94,7 @@ async def create(
         type: str = "cpu:default",
         size: str | dict[str, Any] | None = None,
         max_lifetime: int = 3600,
+        ports: list[int] | None = None,
         envs: dict[str, str] | None = None,
         policy: Policy | None = None,
         cluster: str | None = None,
@@ -118,6 +120,7 @@ async def create(
             type: Sandbox type (default: ``'cpu:default'``).
             size: Size tier name or spec dict.
             max_lifetime: Maximum lifetime in seconds.
+            ports: TCP ports to expose via preview URLs (default: ``[]``).
             envs: Environment variables to inject into the sandbox.
             policy: Network policy (e.g. egress allowlist).
             cluster: Target cluster name.
@@ -144,6 +147,8 @@ async def create(
             body["envs"] = envs
         if policy is not None:
             body["policy"] = policy
+        if ports is not None:
+            body["ports"] = ports
 
         url = f"{creds['api_host']}/api/v1/namespaces/{creds['namespace']}/sandbox"
         payload = await api_request(
@@ -168,7 +173,7 @@ async def create(
             cluster=payload.get("cluster"),
             region=payload.get("region"),
             max_lifetime=payload.get("maxLifetime", 3600),
-            public_url_template=payload.get("publicUrlTemplate"),
+            preview_urls=_parse_preview_urls(payload.get("previewUrls")),
             management_endpoint=payload.get("managementEndpoint"),
             namespace=creds["namespace"],
             api_host=creds["api_host"],
@@ -222,6 +227,7 @@ async def get(
             cluster=payload.get("cluster"),
             region=payload.get("region"),
             max_lifetime=payload.get("maxLifetime", 3600),
+            preview_urls=_parse_preview_urls(payload.get("previewUrls")),
             namespace=creds["namespace"],
             api_host=creds["api_host"],
             api_key=creds["api_key"],
@@ -369,31 +375,36 @@ async def file_op(self, frame_type: str, **extra: Any) -> Any:
     # URL
     # ------------------------------------------------------------------
 
-    async def get_url(self, *, port: int, protocol: str | None = None) -> str:
+    def get_url(self, port: int) -> str:
         """Return the public preview URL for a given port on this sandbox.
 
+        Synchronous local lookup against the ``preview_urls`` dict returned by
+        the server at create time. The URL is opaque — do not parse or
+        reconstruct it.
+
         Args:
-            port: Port number (1–65535).
-            protocol: Optional scheme override (e.g. ``'wss'``).
+            port: Port number (1–65535) declared in ``create(ports=[...])``.
 
         Returns:
-            The resolved preview URL string.
-        """
-        if not self.public_url_template:
-            raise SandboxClientError(
-                f"Cannot get URL for sandbox '{self.id}': public_url_template is not available"
-            )
+            The preview URL string for that port.
 
+        Raises:
+            SandboxInvalidPortError: When ``port`` is not an integer in the
+                range 1–65535.
+            SandboxPortNotProvisionedError: When ``port`` is valid but was not
+                declared in ``create(ports=[...])``.
+        """
         if not isinstance(port, int) or port < 1 or port > 65535:
-            raise SandboxClientError(
+            raise SandboxInvalidPortError(
                 f"Invalid port '{port}': must be an integer between 1 and 65535"
             )
 
-        url = self.public_url_template.replace("{sandboxId}", self.id).replace(
-            "{port}", str(port)
-        )
-        if protocol:
-            url = re.sub(r"^https?://", f"{protocol}://", url)
+        url = self.preview_urls.get(port)
+        if url is None:
+            raise SandboxPortNotProvisionedError(
+                f"Port {port} was not provisioned for sandbox '{self.id}'. "
+                "Declare it in create(ports=[...]) to get a preview URL."
+            )
         return url
 
     # ------------------------------------------------------------------
@@ -483,3 +494,26 @@ def resolve_credentials(
             "namespace": resolved_ns,  # type: ignore[return-value]
             "api_key": resolved_key,  # type: ignore[return-value]
         }
+
+
+def _parse_preview_urls(raw: Any) -> dict[int, str]:
+    """Parse the ``previewUrls`` JSON object from the API response.
+
+    Converts string keys (port numbers) to integers and treats URL values as
+    opaque strings — they are not parsed or reconstructed.
+
+    Returns an empty dict when the server response omits ``previewUrls``
+    (fail-closed: every ``get_url()`` call raises
+    :exc:`SandboxPortNotProvisionedError`).
+    """
+    if not isinstance(raw, dict):
+        return {}
+    result: dict[int, str] = {}
+    for key, value in raw.items():
+        try:
+            port = int(key)
+        except (ValueError, TypeError):
+            continue
+        if 1 <= port <= 65535 and isinstance(value, str):
+            result[port] = value
+    return result

tests/test_sandbox.py

@@ -19,12 +19,15 @@
 from aio_lib_sandbox.errors import (
     SandboxClientError,
     SandboxInitializationError,
+    SandboxInvalidPortError,
     SandboxNotFoundError,
+    SandboxPortNotProvisionedError,
     SandboxTimeoutError,
     SandboxUnauthorizedError,
     SandboxWebSocketError,
 )
 from aio_lib_sandbox.frames import normalize_size
+from aio_lib_sandbox.sandbox import _parse_preview_urls
 from aio_lib_sandbox.ws import PendingExec, PendingFileOp, WsSession
 
 # ---------------------------------------------------------------------------
@@ -256,6 +259,38 @@ async def test_create_builds_ws_endpoint_when_absent(self, monkeypatch):
         assert "sb-noep" in sandbox.endpoint
         assert sandbox.endpoint.startswith("wss://")
 
+    @pytest.mark.asyncio
+    async def test_create_forwards_ports_and_parses_preview_urls(self):
+        payload = {
+            "sandboxId": "sb-ports",
+            "wsEndpoint": "wss://runtime.example.net/ws/v1/namespaces/ns/sandbox/sb-ports/exec",
+            "status": "ready",
+            "token": "tok-ports",
+            "maxLifetime": 3600,
+            "previewUrls": {
+                "3000": "https://sb-ports-3000.preview.example.net",
+                "8080": "https://sb-ports-8080.preview.example.net",
+            },
+        }
+
+        with patch("aio_lib_sandbox.sandbox.api_request", new=AsyncMock(return_value=payload)) as mock_req, \
+             patch.object(Sandbox, "connect", new=AsyncMock()):
+            sandbox = await Sandbox.create(
+                name="ports-sandbox",
+                api_host="https://runtime.example.net",
+                namespace="ns",
+                auth="uuid:key",
+                ports=[3000, 8080],
+            )
+
+        _, kwargs = mock_req.call_args
+        assert kwargs["body"]["ports"] == [3000, 8080]
+        assert sandbox.preview_urls == {
+            3000: "https://sb-ports-3000.preview.example.net",
+            8080: "https://sb-ports-8080.preview.example.net",
+        }
+        assert sandbox.get_url(3000) == "https://sb-ports-3000.preview.example.net"
+
 
 # ---------------------------------------------------------------------------
 # Sandbox.get()
@@ -535,31 +570,29 @@ async def test_list_files_empty(self):
 
 
 class TestGetUrl:
-    @pytest.mark.asyncio
-    async def test_resolves_url_from_template(self):
-        sandbox = _make_sandbox(public_url_template="https://{sandboxId}-{port}.preview.example.net")
-        url = await sandbox.get_url(port=3000)
+    def test_resolves_url_from_preview_urls(self):
+        sandbox = _make_sandbox(preview_urls={3000: "https://sb-test-3000.preview.example.net"})
+        url = sandbox.get_url(3000)
         assert url == "https://sb-test-3000.preview.example.net"
 
-    @pytest.mark.asyncio
-    async def test_overrides_protocol(self):
-        sandbox = _make_sandbox(public_url_template="https://{sandboxId}-{port}.preview.example.net")
-        url = await sandbox.get_url(port=3000, protocol="wss")
-        assert url == "wss://sb-test-3000.preview.example.net"
-
-    @pytest.mark.asyncio
-    async def test_raises_without_template(self):
+    def test_raises_when_port_not_provisioned(self):
         sandbox = _make_sandbox()
-        with pytest.raises(SandboxClientError):
-            await sandbox.get_url(port=3000)
+        with pytest.raises(SandboxPortNotProvisionedError):
+            sandbox.get_url(3000)
 
-    @pytest.mark.asyncio
-    async def test_raises_on_invalid_port(self):
-        sandbox = _make_sandbox(public_url_template="https://{sandboxId}-{port}.preview.example.net")
-        with pytest.raises(SandboxClientError):
-            await sandbox.get_url(port=0)
-        with pytest.raises(SandboxClientError):
-            await sandbox.get_url(port=70000)
+    def test_raises_on_out_of_range_port(self):
+        sandbox = _make_sandbox(preview_urls={3000: "https://sb-test-3000.preview.example.net"})
+        with pytest.raises(SandboxInvalidPortError):
+            sandbox.get_url(0)
+        with pytest.raises(SandboxInvalidPortError):
+            sandbox.get_url(65536)
+
+    def test_raises_on_non_integer_port(self):
+        sandbox = _make_sandbox(preview_urls={3000: "https://sb-test-3000.preview.example.net"})
+        with pytest.raises(SandboxInvalidPortError):
+            sandbox.get_url("3000")
+        with pytest.raises(SandboxInvalidPortError):
+            sandbox.get_url(3000.5)
 
 
 # ---------------------------------------------------------------------------
@@ -703,3 +736,43 @@ async def _mock_req(method, url, *, api_key, body=None, **kw):
             )
 
         assert "policy" not in captured["body"]
+
+
+# ---------------------------------------------------------------------------
+# _parse_preview_urls
+# ---------------------------------------------------------------------------
+
+
+class TestParsePreviewUrls:
+    def test_returns_empty_for_non_dict(self):
+        assert _parse_preview_urls(None) == {}
+        assert _parse_preview_urls("string") == {}
+        assert _parse_preview_urls(42) == {}
+        assert _parse_preview_urls([]) == {}
+
+    def test_parses_string_keys_to_int(self):
+        raw = {"3000": "https://sb-3000.example.net", "8080": "https://sb-8080.example.net"}
+        result = _parse_preview_urls(raw)
+        assert result == {
+            3000: "https://sb-3000.example.net",
+            8080: "https://sb-8080.example.net",
+        }
+
+    def test_skips_non_integer_keys(self):
+        raw = {"3000": "https://sb-3000.example.net", "notaport": "https://sb-x.example.net"}
+        result = _parse_preview_urls(raw)
+        assert result == {3000: "https://sb-3000.example.net"}
+
+    def test_skips_out_of_range_ports(self):
+        raw = {"0": "https://zero.example.net", "65536": "https://toobig.example.net",
+               "3000": "https://sb-3000.example.net"}
+        result = _parse_preview_urls(raw)
+        assert result == {3000: "https://sb-3000.example.net"}
+
+    def test_skips_non_string_url_values(self):
+        raw = {"3000": 12345, "8080": "https://sb-8080.example.net"}
+        result = _parse_preview_urls(raw)
+        assert result == {8080: "https://sb-8080.example.net"}
+
+    def test_returns_empty_for_empty_dict(self):
+        assert _parse_preview_urls({}) == {}