Spaces:
Sleeping
Sleeping
File size: 3,755 Bytes
aff5d04 a3dfc07 aff5d04 a3dfc07 aff5d04 a3dfc07 aff5d04 a3dfc07 aff5d04 a3dfc07 aff5d04 a3dfc07 aff5d04 a3dfc07 aff5d04 a3dfc07 aff5d04 a725155 a3dfc07 aff5d04 a3dfc07 aff5d04 a3dfc07 aff5d04 a3dfc07 aff5d04 a3dfc07 aff5d04 a3dfc07 |
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 |
"""
Test for enhanced ingestion Flask endpoint
"""
import json
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from app import app
class TestEnhancedIngestionEndpoint(unittest.TestCase):
"""Test cases for enhanced ingestion Flask endpoint"""
def setUp(self):
"""Set up test fixtures"""
app.config["TESTING"] = True
self.app = app.test_client()
# Create temporary directory and files for testing
self.temp_dir = tempfile.mkdtemp()
self.test_dir = Path(self.temp_dir)
self.test_file = self.test_dir / "test.md"
self.test_file.write_text(
"# Test Document\n\nThis is test content for enhanced ingestion."
)
def test_ingest_endpoint_with_embeddings_default(self):
"""Test ingestion endpoint with default embeddings enabled"""
with patch("src.config.CORPUS_DIRECTORY", str(self.test_dir)):
response = self.app.post("/ingest")
self.assertEqual(response.status_code, 200)
data = json.loads(response.data)
# Check enhanced response structure
self.assertEqual(data["status"], "success")
self.assertIn("chunks_processed", data)
self.assertIn("files_processed", data)
self.assertIn("embeddings_stored", data)
self.assertIn("store_embeddings", data)
self.assertTrue(data["store_embeddings"]) # Default is True
self.assertGreater(data["chunks_processed"], 0)
self.assertGreater(data["files_processed"], 0)
def test_ingest_endpoint_with_embeddings_disabled(self):
"""Test ingestion endpoint with embeddings disabled"""
with patch("src.config.CORPUS_DIRECTORY", str(self.test_dir)):
response = self.app.post(
"/ingest",
data=json.dumps({"store_embeddings": False}),
content_type="application/json",
)
self.assertEqual(response.status_code, 200)
data = json.loads(response.data)
# Check response structure with embeddings disabled
self.assertEqual(data["status"], "success")
self.assertIn("chunks_processed", data)
self.assertIn("files_processed", data)
self.assertIn("embeddings_stored", data)
self.assertIn("store_embeddings", data)
self.assertFalse(data["store_embeddings"])
self.assertEqual(data["embeddings_stored"], 0)
self.assertGreater(data["chunks_processed"], 0)
self.assertGreater(data["files_processed"], 0)
def test_ingest_endpoint_with_no_json(self):
"""Test ingestion endpoint with no JSON payload (should default to
embeddings enabled)"""
with patch("src.config.CORPUS_DIRECTORY", str(self.test_dir)):
response = self.app.post("/ingest")
self.assertEqual(response.status_code, 200)
data = json.loads(response.data)
# Should default to embeddings enabled
self.assertTrue(data["store_embeddings"])
def test_ingest_endpoint_error_handling(self):
"""Test ingestion endpoint error handling"""
with patch("src.config.CORPUS_DIRECTORY", "/nonexistent/directory"):
response = self.app.post("/ingest")
self.assertEqual(response.status_code, 500)
data = json.loads(response.data)
self.assertEqual(data["status"], "error")
self.assertIn("message", data)
def tearDown(self):
"""Clean up test fixtures"""
import shutil
shutil.rmtree(self.temp_dir, ignore_errors=True)
if __name__ == "__main__":
unittest.main()
|