Spaces:
Sleeping
Sleeping
File size: 3,202 Bytes
29c3655 74e758d 2eb9a5f 74e758d 2eb9a5f |
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 |
import os
import sys
# Ensure project root and src are on sys.path for tests
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
SRC_PATH = os.path.join(PROJECT_ROOT, "src")
if PROJECT_ROOT not in sys.path:
sys.path.insert(0, PROJECT_ROOT)
if SRC_PATH not in sys.path:
sys.path.insert(0, SRC_PATH)
# Set environment variables to disable ChromaDB telemetry
os.environ["ANONYMIZED_TELEMETRY"] = "False"
os.environ["CHROMA_TELEMETRY"] = "False"
from unittest.mock import MagicMock, patch # noqa: E402
import pytest # noqa: E402
from app import app as flask_app # noqa: E402
@pytest.fixture(scope="session", autouse=True)
def disable_chromadb_telemetry():
"""Disable ChromaDB telemetry to avoid errors in tests"""
patches = []
try:
# Patch multiple telemetry-related functions
patches.extend(
[
patch("chromadb.telemetry.product.posthog.capture", return_value=None),
patch(
"chromadb.telemetry.product.posthog.Posthog.capture",
return_value=None,
),
patch(
"chromadb.telemetry.product.posthog.Posthog",
return_value=MagicMock(),
),
patch("chromadb.configure", return_value=None),
]
)
for p in patches:
p.start()
yield
except (ImportError, AttributeError):
# If modules don't exist, continue without patching
yield
finally:
for p in patches:
try:
p.stop()
except Exception:
pass
@pytest.fixture
def app():
"""Flask application fixture."""
# Clear any cached services before each test to prevent state contamination
flask_app.config["RAG_PIPELINE"] = None
flask_app.config["INGESTION_PIPELINE"] = None
flask_app.config["SEARCH_SERVICE"] = None
# Also clear any module-level caches that might exist
import sys
modules_to_clear = [
"src.rag.rag_pipeline",
"src.llm.llm_service",
"src.search.search_service",
"src.embedding.embedding_service",
"src.vector_store.vector_db",
]
for module_name in modules_to_clear:
if module_name in sys.modules:
# Clear any cached instances on the module
module = sys.modules[module_name]
for attr_name in dir(module):
attr = getattr(module, attr_name)
if hasattr(attr, "__dict__") and not attr_name.startswith("_"):
# Clear instance dictionaries that might contain cached data
if hasattr(attr, "_instances"):
attr._instances = {}
yield flask_app
@pytest.fixture
def client(app):
"""Flask test client fixture."""
return app.test_client()
@pytest.fixture(autouse=True)
def reset_mock_state():
"""Fixture to reset any global mock state between tests."""
yield
# Clean up any lingering mock state after each test
import unittest.mock
# Clear any patches that might have been left hanging
unittest.mock.patch.stopall()
|