GitHub Copilot
682a85c6eb
- 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
57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
import pytest
|
|
from sqlalchemy.orm import Session
|
|
from src import crud, models
|
|
from src.simulation import run_simulation_cycle
|
|
|
|
def test_simulation_run_completes_task(db: Session):
|
|
# Arrange: Create a task group and a queued task
|
|
task_group = crud.create_task_group_simple(db, "test_group")
|
|
task = crud.create_task(db, task_group.id, "test_task_1")
|
|
assert task.status == "queued"
|
|
assert task_group.status == "queued"
|
|
|
|
# Act: Run the simulation cycle
|
|
run_simulation_cycle(db)
|
|
|
|
# Assert: Check if the task and task group are completed
|
|
db.refresh(task)
|
|
db.refresh(task_group)
|
|
assert task.status == "completed"
|
|
assert task_group.status == "completed"
|
|
|
|
def test_simulation_run_handles_multiple_tasks(db: Session):
|
|
# Arrange: Create a task group with two tasks
|
|
task_group = crud.create_task_group_simple(db, "multi_task_group")
|
|
task1 = crud.create_task(db, task_group.id, "multi_task_1")
|
|
task2 = crud.create_task(db, task_group.id, "multi_task_2")
|
|
|
|
# Act: Run the simulation cycle once
|
|
run_simulation_cycle(db)
|
|
|
|
# Assert: Both tasks should be completed, and the group status updated
|
|
db.refresh(task1)
|
|
db.refresh(task2)
|
|
db.refresh(task_group)
|
|
assert task1.status == "completed"
|
|
assert task2.status == "completed"
|
|
assert task_group.status == "completed"
|
|
|
|
def test_simulation_does_not_complete_group_if_tasks_remain(db: Session):
|
|
# Arrange: Create a task group with one completed and one queued task
|
|
task_group = crud.create_task_group_simple(db, "partial_group")
|
|
task1 = crud.create_task(db, task_group.id, "partial_task_1")
|
|
task2 = crud.create_task(db, task_group.id, "partial_task_2")
|
|
|
|
# Manually set one task to completed to simulate a partial run
|
|
task1.status = "completed"
|
|
db.commit()
|
|
|
|
# Act: Run the simulation cycle
|
|
run_simulation_cycle(db)
|
|
|
|
# Assert: The second task should be completed, but the group should also be completed now
|
|
db.refresh(task2)
|
|
db.refresh(task_group)
|
|
assert task2.status == "completed"
|
|
assert task_group.status == "completed"
|