210 lines
5.8 KiB
Python
210 lines
5.8 KiB
Python
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
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
|
|
|
|
|
|
# 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():
|
|
"""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():
|
|
"""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):
|
|
"""Create FastAPI test client with mocked database."""
|
|
|
|
def override_get_db():
|
|
try:
|
|
yield test_db
|
|
finally:
|
|
pass
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
|
|
client = TestClient(app)
|
|
yield client
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_ldap():
|
|
"""Mock LDAP authentication."""
|
|
with patch("backend.auth.ldap3.Server") as mock_server, \
|
|
patch("backend.auth.ldap3.Connection") as mock_conn_class:
|
|
|
|
mock_conn = MagicMock()
|
|
mock_conn.bind.return_value = True
|
|
mock_conn.search.return_value = True
|
|
mock_conn.entries = [
|
|
MagicMock(entry_dn="uid=testuser,ou=people,dc=example,dc=com",
|
|
uid=["testuser"])
|
|
]
|
|
mock_conn_class.return_value = mock_conn
|
|
|
|
yield mock_conn
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_gemini():
|
|
"""Mock Google Gemini AI extraction."""
|
|
with patch("backend.ai.gemini_extractor.generate_content") as mock_gen:
|
|
mock_response = MagicMock()
|
|
mock_response.text = '''{
|
|
"name": "Test Item",
|
|
"part_number": "PN-12345",
|
|
"category": "Electronics",
|
|
"quantity": 5
|
|
}'''
|
|
mock_gen.return_value = mock_response
|
|
yield mock_gen
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_claude():
|
|
"""Mock Anthropic Claude AI extraction."""
|
|
with patch("backend.ai.claude_extractor.generate_content") as mock_gen:
|
|
mock_content = MagicMock()
|
|
mock_content.text = '''{
|
|
"name": "Test Item",
|
|
"part_number": "PN-12345",
|
|
"category": "Electronics",
|
|
"quantity": 5
|
|
}'''
|
|
mock_response = MagicMock()
|
|
mock_response.content = [mock_content]
|
|
mock_gen.return_value = mock_response
|
|
yield mock_gen
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_config():
|
|
"""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 admin_token(test_db):
|
|
"""Create a test admin user and return JWT token string."""
|
|
from backend.auth import create_access_token
|
|
|
|
admin = User(
|
|
username="admin",
|
|
hashed_password="hashed_password",
|
|
role="admin",
|
|
origin="local"
|
|
)
|
|
test_db.add(admin)
|
|
test_db.commit()
|
|
test_db.refresh(admin)
|
|
|
|
# Return JWT token string (for use in "Bearer {token}" headers)
|
|
token = create_access_token(user_id=admin.id, username="admin", role="admin")
|
|
return token
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def user_token(test_db):
|
|
"""Create a test regular user and return JWT token string."""
|
|
from backend.auth import create_access_token
|
|
|
|
user = User(
|
|
username="user",
|
|
hashed_password="hashed_password",
|
|
role="user",
|
|
origin="local"
|
|
)
|
|
test_db.add(user)
|
|
test_db.commit()
|
|
test_db.refresh(user)
|
|
|
|
# Return JWT token string (for use in "Bearer {token}" headers)
|
|
token = create_access_token(user_id=user.id, username="user", role="user")
|
|
return token
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def admin_client(test_client, test_db, admin_token):
|
|
"""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="admin",
|
|
role="admin",
|
|
exp=datetime.now(timezone.utc)
|
|
)
|
|
|
|
def override_get_current_user():
|
|
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, test_db, user_token):
|
|
"""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="user",
|
|
role="user",
|
|
exp=datetime.now(timezone.utc)
|
|
)
|
|
|
|
def override_get_current_user():
|
|
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)
|