209 lines
6.6 KiB
Python
209 lines
6.6 KiB
Python
import pytest
|
|
import json
|
|
from typing import Generator, Callable, Optional
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker, Session
|
|
from sqlalchemy.pool import StaticPool
|
|
from fastapi.testclient import TestClient
|
|
from unittest.mock import patch, MagicMock
|
|
from datetime import datetime, timezone
|
|
|
|
from backend.main import app
|
|
from backend.database import Base, get_db
|
|
from backend.models import User
|
|
from backend.config_manager import ConfigManager
|
|
from backend.auth import get_current_admin, TokenData
|
|
import backend.routers.users as users_router
|
|
import backend.routers.auth as auth_router
|
|
import backend.routers.categories as categories_router
|
|
|
|
|
|
# Test data constants
|
|
TEST_ADMIN_USERNAME = "admin"
|
|
TEST_USER_USERNAME = "user"
|
|
TEST_HASHED_PASSWORD = "hashed_password"
|
|
TEST_LDAP_DN = "uid=testuser,ou=people,dc=example,dc=com"
|
|
TEST_AI_RESPONSE = {
|
|
"name": "Test Item",
|
|
"part_number": "PN-12345",
|
|
"category": "Electronics",
|
|
"quantity": 5
|
|
}
|
|
|
|
# Use in-memory SQLite for tests
|
|
SQLALCHEMY_DATABASE_URL = "sqlite://"
|
|
|
|
engine = create_engine(
|
|
SQLALCHEMY_DATABASE_URL,
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
|
|
@pytest.fixture(scope="function", autouse=True)
|
|
def mock_scheduler() -> Generator[None, None, None]:
|
|
"""Shutdown scheduler before and after each test to avoid conflicts."""
|
|
from backend.scheduler import scheduler
|
|
if scheduler.running:
|
|
scheduler.shutdown()
|
|
yield
|
|
if scheduler.running:
|
|
scheduler.shutdown()
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def test_db() -> Generator[Session, None, None]:
|
|
"""Create in-memory SQLite database for tests."""
|
|
Base.metadata.create_all(bind=engine)
|
|
session = TestingSessionLocal()
|
|
yield session
|
|
session.close()
|
|
Base.metadata.drop_all(bind=engine)
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def test_client(test_db: Session) -> Generator[TestClient, None, None]:
|
|
"""Create FastAPI test client with mocked database."""
|
|
|
|
def override_get_db() -> Generator[Session, None, None]:
|
|
try:
|
|
yield test_db
|
|
finally:
|
|
pass
|
|
|
|
# Override all local get_db functions across all routers
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
app.dependency_overrides[users_router.get_db] = override_get_db
|
|
app.dependency_overrides[categories_router.get_db] = override_get_db
|
|
|
|
client = TestClient(app)
|
|
yield client
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_ldap() -> Generator[MagicMock, None, None]:
|
|
"""Mock LDAP authentication."""
|
|
with patch("backend.routers.auth.ldap3.Server") as mock_server, \
|
|
patch("backend.routers.auth.ldap3.Connection") as mock_conn_class:
|
|
|
|
mock_conn = MagicMock()
|
|
mock_conn.bind.return_value = True
|
|
mock_conn.search.return_value = True
|
|
mock_entry = MagicMock()
|
|
mock_entry.entry_dn = TEST_LDAP_DN
|
|
mock_entry.uid = MagicMock()
|
|
mock_entry.uid.values = ["testuser"]
|
|
mock_entry.memberOf = MagicMock()
|
|
mock_entry.memberOf.values = ["cn=inventory_users,ou=groups,dc=ainventory,dc=local"]
|
|
mock_conn.entries = [mock_entry]
|
|
mock_conn_class.return_value = mock_conn
|
|
|
|
yield mock_conn
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_gemini() -> Generator[MagicMock, None, None]:
|
|
"""Mock AI label extraction at the ai_vision level."""
|
|
with patch("backend.ai_vision.extract_label_info") as mock_gen:
|
|
mock_gen.return_value = TEST_AI_RESPONSE
|
|
yield mock_gen
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_claude() -> Generator[MagicMock, None, None]:
|
|
"""Mock AI label extraction at the ai_vision level (Claude)."""
|
|
with patch("backend.ai_vision.extract_label_info") as mock_gen:
|
|
mock_gen.return_value = TEST_AI_RESPONSE
|
|
yield mock_gen
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_config() -> Generator[MagicMock, None, None]:
|
|
"""Mock ConfigManager."""
|
|
with patch.object(ConfigManager, "get_config") as mock_get:
|
|
mock_get.return_value = {
|
|
"ai_provider": "gemini",
|
|
"api_key": "test-key"
|
|
}
|
|
yield mock_get
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def create_test_user(test_db: Session) -> Callable[[str, str], str]:
|
|
"""Factory fixture to create test users with tokens."""
|
|
def _create_user(username: str, role: str) -> str:
|
|
from backend.auth import create_access_token
|
|
|
|
user = User(
|
|
username=username,
|
|
hashed_password=TEST_HASHED_PASSWORD,
|
|
role=role,
|
|
origin="local"
|
|
)
|
|
test_db.add(user)
|
|
test_db.commit()
|
|
test_db.refresh(user)
|
|
return create_access_token(user_id=user.id, username=username, role=role)
|
|
|
|
return _create_user
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def admin_token(create_test_user: Callable[[str, str], str]) -> str:
|
|
"""Create test admin user and return JWT token."""
|
|
return create_test_user(TEST_ADMIN_USERNAME, "admin")
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def user_token(create_test_user: Callable[[str, str], str]) -> str:
|
|
"""Create test regular user and return JWT token."""
|
|
return create_test_user(TEST_USER_USERNAME, "user")
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def admin_client(test_client: TestClient, test_db: Session, admin_token: str) -> Generator[TestClient, None, None]:
|
|
"""Create a test client authenticated as admin."""
|
|
from backend.auth import get_current_user
|
|
|
|
# Create TokenData from the JWT token's user info
|
|
admin_token_data = TokenData(
|
|
sub=1, # First admin user created has id=1
|
|
username=TEST_ADMIN_USERNAME,
|
|
role="admin",
|
|
exp=datetime.now(timezone.utc)
|
|
)
|
|
|
|
def override_get_current_user() -> TokenData:
|
|
return admin_token_data
|
|
|
|
app.dependency_overrides[get_current_user] = override_get_current_user
|
|
|
|
yield test_client
|
|
|
|
app.dependency_overrides.pop(get_current_user, None)
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def user_client(test_client: TestClient, test_db: Session, user_token: str) -> Generator[TestClient, None, None]:
|
|
"""Create a test client authenticated as regular user."""
|
|
from backend.auth import get_current_user
|
|
|
|
# Create TokenData from the JWT token's user info
|
|
user_token_data = TokenData(
|
|
sub=2, # Second user created has id=2
|
|
username=TEST_USER_USERNAME,
|
|
role="user",
|
|
exp=datetime.now(timezone.utc)
|
|
)
|
|
|
|
def override_get_current_user() -> TokenData:
|
|
return user_token_data
|
|
|
|
app.dependency_overrides[get_current_user] = override_get_current_user
|
|
|
|
yield test_client
|
|
|
|
app.dependency_overrides.pop(get_current_user, None)
|