diff --git a/litellm-proxy-extras/litellm_proxy_extras/utils.py b/litellm-proxy-extras/litellm_proxy_extras/utils.py index 369b6561931..a89ac431ea5 100644 --- a/litellm-proxy-extras/litellm_proxy_extras/utils.py +++ b/litellm-proxy-extras/litellm_proxy_extras/utils.py @@ -4,6 +4,8 @@ import re import shutil import subprocess +import sys +import sysconfig import tempfile import time from pathlib import Path @@ -19,7 +21,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 @@ -27,6 +37,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 @@ -924,6 +953,7 @@ def setup_database( [_get_prisma_command(), "db", "push", "--accept-data-loss"], timeout=60, check=True, + env=_get_prisma_env(), ) return True except subprocess.TimeoutExpired: diff --git a/litellm-proxy-extras/tests/test_prisma_env_path_injection.py b/litellm-proxy-extras/tests/test_prisma_env_path_injection.py new file mode 100644 index 00000000000..1d17d270eda --- /dev/null +++ b/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 diff --git a/litellm/proxy/client/cli/commands/db.py b/litellm/proxy/client/cli/commands/db.py new file mode 100644 index 00000000000..27233806e99 --- /dev/null +++ b/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, + _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 + + 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) diff --git a/litellm/proxy/client/cli/main.py b/litellm/proxy/client/cli/main.py index b8c483f4b08..398db44bb7c 100644 --- a/litellm/proxy/client/cli/main.py +++ b/litellm/proxy/client/cli/main.py @@ -11,6 +11,7 @@ from .commands.auth import get_stored_api_key, login, logout, whoami from .commands.chat import chat from .commands.credentials import credentials +from .commands.db import db from .commands.http import http from .commands.keys import keys @@ -99,6 +100,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 models command group cli.add_command(models) # Add the credentials command group diff --git a/pyproject.toml b/pyproject.toml index 8b1386aaf87..7e2883a1fd3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -286,6 +286,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", ] diff --git a/tests/proxy_cli_integration_tests/README.md b/tests/proxy_cli_integration_tests/README.md new file mode 100644 index 00000000000..053abbc3419 --- /dev/null +++ b/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 +``` diff --git a/tests/proxy_cli_integration_tests/test_db_generate_integration.py b/tests/proxy_cli_integration_tests/test_db_generate_integration.py new file mode 100644 index 00000000000..9488011b449 --- /dev/null +++ b/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 diff --git a/tests/test_litellm/proxy/client/cli/test_db_commands.py b/tests/test_litellm/proxy/client/cli/test_db_commands.py new file mode 100644 index 00000000000..c93952e6095 --- /dev/null +++ b/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 diff --git a/tests/test_litellm/proxy/client/cli/test_db_generate_path_verifier.py b/tests/test_litellm/proxy/client/cli/test_db_generate_path_verifier.py new file mode 100644 index 00000000000..022fc6bab68 --- /dev/null +++ b/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