Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 44 additions & 0 deletions src/runloop_api_client/sdk/async_devbox.py
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,50 @@ async def get_info(
**options,
)

async def get_tunnel(
self,
**options: Unpack[BaseRequestOptions],
) -> TunnelView | None:
"""Retrieve the V2 tunnel information for this devbox.

:param options: Optional request configuration
:return: Tunnel details if a tunnel is enabled, None otherwise
:rtype: TunnelView | None

Example:
>>> tunnel = await devbox.get_tunnel()
>>> if tunnel:
... print(f"Tunnel key: {tunnel.tunnel_key}")
"""
info = await self.get_info(**options)
return info.tunnel

async def get_tunnel_url(
self,
port: int,
**options: Unpack[BaseRequestOptions],
) -> str | None:
"""Get the public tunnel URL for a specific port.

Constructs the tunnel URL using the format:
``https://{port}-{tunnel_key}.tunnel.runloop.ai``

:param port: The port number to construct the URL for
:type port: int
:param options: Optional request configuration
:return: The public tunnel URL if a tunnel is enabled, None otherwise
:rtype: str | None

Example:
>>> url = await devbox.get_tunnel_url(8080)
>>> if url:
... print(f"Access your service at: {url}")
"""
tunnel_view = await self.get_tunnel(**options)
if tunnel_view is None:
return None
return f"https://{port}-{tunnel_view.tunnel_key}.tunnel.runloop.ai"

async def await_running(self, *, polling_config: PollingConfig | None = None) -> DevboxView:
"""Wait for the devbox to reach running state.

Expand Down
44 changes: 44 additions & 0 deletions src/runloop_api_client/sdk/devbox.py
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,50 @@ def get_info(
**options,
)

def get_tunnel(
self,
**options: Unpack[BaseRequestOptions],
) -> TunnelView | None:
"""Retrieve the V2 tunnel information for this devbox.

:param options: Optional request configuration
:return: Tunnel details if a tunnel is enabled, None otherwise
:rtype: :class:`~runloop_api_client.types.tunnel_view.TunnelView` | None

Example:
>>> tunnel = devbox.get_tunnel()
>>> if tunnel:
... print(f"Tunnel key: {tunnel.tunnel_key}")
"""
info = self.get_info(**options)
return info.tunnel

def get_tunnel_url(
self,
port: int,
**options: Unpack[BaseRequestOptions],
) -> str | None:
"""Get the public tunnel URL for a specific port.

Constructs the tunnel URL using the format:
``https://{port}-{tunnel_key}.tunnel.runloop.ai``

:param port: The port number to construct the URL for
:type port: int
:param options: Optional request configuration
:return: The public tunnel URL if a tunnel is enabled, None otherwise
:rtype: str | None

Example:
>>> url = devbox.get_tunnel_url(8080)
>>> if url:
... print(f"Access your service at: {url}")
"""
tunnel_view = self.get_tunnel(**options)
if tunnel_view is None:
return None
return f"https://{port}-{tunnel_view.tunnel_key}.tunnel.runloop.ai"

def await_running(self, *, polling_config: PollingConfig | None = None) -> DevboxView:
"""Wait for the devbox to reach running state.

Expand Down
76 changes: 76 additions & 0 deletions tests/sdk/async_devbox/test_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -323,3 +323,79 @@ def test_net_property(self, mock_async_client: AsyncMock) -> None:
net = devbox.net
assert isinstance(net, AsyncNetworkInterface)
assert net._devbox is devbox

@pytest.mark.asyncio
async def test_get_tunnel_returns_tunnel_view(self, mock_async_client: AsyncMock) -> None:
"""Test get_tunnel returns the tunnel from get_info."""
tunnel_view = SimpleNamespace(
tunnel_key="abc123xyz",
auth_mode="open",
create_time_ms=1234567890000,
)
devbox_view_with_tunnel = SimpleNamespace(
id="dbx_123",
status="running",
tunnel=tunnel_view,
)
mock_async_client.devboxes.retrieve = AsyncMock(return_value=devbox_view_with_tunnel)

devbox = AsyncDevbox(mock_async_client, "dbx_123")
result = await devbox.get_tunnel()

assert result is not None
assert result == tunnel_view
assert result.tunnel_key == "abc123xyz"
mock_async_client.devboxes.retrieve.assert_called_once_with("dbx_123")

@pytest.mark.asyncio
async def test_get_tunnel_returns_none_when_no_tunnel(self, mock_async_client: AsyncMock) -> None:
"""Test get_tunnel returns None when no tunnel is enabled."""
devbox_view_no_tunnel = SimpleNamespace(
id="dbx_123",
status="running",
tunnel=None,
)
mock_async_client.devboxes.retrieve = AsyncMock(return_value=devbox_view_no_tunnel)

devbox = AsyncDevbox(mock_async_client, "dbx_123")
result = await devbox.get_tunnel()

assert result is None
mock_async_client.devboxes.retrieve.assert_called_once_with("dbx_123")

@pytest.mark.asyncio
async def test_get_tunnel_url_constructs_url(self, mock_async_client: AsyncMock) -> None:
"""Test get_tunnel_url constructs the correct URL."""
tunnel_view = SimpleNamespace(
tunnel_key="abc123xyz",
auth_mode="open",
create_time_ms=1234567890000,
)
devbox_view_with_tunnel = SimpleNamespace(
id="dbx_123",
status="running",
tunnel=tunnel_view,
)
mock_async_client.devboxes.retrieve = AsyncMock(return_value=devbox_view_with_tunnel)

devbox = AsyncDevbox(mock_async_client, "dbx_123")
result = await devbox.get_tunnel_url(8080)

assert result == "https://8080-abc123xyz.tunnel.runloop.ai"
mock_async_client.devboxes.retrieve.assert_called_once_with("dbx_123")

@pytest.mark.asyncio
async def test_get_tunnel_url_returns_none_when_no_tunnel(self, mock_async_client: AsyncMock) -> None:
"""Test get_tunnel_url returns None when no tunnel is enabled."""
devbox_view_no_tunnel = SimpleNamespace(
id="dbx_123",
status="running",
tunnel=None,
)
mock_async_client.devboxes.retrieve = AsyncMock(return_value=devbox_view_no_tunnel)

devbox = AsyncDevbox(mock_async_client, "dbx_123")
result = await devbox.get_tunnel_url(8080)

assert result is None
mock_async_client.devboxes.retrieve.assert_called_once_with("dbx_123")
72 changes: 72 additions & 0 deletions tests/sdk/devbox/test_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -322,3 +322,75 @@ def test_net_property(self, mock_client: Mock) -> None:
net = devbox.net
assert isinstance(net, NetworkInterface)
assert net._devbox is devbox

def test_get_tunnel_returns_tunnel_view(self, mock_client: Mock) -> None:
"""Test get_tunnel returns the tunnel from get_info."""
tunnel_view = SimpleNamespace(
tunnel_key="abc123xyz",
auth_mode="open",
create_time_ms=1234567890000,
)
devbox_view_with_tunnel = SimpleNamespace(
id="dbx_123",
status="running",
tunnel=tunnel_view,
)
mock_client.devboxes.retrieve.return_value = devbox_view_with_tunnel

devbox = Devbox(mock_client, "dbx_123")
result = devbox.get_tunnel()

assert result is not None
assert result == tunnel_view
assert result.tunnel_key == "abc123xyz"
mock_client.devboxes.retrieve.assert_called_once_with("dbx_123")

def test_get_tunnel_returns_none_when_no_tunnel(self, mock_client: Mock) -> None:
"""Test get_tunnel returns None when no tunnel is enabled."""
devbox_view_no_tunnel = SimpleNamespace(
id="dbx_123",
status="running",
tunnel=None,
)
mock_client.devboxes.retrieve.return_value = devbox_view_no_tunnel

devbox = Devbox(mock_client, "dbx_123")
result = devbox.get_tunnel()

assert result is None
mock_client.devboxes.retrieve.assert_called_once_with("dbx_123")

def test_get_tunnel_url_constructs_url(self, mock_client: Mock) -> None:
"""Test get_tunnel_url constructs the correct URL."""
tunnel_view = SimpleNamespace(
tunnel_key="abc123xyz",
auth_mode="open",
create_time_ms=1234567890000,
)
devbox_view_with_tunnel = SimpleNamespace(
id="dbx_123",
status="running",
tunnel=tunnel_view,
)
mock_client.devboxes.retrieve.return_value = devbox_view_with_tunnel

devbox = Devbox(mock_client, "dbx_123")
result = devbox.get_tunnel_url(8080)

assert result == "https://8080-abc123xyz.tunnel.runloop.ai"
mock_client.devboxes.retrieve.assert_called_once_with("dbx_123")

def test_get_tunnel_url_returns_none_when_no_tunnel(self, mock_client: Mock) -> None:
"""Test get_tunnel_url returns None when no tunnel is enabled."""
devbox_view_no_tunnel = SimpleNamespace(
id="dbx_123",
status="running",
tunnel=None,
)
mock_client.devboxes.retrieve.return_value = devbox_view_no_tunnel

devbox = Devbox(mock_client, "dbx_123")
result = devbox.get_tunnel_url(8080)

assert result is None
mock_client.devboxes.retrieve.assert_called_once_with("dbx_123")