Merge 528653d4c2 into 0fba05800d

litellm-proxy-extras/litellm_proxy_extras/utils.py

@@ -4,6 +4,8 @@ import random
 import re
 import shutil
 import subprocess
+import sys
+import sysconfig
 import tempfile
 import time
 from pathlib import Path
@@ -27,7 +29,15 @@ def str_to_bool(value: Optional[str]) -> bool:
 
 
 def _get_prisma_env() -> dict:
-    """Get environment variables for Prisma, handling offline mode if configured."""
+    """Get environment variables for Prisma, handling offline mode if configured.
+
+    Also injects the running interpreter's scripts directory (the venv ``bin/``
+    dir) to the front of PATH so the ``prisma-client-py`` console-script is
+    found by ``/bin/sh`` without needing to activate the venv. This fixes the
+    silent failure where ``prisma db push`` and ``prisma migrate deploy`` exit 0
+    but never create tables when invoked by absolute path outside an active venv
+    (Symptom 3 of issue #26097).
+    """
     prisma_env = os.environ.copy()
     if str_to_bool(os.getenv("PRISMA_OFFLINE_MODE")):
         # These env vars prevent Prisma from attempting downloads
@@ -35,6 +45,25 @@ def _get_prisma_env() -> dict:
         prisma_env["NPM_CONFIG_CACHE"] = os.getenv(
             "NPM_CONFIG_CACHE", "/app/.cache/npm"
         )
+
+    # Inject the venv scripts dir so prisma-client-py resolves without
+    # venv activation. Needed for db push and migrate deploy just as much
+    # as for db generate — all three shell out via /bin/sh which only finds
+    # the binary if it is on PATH.
+    scripts_dir = sysconfig.get_path("scripts")
+    if not scripts_dir:
+        scripts_dir = os.path.dirname(os.path.abspath(sys.executable))
+    if scripts_dir:
+        existing_path = prisma_env.get("PATH", "")
+        path_entries = existing_path.split(os.pathsep) if existing_path else []
+        if scripts_dir not in path_entries or path_entries[0] != scripts_dir:
+            # Remove any existing occurrence and place scripts_dir at index 0.
+            filtered = [p for p in path_entries if p != scripts_dir]
+            remaining = os.pathsep.join(filtered)
+            prisma_env["PATH"] = (
+                scripts_dir + os.pathsep + remaining if remaining else scripts_dir
+            )
+
     return prisma_env
 
 
@@ -977,6 +1006,7 @@ class ProxyExtrasDBManager:
                         [_get_prisma_command(), "db", "push", "--accept-data-loss"],
                         timeout=prisma_command_timeout(),
                         check=True,
+                        env=_get_prisma_env(),
                     )
                     return True
             except subprocess.TimeoutExpired:

litellm-proxy-extras/tests/test_prisma_env_path_injection.py

@@ -0,0 +1,155 @@
+"""Tests for _get_prisma_env() venv PATH injection (Symptom 3 of issue #26097).
+
+Symptom 3: after `pip install 'litellm[proxy]'`, running the proxy by absolute
+path in a non-activated venv causes `prisma db push` to exit 0 without creating
+tables because `prisma-client-py` is not on PATH when /bin/sh forks for the
+generator. The fix: _get_prisma_env() prepends the interpreter's scripts dir.
+"""
+
+import os
+from unittest.mock import MagicMock, patch
+
+from litellm_proxy_extras.utils import ProxyExtrasDBManager, _get_prisma_env
+
+# ---------------------------------------------------------------------------
+# Unit tests for _get_prisma_env() PATH injection
+# ---------------------------------------------------------------------------
+
+
+def test_get_prisma_env_injects_scripts_dir_at_index_0(monkeypatch):
+    """Scripts dir appears at PATH index 0 when venv is not activated."""
+    fake_scripts = "/fake/venv/bin"
+    # Remove scripts dir from PATH so it's not already there
+    stripped_path = "/usr/local/bin:/usr/bin:/bin"
+    monkeypatch.setenv("PATH", stripped_path)
+    with (patch("sysconfig.get_path", return_value=fake_scripts),):
+        env = _get_prisma_env()
+
+    path_entries = env["PATH"].split(os.pathsep)
+    assert (
+        path_entries[0] == fake_scripts
+    ), f"Expected scripts dir at index 0, got: {env['PATH']}"
+    # Original PATH entries preserved after the injected dir
+    assert "/usr/local/bin" in path_entries
+    assert "/usr/bin" in path_entries
+
+
+def test_get_prisma_env_promotes_scripts_dir_to_index_0(monkeypatch):
+    """Scripts dir is promoted to index 0 if already in PATH at a later position."""
+    fake_scripts = "/fake/venv/bin"
+    path_with_scripts_at_end = f"/usr/bin:/bin:{fake_scripts}"
+    monkeypatch.setenv("PATH", path_with_scripts_at_end)
+    with patch("sysconfig.get_path", return_value=fake_scripts):
+        env = _get_prisma_env()
+
+    path_entries = env["PATH"].split(os.pathsep)
+    assert path_entries[0] == fake_scripts
+    # Must appear exactly once
+    assert path_entries.count(fake_scripts) == 1
+
+
+def test_get_prisma_env_handles_empty_path(monkeypatch):
+    """Empty PATH results in PATH containing only the scripts dir."""
+    fake_scripts = "/fake/venv/bin"
+    monkeypatch.setenv("PATH", "")
+    with patch("sysconfig.get_path", return_value=fake_scripts):
+        env = _get_prisma_env()
+
+    assert env["PATH"] == fake_scripts
+    assert (
+        os.pathsep not in env["PATH"] or env["PATH"].rstrip(os.pathsep) == fake_scripts
+    )
+
+
+def test_get_prisma_env_falls_back_to_dirname_when_sysconfig_falsy(monkeypatch):
+    """Falls back to dirname(sys.executable) when sysconfig returns falsy."""
+    fake_exe = "/fake/venv/bin/python3"
+    monkeypatch.setenv("PATH", "/usr/bin")
+    with (
+        patch("sysconfig.get_path", return_value=None),
+        patch("sys.executable", fake_exe),
+    ):
+        env = _get_prisma_env()
+
+    expected_scripts = os.path.dirname(os.path.abspath(fake_exe))
+    path_entries = env["PATH"].split(os.pathsep)
+    assert path_entries[0] == expected_scripts
+
+
+def test_get_prisma_env_sets_offline_vars_when_enabled(monkeypatch):
+    """PRISMA_OFFLINE_MODE vars are still set alongside the PATH injection."""
+    monkeypatch.setenv("PRISMA_OFFLINE_MODE", "true")
+    monkeypatch.setenv("PATH", "/usr/bin")
+    with patch("sysconfig.get_path", return_value="/venv/bin"):
+        env = _get_prisma_env()
+
+    assert env.get("NPM_CONFIG_PREFER_OFFLINE") == "true"
+    assert "NPM_CONFIG_CACHE" in env
+    # PATH injection still happens
+    assert env["PATH"].startswith("/venv/bin")
+
+
+def test_get_prisma_env_noop_when_scripts_dir_already_at_index_0(monkeypatch):
+    """No duplicate entry when scripts dir is already at index 0."""
+    fake_scripts = "/fake/venv/bin"
+    monkeypatch.setenv("PATH", f"{fake_scripts}:/usr/bin:/bin")
+    with patch("sysconfig.get_path", return_value=fake_scripts):
+        env = _get_prisma_env()
+
+    path_entries = env["PATH"].split(os.pathsep)
+    assert path_entries[0] == fake_scripts
+    assert path_entries.count(fake_scripts) == 1
+
+
+# ---------------------------------------------------------------------------
+# Integration: db push subprocess receives patched env with scripts_dir
+# ---------------------------------------------------------------------------
+
+
+def test_db_push_subprocess_receives_scripts_dir_on_path(monkeypatch, tmp_path):
+    """setup_database(use_migrate=False) passes env with scripts_dir to subprocess.
+
+    This is the Symptom 3 integration guard: the subprocess call must receive
+    the patched env dict that has the venv scripts dir at PATH index 0. Without
+    it, prisma db push exits 0 but the generator fails silently and tables are
+    never created.
+    """
+    fake_scripts = "/fake/venv/bin"
+    monkeypatch.setenv("PATH", "/usr/bin:/bin")
+    monkeypatch.setenv("DATABASE_URL", "postgresql://u:p@localhost:5432/testdb")
+
+    (tmp_path / "schema.prisma").write_text("// stub schema\n")
+
+    captured_calls = []
+
+    def fake_subprocess_run(args, **kwargs):
+        captured_calls.append({"args": list(args), "env": kwargs.get("env", {})})
+        return MagicMock(returncode=0)
+
+    with (
+        patch("sysconfig.get_path", return_value=fake_scripts),
+        patch.object(
+            ProxyExtrasDBManager, "_get_prisma_dir", return_value=str(tmp_path)
+        ),
+        patch("litellm_proxy_extras.utils._get_prisma_command", return_value="prisma"),
+        patch("subprocess.run", side_effect=fake_subprocess_run),
+    ):
+        result = ProxyExtrasDBManager.setup_database(use_migrate=False)
+
+    assert result is True, "setup_database should return True on success"
+    assert (
+        len(captured_calls) == 1
+    ), f"Expected 1 subprocess call, got {len(captured_calls)}"
+
+    received_env = captured_calls[0]["env"]
+    assert received_env is not None, "subprocess.run must be called with env="
+    path_entries = received_env.get("PATH", "").split(os.pathsep)
+    assert path_entries[0] == fake_scripts, (
+        f"Expected scripts dir '{fake_scripts}' at PATH index 0 in subprocess env. "
+        f"Got PATH: {received_env.get('PATH')}"
+    )
+
+    call_args = captured_calls[0]["args"]
+    assert "db" in call_args
+    assert "push" in call_args
+    assert "--accept-data-loss" in call_args

litellm/proxy/client/cli/commands/db.py

@@ -0,0 +1,139 @@
+"""Database management commands for the LiteLLM proxy CLI."""
+
+import os
+import subprocess
+import sys
+import sysconfig
+from typing import Callable, Optional
+
+import click
+
+# The Prisma schema ships inside the litellm package itself (litellm/proxy/
+# schema.prisma); PrismaManager resolves that directory. This is the same
+# schema `db push` operates on, so `db generate` stays consistent with it.
+from litellm.proxy.db.prisma_client import PrismaManager
+
+# The prisma *binary* discovery and PATH-injection helpers live in
+# litellm-proxy-extras. These are optional dependencies; the ImportError is
+# handled at call time so `litellm` works without the proxy extras installed.
+try:
+    from litellm_proxy_extras.utils import (
+        _get_prisma_command as _imported_prisma_command,
+    )
+    from litellm_proxy_extras.utils import (
+        _get_prisma_env as _imported_prisma_env,
+    )
+
+    _PROXY_EXTRAS_AVAILABLE = True
+except ImportError:
+    _imported_prisma_command = None
+    _imported_prisma_env = None
+    _PROXY_EXTRAS_AVAILABLE = False
+
+# Bind to explicitly typed module-level names. litellm-proxy-extras ships no
+# py.typed marker, so its exports are untyped (Any) at this import boundary;
+# the annotations pin the signatures back down. These are the names the tests
+# patch. The `any-ok` markers acknowledge the unavoidable untyped third-party
+# boundary (the annotation is the concrete type the rest of the file relies on).
+_get_prisma_command: Optional[Callable[[], str]] = (
+    _imported_prisma_command  # any-ok: untyped optional import from litellm-proxy-extras
+)
+_get_prisma_env: Optional[Callable[[], dict[str, str]]] = (
+    _imported_prisma_env  # any-ok: untyped optional import from litellm-proxy-extras
+)
+
+
+def _get_venv_scripts_dir() -> str:
+    """Return the directory that holds console scripts for the running interpreter.
+
+    The prisma engine shells out to the ``prisma-client-py`` console script via
+    ``/bin/sh``, which only finds it when that directory is on PATH. In an
+    *activated* venv it is; when ``litellm-proxy`` is invoked by absolute path
+    without activation, it is not. We derive the directory from the current
+    interpreter so the caller never has to activate the venv.
+
+    ``sysconfig.get_path("scripts")`` is the authoritative answer; we fall back
+    to ``dirname(sys.executable)`` if it is empty (defensive: should not happen
+    on a normal CPython install).
+    """
+    scripts_dir = sysconfig.get_path("scripts")
+    if not scripts_dir:
+        scripts_dir = os.path.dirname(os.path.abspath(sys.executable))
+    return scripts_dir
+
+
+def _get_generate_env() -> dict[str, str]:
+    """Build the subprocess env for ``db generate``.
+
+    Delegates to the shared ``_get_prisma_env()`` which now injects the venv
+    scripts dir for all Prisma subprocesses (generate, db push, migrate deploy).
+    Kept as a named wrapper for backward-compat with existing tests.
+    """
+    if _get_prisma_env is not None:
+        result = _get_prisma_env()
+        if result is not None:
+            return result
+    return os.environ.copy()
+
+
+@click.group()
+def db() -> None:
+    """Database management commands."""
+
+
+@db.command(name="generate")
+def db_generate() -> None:
+    """Generate the Prisma client using litellm's bundled schema.
+
+    Runs: prisma generate --schema <path_to_schema.prisma>
+
+    The schema is resolved from the litellm package (litellm/proxy/schema.prisma)
+    via PrismaManager, which is the same schema `db push` operates on. This keeps
+    generate consistent with the existing push path and closes the gap where there
+    was no out-of-the-box `generate` step after a plain pip install.
+    """
+    if not _PROXY_EXTRAS_AVAILABLE:
+        click.echo(
+            "Error: litellm-proxy-extras is not installed. Run: pip install 'litellm[proxy]'",
+            err=True,
+        )
+        raise SystemExit(1)
+
+    prisma_dir = PrismaManager._get_prisma_dir()
+    schema_path = os.path.join(prisma_dir, "schema.prisma")
+
+    if not os.path.exists(schema_path):
+        click.echo(
+            f"Error: schema.prisma not found at {schema_path}. "
+            "Your litellm-proxy-extras installation may be incomplete.",
+            err=True,
+        )
+        raise SystemExit(1)
+
+    if _get_prisma_command is None:
+        click.echo(
+            "Error: litellm-proxy-extras is not installed. Run: pip install 'litellm[proxy]'",
+            err=True,
+        )
+        raise SystemExit(1)
+
+    click.echo(f"Generating Prisma client from {schema_path} ...")
+    command: list[str] = [
+        _get_prisma_command(),
+        "generate",
+        "--schema",
+        schema_path,
+    ]
+    try:
+        subprocess.run(
+            command,
+            check=True,
+            env=_get_generate_env(),
+        )
+        click.echo("Prisma client generated successfully.")
+    except subprocess.CalledProcessError as e:
+        click.echo(
+            f"Error: prisma generate failed (exit {e.returncode}).",
+            err=True,
+        )
+        raise SystemExit(e.returncode)

litellm/proxy/client/cli/main.py

@@ -14,6 +14,7 @@ from .commands.autoroute.commands import autoroute_group
 from .commands.chat import chat
 from .commands.config import config_commands, get_config_value, hidden_command_names
 from .commands.credentials import credentials
+from .commands.db import db
 from .commands.encryption import encryption
 from .commands.http import http
 from .commands.keys import keys
@@ -131,6 +132,8 @@ def version(ctx: click.Context):
 cli.add_command(login)
 cli.add_command(logout)
 cli.add_command(whoami)
+# Add the db command group
+cli.add_command(db)
 # Add the auth command group (e.g. `lite auth print-token`, used as Claude Code's apiKeyHelper)
 cli.add_command(auth_group, name="auth")
 # Add the models command group

pyproject.toml

@@ -321,6 +321,7 @@ asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "session"
 markers = [
     "asyncio: mark test as an asyncio test",
+    "integration: mark test as an integration test requiring real binaries/processes",
     "limit_leaks: mark test with memory limit for leak detection (e.g., '40 MB')",
     "no_parallel: mark test to run sequentially (not in parallel) - typically for memory measurement tests",
 ]

tests/proxy_cli_integration_tests/README.md

@@ -0,0 +1,33 @@
+# Proxy CLI integration tests
+
+These tests invoke real binaries and subprocesses (for example the installed
+`litellm-proxy` console script) rather than mocks. They live here, outside
+`tests/test_litellm/`, because that tree is reserved for mock-only unit tests
+that must run deterministically in CI without external binaries.
+
+## Running
+
+They are skipped by default. Opt in with an environment variable:
+
+```bash
+LITELLM_RUN_INTEGRATION_TESTS=1 pytest tests/proxy_cli_integration_tests/
+```
+
+You also need the package installed so the console scripts exist on PATH:
+
+```bash
+pip install -e '.[proxy]'
+```
+
+Individual tests additionally `pytest.skip` themselves when a required binary
+(such as `litellm-proxy`) is not present in the active environment, so a partial
+install degrades to a skip rather than a failure.
+
+## Markers
+
+Tests here are tagged with the `integration` marker (registered in
+`pyproject.toml`). To run only integration-marked tests:
+
+```bash
+LITELLM_RUN_INTEGRATION_TESTS=1 pytest -m integration
+```

tests/proxy_cli_integration_tests/test_db_generate_integration.py

@@ -0,0 +1,98 @@
+"""Integration tests for `litellm-proxy db generate` (Symptom 2).
+
+Verifies that invoking the binary by absolute path, without activating the venv,
+still finds the bundled `prisma` script because `_get_generate_env()` prepends
+the interpreter's scripts directory to PATH.
+
+These tests invoke real binaries, so they live outside `tests/test_litellm/`
+(the mock-only unit tree) and are skipped by default. Set
+`LITELLM_RUN_INTEGRATION_TESTS=1` to run them. See the README in this directory
+for details.
+"""
+
+import os
+import stat
+import subprocess
+import sysconfig
+
+import pytest
+
+
+@pytest.fixture
+def non_activated_env():
+    venv_root = sysconfig.get_path("data") or os.path.dirname(
+        sysconfig.get_path("scripts") or ""
+    )
+    env = os.environ.copy()
+    env.pop("VIRTUAL_ENV", None)
+    env.pop("VIRTUAL_ENV_PROMPT", None)
+    path_entries = env.get("PATH", "").split(os.pathsep)
+    env["PATH"] = os.pathsep.join(
+        e for e in path_entries if not e.startswith(venv_root)
+    )
+    return env
+
+
+@pytest.fixture
+def fake_prisma_bin(tmp_path):
+    fake_bin = tmp_path / "fake_bin"
+    fake_bin.mkdir()
+    prisma = fake_bin / "prisma"
+    prisma.write_text('#!/bin/sh\necho "fake-prisma called: $@"\nexit 0\n')
+    prisma.chmod(prisma.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
+    return fake_bin
+
+
+@pytest.mark.integration
+@pytest.mark.skipif(
+    not os.environ.get("LITELLM_RUN_INTEGRATION_TESTS"),
+    reason="set LITELLM_RUN_INTEGRATION_TESTS=1 to run integration tests",
+)
+def test_db_generate_absolute_path_non_activated_venv(non_activated_env):
+    scripts_dir = sysconfig.get_path("scripts")
+    litellm_proxy_bin = os.path.join(scripts_dir, "litellm-proxy")
+    if not os.path.isfile(litellm_proxy_bin):
+        pytest.skip("litellm-proxy console script not installed in venv")
+
+    result = subprocess.run(
+        [litellm_proxy_bin, "db", "generate"],
+        env=non_activated_env,
+        capture_output=True,
+        text=True,
+    )
+
+    assert result.returncode == 0, f"stdout={result.stdout!r}\nstderr={result.stderr!r}"
+    assert "Generating Prisma client" in result.stdout
+    assert "Prisma client generated successfully." in result.stdout
+
+
+@pytest.mark.integration
+@pytest.mark.skipif(
+    not os.environ.get("LITELLM_RUN_INTEGRATION_TESTS"),
+    reason="set LITELLM_RUN_INTEGRATION_TESTS=1 to run integration tests",
+)
+def test_db_generate_succeeds_with_scripts_dir_off_caller_path(non_activated_env):
+    """The fix injects the interpreter scripts dir into PATH internally, so
+    ``db generate`` finds the bundled ``prisma`` script even when the caller's
+    PATH does not contain the venv scripts dir. This is the positive proof of
+    Symptom 2: a stripped caller PATH no longer breaks generation.
+    """
+    scripts_dir = sysconfig.get_path("scripts")
+    litellm_proxy_bin = os.path.join(scripts_dir, "litellm-proxy")
+    if not os.path.isfile(litellm_proxy_bin):
+        pytest.skip("litellm-proxy console script not installed in venv")
+
+    # Caller PATH deliberately excludes the venv scripts dir (and thus prisma).
+    test_env = non_activated_env.copy()
+    test_env["PATH"] = "/usr/bin:/bin"
+
+    result = subprocess.run(
+        [litellm_proxy_bin, "db", "generate"],
+        env=test_env,
+        capture_output=True,
+        text=True,
+    )
+
+    assert result.returncode == 0, f"stdout={result.stdout!r}\nstderr={result.stderr!r}"
+    assert "Generating Prisma client" in result.stdout
+    assert "Prisma client generated successfully." in result.stdout

tests/test_litellm/proxy/client/cli/test_db_commands.py

@@ -0,0 +1,262 @@
+# stdlib imports
+import os
+import subprocess
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# third party imports
+from click.testing import CliRunner
+
+sys.path.insert(0, os.path.abspath("../../.."))
+
+# local imports
+from litellm.proxy.client.cli import cli
+
+
+@pytest.fixture
+def cli_runner():
+    return CliRunner()
+
+
+@pytest.fixture(autouse=True)
+def mock_env():
+    with patch.dict(
+        os.environ,
+        {
+            "LITELLM_PROXY_URL": "http://localhost:4000",
+            "LITELLM_PROXY_API_KEY": "sk-test",
+        },
+    ):
+        yield
+
+
+def test_db_generate_success(cli_runner):
+    mock_run = MagicMock(return_value=MagicMock(returncode=0))
+    with (
+        patch(
+            "litellm.proxy.client.cli.commands.db.PrismaManager._get_prisma_dir",
+            return_value="/fake/prisma/dir",
+        ),
+        patch(
+            "litellm.proxy.client.cli.commands.db._get_prisma_command",
+            return_value="prisma",
+        ),
+        patch(
+            "litellm.proxy.client.cli.commands.db._get_prisma_env", return_value=None
+        ),
+        patch("litellm.proxy.client.cli.commands.db.os.path.exists", return_value=True),
+        patch("litellm.proxy.client.cli.commands.db.subprocess.run", mock_run),
+    ):
+        result = cli_runner.invoke(cli, ["db", "generate"])
+
+    assert result.exit_code == 0, result.output
+    call_args = mock_run.call_args[0][0]
+    assert "prisma" in call_args
+    assert "generate" in call_args
+    assert "--schema" in call_args
+    schema_arg = call_args[call_args.index("--schema") + 1]
+    assert schema_arg.endswith("schema.prisma")
+    assert "Prisma client generated successfully." in result.output
+
+
+def test_db_generate_schema_path_uses_get_prisma_dir(cli_runner):
+    mock_run = MagicMock(return_value=MagicMock(returncode=0))
+    with (
+        patch(
+            "litellm.proxy.client.cli.commands.db.PrismaManager._get_prisma_dir",
+            return_value="/custom/litellm/proxy/dir",
+        ),
+        patch(
+            "litellm.proxy.client.cli.commands.db._get_prisma_command",
+            return_value="prisma",
+        ),
+        patch(
+            "litellm.proxy.client.cli.commands.db._get_prisma_env", return_value=None
+        ),
+        patch("litellm.proxy.client.cli.commands.db.os.path.exists", return_value=True),
+        patch("litellm.proxy.client.cli.commands.db.subprocess.run", mock_run),
+    ):
+        result = cli_runner.invoke(cli, ["db", "generate"])
+
+    assert result.exit_code == 0, result.output
+    call_args = mock_run.call_args[0][0]
+    schema_arg = call_args[call_args.index("--schema") + 1]
+    assert schema_arg == "/custom/litellm/proxy/dir/schema.prisma"
+
+
+def test_db_generate_prisma_failure(cli_runner):
+    mock_run = MagicMock(side_effect=subprocess.CalledProcessError(1, "prisma"))
+    with (
+        patch(
+            "litellm.proxy.client.cli.commands.db.PrismaManager._get_prisma_dir",
+            return_value="/fake/prisma/dir",
+        ),
+        patch(
+            "litellm.proxy.client.cli.commands.db._get_prisma_command",
+            return_value="prisma",
+        ),
+        patch(
+            "litellm.proxy.client.cli.commands.db._get_prisma_env", return_value=None
+        ),
+        patch("litellm.proxy.client.cli.commands.db.os.path.exists", return_value=True),
+        patch("litellm.proxy.client.cli.commands.db.subprocess.run", mock_run),
+    ):
+        result = cli_runner.invoke(cli, ["db", "generate"])
+
+    assert result.exit_code != 0
+    assert "prisma generate failed" in result.output
+
+
+def test_db_generate_schema_missing(cli_runner):
+    with (
+        patch(
+            "litellm.proxy.client.cli.commands.db.PrismaManager._get_prisma_dir",
+            return_value="/fake/prisma/dir",
+        ),
+        patch(
+            "litellm.proxy.client.cli.commands.db._get_prisma_command",
+            return_value="prisma",
+        ),
+        patch(
+            "litellm.proxy.client.cli.commands.db._get_prisma_env", return_value=None
+        ),
+        patch(
+            "litellm.proxy.client.cli.commands.db.os.path.exists", return_value=False
+        ),
+    ):
+        result = cli_runner.invoke(cli, ["db", "generate"])
+
+    assert result.exit_code != 0
+    assert "schema.prisma not found" in result.output
+
+
+def test_db_generate_proxy_extras_not_installed(cli_runner):
+    with patch.dict(
+        sys.modules,
+        {
+            "litellm_proxy_extras": None,
+            "litellm_proxy_extras.utils": None,
+        },
+    ):
+        # Patch the module-level flag directly since the import already happened
+        with patch(
+            "litellm.proxy.client.cli.commands.db._PROXY_EXTRAS_AVAILABLE", False
+        ):
+            result = cli_runner.invoke(cli, ["db", "generate"])
+
+    assert result.exit_code != 0
+    assert "not installed" in result.output
+
+
+def test_db_group_registered_in_cli(cli_runner):
+    result = cli_runner.invoke(cli, ["db", "--help"])
+    assert result.exit_code == 0
+    assert "generate" in result.output
+
+
+def test_db_generate_help(cli_runner):
+    result = cli_runner.invoke(cli, ["db", "generate", "--help"])
+    assert result.exit_code == 0
+    assert any(word in result.output for word in ("Prisma", "schema", "prisma"))
+
+
+def test_db_generate_env_includes_scripts_dir_on_path(cli_runner):
+    """The generate subprocess env must carry the interpreter's scripts dir on
+    PATH so the bundled `prisma-client-py` generator resolves without the caller
+    activating the venv (Symptoms 2 & 3).
+
+    PATH injection is now the responsibility of _get_prisma_env() (not
+    _get_generate_env()), so this test mocks _get_prisma_env to return a
+    realistic env that already has scripts_dir prepended — matching what the
+    real implementation produces — and verifies the env reaches subprocess.run
+    intact."""
+    import sysconfig
+
+    captured = {}
+
+    def _capture(*args, **kwargs):
+        captured["env"] = kwargs.get("env")
+        return MagicMock(returncode=0)
+
+    scripts_dir = sysconfig.get_path("scripts") or os.path.dirname(
+        os.path.abspath(sys.executable)
+    )
+
+    # Simulate what _get_prisma_env() actually returns: scripts_dir prepended,
+    # base system PATH entries preserved after it.
+    injected_env = {"PATH": os.pathsep.join([scripts_dir, "/usr/bin", "/bin"])}
+
+    with (
+        patch(
+            "litellm.proxy.client.cli.commands.db.PrismaManager._get_prisma_dir",
+            return_value="/fake/prisma/dir",
+        ),
+        patch(
+            "litellm.proxy.client.cli.commands.db._get_prisma_command",
+            return_value="prisma",
+        ),
+        # _get_prisma_env already injects scripts_dir — mock returns realistic output
+        patch(
+            "litellm.proxy.client.cli.commands.db._get_prisma_env",
+            return_value=injected_env,
+        ),
+        patch("litellm.proxy.client.cli.commands.db.os.path.exists", return_value=True),
+        patch(
+            "litellm.proxy.client.cli.commands.db.subprocess.run", side_effect=_capture
+        ),
+    ):
+        result = cli_runner.invoke(cli, ["db", "generate"])
+
+    assert result.exit_code == 0, result.output
+    env = captured["env"]
+    assert env is not None, "subprocess.run was called without an explicit env"
+    path_entries = env["PATH"].split(os.pathsep)
+    assert (
+        scripts_dir in path_entries
+    ), f"scripts dir {scripts_dir!r} not on child PATH {env['PATH']!r}"
+    # scripts_dir must be first so it shadows any system-installed prisma binary
+    assert path_entries[0] == scripts_dir
+    # base env entries are preserved, not clobbered
+    assert "/usr/bin" in path_entries
+
+
+def test_db_generate_env_does_not_duplicate_scripts_dir(cli_runner):
+    """If the scripts dir is already on PATH (e.g. an activated venv), the fix
+    must not add a duplicate entry."""
+    import sysconfig
+
+    captured = {}
+
+    def _capture(*args, **kwargs):
+        captured["env"] = kwargs.get("env")
+        return MagicMock(returncode=0)
+
+    scripts_dir = sysconfig.get_path("scripts") or os.path.dirname(
+        os.path.abspath(sys.executable)
+    )
+
+    with (
+        patch(
+            "litellm.proxy.client.cli.commands.db.PrismaManager._get_prisma_dir",
+            return_value="/fake/prisma/dir",
+        ),
+        patch(
+            "litellm.proxy.client.cli.commands.db._get_prisma_command",
+            return_value="prisma",
+        ),
+        patch(
+            "litellm.proxy.client.cli.commands.db._get_prisma_env",
+            return_value={"PATH": scripts_dir + os.pathsep + "/usr/bin"},
+        ),
+        patch("litellm.proxy.client.cli.commands.db.os.path.exists", return_value=True),
+        patch(
+            "litellm.proxy.client.cli.commands.db.subprocess.run", side_effect=_capture
+        ),
+    ):
+        result = cli_runner.invoke(cli, ["db", "generate"])
+
+    assert result.exit_code == 0, result.output
+    path_entries = captured["env"]["PATH"].split(os.pathsep)
+    assert path_entries.count(scripts_dir) == 1

tests/test_litellm/proxy/client/cli/test_db_generate_path_verifier.py

@@ -0,0 +1,203 @@
+# Adversarial verifier tests for litellm-prisma-generate-oob (venv PATH fix)
+import os
+import shutil
+import stat
+import tempfile
+from unittest.mock import patch
+
+import pytest
+
+from litellm.proxy.client.cli.commands.db import (
+    _get_generate_env,
+    _get_venv_scripts_dir,
+)
+
+_DB_MODULE = "litellm.proxy.client.cli.commands.db"
+
+
+# ---------------------------------------------------------------------------
+# _get_venv_scripts_dir unit tests
+# ---------------------------------------------------------------------------
+
+
+def test_get_venv_scripts_dir_returns_sysconfig_path():
+    with patch("sysconfig.get_path", return_value="/home/user/venv/bin") as mock_get:
+        result = _get_venv_scripts_dir()
+    mock_get.assert_called_once_with("scripts")
+    assert result == "/home/user/venv/bin"
+
+
+@pytest.mark.parametrize("falsy_value", [None, ""])
+def test_get_venv_scripts_dir_falls_back_when_sysconfig_falsy(falsy_value):
+    fake_exe = "/home/user/venv/bin/python"
+    with (
+        patch("sysconfig.get_path", return_value=falsy_value),
+        patch("sys.executable", fake_exe),
+    ):
+        result = _get_venv_scripts_dir()
+    assert result == os.path.dirname(os.path.abspath(fake_exe))
+
+
+# ---------------------------------------------------------------------------
+# _get_generate_env unit tests
+# ---------------------------------------------------------------------------
+
+
+def test_get_generate_env_no_path_key_in_base_env():
+    fake_scripts = "/home/user/venv/bin"
+    base_env_with_path = {
+        "SOME_VAR": "value",
+        "PATH": fake_scripts,
+    }
+    with patch(f"{_DB_MODULE}._get_prisma_env", return_value=base_env_with_path):
+        env = _get_generate_env()
+    assert env["PATH"] == fake_scripts
+    assert not env["PATH"].startswith(os.pathsep)
+
+
+def test_get_generate_env_prisma_env_attribute_is_none():
+    with (
+        patch(f"{_DB_MODULE}._get_prisma_env", None),
+        patch.dict(os.environ, {"MY_CUSTOM_VAR": "hello"}, clear=False),
+    ):
+        env = _get_generate_env()
+    assert env.get("MY_CUSTOM_VAR") == "hello"
+    assert isinstance(env, dict)
+
+
+def test_get_generate_env_prisma_env_callable_returns_none():
+    sentinel_var = "_VERIFIER_TEST_SENTINEL_12345"
+    with (
+        patch(f"{_DB_MODULE}._get_prisma_env", return_value=None),
+        patch.dict(os.environ, {sentinel_var: "sentinel_value"}, clear=False),
+    ):
+        env = _get_generate_env()
+    assert env.get(sentinel_var) == "sentinel_value"
+
+
+def test_get_generate_env_preserves_non_path_env_vars():
+    base_env = {
+        "PATH": "/usr/bin:/bin",
+        "DATABASE_URL": "postgres://localhost/db",
+        "NODE_OPTIONS": "--max-old-space-size=512",
+        "NPM_CONFIG_CACHE": "/tmp/npm",
+    }
+    fake_scripts = "/home/user/venv/bin"
+    with (
+        patch(f"{_DB_MODULE}._get_prisma_env", return_value=base_env),
+        patch(f"{_DB_MODULE}._get_venv_scripts_dir", return_value=fake_scripts),
+    ):
+        env = _get_generate_env()
+    assert env["DATABASE_URL"] == "postgres://localhost/db"
+    assert env["NODE_OPTIONS"] == "--max-old-space-size=512"
+    assert env["NPM_CONFIG_CACHE"] == "/tmp/npm"
+
+
+# ---------------------------------------------------------------------------
+# Entrypoint resolvability test
+# ---------------------------------------------------------------------------
+
+
+def test_prisma_client_py_resolvable_via_path():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        fake_exe = os.path.join(tmpdir, "prisma-client-py")
+        with open(fake_exe, "w") as f:
+            f.write("#!/bin/sh\necho fake\n")
+        os.chmod(
+            fake_exe,
+            os.stat(fake_exe).st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH,
+        )
+
+        with patch(
+            f"{_DB_MODULE}._get_prisma_env", return_value={"PATH": f"{tmpdir}:/usr/bin"}
+        ):
+            env = _get_generate_env()
+
+        path_entries = env["PATH"].split(os.pathsep)
+        assert (
+            path_entries[0] == tmpdir
+        ), f"scripts dir must be first on PATH; got {path_entries!r}"
+        found = shutil.which("prisma-client-py", path=env["PATH"])
+        assert found is not None, "prisma-client-py not findable via PATH lookup"
+        assert os.path.dirname(found) == tmpdir
+
+
+# ---------------------------------------------------------------------------
+# Adversarial: scripts_dir already at non-zero index
+#
+# _get_prisma_env() must promote scripts_dir to index 0 even when it appears
+# later in PATH. _get_generate_env() delegates entirely to _get_prisma_env(),
+# so this test verifies the delegation chain preserves the promotion.
+# ---------------------------------------------------------------------------
+
+
+def test_scripts_dir_promoted_to_index_zero_when_at_later_position():
+    fake_scripts = "/home/user/venv/bin"
+    promoted_path = os.pathsep.join(
+        [
+            fake_scripts,
+            "/usr/bin",
+            "/opt/other",
+            "/bin",
+        ]
+    )
+    promoted_env = {"PATH": promoted_path}
+
+    with patch(f"{_DB_MODULE}._get_prisma_env", return_value=promoted_env):
+        env = _get_generate_env()
+
+    path_entries = env["PATH"].split(os.pathsep)
+    assert path_entries[0] == fake_scripts, (
+        f"scripts_dir must be at index 0 in the env returned by _get_generate_env(); "
+        f"got {path_entries!r}. "
+        f"_get_prisma_env() is responsible for the promotion; "
+        f"_get_generate_env() must faithfully delegate."
+    )
+    assert path_entries.count(fake_scripts) == 1, "scripts_dir must not be duplicated"
+
+
+# ---------------------------------------------------------------------------
+# Delegation regression tests (added for Symptom 3 consolidation)
+# After _get_prisma_env() absorbed PATH injection, _get_generate_env() is a
+# thin delegation wrapper. These tests ensure the delegation chain works end-
+# to-end and the None-guard fires correctly when extras are not installed.
+# ---------------------------------------------------------------------------
+
+
+def test_get_generate_env_delegation_chain_returns_scripts_dir_at_index_0():
+    """End-to-end: _get_generate_env() produces PATH with scripts_dir at index 0.
+
+    Validates the full delegation path: _get_generate_env() -> _get_prisma_env()
+    (which now handles PATH injection). This is the regression guard for the
+    Symptom 3 consolidation: if the delegation chain breaks, db generate would
+    regress to the old scoped-only fix and db push would still be unfixed.
+    """
+    fake_scripts = "/fake/venv/bin"
+    base_env = {"PATH": "/usr/bin:/bin"}
+    with (
+        patch(
+            f"{_DB_MODULE}._get_prisma_env",
+            return_value={**base_env, "PATH": f"{fake_scripts}:/usr/bin:/bin"},
+        ),
+    ):
+        env = _get_generate_env()
+
+    path_entries = env["PATH"].split(os.pathsep)
+    assert path_entries[0] == fake_scripts, (
+        f"_get_generate_env() must return env with scripts_dir at PATH index 0. "
+        f"Got: {env.get('PATH')}"
+    )
+
+
+def test_get_generate_env_falls_back_to_os_environ_when_prisma_env_is_none():
+    """Falls back to os.environ.copy() when _get_prisma_env is None (extras not installed).
+
+    In the not-installed case, _PROXY_EXTRAS_AVAILABLE=False and _get_prisma_env
+    is bound to None (not a callable). The wrapper must not raise AttributeError.
+    """
+    with patch(f"{_DB_MODULE}._get_prisma_env", None):
+        env = _get_generate_env()
+
+    # Must return a dict (a copy of os.environ) without raising
+    assert isinstance(env, dict)
+    assert len(env) > 0