File size: 3,838 Bytes
310260a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | """
Task Helper Utilities for Testing
This module provides utility functions for creating and managing tasks in tests.
"""
from typing import Optional, List
from datetime import datetime
from sqlmodel import Session
from src.models.task import Task
def create_test_task(
session: Session,
user_id: int,
title: str = "Test Task",
description: Optional[str] = None,
completed: bool = False
) -> Task:
"""
Create a test task in the database.
Args:
session: Database session
user_id: User ID who owns the task
title: Task title
description: Optional task description
completed: Completion status
Returns:
Created Task instance
"""
task = Task(
user_id=user_id,
title=title,
description=description,
completed=completed,
created_at=datetime.utcnow(),
updated_at=datetime.utcnow()
)
session.add(task)
session.commit()
session.refresh(task)
return task
def create_multiple_tasks(
session: Session,
user_id: int,
count: int,
title_prefix: str = "Task",
completed: bool = False
) -> List[Task]:
"""
Create multiple test tasks for a user.
Args:
session: Database session
user_id: User ID who owns the tasks
count: Number of tasks to create
title_prefix: Prefix for task titles
completed: Completion status for all tasks
Returns:
List of created Task instances
"""
tasks = []
for i in range(count):
task = create_test_task(
session=session,
user_id=user_id,
title=f"{title_prefix} {i+1}",
description=f"Description for {title_prefix} {i+1}",
completed=completed
)
tasks.append(task)
return tasks
def get_task_by_id(session: Session, task_id: int) -> Optional[Task]:
"""
Retrieve a task by ID.
Args:
session: Database session
task_id: Task ID to retrieve
Returns:
Task instance or None if not found
"""
return session.get(Task, task_id)
def delete_all_tasks(session: Session, user_id: Optional[int] = None):
"""
Delete all tasks, optionally filtered by user.
Args:
session: Database session
user_id: Optional user ID to filter by
"""
if user_id:
session.query(Task).filter(Task.user_id == user_id).delete()
else:
session.query(Task).delete()
session.commit()
def count_tasks(session: Session, user_id: int, completed: Optional[bool] = None) -> int:
"""
Count tasks for a user, optionally filtered by completion status.
Args:
session: Database session
user_id: User ID to count tasks for
completed: Optional completion status filter
Returns:
Number of tasks matching criteria
"""
query = session.query(Task).filter(Task.user_id == user_id)
if completed is not None:
query = query.filter(Task.completed == completed)
return query.count()
def assert_task_equals(task: Task, expected_title: str, expected_user_id: int, expected_completed: bool = False):
"""
Assert that a task matches expected values.
Args:
task: Task instance to check
expected_title: Expected task title
expected_user_id: Expected user ID
expected_completed: Expected completion status
Raises:
AssertionError: If task doesn't match expected values
"""
assert task.title == expected_title, f"Expected title '{expected_title}', got '{task.title}'"
assert task.user_id == expected_user_id, f"Expected user_id {expected_user_id}, got {task.user_id}"
assert task.completed == expected_completed, f"Expected completed {expected_completed}, got {task.completed}"
|