94 lines
3.3 KiB
Python
94 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import threading
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from imagefind import api as api_module
|
|
from imagefind.database import Database
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolated_postgres_database(monkeypatch):
|
|
"""Give every distinct test data directory its own PostgreSQL database.
|
|
|
|
Set ``IMAGEFIND_TEST_POSTGRES_ADMIN_DSN`` to a PostgreSQL/pgvector server
|
|
where the configured user may create and drop databases. Tests which do
|
|
not open ``Database`` remain usable without that environment variable.
|
|
"""
|
|
|
|
admin_dsn = os.environ.get("IMAGEFIND_TEST_POSTGRES_ADMIN_DSN", "").strip()
|
|
if not admin_dsn:
|
|
yield
|
|
return
|
|
|
|
import psycopg
|
|
from psycopg import sql
|
|
from psycopg.conninfo import conninfo_to_dict
|
|
|
|
admin_parameters = conninfo_to_dict(admin_dsn)
|
|
admin_parameters.setdefault("dbname", "postgres")
|
|
configurations: dict[Path, tuple[str, Path]] = {}
|
|
guard = threading.Lock()
|
|
|
|
def postgres_conf(database: Database) -> Path:
|
|
data_dir = database.path.parent.resolve()
|
|
with guard:
|
|
existing = configurations.get(data_dir)
|
|
if existing:
|
|
return existing[1]
|
|
database_name = f"imagefind_test_{uuid.uuid4().hex[:24]}"
|
|
with psycopg.connect(**admin_parameters, autocommit=True) as admin:
|
|
admin.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(database_name)))
|
|
conf_path = data_dir / ".postgres-client.conf"
|
|
conf_path.parent.mkdir(parents=True, exist_ok=True)
|
|
conf_path.write_text(
|
|
"\n".join(
|
|
(
|
|
f"host={admin_parameters.get('host', '127.0.0.1')}",
|
|
f"port={admin_parameters.get('port', '5432')}",
|
|
f"database={database_name}",
|
|
f"username={admin_parameters.get('user', 'postgres')}",
|
|
f"password={admin_parameters.get('password', '')}",
|
|
f"sslmode={admin_parameters.get('sslmode', 'disable')}",
|
|
"",
|
|
)
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
configurations[data_dir] = (database_name, conf_path)
|
|
return conf_path
|
|
|
|
monkeypatch.setattr(Database, "postgres_conf_path", property(postgres_conf))
|
|
try:
|
|
yield
|
|
finally:
|
|
for database_name, _ in configurations.values():
|
|
with psycopg.connect(**admin_parameters, autocommit=True) as admin:
|
|
admin.execute(
|
|
sql.SQL("DROP DATABASE IF EXISTS {} WITH (FORCE)").format(
|
|
sql.Identifier(database_name)
|
|
)
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _python_313_asgi_thread_compatibility(monkeypatch):
|
|
"""Avoid a Python 3.13 sandbox-only ASGI selector deadlock.
|
|
|
|
Production fnOS uses Python 3.12 and keeps blocking API reads in worker
|
|
threads. The repository test sandbox can deadlock when ASGITransport awaits
|
|
``asyncio.to_thread`` on 3.13, so tests execute only this wrapper inline.
|
|
"""
|
|
|
|
if sys.version_info < (3, 13):
|
|
return
|
|
|
|
async def inline(function, /, *args, **kwargs):
|
|
return function(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(api_module, "_background_api", inline)
|