senti/tests/unit/test_databases.py

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