Spaces:
Running
Running
File size: 4,055 Bytes
021e065 | 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 121 122 123 124 125 126 127 128 129 130 131 132 133 | import pytest
import sys
import os
sys.path.insert(0, '.')
class TestPostgreSQL:
@pytest.fixture(autouse=True, scope="class")
def setup_class(self):
from backend.database.postgres.models import Base, engine
Base.metadata.create_all(bind=engine)
yield
engine.dispose()
def test_connection(self):
from backend.database.postgres.models import SessionLocal
db = SessionLocal()
from sqlalchemy import text
result = db.execute(text("SELECT 1")).scalar()
db.close()
assert result == 1
def test_tables_exist(self):
from backend.database.postgres.models.base import engine
from sqlalchemy import inspect
inspector = inspect(engine)
tables = inspector.get_table_names()
required = [
"users", "chat_sessions", "messages",
"user_memory", "financial_state", "projects",
"generated_files", "audit_log"
]
for table in required:
assert table in tables, f"Missing table: {table}"
def test_indexes_exist(self):
from backend.database.postgres.models.base import engine
from sqlalchemy import inspect
inspector = inspect(engine)
# Check critical indexes
chat_indexes = [
idx['name'] for idx in
inspector.get_indexes('chat_sessions')
]
assert any('user' in idx for idx in chat_indexes), \
"Missing user index on chat_sessions"
class TestRedis:
def test_connection(self):
from backend.database.redis.client import cache
assert cache.health_check() is True
def test_set_get(self):
from backend.database.redis.client import cache
cache.set("response", "test_key", {"value": 42})
result = cache.get("response", "test_key")
assert result == {"value": 42}
cache.delete("response", "test_key")
def test_rate_limiter(self):
from backend.database.redis.client import cache
cache.flush_category("rate_limit")
count = cache.incr("rate_limit", "test_user_rl")
assert count == 1
count = cache.incr("rate_limit", "test_user_rl")
assert count == 2
cache.flush_category("rate_limit")
def test_idempotency(self):
from backend.database.redis.client import cache
key = "test_idem_001"
result = {"status": "processed"}
assert cache.check_idempotency(key) is None
cache.mark_idempotency(key, result)
assert cache.check_idempotency(key) == result
cache.delete("idempotency", key)
class TestQdrant:
def test_connection(self):
from backend.database.vector.client import vector_store
assert vector_store.health_check() is True
def test_collections_exist(self):
from backend.database.vector.client import vector_store, get_qdrant
vector_store.create_collections()
qdrant = get_qdrant()
collections = [
c.name for c in qdrant.get_collections().collections
]
assert "senti_knowledge" in collections
def test_document_count_positive(self):
from backend.database.vector.client import vector_store
vector_store.create_collections()
count = vector_store.get_count("knowledge")
print(f"Knowledge documents: {count}")
assert count >= 0 # OK if 0 until corpus is built
class TestAuth:
def test_token_create_verify(self):
from backend.database.redis.sessions import create_token, verify_token
token = create_token("test_hash_123", "free")
assert token is not None
payload = verify_token(token)
assert payload is not None
assert payload["sub"] == "test_hash_123"
assert payload["tier"] == "free"
def test_invalid_token_fails(self):
from backend.database.redis.sessions import verify_token
result = verify_token("not.a.valid.token")
assert result is None
def test_phone_hashing(self):
from backend.database.redis.sessions import hash_phone
h1 = hash_phone("+254712345678")
h2 = hash_phone("254712345678")
h3 = hash_phone("0712345678")
# Same number, different formats
# Normalize removes +/spaces
assert h1 == h2 # Both become 254712345678
|