creativesar/taskflow

1"""2Unit tests for task service layer3"""4import pytest5from unittest.mock import AsyncMock, MagicMock6from datetime import datetime7 8# Import the service functions9from services.task_service import (10    create_task,11    list_tasks,12    get_task,13    update_task,14    delete_task,15    toggle_completion16)17from models import Task18 19 20class TestTaskService:21    """Test cases for task service functions."""22 23    @pytest.mark.asyncio24    async def test_create_task_success(self, mock_session, sample_task):25        """Test successful task creation."""26        # Setup mock27        mock_session.refresh = AsyncMock()28 29        # Execute30        result = await create_task(31            session=mock_session,32            user_id="test-user-id",33            title="Test Task",34            description="Test description"35        )36 37        # Verify38        assert mock_session.add.called39        assert mock_session.commit.called40        assert mock_session.refresh.called41        assert result.title == "Test Task"42        assert result.description == "Test description"43        assert result.user_id == "test-user-id"44        assert result.completed is False45 46    @pytest.mark.asyncio47    async def test_create_task_minimal(self, mock_session):48        """Test task creation with only required fields."""49        mock_session.refresh = AsyncMock()50 51        result = await create_task(52            session=mock_session,53            user_id="test-user-id",54            title="Minimal Task"55        )56 57        assert result.title == "Minimal Task"58        assert result.description is None59 60    @pytest.mark.asyncio61    async def test_list_tasks_empty(self, mock_session):62        """Test listing tasks when none exist."""63        from sqlmodel import select64 65        # Mock empty result66        mock_result = MagicMock()67        mock_result.scalars().all.return_value = []68        mock_session.execute = AsyncMock(return_value=mock_result)69 70        result = await list_tasks(mock_session, "test-user-id")71 72        assert result == []73        assert mock_session.execute.called74 75    @pytest.mark.asyncio76    async def test_list_tasks_with_data(self, mock_session, sample_task):77        """Test listing tasks when some exist."""78        from sqlmodel import select79 80        # Mock result with tasks81        mock_result = MagicMock()82        mock_result.scalars().all.return_value = [sample_task]83        mock_session.execute = AsyncMock(return_value=mock_result)84 85        result = await list_tasks(mock_session, "test-user-id")86 87        assert len(result) == 188        assert result[0].title == "Test Task"89 90    @pytest.mark.asyncio91    async def test_get_task_exists(self, mock_session, sample_task):92        """Test getting an existing task."""93        from sqlmodel import select94 95        mock_result = MagicMock()96        mock_result.scalar_one_or_none.return_value = sample_task97        mock_session.execute = AsyncMock(return_value=mock_result)98 99        result = await get_task(mock_session, 1, "test-user-id")100 101        assert result is not None102        assert result.id == 1103        assert result.title == "Test Task"104 105    @pytest.mark.asyncio106    async def test_get_task_not_found(self, mock_session):107        """Test getting a non-existent task."""108        from sqlmodel import select109 110        mock_result = MagicMock()111        mock_result.scalar_one_or_none.return_value = None112        mock_session.execute = AsyncMock(return_value=mock_result)113 114        result = await get_task(mock_session, 999, "test-user-id")115 116        assert result is None117 118    @pytest.mark.asyncio119    async def test_update_task_success(self, mock_session, sample_task):120        """Test successful task update."""121        from sqlmodel import select122 123        # First call returns the task, second call after refresh124        mock_result = MagicMock()125        mock_result.scalar_one_or_none.return_value = sample_task126        mock_session.execute = AsyncMock(return_value=mock_result)127        mock_session.refresh = AsyncMock()128 129        result = await update_task(130            mock_session,131            task_id=1,132            user_id="test-user-id",133            title="Updated Title",134            description="Updated Description"135        )136 137        assert mock_session.commit.called138        assert mock_session.refresh.called139        assert result is not None140 141    @pytest.mark.asyncio142    async def test_update_task_not_found(self, mock_session):143        """Test updating a non-existent task."""144        from sqlmodel import select145 146        mock_result = MagicMock()147        mock_result.scalar_one_or_none.return_value = None148        mock_session.execute = AsyncMock(return_value=mock_result)149 150        result = await update_task(151            mock_session,152            task_id=999,153            user_id="test-user-id",154            title="Updated Title"155        )156 157        assert result is None158        assert not mock_session.commit.called159 160    @pytest.mark.asyncio161    async def test_delete_task_success(self, mock_session, sample_task):162        """Test successful task deletion."""163        from sqlmodel import select164 165        mock_session.delete = AsyncMock()166 167        mock_result = MagicMock()168        mock_result.scalar_one_or_none.return_value = sample_task169        mock_session.execute = AsyncMock(return_value=mock_result)170 171        result = await delete_task(mock_session, 1, "test-user-id")172 173        assert result is True174        assert mock_session.delete.called175        assert mock_session.commit.called176 177    @pytest.mark.asyncio178    async def test_delete_task_not_found(self, mock_session):179        """Test deleting a non-existent task."""180        from sqlmodel import select181 182        mock_result = MagicMock()183        mock_result.scalar_one_or_none.return_value = None184        mock_session.execute = AsyncMock(return_value=mock_result)185 186        result = await delete_task(mock_session, 999, "test-user-id")187 188        assert result is False189        assert not mock_session.delete.called190 191    @pytest.mark.asyncio192    async def test_toggle_completion(self, mock_session, sample_task):193        """Test toggling task completion."""194        from sqlmodel import select195 196        # Create a fresh task with completed=False for the test197        from datetime import datetime198        incomplete_task = Task(199            id=1,200            user_id="test-user-id",201            title="Test Task",202            description="Test description",203            completed=False,204            created_at=datetime(2025, 1, 1),205            updated_at=datetime(2025, 1, 1)206        )207 208        mock_result = MagicMock()209        mock_result.scalar_one_or_none.return_value = incomplete_task210        mock_session.execute = AsyncMock(return_value=mock_result)211        mock_session.refresh = AsyncMock()212 213        result = await toggle_completion(mock_session, 1, "test-user-id")214 215        assert mock_session.commit.called216        assert mock_session.refresh.called217 218 219class TestTaskModel:220    """Test cases for Task model."""221 222    def test_task_creation(self):223        """Test Task model instantiation."""224        task = Task(225            user_id="user-123",226            title="Test Task",227            description="Description"228        )229 230        assert task.user_id == "user-123"231        assert task.title == "Test Task"232        assert task.description == "Description"233        assert task.completed is False234 235    def test_task_default_completed(self):236        """Test that completed defaults to False."""237        task = Task(user_id="user-123", title="Test")238 239        assert task.completed is False240