Spaces:
Runtime error
Runtime error
File size: 4,071 Bytes
60444f3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
import pytest
import uuid
from pathlib import Path
import shutil
from qdrant_singleton import QdrantClientSingleton, CURRENT_SCHEMA_VERSION
from qdrant_client.http import models
@pytest.fixture(autouse=True)
def setup_teardown():
"""Setup and teardown for each test"""
# Store original state
original_path = QdrantClientSingleton._storage_path
original_instance = QdrantClientSingleton._instance
# Create temporary storage
temp_path = Path("test_qdrant_data")
QdrantClientSingleton._storage_path = temp_path
QdrantClientSingleton._instance = None
yield
# Cleanup
if QdrantClientSingleton._instance:
QdrantClientSingleton._instance.close()
# Restore original state
QdrantClientSingleton._instance = original_instance
QdrantClientSingleton._storage_path = original_path
# Remove test directory if it exists
if temp_path.exists():
shutil.rmtree(temp_path)
def test_singleton_pattern():
"""Test that get_instance returns the same instance"""
instance1 = QdrantClientSingleton.get_instance()
instance2 = QdrantClientSingleton.get_instance()
assert instance1 is instance2
def test_storage_path_creation():
"""Test that storage path is created if it doesn't exist"""
assert not QdrantClientSingleton._storage_path.exists()
QdrantClientSingleton.get_instance()
assert QdrantClientSingleton._storage_path.exists()
def test_collection_creation():
"""Test collection creation"""
client = QdrantClientSingleton.get_instance()
collection_name = "test_collection"
# Create collection
QdrantClientSingleton.initialize_collection(collection_name)
# Check collection exists
collections = client.get_collections().collections
collection_names = [collection.name for collection in collections]
assert collection_name in collection_names
def test_schema_version_check():
"""Test schema version checking and updating"""
client = QdrantClientSingleton.get_instance()
collection_name = "test_schema_collection"
# Create collection
QdrantClientSingleton.initialize_collection(collection_name)
# Add a point with current schema version
point_id = str(uuid.uuid4())
client.upsert(
collection_name=collection_name,
points=[
models.PointStruct(
id=point_id,
vector=[0.0] * 512, # VECTOR_SIZE
payload={
"path": "test.jpg",
"absolute_path": "/test/test.jpg",
"schema_version": CURRENT_SCHEMA_VERSION,
"indexed_at": 123456789
}
)
]
)
# Verify point was added
search_result = client.scroll(
collection_name=collection_name,
limit=1
)
assert len(search_result[0]) == 1
assert search_result[0][0].id == point_id
assert search_result[0][0].payload["schema_version"] == CURRENT_SCHEMA_VERSION
def test_payload_indexes():
"""Test that payload indexes are created correctly"""
client = QdrantClientSingleton.get_instance()
collection_name = "test_indexes"
# Create collection
QdrantClientSingleton.initialize_collection(collection_name)
# Get collection info
collection_info = client.get_collection(collection_name)
# Check that collection exists and has correct vector size
assert collection_info.config.params.vectors.size == 512
assert collection_info.config.params.vectors.distance == models.Distance.COSINE
def test_empty_collection_schema_check():
"""Test schema check behavior with empty collection"""
client = QdrantClientSingleton.get_instance()
collection_name = "test_empty_collection"
# Create collection
QdrantClientSingleton.initialize_collection(collection_name)
# Verify collection exists
collections = client.get_collections().collections
collection_names = [collection.name for collection in collections]
assert collection_name in collection_names |