mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-10 22:41:41 +00:00
fix(proxy-extras): rebuild indexes left INVALID by a migration deadlock
Two replicas racing prisma migrate deploy can deadlock, and the loser dies mid CREATE INDEX CONCURRENTLY, leaving the index INVALID. The retried migration's IF NOT EXISTS then skips it, so the planner never uses it. After migrations succeed, look for INVALID indexes on LiteLLM tables and have one replica (advisory try-lock) REINDEX INDEX CONCURRENTLY each of them, dropping _ccnew/_ccold leftovers of an interrupted rebuild instead. The repair never blocks startup: a failed rebuild is logged and retried on the next boot. Also encode DATABASE_URL query values with quote instead of quote_plus so options=-c%20... reaches psycopg intact.
This commit is contained in:
mateo-berri
parent
a677242d6f
commit
55c7872496
2 changed files with 359 additions and 7 deletions
|
|
@ -6,18 +6,23 @@ import shutil
|
|||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from typing import TYPE_CHECKING, Final, Optional
|
||||
|
||||
from litellm_proxy_extras._logging import logger
|
||||
from litellm_proxy_extras.replica_identity import (
|
||||
REPLICA_IDENTITY_FULL_ENV_VAR,
|
||||
apply_replica_identity_full,
|
||||
)
|
||||
from litellm_proxy_extras.prisma_toolchain import (
|
||||
ensure_prisma_toolchain,
|
||||
prisma_command_timeout,
|
||||
)
|
||||
from litellm_proxy_extras.replica_identity import (
|
||||
REPLICA_IDENTITY_FULL_ENV_VAR,
|
||||
apply_replica_identity_full,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import psycopg
|
||||
import psycopg.sql
|
||||
|
||||
|
||||
def str_to_bool(value: Optional[str]) -> bool:
|
||||
|
|
@ -40,6 +45,29 @@ def _get_prisma_env() -> dict:
|
|||
|
||||
_MIGRATION_TS_RE = re.compile(r"^(\d{14})_")
|
||||
|
||||
INDEX_REPAIR_ADVISORY_LOCK_KEY: Final = int.from_bytes(b"litellm", "big")
|
||||
_TRANSIENT_INDEX_SUFFIX_RE: Final = re.compile(r"_cc(?:new|old)\d*$")
|
||||
_INVALID_LITELLM_INDEXES_SQL: Final = (
|
||||
"SELECT n.nspname, c.relname, pg_size_pretty(pg_table_size(t.oid)) "
|
||||
"FROM pg_index i "
|
||||
"JOIN pg_class c ON c.oid = i.indexrelid "
|
||||
"JOIN pg_class t ON t.oid = i.indrelid "
|
||||
"JOIN pg_namespace n ON n.oid = t.relnamespace "
|
||||
"WHERE NOT i.indisvalid "
|
||||
" AND c.relkind = 'i' "
|
||||
" AND n.nspname = %s "
|
||||
" AND t.relname LIKE %s "
|
||||
" AND NOT EXISTS (SELECT 1 FROM pg_constraint k WHERE k.conindid = i.indexrelid) "
|
||||
"ORDER BY c.relname"
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _InvalidIndex:
|
||||
schema: str
|
||||
name: str
|
||||
table_size: str
|
||||
|
||||
_SPEND_LOGS_ALTER_RE = re.compile(r'^ALTER\s+TABLE\s+"LiteLLM_SpendLogs"\s', re.IGNORECASE)
|
||||
_SPEND_LOGS_ARTIFACT_DROP_RE = re.compile(
|
||||
r'^DROP\s+TABLE\s+"LiteLLM_SpendLogs_[^"]*"', re.IGNORECASE
|
||||
|
|
@ -557,7 +585,7 @@ class ProxyExtrasDBManager:
|
|||
def _strip_prisma_query_params(url: str) -> str:
|
||||
"""Remove Prisma-specific query params (connection_limit, pool_timeout,
|
||||
schema, etc.) from DATABASE_URL so psycopg can parse it."""
|
||||
from urllib.parse import urlparse, urlunparse, parse_qsl, urlencode
|
||||
from urllib.parse import parse_qsl, quote, urlencode, urlparse, urlunparse
|
||||
|
||||
parsed = urlparse(url)
|
||||
if not parsed.query:
|
||||
|
|
@ -578,7 +606,7 @@ class ProxyExtrasDBManager:
|
|||
"target_session_attrs",
|
||||
}
|
||||
kept = [(k, v) for k, v in parse_qsl(parsed.query) if k in libpq_params]
|
||||
return urlunparse(parsed._replace(query=urlencode(kept)))
|
||||
return urlunparse(parsed._replace(query=urlencode(kept, quote_via=quote)))
|
||||
|
||||
@staticmethod
|
||||
def _warn_if_db_ahead_of_head(migrations_dir: str) -> None:
|
||||
|
|
@ -652,6 +680,91 @@ class ProxyExtrasDBManager:
|
|||
", ".join(sorted_hostile[:5]) + (" ..." if len(sorted_hostile) > 5 else ""),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _invalid_litellm_indexes(
|
||||
conn: "psycopg.Connection[tuple[str, str, str]]", schema: str
|
||||
) -> tuple[_InvalidIndex, ...]:
|
||||
rows: Final = conn.execute(_INVALID_LITELLM_INDEXES_SQL, (schema, "LiteLLM\\_%")).fetchall()
|
||||
return tuple(_InvalidIndex(*row) for row in rows)
|
||||
|
||||
@staticmethod
|
||||
def _index_repair(index: _InvalidIndex) -> tuple["psycopg.sql.Composed", str]:
|
||||
from psycopg import sql
|
||||
|
||||
target: Final = sql.Identifier(index.schema, index.name)
|
||||
if _TRANSIENT_INDEX_SUFFIX_RE.search(index.name):
|
||||
return sql.SQL("DROP INDEX CONCURRENTLY IF EXISTS {}").format(target), "Dropped leftover"
|
||||
return sql.SQL("REINDEX INDEX CONCURRENTLY {}").format(target), "Rebuilt"
|
||||
|
||||
@staticmethod
|
||||
def _repair_index(conn: "psycopg.Connection[tuple[str, str, str]]", index: _InvalidIndex) -> None:
|
||||
import psycopg
|
||||
|
||||
statement, action = ProxyExtrasDBManager._index_repair(index)
|
||||
try:
|
||||
conn.execute(statement)
|
||||
except psycopg.Error as e:
|
||||
logger.warning(
|
||||
"Could not repair invalid index %s.%s, will retry on the next startup. "
|
||||
"If this keeps happening, run `%s` by hand as the index owner. Error: %s",
|
||||
index.schema,
|
||||
index.name,
|
||||
statement.as_string(conn),
|
||||
e,
|
||||
)
|
||||
return
|
||||
logger.info("%s invalid index %s.%s", action, index.schema, index.name)
|
||||
|
||||
@staticmethod
|
||||
def repair_invalid_indexes(lock_timeout: str = "30s") -> bool:
|
||||
"""Rebuild LiteLLM indexes an interrupted CREATE INDEX CONCURRENTLY left
|
||||
INVALID (a migration deadlock between replicas is the usual cause; the
|
||||
retried migration skips them because of IF NOT EXISTS). Never raises:
|
||||
returns True when no invalid index remains, False when the repair was
|
||||
skipped or failed and will be retried on the next startup."""
|
||||
database_url: Final = os.getenv("DATABASE_URL")
|
||||
if not database_url:
|
||||
return False
|
||||
|
||||
try:
|
||||
import psycopg
|
||||
from psycopg import sql
|
||||
except ImportError:
|
||||
logger.warning(
|
||||
"psycopg is not installed; skipping the invalid index check. "
|
||||
"Install the litellm[extra_proxy] extra, which includes psycopg."
|
||||
)
|
||||
return False
|
||||
|
||||
schema: Final = ProxyExtrasDBManager._prisma_schema_param(database_url) or "public"
|
||||
cleaned_url: Final = ProxyExtrasDBManager._strip_prisma_query_params(database_url)
|
||||
try:
|
||||
with psycopg.connect(cleaned_url, connect_timeout=10, autocommit=True) as conn:
|
||||
conn.execute("SET statement_timeout = 0")
|
||||
conn.execute(sql.SQL("SET lock_timeout = {}").format(sql.Literal(lock_timeout)))
|
||||
found: Final = ProxyExtrasDBManager._invalid_litellm_indexes(conn, schema)
|
||||
if not found:
|
||||
return True
|
||||
logger.warning(
|
||||
"Found %d invalid index(es) left by an interrupted CREATE INDEX "
|
||||
"CONCURRENTLY, rebuilding: %s",
|
||||
len(found),
|
||||
", ".join(f"{index.name} (table size {index.table_size})" for index in found),
|
||||
)
|
||||
lock_row: Final = conn.execute(
|
||||
"SELECT pg_try_advisory_lock(%s)", (INDEX_REPAIR_ADVISORY_LOCK_KEY,)
|
||||
).fetchone()
|
||||
if lock_row is None or not lock_row[0]:
|
||||
logger.info("Another replica is already rebuilding the invalid indexes, skipping")
|
||||
return False
|
||||
for index in ProxyExtrasDBManager._invalid_litellm_indexes(conn, schema):
|
||||
ProxyExtrasDBManager._repair_index(conn, index)
|
||||
remaining: Final = ProxyExtrasDBManager._invalid_litellm_indexes(conn, schema)
|
||||
except psycopg.Error as e:
|
||||
logger.warning("Could not check for invalid indexes, will retry on the next startup. Error: %s", e)
|
||||
return False
|
||||
return not remaining
|
||||
|
||||
@staticmethod
|
||||
def _setup_database_v2(use_migrate: bool) -> bool:
|
||||
"""
|
||||
|
|
@ -886,6 +999,7 @@ class ProxyExtrasDBManager:
|
|||
use_migrate=use_migrate, use_v2_resolver=use_v2_resolver
|
||||
)
|
||||
if migrated:
|
||||
ProxyExtrasDBManager.repair_invalid_indexes()
|
||||
ProxyExtrasDBManager.apply_replica_identity_full_if_requested()
|
||||
return migrated
|
||||
|
||||
|
|
|
|||
238
tests/proxy_migration_tests/test_invalid_index_repair.py
Normal file
238
tests/proxy_migration_tests/test_invalid_index_repair.py
Normal file
|
|
@ -0,0 +1,238 @@
|
|||
import os
|
||||
import subprocess
|
||||
import threading
|
||||
import uuid
|
||||
from collections.abc import Iterator, Mapping
|
||||
from pathlib import Path
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from litellm_proxy_extras.utils import INDEX_REPAIR_ADVISORY_LOCK_KEY, ProxyExtrasDBManager
|
||||
|
||||
psycopg = pytest.importorskip("psycopg")
|
||||
|
||||
pytestmark = pytest.mark.timeout(120)
|
||||
|
||||
requires_db: Final = pytest.mark.skipif(
|
||||
"DATABASE_URL" not in os.environ,
|
||||
reason="requires a postgres database (DATABASE_URL)",
|
||||
)
|
||||
|
||||
HEALTH_TABLE: Final = "LiteLLM_HealthCheckTable"
|
||||
HEALTH_INDEX: Final = "LiteLLM_HealthCheckTable_model_id_model_name_checked_at_idx"
|
||||
HEALTH_INDEX_COLUMNS: Final = '"model_id", "model_name", "checked_at" DESC'
|
||||
LOOKALIKE_TABLE: Final = "LiteLLMLookalikeTable"
|
||||
LOOKALIKE_INDEX: Final = "LiteLLMLookalikeTable_id_idx"
|
||||
PARTITIONED_TABLE: Final = "LiteLLM_PartitionedTable"
|
||||
PARTITIONED_INDEX: Final = "LiteLLM_PartitionedTable_id_idx"
|
||||
|
||||
|
||||
def _base_url() -> str:
|
||||
return os.environ["DATABASE_URL"].split("?")[0]
|
||||
|
||||
|
||||
def _index_validity(schema: str) -> Mapping[str, bool]:
|
||||
with psycopg.connect(_base_url(), autocommit=True) as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT c.relname, i.indisvalid FROM pg_index i "
|
||||
"JOIN pg_class c ON c.oid = i.indexrelid "
|
||||
"JOIN pg_namespace n ON n.oid = c.relnamespace "
|
||||
"WHERE n.nspname = %s",
|
||||
(schema,),
|
||||
).fetchall()
|
||||
return MappingProxyType(dict(rows))
|
||||
|
||||
|
||||
def _interrupt_concurrent_build(schema: str, table: str, statement: str) -> None:
|
||||
"""Abort a CONCURRENTLY build while it waits on an older snapshot, the same
|
||||
spot the deadlock loser dies at, so it leaves its index INVALID."""
|
||||
with psycopg.connect(_base_url()) as pin:
|
||||
pin.isolation_level = psycopg.IsolationLevel.REPEATABLE_READ
|
||||
pin.execute(f'SELECT count(*) FROM "{schema}"."{table}"')
|
||||
with psycopg.connect(_base_url(), autocommit=True) as builder:
|
||||
builder.execute("SET statement_timeout = '1s'")
|
||||
with pytest.raises(psycopg.errors.QueryCanceled):
|
||||
builder.execute(statement)
|
||||
|
||||
|
||||
def _leave_invalid_index(schema: str, table: str, index: str, columns: str) -> None:
|
||||
_interrupt_concurrent_build(
|
||||
schema, table, f'CREATE INDEX CONCURRENTLY "{index}" ON "{schema}"."{table}" ({columns})'
|
||||
)
|
||||
|
||||
|
||||
def _leave_invalid_reindex_leftover(schema: str, table: str, index: str) -> None:
|
||||
_interrupt_concurrent_build(schema, table, f'REINDEX INDEX CONCURRENTLY "{schema}"."{index}"')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scratch_schema(monkeypatch: pytest.MonkeyPatch) -> Iterator[str]:
|
||||
schema: Final = f"invalid_index_{uuid.uuid4().hex[:8]}"
|
||||
with psycopg.connect(_base_url(), autocommit=True) as conn:
|
||||
conn.execute(f'CREATE SCHEMA "{schema}"')
|
||||
conn.execute(
|
||||
f'CREATE TABLE "{schema}"."{HEALTH_TABLE}" (model_id TEXT, model_name TEXT, checked_at TIMESTAMPTZ)'
|
||||
)
|
||||
conn.execute(f'CREATE TABLE "{schema}"."{LOOKALIKE_TABLE}" (id TEXT)')
|
||||
|
||||
monkeypatch.setenv("DATABASE_URL", f"{_base_url()}?schema={schema}")
|
||||
yield schema
|
||||
|
||||
with psycopg.connect(_base_url(), autocommit=True) as conn:
|
||||
conn.execute(f'DROP SCHEMA "{schema}" CASCADE')
|
||||
|
||||
|
||||
@requires_db
|
||||
def test_repair_rebuilds_invalid_litellm_indexes_and_leaves_lookalike_tables_alone(scratch_schema: str) -> None:
|
||||
_leave_invalid_index(scratch_schema, HEALTH_TABLE, HEALTH_INDEX, HEALTH_INDEX_COLUMNS)
|
||||
_leave_invalid_index(scratch_schema, LOOKALIKE_TABLE, LOOKALIKE_INDEX, "id")
|
||||
assert _index_validity(scratch_schema) == {HEALTH_INDEX: False, LOOKALIKE_INDEX: False}
|
||||
|
||||
assert ProxyExtrasDBManager.repair_invalid_indexes() is True
|
||||
|
||||
assert _index_validity(scratch_schema) == {HEALTH_INDEX: True, LOOKALIKE_INDEX: False}
|
||||
|
||||
|
||||
@requires_db
|
||||
def test_repair_drops_leftovers_of_interrupted_rebuilds(scratch_schema: str) -> None:
|
||||
_leave_invalid_index(scratch_schema, HEALTH_TABLE, HEALTH_INDEX, HEALTH_INDEX_COLUMNS)
|
||||
_leave_invalid_reindex_leftover(scratch_schema, HEALTH_TABLE, HEALTH_INDEX)
|
||||
_leave_invalid_index(scratch_schema, HEALTH_TABLE, f"{HEALTH_TABLE}_model_id_idx_ccold", '"model_id"')
|
||||
_leave_invalid_index(scratch_schema, HEALTH_TABLE, f"{HEALTH_TABLE}_model_id_idx_ccnew1", '"model_id"')
|
||||
before: Final = _index_validity(scratch_schema)
|
||||
assert len(before) == 4
|
||||
assert set(before.values()) == {False}
|
||||
|
||||
assert ProxyExtrasDBManager.repair_invalid_indexes() is True
|
||||
|
||||
assert _index_validity(scratch_schema) == {HEALTH_INDEX: True}
|
||||
|
||||
|
||||
@requires_db
|
||||
def test_repair_is_a_no_op_when_every_index_is_valid(scratch_schema: str) -> None:
|
||||
with psycopg.connect(_base_url(), autocommit=True) as conn:
|
||||
conn.execute(f'CREATE INDEX "{HEALTH_INDEX}" ON "{scratch_schema}"."{HEALTH_TABLE}" ({HEALTH_INDEX_COLUMNS})')
|
||||
|
||||
assert ProxyExtrasDBManager.repair_invalid_indexes() is True
|
||||
|
||||
assert _index_validity(scratch_schema) == {HEALTH_INDEX: True}
|
||||
|
||||
|
||||
@requires_db
|
||||
def test_repair_leaves_partitioned_parent_indexes_alone(scratch_schema: str) -> None:
|
||||
with psycopg.connect(_base_url(), autocommit=True) as conn:
|
||||
conn.execute(f'CREATE TABLE "{scratch_schema}"."{PARTITIONED_TABLE}" (id INT) PARTITION BY RANGE (id)')
|
||||
conn.execute(
|
||||
f'CREATE TABLE "{scratch_schema}"."{PARTITIONED_TABLE}_p0" '
|
||||
f'PARTITION OF "{scratch_schema}"."{PARTITIONED_TABLE}" FOR VALUES FROM (0) TO (10)'
|
||||
)
|
||||
conn.execute(f'CREATE INDEX "{PARTITIONED_INDEX}" ON ONLY "{scratch_schema}"."{PARTITIONED_TABLE}" (id)')
|
||||
assert _index_validity(scratch_schema) == {PARTITIONED_INDEX: False}
|
||||
|
||||
assert ProxyExtrasDBManager.repair_invalid_indexes() is True
|
||||
|
||||
assert _index_validity(scratch_schema) == {PARTITIONED_INDEX: False}
|
||||
|
||||
|
||||
@requires_db
|
||||
def test_repair_yields_to_the_replica_holding_the_repair_lock(scratch_schema: str) -> None:
|
||||
_leave_invalid_index(scratch_schema, HEALTH_TABLE, HEALTH_INDEX, HEALTH_INDEX_COLUMNS)
|
||||
|
||||
with psycopg.connect(_base_url(), autocommit=True) as other_replica:
|
||||
other_replica.execute("SELECT pg_advisory_lock(%s)", (INDEX_REPAIR_ADVISORY_LOCK_KEY,))
|
||||
assert ProxyExtrasDBManager.repair_invalid_indexes() is False
|
||||
assert _index_validity(scratch_schema) == {HEALTH_INDEX: False}
|
||||
|
||||
assert ProxyExtrasDBManager.repair_invalid_indexes() is True
|
||||
assert _index_validity(scratch_schema) == {HEALTH_INDEX: True}
|
||||
|
||||
|
||||
@requires_db
|
||||
def test_repair_gives_up_on_a_blocked_rebuild_and_finishes_it_on_the_next_startup(scratch_schema: str) -> None:
|
||||
_leave_invalid_index(scratch_schema, HEALTH_TABLE, HEALTH_INDEX, HEALTH_INDEX_COLUMNS)
|
||||
|
||||
with psycopg.connect(_base_url()) as pin:
|
||||
pin.isolation_level = psycopg.IsolationLevel.REPEATABLE_READ
|
||||
pin.execute(f'SELECT count(*) FROM "{scratch_schema}"."{HEALTH_TABLE}"')
|
||||
assert ProxyExtrasDBManager.repair_invalid_indexes(lock_timeout="1s") is False
|
||||
blocked: Final = _index_validity(scratch_schema)
|
||||
assert blocked[HEALTH_INDEX] is False
|
||||
assert [name for name in blocked if name.endswith("_ccnew")]
|
||||
|
||||
assert ProxyExtrasDBManager.repair_invalid_indexes() is True
|
||||
assert _index_validity(scratch_schema) == {HEALTH_INDEX: True}
|
||||
|
||||
|
||||
def _hold_snapshot(schema: str, table: str, pinned: threading.Event, seconds: float) -> None:
|
||||
with psycopg.connect(_base_url()) as pin:
|
||||
pin.isolation_level = psycopg.IsolationLevel.REPEATABLE_READ
|
||||
pin.execute(f'SELECT count(*) FROM "{schema}"."{table}"')
|
||||
pinned.set()
|
||||
pin.execute("SELECT pg_sleep(%s)", (seconds,))
|
||||
|
||||
|
||||
@requires_db
|
||||
def test_repair_outlives_a_statement_timeout_passed_through_database_url_options(
|
||||
scratch_schema: str, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
_leave_invalid_index(scratch_schema, HEALTH_TABLE, HEALTH_INDEX, HEALTH_INDEX_COLUMNS)
|
||||
monkeypatch.setenv("DATABASE_URL", f"{_base_url()}?schema={scratch_schema}&options=-c%20statement_timeout%3D2000")
|
||||
pinned: Final = threading.Event()
|
||||
holder: Final = threading.Thread(target=_hold_snapshot, args=(scratch_schema, HEALTH_TABLE, pinned, 5.0))
|
||||
holder.start()
|
||||
pinned.wait()
|
||||
try:
|
||||
assert ProxyExtrasDBManager.repair_invalid_indexes() is True
|
||||
finally:
|
||||
holder.join()
|
||||
|
||||
assert _index_validity(scratch_schema) == {HEALTH_INDEX: True}
|
||||
|
||||
|
||||
@requires_db
|
||||
def test_repair_defaults_to_the_public_schema(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
table: Final = f"LiteLLM_ScratchTable_{uuid.uuid4().hex[:8]}"
|
||||
index: Final = f"{table}_id_idx"
|
||||
monkeypatch.setenv("DATABASE_URL", _base_url())
|
||||
with psycopg.connect(_base_url(), autocommit=True) as conn:
|
||||
conn.execute(f'CREATE TABLE public."{table}" (id TEXT)')
|
||||
try:
|
||||
_leave_invalid_index("public", table, index, "id")
|
||||
assert _index_validity("public")[index] is False
|
||||
|
||||
assert ProxyExtrasDBManager.repair_invalid_indexes() is True
|
||||
|
||||
assert _index_validity("public")[index] is True
|
||||
finally:
|
||||
with psycopg.connect(_base_url(), autocommit=True) as conn:
|
||||
conn.execute(f'DROP TABLE public."{table}"')
|
||||
|
||||
|
||||
def test_repair_survives_an_unreachable_database(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("DATABASE_URL", "postgresql://u:p@127.0.0.1:9/x?schema=whatever")
|
||||
|
||||
assert ProxyExtrasDBManager.repair_invalid_indexes() is False
|
||||
|
||||
|
||||
class _MigrateDeployApplied:
|
||||
stdout = "Applied migration.\n"
|
||||
stderr = ""
|
||||
returncode = 0
|
||||
|
||||
|
||||
@requires_db
|
||||
@pytest.mark.parametrize("use_v2_resolver", [True, False])
|
||||
def test_setup_database_repairs_the_index_after_a_recovered_deploy(
|
||||
scratch_schema: str, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, use_v2_resolver: bool
|
||||
) -> None:
|
||||
_leave_invalid_index(scratch_schema, HEALTH_TABLE, HEALTH_INDEX, HEALTH_INDEX_COLUMNS)
|
||||
(tmp_path / "schema.prisma").write_text("// stub")
|
||||
monkeypatch.setattr(ProxyExtrasDBManager, "_get_prisma_dir", lambda: str(tmp_path))
|
||||
monkeypatch.setattr(ProxyExtrasDBManager, "_warn_if_db_ahead_of_head", lambda _: None)
|
||||
monkeypatch.setattr(ProxyExtrasDBManager, "_resolve_all_migrations", lambda *_, **__: True)
|
||||
monkeypatch.setattr(subprocess, "run", lambda *_, **__: _MigrateDeployApplied())
|
||||
|
||||
assert ProxyExtrasDBManager.setup_database(use_migrate=True, use_v2_resolver=use_v2_resolver) is True
|
||||
|
||||
assert _index_validity(scratch_schema) == {HEALTH_INDEX: True}
|
||||
Loading…
Add table
Reference in a new issue