- Add automated background task processing every 5 seconds - Implement run_simulation_cycle() function to process queued tasks - Update task and task group status from 'queued' to 'completed' - Add comprehensive tests for simulation logic in test_simulation.py - Integrate simulation with FastAPI lifecycle management - Unify status handling across the application - Update USER_MANUAL.md with detailed simulation documentation Resolves #7
27 lines
771 B
Python
27 lines
771 B
Python
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from src.database import Base
|
|
|
|
# Test Database Setup
|
|
TEST_DATABASE_URL = "sqlite:///./test.db"
|
|
|
|
engine = create_engine(
|
|
TEST_DATABASE_URL, connect_args={"check_same_thread": False}
|
|
)
|
|
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
@pytest.fixture(scope="function")
|
|
def db():
|
|
"""
|
|
Creates a clean database and a new session for each test function.
|
|
This ensures test isolation by creating and dropping tables for each test.
|
|
"""
|
|
Base.metadata.create_all(bind=engine)
|
|
db_session = TestingSessionLocal()
|
|
try:
|
|
yield db_session
|
|
finally:
|
|
db_session.close()
|
|
Base.metadata.drop_all(bind=engine)
|