169 lines
4.5 KiB
Python
169 lines
4.5 KiB
Python
import os
|
|
|
|
# Set testing environment before importing app modules
|
|
os.environ["TESTING"] = "true"
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.main import app
|
|
from app.core.database import Base, get_db
|
|
from app.core.redis import get_redis
|
|
from app.models import User, Role, Department
|
|
|
|
# Use in-memory SQLite for testing
|
|
SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"
|
|
|
|
engine = create_engine(
|
|
SQLALCHEMY_DATABASE_URL,
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
# Ensure app code paths that use SessionLocal directly hit the test DB
|
|
from app.core import database as database_module
|
|
database_module.engine = engine
|
|
database_module.SessionLocal = TestingSessionLocal
|
|
|
|
|
|
class MockRedis:
|
|
"""Mock Redis client for testing."""
|
|
|
|
def __init__(self):
|
|
self.store = {}
|
|
|
|
def get(self, key):
|
|
return self.store.get(key)
|
|
|
|
def set(self, key, value):
|
|
self.store[key] = value
|
|
|
|
def setex(self, key, seconds, value):
|
|
self.store[key] = value
|
|
|
|
def delete(self, key):
|
|
if key in self.store:
|
|
del self.store[key]
|
|
|
|
def scan_iter(self, match=None):
|
|
"""Iterate over keys matching a pattern."""
|
|
import fnmatch
|
|
if match is None:
|
|
yield from self.store.keys()
|
|
else:
|
|
pattern = match.replace("*", "**")
|
|
for key in self.store.keys():
|
|
if fnmatch.fnmatch(key, match):
|
|
yield key
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def db():
|
|
"""Create a fresh database for each test."""
|
|
Base.metadata.create_all(bind=engine)
|
|
db = TestingSessionLocal()
|
|
|
|
# Create default role
|
|
admin_role = Role(
|
|
id="00000000-0000-0000-0000-000000000001",
|
|
name="super_admin",
|
|
permissions={"all": True},
|
|
is_system_role=True,
|
|
)
|
|
db.add(admin_role)
|
|
|
|
engineer_role = Role(
|
|
id="00000000-0000-0000-0000-000000000003",
|
|
name="engineer",
|
|
permissions={"projects.read": True, "tasks.read": True, "tasks.write": True},
|
|
is_system_role=False,
|
|
)
|
|
db.add(engineer_role)
|
|
|
|
# Create system admin user
|
|
admin_user = User(
|
|
id="00000000-0000-0000-0000-000000000001",
|
|
email="ymirliu@panjit.com.tw",
|
|
name="System Administrator",
|
|
role_id="00000000-0000-0000-0000-000000000001",
|
|
is_active=True,
|
|
is_system_admin=True,
|
|
)
|
|
db.add(admin_user)
|
|
|
|
db.commit()
|
|
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
Base.metadata.drop_all(bind=engine)
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_redis():
|
|
"""Create mock Redis for testing."""
|
|
from app.core import redis as redis_module
|
|
client = redis_module.redis_client
|
|
if hasattr(client, "store"):
|
|
client.store.clear()
|
|
return client
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def client(db, mock_redis):
|
|
"""Create test client with overridden dependencies."""
|
|
# Reset rate limiter storage before each test
|
|
from app.core.rate_limiter import limiter
|
|
if hasattr(limiter, '_storage') and limiter._storage:
|
|
try:
|
|
limiter._storage.reset()
|
|
except Exception:
|
|
pass # Memory storage might not have reset method
|
|
# For memory storage, clear internal state
|
|
if hasattr(limiter, '_limiter') and hasattr(limiter._limiter, '_storage'):
|
|
storage = limiter._limiter._storage
|
|
if hasattr(storage, 'storage'):
|
|
storage.storage.clear()
|
|
|
|
def override_get_db():
|
|
try:
|
|
yield db
|
|
finally:
|
|
pass
|
|
|
|
def override_get_redis():
|
|
return mock_redis
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
app.dependency_overrides[get_redis] = override_get_redis
|
|
|
|
with TestClient(app) as test_client:
|
|
yield test_client
|
|
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def admin_token(client, mock_redis):
|
|
"""Get an admin token for testing."""
|
|
from app.core.security import create_access_token, create_token_payload
|
|
|
|
token_data = create_token_payload(
|
|
user_id="00000000-0000-0000-0000-000000000001",
|
|
email="ymirliu@panjit.com.tw",
|
|
role="super_admin",
|
|
department_id=None,
|
|
is_system_admin=True,
|
|
)
|
|
token = create_access_token(token_data)
|
|
|
|
# Store in mock Redis
|
|
mock_redis.setex("session:00000000-0000-0000-0000-000000000001", 900, token)
|
|
|
|
return token
|