import sys
from typing import Any, Dict, List, Optional

import pytest
from fastapi import FastAPI

from ray import serve
from ray.serve._private.build_app import (
    CUSTOM_INGRESS_REQUEST_ROUTER_UNSUPPORTED_ERROR,
    BuiltApplication,
    build_app,
)
from ray.serve._private.client import ServeControllerClient
from ray.serve._private.common import DeploymentID
from ray.serve.config import RequestRouterConfig
from ray.serve.deployment import Application, Deployment
from ray.serve.exceptions import RayServeException
from ray.serve.experimental.round_robin_router import RoundRobinRouter
from ray.serve.handle import DeploymentHandle


class FakeDeploymentHandle:
    def __init__(self, deployment_name: str, app_name: str):
        self.deployment_id = DeploymentID(deployment_name, app_name)

    @classmethod
    def from_deployment(cls, deployment, app_name: str) -> "FakeDeploymentHandle":
        return cls(deployment.name, app_name)

    def __eq__(self, other: Any) -> bool:
        return (
            isinstance(other, FakeDeploymentHandle)
            and self.deployment_id == other.deployment_id
        )


def _build_and_check(
    app: Application,
    *,
    expected_ingress_name: str,
    expected_deployments: List[Deployment],
    app_name: str = "default",
    default_runtime_env: Optional[Dict[str, Any]] = None,
):
    built_app: BuiltApplication = build_app(
        app,
        name=app_name,
        # Each real DeploymentHandle has a unique ID (intentionally), so the below
        # equality checks don't work. Use a fake implementation instead.
        make_deployment_handle=FakeDeploymentHandle.from_deployment,
        default_runtime_env=default_runtime_env,
    )
    assert built_app.name == app_name
    assert built_app.ingress_deployment_name == expected_ingress_name
    assert len(built_app.deployments) == len(expected_deployments)

    # Check that the returned deployment_handles are populated properly.
    assert len(built_app.deployment_handles) == len(expected_deployments)
    for d in expected_deployments:
        h = built_app.deployment_handles.get(d.name, None)
        assert h is not None, f"No handle returned for deployment {d.name}."
        assert isinstance(h, FakeDeploymentHandle)
        assert h.deployment_id == DeploymentID(d.name, app_name=app_name)

    for expected_deployment in expected_deployments:
        generated_deployment = None
        for d in built_app.deployments:
            if d.name == expected_deployment.name:
                generated_deployment = d

        assert generated_deployment is not None, (
            f"Expected a deployment with name '{expected_deployment.name}' "
            "to be generated but none was found. All generated names: "
            + str([d.name for d in built_app.deployments])
        )

        assert expected_deployment == generated_deployment


def test_real_deployment_handle_default():
    """Other tests inject a FakeDeploymentHandle, so check the default behavior."""

    @serve.deployment
    class D:
        pass

    built_app: BuiltApplication = build_app(
        D.bind(D.options(name="Inner").bind()),
        name="app-name",
    )
    assert len(built_app.deployments) == 2
    assert len(built_app.deployments[1].init_args) == 1
    assert isinstance(built_app.deployments[1].init_args[0], DeploymentHandle)
    assert built_app.deployments[1].init_args[0].deployment_id == DeploymentID(
        "Inner", app_name="app-name"
    )


def test_single_deployment_basic():
    @serve.deployment(
        num_replicas=123,
        max_ongoing_requests=10,
        max_queued_requests=10,
    )
    class D:
        pass

    app = D.bind("hello world!", hello="world")
    _build_and_check(
        app,
        expected_ingress_name="D",
        expected_deployments=[
            D.options(
                name="D", _init_args=("hello world!",), _init_kwargs={"hello": "world"}
            )
        ],
    )


def test_single_deployment_custom_name():
    @serve.deployment(
        num_replicas=123,
        max_ongoing_requests=10,
        max_queued_requests=10,
    )
    class D:
        pass

    app = D.options(name="foobar123").bind("hello world!", hello="world")
    _build_and_check(
        app,
        expected_ingress_name="foobar123",
        expected_deployments=[
            D.options(
                name="foobar123",
                _init_args=("hello world!",),
                _init_kwargs={"hello": "world"},
            )
        ],
    )

    # Change to `with pytest.raises(ValueError)` when the warning is removed.
    with pytest.warns(UserWarning):

        @serve.deployment(name="test#deployment")
        def my_deployment():
            return "Hello!"

    with pytest.warns(UserWarning):

        @serve.deployment()
        def my_deployment():
            return "Hello!"

        my_deployment.options(name="test#deployment")


def test_multi_deployment_basic():
    @serve.deployment(num_replicas=3)
    class Inner:
        pass

    @serve.deployment(num_replicas=1)
    class Outer:
        pass

    app = Outer.bind(Inner.bind(), other=Inner.options(name="Other").bind())
    _build_and_check(
        app,
        expected_ingress_name="Outer",
        expected_deployments=[
            Inner.options(name="Inner", _init_args=tuple(), _init_kwargs={}),
            Inner.options(name="Other", _init_args=tuple(), _init_kwargs={}),
            Outer.options(
                name="Outer",
                _init_args=(
                    FakeDeploymentHandle(
                        "Inner",
                        app_name="default",
                    ),
                ),
                _init_kwargs={
                    "other": FakeDeploymentHandle(
                        "Other",
                        app_name="default",
                    ),
                },
            ),
        ],
    )


def test_multi_deployment_handle_in_nested_obj():
    @serve.deployment(num_replicas=3)
    class Inner:
        pass

    @serve.deployment(num_replicas=1)
    class Outer:
        pass

    app = Outer.bind([Inner.bind()])
    _build_and_check(
        app,
        expected_ingress_name="Outer",
        expected_deployments=[
            Inner.options(name="Inner", _init_args=tuple(), _init_kwargs={}),
            Outer.options(
                name="Outer",
                _init_args=(
                    [
                        FakeDeploymentHandle(
                            "Inner",
                            app_name="default",
                        ),
                    ],
                ),
                _init_kwargs={},
            ),
        ],
    )


def test_multi_deployment_custom_app_name():
    @serve.deployment(num_replicas=3)
    class Inner:
        pass

    @serve.deployment(num_replicas=1)
    class Outer:
        pass

    app = Outer.bind(Inner.bind())
    _build_and_check(
        app,
        app_name="custom",
        expected_ingress_name="Outer",
        expected_deployments=[
            Inner.options(name="Inner", _init_args=tuple(), _init_kwargs={}),
            Outer.options(
                name="Outer",
                _init_args=(
                    FakeDeploymentHandle(
                        "Inner",
                        app_name="custom",
                    ),
                ),
                _init_kwargs={},
            ),
        ],
    )


def test_multi_deployment_name_collision():
    @serve.deployment
    class Inner:
        pass

    @serve.deployment
    class Outer:
        pass

    app = Outer.bind(
        Inner.bind("arg1"),
        Inner.bind("arg2"),
    )
    _build_and_check(
        app,
        expected_ingress_name="Outer",
        expected_deployments=[
            Inner.options(name="Inner", _init_args=("arg1",), _init_kwargs={}),
            Inner.options(name="Inner_1", _init_args=("arg2",), _init_kwargs={}),
            Outer.options(
                name="Outer",
                _init_args=(
                    FakeDeploymentHandle(
                        "Inner",
                        app_name="default",
                    ),
                    FakeDeploymentHandle(
                        "Inner_1",
                        app_name="default",
                    ),
                ),
                _init_kwargs={},
            ),
        ],
    )


def test_multi_deployment_same_app_passed_twice():
    @serve.deployment
    class Shared:
        pass

    @serve.deployment(num_replicas=3)
    class Inner:
        pass

    @serve.deployment(num_replicas=1)
    class Outer:
        pass

    shared = Shared.bind()
    app = Outer.bind(Inner.bind(shared), shared)
    shared_handle = FakeDeploymentHandle(
        "Shared",
        app_name="default",
    )
    _build_and_check(
        app,
        expected_ingress_name="Outer",
        expected_deployments=[
            Shared.options(
                name="Shared",
                _init_args=tuple(),
                _init_kwargs={},
            ),
            Inner.options(name="Inner", _init_args=(shared_handle,), _init_kwargs={}),
            Outer.options(
                name="Outer",
                _init_args=(
                    FakeDeploymentHandle(
                        "Inner",
                        app_name="default",
                    ),
                    shared_handle,
                ),
                _init_kwargs={},
            ),
        ],
    )


def test_default_runtime_env():
    @serve.deployment
    class NoRayActorOptions:
        pass

    @serve.deployment(ray_actor_options={"num_cpus": 0, "num_gpus": 1})
    class NoRuntimeEnv:
        pass

    @serve.deployment(ray_actor_options={"runtime_env": {"env_vars": {"ENV": "VAR"}}})
    class RuntimeEnvNoWorkingDir:
        pass

    @serve.deployment(
        ray_actor_options={"runtime_env": {"working_dir": "s3://test.zip"}}
    )
    class RuntimeEnvWithWorkingDir:
        pass

    @serve.deployment
    class Outer:
        pass

    app = Outer.bind(
        NoRayActorOptions.bind(),
        NoRuntimeEnv.bind(),
        RuntimeEnvNoWorkingDir.bind(),
        RuntimeEnvWithWorkingDir.bind(),
    )

    handles = tuple(
        FakeDeploymentHandle(name, app_name="default")
        for name in [
            "NoRayActorOptions",
            "NoRuntimeEnv",
            "RuntimeEnvNoWorkingDir",
            "RuntimeEnvWithWorkingDir",
        ]
    )

    # 1) Test behavior when no default_runtime_env is passed.
    _build_and_check(
        app,
        expected_ingress_name="Outer",
        expected_deployments=[
            Outer.options(name="Outer", _init_args=handles, _init_kwargs={}),
            NoRayActorOptions.options(
                name="NoRayActorOptions", _init_args=tuple(), _init_kwargs={}
            ),
            NoRuntimeEnv.options(
                name="NoRuntimeEnv", _init_args=tuple(), _init_kwargs={}
            ),
            RuntimeEnvNoWorkingDir.options(
                name="RuntimeEnvNoWorkingDir", _init_args=tuple(), _init_kwargs={}
            ),
            RuntimeEnvWithWorkingDir.options(
                name="RuntimeEnvWithWorkingDir", _init_args=tuple(), _init_kwargs={}
            ),
        ],
    )

    # 2) Test behavior when a default_runtime_env is passed without a working_dir.
    default_runtime_env = {"env_vars": {"DEFAULT": "ENV"}}
    _build_and_check(
        app,
        expected_ingress_name="Outer",
        expected_deployments=[
            Outer.options(
                name="Outer",
                _init_args=handles,
                _init_kwargs={},
                ray_actor_options={"num_cpus": 1, "runtime_env": default_runtime_env},
            ),
            NoRayActorOptions.options(
                name="NoRayActorOptions",
                _init_args=tuple(),
                _init_kwargs={},
                ray_actor_options={"num_cpus": 1, "runtime_env": default_runtime_env},
            ),
            NoRuntimeEnv.options(
                name="NoRuntimeEnv",
                _init_args=tuple(),
                _init_kwargs={},
                ray_actor_options={
                    "num_cpus": 0,
                    "num_gpus": 1,
                    "runtime_env": default_runtime_env,
                },
            ),
            # ray_actor_options shouldn't be affected.
            RuntimeEnvNoWorkingDir.options(
                name="RuntimeEnvNoWorkingDir", _init_args=tuple(), _init_kwargs={}
            ),
            # ray_actor_options shouldn't be affected.
            RuntimeEnvWithWorkingDir.options(
                name="RuntimeEnvWithWorkingDir", _init_args=tuple(), _init_kwargs={}
            ),
        ],
        default_runtime_env=default_runtime_env,
    )

    # 3) Test behavior when a default_runtime_env is passed with a working_dir.
    default_runtime_env = {
        "working_dir": "s3://default.zip",
        "env_vars": {"DEFAULT": "ENV"},
    }
    _build_and_check(
        app,
        expected_ingress_name="Outer",
        expected_deployments=[
            Outer.options(
                name="Outer",
                _init_args=handles,
                _init_kwargs={},
                ray_actor_options={"num_cpus": 1, "runtime_env": default_runtime_env},
            ),
            NoRayActorOptions.options(
                name="NoRayActorOptions",
                _init_args=tuple(),
                _init_kwargs={},
                ray_actor_options={
                    "num_cpus": 1,
                    "runtime_env": default_runtime_env,
                },
            ),
            NoRuntimeEnv.options(
                name="NoRuntimeEnv",
                _init_args=tuple(),
                _init_kwargs={},
                ray_actor_options={
                    "num_cpus": 0,
                    "num_gpus": 1,
                    "runtime_env": default_runtime_env,
                },
            ),
            # Only the working_dir field should be overridden.
            RuntimeEnvNoWorkingDir.options(
                name="RuntimeEnvNoWorkingDir",
                _init_args=tuple(),
                _init_kwargs={},
                ray_actor_options={
                    "num_cpus": 1,
                    "runtime_env": {
                        "working_dir": "s3://default.zip",
                        **RuntimeEnvNoWorkingDir.ray_actor_options["runtime_env"],
                    },
                },
            ),
            # ray_actor_options shouldn't be affected.
            RuntimeEnvWithWorkingDir.options(
                name="RuntimeEnvWithWorkingDir", _init_args=tuple(), _init_kwargs={}
            ),
        ],
        default_runtime_env=default_runtime_env,
    )


def test_external_scaler_enabled():
    """Test that external_scaler_enabled is correctly set in BuiltApplication.

    This test verifies that when build_app is called with external_scaler_enabled=True
    or external_scaler_enabled=False, the resulting BuiltApplication object correctly
    stores the external_scaler_enabled value.
    """

    @serve.deployment
    class D:
        pass

    app = D.bind()

    # Test with external_scaler_enabled=True
    built_app_with_scaler: BuiltApplication = build_app(
        app,
        name="app-with-scaler",
        external_scaler_enabled=True,
    )
    assert built_app_with_scaler.external_scaler_enabled is True
    assert built_app_with_scaler.name == "app-with-scaler"

    # Test with external_scaler_enabled=False (explicit)
    built_app_without_scaler: BuiltApplication = build_app(
        app,
        name="app-without-scaler",
        external_scaler_enabled=False,
    )
    assert built_app_without_scaler.external_scaler_enabled is False
    assert built_app_without_scaler.name == "app-without-scaler"

    # Test with default value (should be False)
    built_app_default: BuiltApplication = build_app(
        app,
        name="app-default",
    )
    assert built_app_default.external_scaler_enabled is False
    assert built_app_default.name == "app-default"


def test_ingress_name_got_modified():
    """Test that the ingress deployment name is modified when there are name
    collisions.
    """

    @serve.deployment
    class D:
        pass

    app = D.bind(D.bind(D.bind()))
    built_app: BuiltApplication = build_app(
        app,
        name="default",
        make_deployment_handle=FakeDeploymentHandle.from_deployment,
    )

    # The ingress deployment name should be modified to avoid collision.
    assert built_app.ingress_deployment_name == "D_2"

    # The ingress deployment should be the last one.
    assert len(built_app.deployments) == 3
    assert built_app.deployments[-1].name == "D_2"


def test_build_app_keeps_ingress_request_router_separate_from_app_deployments(
    monkeypatch,
):
    monkeypatch.setattr("ray.serve._private.build_app.RAY_SERVE_ENABLE_HA_PROXY", True)

    @serve.deployment
    class Ingress:
        pass

    @serve.deployment
    class IngressRequestRouter:
        pass

    ingress_app = Ingress.bind()
    app = ingress_app._with_ingress_request_router(
        IngressRequestRouter.bind(llm_deployment=ingress_app)
    )

    built_app: BuiltApplication = build_app(
        app,
        name="default",
        make_deployment_handle=FakeDeploymentHandle.from_deployment,
    )

    assert [deployment.name for deployment in built_app.deployments] == ["Ingress"]
    assert set(built_app.deployment_handles) == {"Ingress"}
    assert built_app.ingress_request_router_deployment is not None
    assert built_app.ingress_request_router_deployment.name == "IngressRequestRouter"


def test_build_app_requires_ingress_request_router_to_be_single_deployment(
    monkeypatch,
):
    monkeypatch.setattr("ray.serve._private.build_app.RAY_SERVE_ENABLE_HA_PROXY", True)

    @serve.deployment
    class Ingress:
        pass

    @serve.deployment
    class RouterChild:
        pass

    @serve.deployment
    class IngressRequestRouter:
        def __init__(self, child):
            self._child = child

    with pytest.raises(
        ValueError,
        match=(
            "`ingress_request_router` to build into exactly one standalone "
            "deployment"
        ),
    ):
        ingress_app = Ingress.bind()
        app = ingress_app._with_ingress_request_router(
            IngressRequestRouter.bind(RouterChild.bind())
        )
        build_app(
            app,
            name="default",
            make_deployment_handle=FakeDeploymentHandle.from_deployment,
        )


def test_build_app_rejects_ingress_request_router_in_main_app_graph(monkeypatch):
    monkeypatch.setattr("ray.serve._private.build_app.RAY_SERVE_ENABLE_HA_PROXY", True)

    @serve.deployment
    class IngressRequestRouter:
        pass

    @serve.deployment
    class Ingress:
        def __init__(self, router):
            self._router = router

    ingress_request_router = IngressRequestRouter.bind()
    ingress_app = Ingress.bind(ingress_request_router)
    app = ingress_app._with_ingress_request_router(ingress_request_router)

    with pytest.raises(
        ValueError,
        match="did not produce any new deployments",
    ):
        build_app(
            app,
            name="default",
            make_deployment_handle=FakeDeploymentHandle.from_deployment,
        )


def test_ingress_validation_excludes_ingress_request_router_fastapi_app(monkeypatch):
    monkeypatch.setattr("ray.serve._private.build_app.RAY_SERVE_ENABLE_HA_PROXY", True)

    ingress_api = FastAPI()
    router_api = FastAPI()

    @serve.deployment
    @serve.ingress(ingress_api)
    class Ingress:
        pass

    @serve.deployment
    @serve.ingress(router_api)
    class IngressRequestRouter:
        pass

    ingress_app = Ingress.bind()
    app = ingress_app._with_ingress_request_router(
        IngressRequestRouter.bind(llm_deployment=ingress_app)
    )

    built_app: BuiltApplication = build_app(
        app,
        name="default",
        make_deployment_handle=FakeDeploymentHandle.from_deployment,
    )

    client = object.__new__(ServeControllerClient)
    client._check_ingress_deployments([built_app])


def test_callable_uses_multiplexing_static():
    """Detects `@serve.multiplexed` on a method/function statically (class input)."""
    from ray.serve._private.utils import _callable_uses_multiplexing

    class MultiplexedClass:
        @serve.multiplexed(max_num_models_per_replica=2)
        async def load_model(self, model_id: str) -> str:
            return model_id

    @serve.multiplexed
    async def standalone(model_id: str) -> str:
        return model_id

    class Plain:
        async def __call__(self, request) -> str:
            return "ok"

    assert _callable_uses_multiplexing(MultiplexedClass)
    assert _callable_uses_multiplexing(standalone)
    assert not _callable_uses_multiplexing(Plain)


def test_callable_uses_multiplexing_dynamic_instance():
    """Detects multiplexing wired up dynamically at init (e.g. LLMServer LoRA).

    This is only visible on a constructed instance, not on the class statically.
    """
    from ray.serve._private.utils import _callable_uses_multiplexing

    class DynamicMultiplexed:
        def __init__(self):
            async def _load(model_id: str) -> str:
                return model_id

            self._load_model = serve.multiplexed(max_num_models_per_replica=2)(_load)

    # Not detectable on the class (the wrapper only exists after construction)...
    assert not _callable_uses_multiplexing(DynamicMultiplexed)
    # ...but detectable on the constructed instance.
    assert _callable_uses_multiplexing(DynamicMultiplexed())


def test_callable_uses_multiplexing_ignores_handle_attrs():
    """A composed ingress holding DeploymentHandles must not be flagged.

    `DeploymentHandle.__getattr__` returns a (truthy) handle for any attribute name,
    so the instance-attribute scan must check the marker by identity (`is True`) to
    avoid false positives that would break every composed app under direct ingress.
    """
    from ray.serve._private.utils import _callable_uses_multiplexing

    class FakeHandle:
        # Mirrors DeploymentHandle: returns a truthy object for any attribute.
        def __getattr__(self, name):
            return self

    class Ingress:
        def __init__(self):
            self._backend = FakeHandle()

        async def __call__(self, request):
            return await self._backend.remote()

    assert not _callable_uses_multiplexing(Ingress())


@pytest.mark.parametrize(
    "haproxy_enabled, request_router_class, rejected",
    [
        # A custom router on the ingress is rejected only under HAProxy, which
        # load-balances ingress traffic and bypasses the Serve request router.
        (True, RoundRobinRouter, True),
        (False, RoundRobinRouter, False),
        # The default router (left unset) is not custom.
        (True, None, False),
    ],
)
def test_build_app_rejects_only_custom_ingress_request_router_under_haproxy(
    monkeypatch, haproxy_enabled, request_router_class, rejected
):
    """A custom ingress router is rejected only under HAProxy. The default
    router and the no-HAProxy case build."""
    monkeypatch.setattr(
        "ray.serve._private.build_app.RAY_SERVE_ENABLE_HA_PROXY", haproxy_enabled
    )

    options = {}
    if request_router_class is not None:
        options["request_router_config"] = RequestRouterConfig(
            request_router_class=request_router_class
        )

    @serve.deployment
    class Ingress:
        pass

    app = Ingress.options(**options).bind()

    def build():
        return build_app(
            app,
            name="default",
            make_deployment_handle=FakeDeploymentHandle.from_deployment,
        )

    if rejected:
        with pytest.raises(
            RayServeException, match=CUSTOM_INGRESS_REQUEST_ROUTER_UNSUPPORTED_ERROR
        ):
            build()
    else:
        assert [deployment.name for deployment in build().deployments] == ["Ingress"]


def test_build_app_allows_custom_ingress_request_router_in_direct_streaming(
    monkeypatch,
):
    """Direct streaming attaches an ingress_request_router that delegates replica
    selection back to the ingress deployment's router, so a custom router is
    allowed even under HAProxy."""
    monkeypatch.setattr("ray.serve._private.build_app.RAY_SERVE_ENABLE_HA_PROXY", True)

    @serve.deployment(
        request_router_config=RequestRouterConfig(request_router_class=RoundRobinRouter)
    )
    class Ingress:
        pass

    @serve.deployment
    class IngressRequestRouter:
        pass

    ingress_app = Ingress.bind()
    app = ingress_app._with_ingress_request_router(
        IngressRequestRouter.bind(server=ingress_app)
    )

    built_app: BuiltApplication = build_app(
        app,
        name="default",
        make_deployment_handle=FakeDeploymentHandle.from_deployment,
    )

    assert [deployment.name for deployment in built_app.deployments] == ["Ingress"]
    assert built_app.ingress_request_router_deployment is not None


def test_build_app_haproxy_allows_custom_router_on_non_ingress_deployment(monkeypatch):
    """The guard targets only the ingress, so a custom router on a downstream
    deployment is honored under HAProxy."""
    monkeypatch.setattr("ray.serve._private.build_app.RAY_SERVE_ENABLE_HA_PROXY", True)

    @serve.deployment(
        request_router_config=RequestRouterConfig(request_router_class=RoundRobinRouter)
    )
    class Downstream:
        pass

    @serve.deployment
    class Ingress:
        def __init__(self, child):
            pass

    built_app: BuiltApplication = build_app(
        Ingress.bind(Downstream.bind()),
        name="default",
        make_deployment_handle=FakeDeploymentHandle.from_deployment,
    )

    assert {deployment.name for deployment in built_app.deployments} == {
        "Ingress",
        "Downstream",
    }


if __name__ == "__main__":
    sys.exit(pytest.main(["-v", "-s", __file__]))