opensandbox-group--opensandbox/server/tests/test_routes_endpoint_behavior.py

# Copyright 2025 Alibaba Group Holding Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from types import SimpleNamespace

from fastapi.testclient import TestClient

from opensandbox_server.api import lifecycle
from opensandbox_server.api.schema import Endpoint


def test_get_endpoint_returns_service_result(
    client: TestClient,
    auth_headers: dict,
    monkeypatch,
) -> None:
    calls: list[tuple[str, int]] = []

    class StubService:
        @staticmethod
        def get_endpoint(sandbox_id: str, port: int, **kwargs) -> Endpoint:
            calls.append((sandbox_id, port))
            return Endpoint(endpoint="10.57.1.91:40109/proxy/44772")

    monkeypatch.setattr(lifecycle, "sandbox_service", StubService())

    response = client.get(
        "/v1/sandboxes/sbx-001/endpoints/44772",
        headers=auth_headers,
    )

    assert response.status_code == 200
    assert response.json()["endpoint"] == "10.57.1.91:40109/proxy/44772"
    assert calls == [("sbx-001", 44772)]


def test_get_endpoint_use_server_proxy_rewrites_url(
    client: TestClient,
    auth_headers: dict,
    monkeypatch,
) -> None:
    class StubService:
        @staticmethod
        def get_endpoint(sandbox_id: str, port: int, **kwargs) -> Endpoint:
            return Endpoint(endpoint="10.57.1.91:40109/proxy/44772")

    monkeypatch.setattr(lifecycle, "sandbox_service", StubService())

    response = client.get(
        "/v1/sandboxes/sbx-001/endpoints/44772",
        params={"use_server_proxy": "true"},
        headers=auth_headers,
    )

    assert response.status_code == 200
    assert response.json()["endpoint"] == "testserver/sandboxes/sbx-001/proxy/44772"


def test_get_endpoint_use_server_proxy_prefers_server_eip(
    client: TestClient,
    auth_headers: dict,
    monkeypatch,
) -> None:
    class StubService:
        @staticmethod
        def get_endpoint(sandbox_id: str, port: int, **kwargs) -> Endpoint:
            return Endpoint(endpoint="10.57.1.91:40109/proxy/44772")

    monkeypatch.setattr(lifecycle, "sandbox_service", StubService())
    monkeypatch.setattr(
        lifecycle,
        "get_config",
        lambda: SimpleNamespace(server=SimpleNamespace(eip="sandbox.example.com/opensandbox/")),
    )

    response = client.get(
        "/v1/sandboxes/sbx-001/endpoints/44772",
        params={"use_server_proxy": "true"},
        headers=auth_headers,
    )

    assert response.status_code == 200
    assert (
        response.json()["endpoint"]
        == "sandbox.example.com/opensandbox/sandboxes/sbx-001/proxy/44772"
    )


def test_get_endpoint_rejects_server_proxy_with_expires(
    client: TestClient,
    auth_headers: dict,
    monkeypatch,
) -> None:
    class StubService:
        @staticmethod
        def get_endpoint(sandbox_id: str, port: int, **kwargs) -> Endpoint:
            raise AssertionError("signed endpoint resolution should not run")

    monkeypatch.setattr(lifecycle, "sandbox_service", StubService())

    response = client.get(
        "/v1/sandboxes/sbx-001/endpoints/44772",
        params={"use_server_proxy": "true", "expires": "2000000000"},
        headers=auth_headers,
    )

    assert response.status_code == 400
    payload = response.json()
    assert payload["code"] == "SANDBOX::INVALID_PARAMETER"
    assert "use_server_proxy cannot be combined with expires" in payload["message"]


def test_get_endpoint_rejects_non_numeric_port(
    client: TestClient,
    auth_headers: dict,
) -> None:
    response = client.get(
        "/v1/sandboxes/sbx-001/endpoints/not-a-port",
        headers=auth_headers,
    )

    assert response.status_code == 422


def test_get_endpoint_passes_expires_to_service(
    client: TestClient,
    auth_headers: dict,
    monkeypatch,
) -> None:
    captured: dict = {}

    class StubService:
        @staticmethod
        def get_endpoint(sandbox_id: str, port: int, **kwargs) -> Endpoint:
            captured.update({"sandbox_id": sandbox_id, "port": port, **kwargs})
            return Endpoint(endpoint="sandbox.example.com")

    monkeypatch.setattr(lifecycle, "sandbox_service", StubService())

    response = client.get(
        "/v1/sandboxes/sbx-001/endpoints/44772",
        params={"expires": "2000000000"},
        headers=auth_headers,
    )

    assert response.status_code == 200
    assert captured.get("expires") == 2000000000


def test_get_endpoint_unsigned_when_expires_omitted(
    client: TestClient,
    auth_headers: dict,
    monkeypatch,
) -> None:
    captured: dict = {}

    class StubService:
        @staticmethod
        def get_endpoint(sandbox_id: str, port: int, **kwargs) -> Endpoint:
            captured.update(kwargs)
            return Endpoint(endpoint="sandbox.example.com")

    monkeypatch.setattr(lifecycle, "sandbox_service", StubService())

    response = client.get(
        "/v1/sandboxes/sbx-001/endpoints/44772",
        headers=auth_headers,
    )

    assert response.status_code == 200
    assert captured.get("expires") is None