1  import json
  2  import tempfile
  3  
  4  import pytest
  5  from litestar.testing import TestClient
  6  from sqlalchemy import create_engine
  7  from sqlalchemy.pool import NullPool
  8  
  9  from evidently._pydantic_compat import BaseModel
 10  from evidently.legacy.core import new_id
 11  from evidently.legacy.utils import NumpyEncoder
 12  from evidently.ui.service.app import create_app
 13  from evidently.ui.service.base import Project
 14  from evidently.ui.service.base import User
 15  from evidently.ui.service.local_service import LocalConfig
 16  from evidently.ui.service.storage.sql.metadata import SQLProjectMetadataStorage
 17  from evidently.ui.service.storage.sql.utils import migrate_database
 18  from evidently.ui.service.type_aliases import ZERO_UUID
 19  
 20  HEADERS = {"Content-Type": "application/json"}
 21  
 22  
 23  @pytest.fixture
 24  def sqlite_engine():
 25      """Create a temporary SQLite database for testing."""
 26      import gc
 27      import os
 28      import sys
 29      import time
 30  
 31      with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
 32          db_path = f.name
 33  
 34      # Use NullPool to avoid connection pooling issues on Windows
 35      engine = create_engine(f"sqlite:///{db_path}", poolclass=NullPool)
 36      migrate_database(f"sqlite:///{db_path}")
 37  
 38      yield engine
 39  
 40      # Close all connections and dispose the engine
 41      engine.dispose(close=True)
 42  
 43      # Force garbage collection to ensure all references are cleared
 44      gc.collect()
 45  
 46      # On Windows, files can't be deleted if they're still open
 47      # Wait longer and retry more times
 48      if sys.platform == "win32":
 49          time.sleep(0.2)
 50  
 51      max_retries = 10
 52      for attempt in range(max_retries):
 53          try:
 54              os.unlink(db_path)
 55              break
 56          except PermissionError:
 57              if attempt < max_retries - 1:
 58                  # Increase wait time with each retry
 59                  time.sleep(0.1 * (attempt + 1))
 60                  # Force another garbage collection
 61                  gc.collect()
 62              else:
 63                  # On Windows, sometimes we need to just skip the cleanup
 64                  # The temp file will be cleaned up by the OS eventually
 65                  if sys.platform == "win32":
 66                      import warnings
 67  
 68                      warnings.warn(
 69                          f"Could not delete SQLite database file {db_path} on Windows. It will be cleaned up by the OS."
 70                      )
 71                  else:
 72                      raise
 73  
 74  
 75  @pytest.fixture
 76  def test_user():
 77      """Create a test user."""
 78      return User(id=ZERO_UUID, name="Test User")
 79  
 80  
 81  @pytest.fixture
 82  def test_project():
 83      """Create a test project."""
 84      return Project(
 85          id=ZERO_UUID,
 86          name="Test Project",
 87          description="A test project",
 88      )
 89  
 90  
 91  @pytest.fixture
 92  def test_project_id():
 93      """Create a test project ID."""
 94      return new_id()
 95  
 96  
 97  @pytest.fixture
 98  def metadata_storage(sqlite_engine):
 99      """Create SQL metadata storage instance."""
100      return SQLProjectMetadataStorage(sqlite_engine)
101  
102  
103  @pytest.fixture
104  def test_client(tmp_path):
105      """Create a test client."""
106      config = LocalConfig()
107      config.storage.path = str(tmp_path)
108      app = create_app(config=config)
109      return TestClient(app=app)
110  
111  
112  @pytest.fixture
113  def mock_project():
114      """Create a mock project."""
115      return Project(name="mock", team_id=None)
116  
117  
118  def _dumps(obj: BaseModel):
119      """Dump object to JSON string."""
120      return json.dumps(obj.dict(), allow_nan=True, cls=NumpyEncoder)