14954586403

Committed 11 May 2025 09:49AM UTC coverage: 93.316%. First build

Build # 14954586403

Build Type

Pull #23

github

Committed by

web-flow

Commit Message

Merge 8e09c10d8 into 80aaa5b38

Pull Request Pull Request #23: fix: all failing tests;

Run Details

223 of 231 new or added lines in 6 files covered. (96.54%)

1396 of 1496 relevant lines covered (93.32%)

0.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.11

/tests/services/test_vector_database_service.py

from unittest.mock import MagicMock
import pytest
from langchain_chroma import Chroma
from langchain_core.documents import Document

from app.services.vector_database_service import ChromaDB, get_vector_database
import uuid
from chromadb.errors import DuplicateIDError


def test_chromadb_initialization(monkeypatch, tmp_path):
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )

    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_db"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(
        persist_directory=str(temp_persist_dir)
    )  # Pass temp_persist_dir
    assert vector_db is not None, "ChromaDB instance should be initialized"
    assert isinstance(vector_db, ChromaDB), "Expected an instance of ChromaDB class"

    # Ensure the database can be initialized before attempting to delete
    db_client = vector_db._get_db()
    assert db_client is not None, "Chroma client should be initialized"

    vector_db._delete()  # Clean up the database after test


def test_chromadb_initialization_with_persist_directory(tmp_path, monkeypatch):
    # Mock the environment variable for ChromaDB with a custom persist directory
    temp_dir = tmp_path / "hihi"
    temp_dir.mkdir()
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    vector_db = ChromaDB(str(temp_dir))
    assert vector_db is not None, "ChromaDB instance should be initialized"
    assert vector_db.persist_directory == str(
        temp_dir
    ), "Persist directory should be '{{str(temp_dir)}}'"
    assert isinstance(vector_db, ChromaDB), "Expected an instance of ChromaDB class"
    vector_db._delete()  # Clean up the database after test


def test_chromadb_get_db(monkeypatch, tmp_path):
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_db_get"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(
        persist_directory=str(temp_persist_dir)
    )  # Pass temp_persist_dir
    db_instance = vector_db._get_db()
    db_instance2 = vector_db._get_db()
    assert db_instance is not None, "ChromaDB instance should be initialized"
    assert isinstance(db_instance, Chroma), "Expected an instance of ChromaDB class"
    assert (
        db_instance is db_instance2
    ), "Should return the same instance on subsequent calls"
    vector_db._delete()  # Clean up the database after test


def test_chromadb_generate_document_ids(tmp_path):
    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_generate_ids"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(
        persist_directory=str(temp_persist_dir)
    )  # Pass temp_persist_dir
    documents = []
    for i in range(5):
        doc = Document(
            page_content="Test content", metadata={"source": f"test_source_{i}"}
        )
        documents.append(doc)
    document_ids = vector_db._generate_document_ids(documents)
    assert len(document_ids) == len(
        documents
    ), "Should generate the same number of IDs as documents"
    for i, doc in enumerate(documents):
        assert document_ids[i] == str(
            uuid.uuid3(uuid.NAMESPACE_DNS, doc.page_content)
        ), f"ID for document {i} should match"

    vector_db._delete()  # Clean up the database after test


def test_chromadb_add_documents(monkeypatch, tmp_path):
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_add_documents"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(
        persist_directory=str(temp_persist_dir)
    )  # Pass temp_persist_dir
    documents = []
    for i in range(2):
        doc = Document(
            page_content="Test content " + str(i),
            metadata={"source": f"test_source_{i}"},
        )
        documents.append(doc)
    vector_db.add_documents(documents)
    db_instance = vector_db._get_db()
    assert db_instance is not None, "ChromaDB instance should be initialized"
    assert vector_db.count() == len(
        documents
    ), "Should add the same number of documents to the database"
    vector_db._delete()  # Clean up the database after test


def test_chromadb_add_documents_empty(monkeypatch, tmp_path):  # Add tmp_path fixture
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_add_empty"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(
        persist_directory=str(temp_persist_dir)
    )  # Pass temp_persist_dir

    vector_db.add_documents([])
    db_instance = vector_db._get_db()
    assert db_instance is not None, "ChromaDB instance should be initialized"
    assert vector_db.count() == 0, "Should not add any documents to the database"
    vector_db._delete()  # Clean up the database after test


def test_chromadb_add_documents_error(monkeypatch):
    # Mock the environment variable for ChromaDB
    mock_get_db = MagicMock(side_effect=Exception("Database error"))

    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    vector_db = ChromaDB()
    monkeypatch.setattr(vector_db, "_get_db", mock_get_db)
    documents = ["123", "222"]

    with pytest.raises(Exception) as excinfo:
        vector_db.add_documents(documents)


def test_chromadb_add_duplicate_document(monkeypatch, tmp_path):
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_add_duplicate"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(
        persist_directory=str(temp_persist_dir)
    )  # Pass temp_persist_dir

    documents = []
    for i in range(5):
        doc = Document(
            page_content="Test content " + str(i),
            metadata={"source": f"test_source_{i}"},
        )
        documents.append(doc)
    vector_db.add_documents(documents)
    db_instance = vector_db._get_db()
    assert db_instance is not None, "ChromaDB instance should be initialized"
    with pytest.raises(DuplicateIDError):
        vector_db.add_documents([documents[0]])
    vector_db._delete()  # Clean up the database after test
    pass


def test_chromadb_delete_documents(monkeypatch, tmp_path):
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_delete_documents"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(
        persist_directory=str(temp_persist_dir)
    )  # Pass temp_persist_dir

    documents = []
    for i in range(5):
        doc = Document(
            page_content="Test content " + str(i),
            metadata={"source": f"test_source_{i}"},
        )
        documents.append(doc)
    vector_db.add_documents(documents)
    db_instance = vector_db._get_db()
    assert vector_db.count() == len(
        documents
    ), "Should add the same number of documents to the database"
    vector_db._delete()  # Clean up the database after test
    assert vector_db.count() == 0, "Should delete all documents from the database"


def test_chromadb_delete_documents_exception(monkeypatch):
    # Mock the environment variable for ChromaDB
    def mock_get_db(self):
        raise Exception("Database error")

    monkeypatch.setattr(
        "app.services.vector_database_service.ChromaDB._get_db", mock_get_db
    )
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    vector_db = ChromaDB()

    with pytest.raises(Exception) as excinfo:
        vector_db.delete_document("test_source_1")


def test_chromadb_delete_documents_empty(monkeypatch, tmp_path):
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_delete_empty"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(
        persist_directory=str(temp_persist_dir)
    )  # Initialize with temp_persist_dir

    # With a fresh temporary database, the count should initially be 0
    assert vector_db.count() == 0, "Database should be empty at the start of the test."

    vector_db.delete_document("")
    assert (
        vector_db.count() == 0
    ), "Database should remain empty after calling delete_document with an empty path on an initially empty database."

    vector_db._delete()  # Clean up the database after test


def test_chromadb_search_context(monkeypatch, tmp_path):
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_search_context"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(
        persist_directory=str(temp_persist_dir)
    )  # Pass temp_persist_dir

    documents = []
    for i in range(5):
        doc = Document(
            page_content="Test content " + str(i),
            metadata={"source": f"test_source_{i}"},
        )
        documents.append(doc)
    vector_db.add_documents(documents)
    results = vector_db.search_context("Test content")
    assert len(results) > 0, "Should return search results"
    assert isinstance(
        results[0], Document
    ), "Search result should be a Document instance"
    vector_db._delete()  # Clean up the database after test


def test_chromadb_search_context_none(monkeypatch):
    # Mock the environment variable for ChromaDB
    def mock_get_db(self):
        raise Exception("Database error")

    monkeypatch.setattr(
        "app.services.vector_database_service.ChromaDB._get_db", mock_get_db
    )
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    vector_db = ChromaDB()

    result = vector_db.search_context("test_source_1")
    print("result", result)
    assert (
        result == []
    ), "Should not return any results when the database is not available"


def test_chromadb_search_context_empty(monkeypatch, tmp_path):
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_search_empty"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(
        persist_directory=str(temp_persist_dir)
    )  # Initialize with temp_persist_dir

    results = vector_db.search_context("Non-existent content")
    assert (
        len(results) == 0
    ), "Should return no search results for non-existent content in an empty database"
    vector_db._delete()  # Clean up the database after test


def test_chromadb_is_empty(monkeypatch, tmp_path):
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_is_empty"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(persist_directory=str(temp_persist_dir))
    assert vector_db.is_empty() is True, "Database should be empty initially"
    documents = []
    for i in range(5):
        doc = Document(
            page_content="Test content " + str(i),
            metadata={"source": f"test_source_{i}"},
        )
        documents.append(doc)
    vector_db.add_documents(documents)
    assert (
        vector_db.is_empty() is False
    ), "Database should not be empty after adding documents"
    vector_db._delete()  # Clean up the database after test


def test_chromadb_count(monkeypatch, tmp_path):
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_count"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(
        persist_directory=str(temp_persist_dir)
    )  # Initialize with temp_persist_dir
    documents = []
    for i in range(5):
        doc = Document(
            page_content="Test content" + str(i),
            metadata={"source": f"test_source_{i}"},
        )
        documents.append(doc)
    vector_db.add_documents(documents)
    count = vector_db.count()
    assert count == len(
        documents
    ), "Should return the same number of documents in the collection"
    vector_db._delete()  # Clean up the database after test


def test_chromadb_delete(monkeypatch, tmp_path):
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_delete"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(
        persist_directory=str(temp_persist_dir)
    )  # Initialize with temp_persist_dir
    documents = []
    for i in range(5):
        doc = Document(
            page_content="Test content" + str(i),
            metadata={"source": f"test_source_{i}"},
        )
        documents.append(doc)
    vector_db.add_documents(documents)
    db_instance = vector_db._get_db()
    assert vector_db.count() == len(
        documents
    ), "Should add the same number of documents to the database"
    vector_db._delete()  # Clean up the database after test
    assert vector_db.is_empty() is True, "Database should be empty after deletion"


def test_get_vector_database_chromadb(monkeypatch, tmp_path):
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )
    # Create a temporary directory for ChromaDB persistence for this test
    temp_persist_dir = tmp_path / "chroma_test_get_vector_db"
    temp_persist_dir.mkdir()
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_DIRECTORY",
        str(temp_persist_dir),
    )

    vector_db = get_vector_database()
    assert isinstance(vector_db, ChromaDB), "Expected an instance of ChromaDB class"
    # Ensure the _delete operation can write to the temporary directory
    try:
        vector_db._delete()  # Clean up the database after test
    except Exception as e:
        pytest.fail(f"_delete failed with temporary directory: {e}")


def test_get_vector_database_invalid_provider(monkeypatch):
    # Mock the environment variable for an invalid provider
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER",
        "invalid_provider",
    )
    with pytest.raises(ValueError):
        get_vector_database()


def test_get_collection_count_with_client_none(monkeypatch, tmp_path):
    # Mock the environment variable for ChromaDB
    monkeypatch.setattr(
        "app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
    )

    # Create a temporary directory for ChromaDB persistence
    temp_persist_dir = tmp_path / "chroma_test_collection_count"
    temp_persist_dir.mkdir()

    vector_db = ChromaDB(persist_directory=str(temp_persist_dir)) # Initialize with temp_persist_dir
    
    fake_db_client = MagicMock()
    # Simulate that the collection is not available or accessible on the client
    fake_db_client._collection = None 

    monkeypatch.setattr(vector_db, "_db", fake_db_client) # vector_db._get_db() will now return fake_db_client

    count = vector_db._get_collection_count()
    assert count == 0, "Should return 0 when collection is not available via the client"

1	from unittest.mock import MagicMock	1✔
2	import pytest	1✔
3	from langchain_chroma import Chroma	1✔
4	from langchain_core.documents import Document	1✔
5
6	from app.services.vector_database_service import ChromaDB, get_vector_database	1✔
7	import uuid	1✔
8	from chromadb.errors import DuplicateIDError	1✔
9
10
11	def test_chromadb_initialization(monkeypatch, tmp_path):	1✔
12	# Mock the environment variable for ChromaDB
13	monkeypatch.setattr(	1✔
14	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
15	)
16
17	# Create a temporary directory for ChromaDB persistence
18	temp_persist_dir = tmp_path / "chroma_test_db"	1✔
19	temp_persist_dir.mkdir()	1✔
20
21	vector_db = ChromaDB(	1✔
22	persist_directory=str(temp_persist_dir)
23	) # Pass temp_persist_dir
24	assert vector_db is not None, "ChromaDB instance should be initialized"	1✔
25	assert isinstance(vector_db, ChromaDB), "Expected an instance of ChromaDB class"	1✔
26
27	# Ensure the database can be initialized before attempting to delete
28	db_client = vector_db._get_db()	1✔
29	assert db_client is not None, "Chroma client should be initialized"	1✔
30
31	vector_db._delete() # Clean up the database after test	1✔
32
33
34	def test_chromadb_initialization_with_persist_directory(tmp_path, monkeypatch):	1✔
35	# Mock the environment variable for ChromaDB with a custom persist directory
36	temp_dir = tmp_path / "hihi"	1✔
37	temp_dir.mkdir()	1✔
38	monkeypatch.setattr(	1✔
39	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
40	)
41	vector_db = ChromaDB(str(temp_dir))	1✔
42	assert vector_db is not None, "ChromaDB instance should be initialized"	1✔
43	assert vector_db.persist_directory == str(	1✔
44	temp_dir
45	), "Persist directory should be '{{str(temp_dir)}}'"
46	assert isinstance(vector_db, ChromaDB), "Expected an instance of ChromaDB class"	1✔
47	vector_db._delete() # Clean up the database after test	1✔
48
49
50	def test_chromadb_get_db(monkeypatch, tmp_path):	1✔
51	# Mock the environment variable for ChromaDB
52	monkeypatch.setattr(	1✔
53	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
54	)
55	# Create a temporary directory for ChromaDB persistence
56	temp_persist_dir = tmp_path / "chroma_test_db_get"	1✔
57	temp_persist_dir.mkdir()	1✔
58
59	vector_db = ChromaDB(	1✔
60	persist_directory=str(temp_persist_dir)
61	) # Pass temp_persist_dir
62	db_instance = vector_db._get_db()	1✔
63	db_instance2 = vector_db._get_db()	1✔
64	assert db_instance is not None, "ChromaDB instance should be initialized"	1✔
65	assert isinstance(db_instance, Chroma), "Expected an instance of ChromaDB class"	1✔
66	assert (	1✔
67	db_instance is db_instance2
68	), "Should return the same instance on subsequent calls"
69	vector_db._delete() # Clean up the database after test	1✔
70
71
72	def test_chromadb_generate_document_ids(tmp_path):	1✔
73	# Create a temporary directory for ChromaDB persistence
74	temp_persist_dir = tmp_path / "chroma_test_generate_ids"	1✔
75	temp_persist_dir.mkdir()	1✔
76
77	vector_db = ChromaDB(	1✔
78	persist_directory=str(temp_persist_dir)
79	) # Pass temp_persist_dir
80	documents = []	1✔
81	for i in range(5):	1✔
82	doc = Document(	1✔
83	page_content="Test content", metadata={"source": f"test_source_{i}"}
84	)
85	documents.append(doc)	1✔
86	document_ids = vector_db._generate_document_ids(documents)	1✔
87	assert len(document_ids) == len(	1✔
88	documents
89	), "Should generate the same number of IDs as documents"
90	for i, doc in enumerate(documents):	1✔
91	assert document_ids[i] == str(	1✔
92	uuid.uuid3(uuid.NAMESPACE_DNS, doc.page_content)
93	), f"ID for document {i} should match"
94
95	vector_db._delete() # Clean up the database after test	1✔
96
97
98	def test_chromadb_add_documents(monkeypatch, tmp_path):	1✔
99	# Mock the environment variable for ChromaDB
100	monkeypatch.setattr(	1✔
101	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
102	)
103	# Create a temporary directory for ChromaDB persistence
104	temp_persist_dir = tmp_path / "chroma_test_add_documents"	1✔
105	temp_persist_dir.mkdir()	1✔
106
107	vector_db = ChromaDB(	1✔
108	persist_directory=str(temp_persist_dir)
109	) # Pass temp_persist_dir
110	documents = []	1✔
111	for i in range(2):	1✔
112	doc = Document(	1✔
113	page_content="Test content " + str(i),
114	metadata={"source": f"test_source_{i}"},
115	)
116	documents.append(doc)	1✔
117	vector_db.add_documents(documents)	1✔
118	db_instance = vector_db._get_db()	1✔
119	assert db_instance is not None, "ChromaDB instance should be initialized"	1✔
120	assert vector_db.count() == len(	1✔
121	documents
122	), "Should add the same number of documents to the database"
123	vector_db._delete() # Clean up the database after test	1✔
124
125
126	def test_chromadb_add_documents_empty(monkeypatch, tmp_path): # Add tmp_path fixture	1✔
127	# Mock the environment variable for ChromaDB
128	monkeypatch.setattr(	1✔
129	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
130	)
131	# Create a temporary directory for ChromaDB persistence
132	temp_persist_dir = tmp_path / "chroma_test_add_empty"	1✔
133	temp_persist_dir.mkdir()	1✔
134
135	vector_db = ChromaDB(	1✔
136	persist_directory=str(temp_persist_dir)
137	) # Pass temp_persist_dir
138
139	vector_db.add_documents([])	1✔
140	db_instance = vector_db._get_db()	1✔
141	assert db_instance is not None, "ChromaDB instance should be initialized"	1✔
142	assert vector_db.count() == 0, "Should not add any documents to the database"	1✔
143	vector_db._delete() # Clean up the database after test	1✔
144
145
146	def test_chromadb_add_documents_error(monkeypatch):	1✔
147	# Mock the environment variable for ChromaDB
148	mock_get_db = MagicMock(side_effect=Exception("Database error"))	1✔
149
150	monkeypatch.setattr(	1✔
151	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
152	)
153	vector_db = ChromaDB()	1✔
154	monkeypatch.setattr(vector_db, "_get_db", mock_get_db)	1✔
155	documents = ["123", "222"]	1✔
156
157	with pytest.raises(Exception) as excinfo:	1✔
158	vector_db.add_documents(documents)	1✔
159
160
161	def test_chromadb_add_duplicate_document(monkeypatch, tmp_path):	1✔
162	# Mock the environment variable for ChromaDB
163	monkeypatch.setattr(	1✔
164	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
165	)
166	# Create a temporary directory for ChromaDB persistence
167	temp_persist_dir = tmp_path / "chroma_test_add_duplicate"	1✔
168	temp_persist_dir.mkdir()	1✔
169
170	vector_db = ChromaDB(	1✔
171	persist_directory=str(temp_persist_dir)
172	) # Pass temp_persist_dir
173
174	documents = []	1✔
175	for i in range(5):	1✔
176	doc = Document(	1✔
177	page_content="Test content " + str(i),
178	metadata={"source": f"test_source_{i}"},
179	)
180	documents.append(doc)	1✔
181	vector_db.add_documents(documents)	1✔
182	db_instance = vector_db._get_db()	1✔
183	assert db_instance is not None, "ChromaDB instance should be initialized"	1✔
184	with pytest.raises(DuplicateIDError):	1✔
185	vector_db.add_documents([documents[0]])	1✔
186	vector_db._delete() # Clean up the database after test	1✔
187	pass	1✔
188
189
190	def test_chromadb_delete_documents(monkeypatch, tmp_path):	1✔
191	# Mock the environment variable for ChromaDB
192	monkeypatch.setattr(	1✔
193	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
194	)
195	# Create a temporary directory for ChromaDB persistence
196	temp_persist_dir = tmp_path / "chroma_test_delete_documents"	1✔
197	temp_persist_dir.mkdir()	1✔
198
199	vector_db = ChromaDB(	1✔
200	persist_directory=str(temp_persist_dir)
201	) # Pass temp_persist_dir
202
203	documents = []	1✔
204	for i in range(5):	1✔
205	doc = Document(	1✔
206	page_content="Test content " + str(i),
207	metadata={"source": f"test_source_{i}"},
208	)
209	documents.append(doc)	1✔
210	vector_db.add_documents(documents)	1✔
211	db_instance = vector_db._get_db()	1✔
212	assert vector_db.count() == len(	1✔
213	documents
214	), "Should add the same number of documents to the database"
215	vector_db._delete() # Clean up the database after test	1✔
216	assert vector_db.count() == 0, "Should delete all documents from the database"	1✔
217
218
219	def test_chromadb_delete_documents_exception(monkeypatch):	1✔
220	# Mock the environment variable for ChromaDB
221	def mock_get_db(self):	1✔
222	raise Exception("Database error")	1✔
223
224	monkeypatch.setattr(	1✔
225	"app.services.vector_database_service.ChromaDB._get_db", mock_get_db
226	)
227	monkeypatch.setattr(	1✔
228	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
229	)
230	vector_db = ChromaDB()	1✔
231
232	with pytest.raises(Exception) as excinfo:	1✔
233	vector_db.delete_document("test_source_1")	1✔
234
235
236	def test_chromadb_delete_documents_empty(monkeypatch, tmp_path):	1✔
237	# Mock the environment variable for ChromaDB
238	monkeypatch.setattr(	1✔
239	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
240	)
241	# Create a temporary directory for ChromaDB persistence
242	temp_persist_dir = tmp_path / "chroma_test_delete_empty"	1✔
243	temp_persist_dir.mkdir()	1✔
244
245	vector_db = ChromaDB(	1✔
246	persist_directory=str(temp_persist_dir)
247	) # Initialize with temp_persist_dir
248
249	# With a fresh temporary database, the count should initially be 0
250	assert vector_db.count() == 0, "Database should be empty at the start of the test."	1✔
251
252	vector_db.delete_document("")	1✔
253	assert (	1✔
254	vector_db.count() == 0
255	), "Database should remain empty after calling delete_document with an empty path on an initially empty database."
256
257	vector_db._delete() # Clean up the database after test	1✔
258
259
260	def test_chromadb_search_context(monkeypatch, tmp_path):	1✔
261	# Mock the environment variable for ChromaDB
262	monkeypatch.setattr(	1✔
263	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
264	)
265	# Create a temporary directory for ChromaDB persistence
266	temp_persist_dir = tmp_path / "chroma_test_search_context"	1✔
267	temp_persist_dir.mkdir()	1✔
268
269	vector_db = ChromaDB(	1✔
270	persist_directory=str(temp_persist_dir)
271	) # Pass temp_persist_dir
272
273	documents = []	1✔
274	for i in range(5):	1✔
275	doc = Document(	1✔
276	page_content="Test content " + str(i),
277	metadata={"source": f"test_source_{i}"},
278	)
279	documents.append(doc)	1✔
280	vector_db.add_documents(documents)	1✔
281	results = vector_db.search_context("Test content")	1✔
282	assert len(results) > 0, "Should return search results"	1✔
283	assert isinstance(	1✔
284	results[0], Document
285	), "Search result should be a Document instance"
286	vector_db._delete() # Clean up the database after test	1✔
287
288
289	def test_chromadb_search_context_none(monkeypatch):	1✔
290	# Mock the environment variable for ChromaDB
291	def mock_get_db(self):	1✔
292	raise Exception("Database error")	1✔
293
294	monkeypatch.setattr(	1✔
295	"app.services.vector_database_service.ChromaDB._get_db", mock_get_db
296	)
297	monkeypatch.setattr(	1✔
298	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
299	)
300	vector_db = ChromaDB()	1✔
301
302	result = vector_db.search_context("test_source_1")	1✔
303	print("result", result)	1✔
304	assert (	1✔
305	result == []
306	), "Should not return any results when the database is not available"
307
308
309	def test_chromadb_search_context_empty(monkeypatch, tmp_path):	1✔
310	# Mock the environment variable for ChromaDB
311	monkeypatch.setattr(	1✔
312	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
313	)
314	# Create a temporary directory for ChromaDB persistence
315	temp_persist_dir = tmp_path / "chroma_test_search_empty"	1✔
316	temp_persist_dir.mkdir()	1✔
317
318	vector_db = ChromaDB(	1✔
319	persist_directory=str(temp_persist_dir)
320	) # Initialize with temp_persist_dir
321
322	results = vector_db.search_context("Non-existent content")	1✔
323	assert (	1✔
324	len(results) == 0
325	), "Should return no search results for non-existent content in an empty database"
326	vector_db._delete() # Clean up the database after test	1✔
327
328
329	def test_chromadb_is_empty(monkeypatch, tmp_path):	1✔
330	# Mock the environment variable for ChromaDB
331	monkeypatch.setattr(	1✔
332	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
333	)
334	# Create a temporary directory for ChromaDB persistence
335	temp_persist_dir = tmp_path / "chroma_test_is_empty"	1✔
336	temp_persist_dir.mkdir()	1✔
337
338	vector_db = ChromaDB(persist_directory=str(temp_persist_dir))	1✔
339	assert vector_db.is_empty() is True, "Database should be empty initially"	1✔
340	documents = []	1✔
341	for i in range(5):	1✔
342	doc = Document(	1✔
343	page_content="Test content " + str(i),
344	metadata={"source": f"test_source_{i}"},
345	)
346	documents.append(doc)	1✔
347	vector_db.add_documents(documents)	1✔
348	assert (	1✔
349	vector_db.is_empty() is False
350	), "Database should not be empty after adding documents"
351	vector_db._delete() # Clean up the database after test	1✔
352
353
354	def test_chromadb_count(monkeypatch, tmp_path):	1✔
355	# Mock the environment variable for ChromaDB
356	monkeypatch.setattr(	1✔
357	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
358	)
359	# Create a temporary directory for ChromaDB persistence
360	temp_persist_dir = tmp_path / "chroma_test_count"	1✔
361	temp_persist_dir.mkdir()	1✔
362
363	vector_db = ChromaDB(	1✔
364	persist_directory=str(temp_persist_dir)
365	) # Initialize with temp_persist_dir
366	documents = []	1✔
367	for i in range(5):	1✔
368	doc = Document(	1✔
369	page_content="Test content" + str(i),
370	metadata={"source": f"test_source_{i}"},
371	)
372	documents.append(doc)	1✔
373	vector_db.add_documents(documents)	1✔
374	count = vector_db.count()	1✔
375	assert count == len(	1✔
376	documents
377	), "Should return the same number of documents in the collection"
378	vector_db._delete() # Clean up the database after test	1✔
379
380
381	def test_chromadb_delete(monkeypatch, tmp_path):	1✔
382	# Mock the environment variable for ChromaDB
383	monkeypatch.setattr(	1✔
384	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
385	)
386	# Create a temporary directory for ChromaDB persistence
387	temp_persist_dir = tmp_path / "chroma_test_delete"	1✔
388	temp_persist_dir.mkdir()	1✔
389
390	vector_db = ChromaDB(	1✔
391	persist_directory=str(temp_persist_dir)
392	) # Initialize with temp_persist_dir
393	documents = []	1✔
394	for i in range(5):	1✔
395	doc = Document(	1✔
396	page_content="Test content" + str(i),
397	metadata={"source": f"test_source_{i}"},
398	)
399	documents.append(doc)	1✔
400	vector_db.add_documents(documents)	1✔
401	db_instance = vector_db._get_db()	1✔
402	assert vector_db.count() == len(	1✔
403	documents
404	), "Should add the same number of documents to the database"
405	vector_db._delete() # Clean up the database after test	1✔
406	assert vector_db.is_empty() is True, "Database should be empty after deletion"	1✔
407
408
409	def test_get_vector_database_chromadb(monkeypatch, tmp_path):	1✔
410	# Mock the environment variable for ChromaDB
411	monkeypatch.setattr(	1✔
412	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
413	)
414	# Create a temporary directory for ChromaDB persistence for this test
415	temp_persist_dir = tmp_path / "chroma_test_get_vector_db"	1✔
416	temp_persist_dir.mkdir()	1✔
417	monkeypatch.setattr(	1✔
418	"app.services.vector_database_service.settings.VECTOR_DB_DIRECTORY",
419	str(temp_persist_dir),
420	)
421
422	vector_db = get_vector_database()	1✔
423	assert isinstance(vector_db, ChromaDB), "Expected an instance of ChromaDB class"	1✔
424	# Ensure the _delete operation can write to the temporary directory
425	try:	1✔
426	vector_db._delete() # Clean up the database after test	1✔
NEW 427	except Exception as e:	×
NEW 428	pytest.fail(f"_delete failed with temporary directory: {e}")	×
429
430
431	def test_get_vector_database_invalid_provider(monkeypatch):	1✔
432	# Mock the environment variable for an invalid provider
433	monkeypatch.setattr(	1✔
434	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER",
435	"invalid_provider",
436	)
437	with pytest.raises(ValueError):	1✔
438	get_vector_database()	1✔
439
440
441	def test_get_collection_count_with_client_none(monkeypatch, tmp_path):	1✔
442	# Mock the environment variable for ChromaDB
443	monkeypatch.setattr(	1✔
444	"app.services.vector_database_service.settings.VECTOR_DB_PROVIDER", "chroma"
445	)
446
447	# Create a temporary directory for ChromaDB persistence
448	temp_persist_dir = tmp_path / "chroma_test_collection_count"	1✔
449	temp_persist_dir.mkdir()	1✔
450
451	vector_db = ChromaDB(persist_directory=str(temp_persist_dir)) # Initialize with temp_persist_dir	1✔
452
453	fake_db_client = MagicMock()	1✔
454	# Simulate that the collection is not available or accessible on the client
455	fake_db_client._collection = None	1✔
456
457	monkeypatch.setattr(vector_db, "_db", fake_db_client) # vector_db._get_db() will now return fake_db_client	1✔
458
459	count = vector_db._get_collection_count()	1✔
460	assert count == 0, "Should return 0 when collection is not available via the client"	1✔

CodeHex16 / LLM-API / 14954586403

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous