george
387 lines
13 KiB
Python
387 lines
13 KiB
Python
"""
|
|
Tests for backend/app/db.py
|
|
|
|
These tests verify SQLite database operations including:
|
|
- Table creation (init_db)
|
|
- Library CRUD operations
|
|
- Document chunk storage and retrieval
|
|
- Full-text search functionality
|
|
|
|
All tests use a temporary test database file.
|
|
"""
|
|
import pytest
|
|
from datetime import datetime
|
|
|
|
|
|
class TestInitDatabase:
|
|
"""Tests for init_db() - table creation."""
|
|
|
|
def test_init_db_creates_tables(self, test_database):
|
|
"""Database should have libraries and documents tables after init."""
|
|
import sqlite3
|
|
from backend.app.db import get_connection, get_db_path
|
|
|
|
conn = get_connection()
|
|
cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")
|
|
tables = [row[0] for row in cursor.fetchall()]
|
|
|
|
# Should have libraries, documents, and FTS virtual table
|
|
assert "libraries" in tables or any("libraries" in t.lower() for t in tables)
|
|
conn.close()
|
|
|
|
def test_init_db_returns_success(self, test_database):
|
|
"""init_db should return success indicator."""
|
|
from backend.app.db import init_db
|
|
|
|
result = init_db()
|
|
assert result["success"] is True
|
|
|
|
|
|
class TestLibraryOperations:
|
|
"""Tests for library CRUD operations."""
|
|
|
|
def test_upsert_library_new(self, test_database):
|
|
"""Upsert should create new library."""
|
|
from backend.app.db import upsert_library
|
|
|
|
result = upsert_library(
|
|
library_id="/local/testlib",
|
|
name="Test Library",
|
|
description="A test library for unit tests"
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["id"] == "/local/testlib"
|
|
|
|
def test_upsert_library_update(self, test_database):
|
|
"""Upsert should update existing library."""
|
|
from backend.app.db import upsert_library
|
|
|
|
# Insert first library
|
|
upsert_library(
|
|
library_id="/local/upsertlib",
|
|
name="Original Name",
|
|
description="Original description"
|
|
)
|
|
|
|
# Update it
|
|
result = upsert_library(
|
|
library_id="/local/upsertlib",
|
|
name="Updated Name",
|
|
description="Updated description"
|
|
)
|
|
|
|
assert result["success"] is True
|
|
|
|
def test_upsert_library_id_normalization(self, test_database):
|
|
"""Library ID normalization - /local/ prefix should be preserved."""
|
|
from backend.app.db import upsert_library
|
|
|
|
# Test various ID formats
|
|
test_ids = [
|
|
"/local/foundryvtt",
|
|
"foundryvtt",
|
|
"/local/mydocs",
|
|
]
|
|
|
|
for lib_id in test_ids:
|
|
result = upsert_library(library_id=lib_id, name="Test", description="Desc")
|
|
assert result["success"] is True
|
|
# Verify we can retrieve it back
|
|
from backend.app.db import get_chunks_for_library
|
|
# Just ensure no errors occur
|
|
|
|
def test_list_libraries(self, test_database):
|
|
"""list_libraries should return list of libraries."""
|
|
from backend.app.db import upsert_library, list_libraries
|
|
|
|
# Create some libraries
|
|
for i in range(3):
|
|
upsert_library(
|
|
library_id=f"/local/lib{i}",
|
|
name=f"Library {i}",
|
|
description=f"Description {i}"
|
|
)
|
|
|
|
libs = list_libraries()
|
|
assert isinstance(libs, list)
|
|
assert len(libs) >= 3
|
|
|
|
def test_search_libraries(self, test_database):
|
|
"""search_libraries should find libraries by name/description."""
|
|
from backend.app.db import upsert_library, search_libraries
|
|
|
|
# Create libraries with searchable names
|
|
upsert_library(library_id="/local/foo1", name="Foo Library", description="Bar baz")
|
|
upsert_library(library_id="/local/foo2", name="Other Library", description="Different content")
|
|
|
|
results = search_libraries("foo")
|
|
assert isinstance(results, list)
|
|
|
|
|
|
class TestDocumentChunkOperations:
|
|
"""Tests for document chunk storage and retrieval."""
|
|
|
|
def test_insert_document_chunk_new(self, test_database):
|
|
"""insert_document_chunk should create new chunk record."""
|
|
from backend.app.db import insert_document_chunk
|
|
|
|
result = insert_document_chunk(
|
|
doc_id="doc-1",
|
|
library_id="/local/testlib",
|
|
path="docs/example.md",
|
|
title="Example Document",
|
|
content="# Example\n\nThis is the content.",
|
|
chunk_index=0,
|
|
token_estimate=100
|
|
)
|
|
|
|
assert result["success"] is True
|
|
|
|
def test_insert_document_chunk_update(self, test_database):
|
|
"""insert_document_chunk should update existing record."""
|
|
from backend.app.db import insert_document_chunk
|
|
|
|
# Insert first
|
|
insert_document_chunk(
|
|
doc_id="doc-update-test",
|
|
library_id="/local/uplib",
|
|
path="old-path.md",
|
|
title="Old Title",
|
|
content="# Old\nContent here.",
|
|
chunk_index=0,
|
|
token_estimate=50
|
|
)
|
|
|
|
# Update it
|
|
result = insert_document_chunk(
|
|
doc_id="doc-update-test",
|
|
library_id="/local/uplib",
|
|
path="new-path.md",
|
|
title="New Title",
|
|
content="# New\nUpdated content.",
|
|
chunk_index=1,
|
|
token_estimate=75
|
|
)
|
|
|
|
assert result["success"] is True
|
|
|
|
def test_get_document_by_id(self, test_database):
|
|
"""get_document_by_id should retrieve document by ID."""
|
|
from backend.app.db import insert_document_chunk, get_document_by_id
|
|
|
|
# Insert document
|
|
doc_id = "unique-doc-id-12345"
|
|
insert_document_chunk(
|
|
doc_id=doc_id,
|
|
library_id="/local/testlib",
|
|
path="docs/test.md",
|
|
title="Test Document",
|
|
content="# Test\n\nTest content here.",
|
|
chunk_index=None,
|
|
token_estimate=200
|
|
)
|
|
|
|
# Retrieve it
|
|
doc = get_document_by_id(doc_id)
|
|
assert doc is not None
|
|
assert doc["id"] == doc_id
|
|
|
|
def test_get_chunks_for_library(self, test_database):
|
|
"""get_chunks_for_library should return all chunks for a library."""
|
|
from backend.app.db import upsert_library, insert_document_chunk, get_chunks_for_library
|
|
|
|
# Create library
|
|
upsert_library(library_id="/local/chunktest", name="Chunk Test", description="Test")
|
|
|
|
# Add some chunks
|
|
for i in range(3):
|
|
insert_document_chunk(
|
|
doc_id=f"chunk-{i}",
|
|
library_id="/local/chunktest",
|
|
path=f"path{i}.md",
|
|
title=f"Section {i}",
|
|
content=f"Content section {i}.",
|
|
chunk_index=i,
|
|
token_estimate=50
|
|
)
|
|
|
|
chunks = get_chunks_for_library("/local/chunktest")
|
|
assert isinstance(chunks, list)
|
|
assert len(chunks) >= 3
|
|
|
|
def test_clear_library_documents(self, test_database):
|
|
"""clear_library_documents should delete all docs for a library."""
|
|
from backend.app.db import upsert_library, insert_document_chunk, clear_library_documents, get_chunks_for_library
|
|
|
|
# Create and populate library
|
|
upsert_library(library_id="/local/cleartest", name="Clear Test", description="Test")
|
|
for i in range(5):
|
|
insert_document_chunk(
|
|
doc_id=f"clear-{i}",
|
|
library_id="/local/cleartest",
|
|
path=f"path{i}.md",
|
|
content=f"Content {i}.",
|
|
token_estimate=20
|
|
)
|
|
|
|
# Clear it
|
|
result = clear_library_documents("/local/cleartest")
|
|
assert result["success"] is True
|
|
|
|
# Verify cleared
|
|
remaining = get_chunks_for_library("/local/cleartest")
|
|
assert len(remaining) == 0
|
|
|
|
def test_replace_library_documents_is_atomic(self, test_database):
|
|
"""Replacing chunks should remove old rows and insert the new set."""
|
|
from backend.app.db import (
|
|
get_chunks_for_library,
|
|
insert_document_chunk,
|
|
replace_library_documents,
|
|
upsert_library,
|
|
)
|
|
|
|
library_id = "/local/replacetest"
|
|
upsert_library(library_id, "Replace test", source_path=library_id)
|
|
insert_document_chunk(
|
|
"old-chunk",
|
|
library_id,
|
|
"old.md",
|
|
content="old content",
|
|
chunk_index=0,
|
|
)
|
|
|
|
result = replace_library_documents(
|
|
library_id,
|
|
[
|
|
{
|
|
"id": "new-chunk",
|
|
"path": "new.md",
|
|
"title": "new",
|
|
"content": "new content",
|
|
"chunk_index": 0,
|
|
"token_estimate": 2,
|
|
}
|
|
],
|
|
)
|
|
|
|
chunks = get_chunks_for_library(library_id)
|
|
assert result["success"] is True
|
|
assert result["deleted"] >= 1
|
|
assert result["inserted"] == 1
|
|
assert [chunk["id"] for chunk in chunks] == ["new-chunk"]
|
|
|
|
def test_failed_replacement_keeps_existing_chunks(self, test_database):
|
|
"""A bad replacement must roll back instead of erasing the old index."""
|
|
from backend.app.db import (
|
|
get_chunks_for_library,
|
|
insert_document_chunk,
|
|
replace_library_documents,
|
|
upsert_library,
|
|
)
|
|
|
|
library_id = "/local/rollbacktest"
|
|
upsert_library(library_id, "Rollback test", source_path=library_id)
|
|
insert_document_chunk(
|
|
"old-chunk",
|
|
library_id,
|
|
"old.md",
|
|
content="old content",
|
|
chunk_index=0,
|
|
)
|
|
|
|
duplicate = {
|
|
"id": "duplicate",
|
|
"path": "new.md",
|
|
"content": "new content",
|
|
"chunk_index": 0,
|
|
}
|
|
result = replace_library_documents(library_id, [duplicate, duplicate])
|
|
|
|
chunks = get_chunks_for_library(library_id)
|
|
assert result["success"] is False
|
|
assert [chunk["id"] for chunk in chunks] == ["old-chunk"]
|
|
|
|
|
|
class TestDatabaseEdgeCases:
|
|
"""Tests for edge cases and error handling."""
|
|
|
|
def test_empty_library_id(self, test_database):
|
|
"""Operations with empty ID should handle gracefully."""
|
|
from backend.app.db import upsert_library
|
|
|
|
result = upsert_library(library_id="", name="Test", description="Desc")
|
|
# Should not crash, though may not be a valid operation
|
|
|
|
def test_special_characters_in_content(self, test_database):
|
|
"""Content with special characters should be stored."""
|
|
from backend.app.db import insert_document_chunk
|
|
|
|
content = "Hello \"world\" <tag /> & amp; 'apostrophe'"
|
|
result = insert_document_chunk(
|
|
doc_id="special-test",
|
|
library_id="/local/speciallib",
|
|
path="special.md",
|
|
content=content,
|
|
token_estimate=100
|
|
)
|
|
|
|
assert result["success"] is True
|
|
|
|
def test_very_long_content(self, test_database):
|
|
"""Long content should be stored."""
|
|
from backend.app.db import insert_document_chunk
|
|
|
|
long_content = "a" * 5000
|
|
result = insert_document_chunk(
|
|
doc_id="long-test",
|
|
library_id="/local/longlib",
|
|
path="long.md",
|
|
content=long_content,
|
|
token_estimate=1000
|
|
)
|
|
|
|
assert result["success"] is True
|
|
|
|
def test_none_description(self, test_database):
|
|
"""Library with None description should work."""
|
|
from backend.app.db import upsert_library
|
|
|
|
result = upsert_library(
|
|
library_id="/local/nonedesc",
|
|
name="No Description Lib",
|
|
description=None
|
|
)
|
|
|
|
assert result["success"] is True
|
|
|
|
|
|
class TestDatabaseInitialization:
|
|
"""Tests for database initialization state."""
|
|
|
|
def test_database_is_empty_after_init(self, test_database):
|
|
"""Database should be empty right after init."""
|
|
from backend.app.db import list_libraries
|
|
|
|
libs = list_libraries()
|
|
assert isinstance(libs, list)
|
|
|
|
|
|
# =============================================================================
|
|
# FIXTURES
|
|
# =============================================================================
|
|
|
|
@pytest.fixture
|
|
def sample_doc():
|
|
"""Sample document chunk for testing."""
|
|
return {
|
|
"doc_id": "sample-doc-1",
|
|
"library_id": "/local/samplelib",
|
|
"path": "docs/guide.md",
|
|
"title": "Getting Started Guide",
|
|
"content": "# Getting Started\n\nWelcome to the guide. This is a sample document for testing.\n\n## Installation\n\nInstall with pip.",
|
|
"chunk_index": 0,
|
|
"token_estimate": 500
|
|
}
|