didier/tests/test_database/test_crud/test_tasks.py

64 lines
2.2 KiB
Python
Raw Normal View History

2022-07-24 01:49:52 +02:00
import datetime
import pytest
from freezegun import freeze_time
from sqlalchemy import select
2022-07-27 21:10:43 +02:00
from sqlalchemy.ext.asyncio import AsyncSession
2022-07-24 01:49:52 +02:00
from database.crud import tasks as crud
from database.enums import TaskType
2022-08-29 20:24:42 +02:00
from database.schemas import Task
2022-07-24 01:49:52 +02:00
@pytest.fixture
def task_type() -> TaskType:
"""Fixture to use the same TaskType in every test"""
return TaskType.BIRTHDAYS
@pytest.fixture
2022-07-27 21:10:43 +02:00
async def task(postgres: AsyncSession, task_type: TaskType) -> Task:
2022-07-24 01:49:52 +02:00
"""Fixture to create a task"""
task = Task(task=task_type)
2022-07-25 19:12:27 +02:00
postgres.add(task)
await postgres.commit()
2022-07-24 01:49:52 +02:00
return task
2022-07-27 21:10:43 +02:00
async def test_get_task_by_enum_present(postgres: AsyncSession, task: Task, task_type: TaskType):
2022-07-24 01:49:52 +02:00
"""Test getting a task by its enum type when it exists"""
2022-07-25 19:12:27 +02:00
result = await crud.get_task_by_enum(postgres, task_type)
2022-07-24 01:49:52 +02:00
assert result is not None
assert result == task
2022-07-27 21:10:43 +02:00
async def test_get_task_by_enum_not_present(postgres: AsyncSession, task_type: TaskType):
2022-07-24 01:49:52 +02:00
"""Test getting a task by its enum type when it doesn't exist"""
2022-07-25 19:12:27 +02:00
result = await crud.get_task_by_enum(postgres, task_type)
2022-07-24 01:49:52 +02:00
assert result is None
@freeze_time("2022/07/24")
2022-07-27 21:10:43 +02:00
async def test_set_execution_time_exists(postgres: AsyncSession, task: Task, task_type: TaskType):
2022-07-24 01:49:52 +02:00
"""Test setting the execution time of an existing task"""
2022-07-25 19:12:27 +02:00
await postgres.refresh(task)
2022-07-24 01:49:52 +02:00
assert task.previous_run is None
2022-07-25 19:12:27 +02:00
await crud.set_last_task_execution_time(postgres, task_type)
await postgres.refresh(task)
2022-09-18 17:45:37 +02:00
assert task.previous_run == datetime.datetime(year=2022, month=7, day=24, tzinfo=datetime.timezone.utc)
2022-07-24 01:49:52 +02:00
@freeze_time("2022/07/24")
2022-07-27 21:10:43 +02:00
async def test_set_execution_time_doesnt_exist(postgres: AsyncSession, task_type: TaskType):
2022-07-24 01:49:52 +02:00
"""Test setting the execution time of a non-existing task"""
statement = select(Task).where(Task.task == task_type)
2022-07-25 19:12:27 +02:00
results = list((await postgres.execute(statement)).scalars().all())
2022-07-24 01:49:52 +02:00
assert len(results) == 0
2022-07-25 19:12:27 +02:00
await crud.set_last_task_execution_time(postgres, task_type)
results = list((await postgres.execute(statement)).scalars().all())
2022-07-24 01:49:52 +02:00
assert len(results) == 1
task = results[0]
2022-09-18 17:45:37 +02:00
assert task.previous_run == datetime.datetime(year=2022, month=7, day=24, tzinfo=datetime.timezone.utc)