test: improve conftest.py code quality - add type hints, docstrings, DRY refactoring
This commit is contained in:
@@ -1,6 +1,8 @@
|
|||||||
import pytest
|
import pytest
|
||||||
|
import json
|
||||||
|
from typing import Generator, Callable, Optional
|
||||||
from sqlalchemy import create_engine
|
from sqlalchemy import create_engine
|
||||||
from sqlalchemy.orm import sessionmaker
|
from sqlalchemy.orm import sessionmaker, Session
|
||||||
from sqlalchemy.pool import StaticPool
|
from sqlalchemy.pool import StaticPool
|
||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
from unittest.mock import patch, MagicMock
|
from unittest.mock import patch, MagicMock
|
||||||
@@ -13,6 +15,18 @@ from backend.config_manager import ConfigManager
|
|||||||
from backend.auth import get_current_admin, TokenData
|
from backend.auth import get_current_admin, TokenData
|
||||||
|
|
||||||
|
|
||||||
|
# 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
|
# Use in-memory SQLite for tests
|
||||||
SQLALCHEMY_DATABASE_URL = "sqlite://"
|
SQLALCHEMY_DATABASE_URL = "sqlite://"
|
||||||
|
|
||||||
@@ -25,7 +39,7 @@ TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engin
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function", autouse=True)
|
@pytest.fixture(scope="function", autouse=True)
|
||||||
def mock_scheduler():
|
def mock_scheduler() -> Generator[None, None, None]:
|
||||||
"""Shutdown scheduler before and after each test to avoid conflicts."""
|
"""Shutdown scheduler before and after each test to avoid conflicts."""
|
||||||
from backend.scheduler import scheduler
|
from backend.scheduler import scheduler
|
||||||
if scheduler.running:
|
if scheduler.running:
|
||||||
@@ -36,7 +50,7 @@ def mock_scheduler():
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def test_db():
|
def test_db() -> Generator[Session, None, None]:
|
||||||
"""Create in-memory SQLite database for tests."""
|
"""Create in-memory SQLite database for tests."""
|
||||||
Base.metadata.create_all(bind=engine)
|
Base.metadata.create_all(bind=engine)
|
||||||
session = TestingSessionLocal()
|
session = TestingSessionLocal()
|
||||||
@@ -46,10 +60,10 @@ def test_db():
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def test_client(test_db):
|
def test_client(test_db: Session) -> Generator[TestClient, None, None]:
|
||||||
"""Create FastAPI test client with mocked database."""
|
"""Create FastAPI test client with mocked database."""
|
||||||
|
|
||||||
def override_get_db():
|
def override_get_db() -> Generator[Session, None, None]:
|
||||||
try:
|
try:
|
||||||
yield test_db
|
yield test_db
|
||||||
finally:
|
finally:
|
||||||
@@ -63,7 +77,7 @@ def test_client(test_db):
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def mock_ldap():
|
def mock_ldap() -> Generator[MagicMock, None, None]:
|
||||||
"""Mock LDAP authentication."""
|
"""Mock LDAP authentication."""
|
||||||
with patch("backend.auth.ldap3.Server") as mock_server, \
|
with patch("backend.auth.ldap3.Server") as mock_server, \
|
||||||
patch("backend.auth.ldap3.Connection") as mock_conn_class:
|
patch("backend.auth.ldap3.Connection") as mock_conn_class:
|
||||||
@@ -72,7 +86,7 @@ def mock_ldap():
|
|||||||
mock_conn.bind.return_value = True
|
mock_conn.bind.return_value = True
|
||||||
mock_conn.search.return_value = True
|
mock_conn.search.return_value = True
|
||||||
mock_conn.entries = [
|
mock_conn.entries = [
|
||||||
MagicMock(entry_dn="uid=testuser,ou=people,dc=example,dc=com",
|
MagicMock(entry_dn=TEST_LDAP_DN,
|
||||||
uid=["testuser"])
|
uid=["testuser"])
|
||||||
]
|
]
|
||||||
mock_conn_class.return_value = mock_conn
|
mock_conn_class.return_value = mock_conn
|
||||||
@@ -81,31 +95,21 @@ def mock_ldap():
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def mock_gemini():
|
def mock_gemini() -> Generator[MagicMock, None, None]:
|
||||||
"""Mock Google Gemini AI extraction."""
|
"""Mock Google Gemini AI extraction."""
|
||||||
with patch("backend.ai.gemini_extractor.generate_content") as mock_gen:
|
with patch("backend.ai.gemini_extractor.generate_content") as mock_gen:
|
||||||
mock_response = MagicMock()
|
mock_response = MagicMock()
|
||||||
mock_response.text = '''{
|
mock_response.text = json.dumps(TEST_AI_RESPONSE)
|
||||||
"name": "Test Item",
|
|
||||||
"part_number": "PN-12345",
|
|
||||||
"category": "Electronics",
|
|
||||||
"quantity": 5
|
|
||||||
}'''
|
|
||||||
mock_gen.return_value = mock_response
|
mock_gen.return_value = mock_response
|
||||||
yield mock_gen
|
yield mock_gen
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def mock_claude():
|
def mock_claude() -> Generator[MagicMock, None, None]:
|
||||||
"""Mock Anthropic Claude AI extraction."""
|
"""Mock Anthropic Claude AI extraction."""
|
||||||
with patch("backend.ai.claude_extractor.generate_content") as mock_gen:
|
with patch("backend.ai.claude_extractor.generate_content") as mock_gen:
|
||||||
mock_content = MagicMock()
|
mock_content = MagicMock()
|
||||||
mock_content.text = '''{
|
mock_content.text = json.dumps(TEST_AI_RESPONSE)
|
||||||
"name": "Test Item",
|
|
||||||
"part_number": "PN-12345",
|
|
||||||
"category": "Electronics",
|
|
||||||
"quantity": 5
|
|
||||||
}'''
|
|
||||||
mock_response = MagicMock()
|
mock_response = MagicMock()
|
||||||
mock_response.content = [mock_content]
|
mock_response.content = [mock_content]
|
||||||
mock_gen.return_value = mock_response
|
mock_gen.return_value = mock_response
|
||||||
@@ -113,7 +117,7 @@ def mock_claude():
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def mock_config():
|
def mock_config() -> Generator[MagicMock, None, None]:
|
||||||
"""Mock ConfigManager."""
|
"""Mock ConfigManager."""
|
||||||
with patch.object(ConfigManager, "get_config") as mock_get:
|
with patch.object(ConfigManager, "get_config") as mock_get:
|
||||||
mock_get.return_value = {
|
mock_get.return_value = {
|
||||||
@@ -124,59 +128,51 @@ def mock_config():
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def admin_token(test_db):
|
def create_test_user(test_db: Session) -> Callable[[str, str], str]:
|
||||||
"""Create a test admin user and return JWT token string."""
|
"""Factory fixture to create test users with tokens."""
|
||||||
from backend.auth import create_access_token
|
def _create_user(username: str, role: str) -> str:
|
||||||
|
from backend.auth import create_access_token
|
||||||
|
|
||||||
admin = User(
|
user = User(
|
||||||
username="admin",
|
username=username,
|
||||||
hashed_password="hashed_password",
|
hashed_password=TEST_HASHED_PASSWORD,
|
||||||
role="admin",
|
role=role,
|
||||||
origin="local"
|
origin="local"
|
||||||
)
|
)
|
||||||
test_db.add(admin)
|
test_db.add(user)
|
||||||
test_db.commit()
|
test_db.commit()
|
||||||
test_db.refresh(admin)
|
test_db.refresh(user)
|
||||||
|
return create_access_token(user_id=user.id, username=username, role=role)
|
||||||
|
|
||||||
# Return JWT token string (for use in "Bearer {token}" headers)
|
return _create_user
|
||||||
token = create_access_token(user_id=admin.id, username="admin", role="admin")
|
|
||||||
return token
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def user_token(test_db):
|
def admin_token(create_test_user: Callable[[str, str], str]) -> str:
|
||||||
"""Create a test regular user and return JWT token string."""
|
"""Create test admin user and return JWT token."""
|
||||||
from backend.auth import create_access_token
|
return create_test_user(TEST_ADMIN_USERNAME, "admin")
|
||||||
|
|
||||||
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")
|
@pytest.fixture(scope="function")
|
||||||
def admin_client(test_client, test_db, admin_token):
|
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."""
|
"""Create a test client authenticated as admin."""
|
||||||
from backend.auth import get_current_user
|
from backend.auth import get_current_user
|
||||||
|
|
||||||
# Create TokenData from the JWT token's user info
|
# Create TokenData from the JWT token's user info
|
||||||
admin_token_data = TokenData(
|
admin_token_data = TokenData(
|
||||||
sub=1, # First admin user created has id=1
|
sub=1, # First admin user created has id=1
|
||||||
username="admin",
|
username=TEST_ADMIN_USERNAME,
|
||||||
role="admin",
|
role="admin",
|
||||||
exp=datetime.now(timezone.utc)
|
exp=datetime.now(timezone.utc)
|
||||||
)
|
)
|
||||||
|
|
||||||
def override_get_current_user():
|
def override_get_current_user() -> TokenData:
|
||||||
return admin_token_data
|
return admin_token_data
|
||||||
|
|
||||||
app.dependency_overrides[get_current_user] = override_get_current_user
|
app.dependency_overrides[get_current_user] = override_get_current_user
|
||||||
@@ -187,19 +183,19 @@ def admin_client(test_client, test_db, admin_token):
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def user_client(test_client, test_db, user_token):
|
def user_client(test_client: TestClient, test_db: Session, user_token: str) -> Generator[TestClient, None, None]:
|
||||||
"""Create a test client authenticated as regular user."""
|
"""Create a test client authenticated as regular user."""
|
||||||
from backend.auth import get_current_user
|
from backend.auth import get_current_user
|
||||||
|
|
||||||
# Create TokenData from the JWT token's user info
|
# Create TokenData from the JWT token's user info
|
||||||
user_token_data = TokenData(
|
user_token_data = TokenData(
|
||||||
sub=2, # Second user created has id=2
|
sub=2, # Second user created has id=2
|
||||||
username="user",
|
username=TEST_USER_USERNAME,
|
||||||
role="user",
|
role="user",
|
||||||
exp=datetime.now(timezone.utc)
|
exp=datetime.now(timezone.utc)
|
||||||
)
|
)
|
||||||
|
|
||||||
def override_get_current_user():
|
def override_get_current_user() -> TokenData:
|
||||||
return user_token_data
|
return user_token_data
|
||||||
|
|
||||||
app.dependency_overrides[get_current_user] = override_get_current_user
|
app.dependency_overrides[get_current_user] = override_get_current_user
|
||||||
|
|||||||
Reference in New Issue
Block a user