Add tasks filters

tasks/models.py

@@ -28,6 +28,29 @@ class Task(models.Model):
     def __str__(self) -> str:
         return self.title
 
+    @classmethod
+    def filter_by_status(cls, status_filter):
+        if not isinstance(status_filter, Status):
+            raise ValueError
+        return cls.objects.filter(status=status_filter)
+
+    @classmethod
+    def filter_by_assignee(cls, assignee_id):
+        if not isinstance(assignee_id, int):
+            raise TypeError
+        try:
+            user = User.objects.get(pk=assignee_id)
+        except User.DoesNotExist:
+            raise ValueError
+        return cls.objects.filter(assignee=user)
+
+    @classmethod
+    def filter_by_symbol(cls, priority_filter):
+        if not isinstance(priority_filter, Priority):
+            raise ValueError
+        print(priority_filter)
+        return cls.objects.filter(priority=priority_filter)
+
 
 class Comment(models.Model):
     user_id = models.ForeignKey(User, on_delete=models.RESTRICT, related_name='comments')

tasks/tests/test_tasks_filters.py

@@ -0,0 +1,140 @@
+from django.contrib.auth.models import User as DjangoUser
+from django.db.models.query import QuerySet
+import pytest
+from tasks.models import Priority, Task, Status
+from users.models import User, Team, Role
+
+pytestmark = pytest.mark.django_db
+
+
+class DBPrepare:
+    """
+    Create an employee test user
+    """
+    @staticmethod
+    def create_example_employee(username):
+        team = Team.objects.first()
+        django_user = DjangoUser.objects.create_user(username=username,
+                                                     email="example@gmail.com",
+                                                     password='xsdDS23',
+                                                     first_name="Test",
+                                                     last_name="Test")
+        employee = User.objects.create(user=django_user,
+                                       role=Role.EMPLOYEE,
+                                       team_id=team)
+        employee.save()
+        return employee
+
+    """
+    Create a manager test user
+    """
+    @staticmethod
+    def create_example_manager(username):
+        team = Team.objects.first()
+        django_user = DjangoUser.objects.create_user(username=username,
+                                                     email="example@gmail.com",
+                                                     password='xsdDS23',
+                                                     first_name="Test",
+                                                     last_name="Test")
+        manager = User.objects.create(user=django_user,
+                                      role=Role.MANAGER,
+                                      team_id=team)
+        manager.save()
+        return manager
+
+    """
+    Create a test team
+    """
+    @staticmethod
+    def create_example_team():
+        team = Team.objects.create(name="TestTeam", description="Test Team")
+        team.save()
+        return team
+
+    """
+    Create a test task
+    """
+    @staticmethod
+    def create_example_task(assignee, assigner, priority, status):
+        title = "Test task"
+        description = 'Test'
+        task = Task.objects.create(
+                                    title=title,
+                                    created_by=assigner,
+                                    assignee=assignee,
+                                    status=status,
+                                    priority=priority,
+                                    description=description)
+        task.save()
+        return task
+
+
+@pytest.mark.django_db
+class TestTasksFilters:
+    """
+    Add test data to database
+    """
+    @pytest.fixture
+    def prepare_database(self):
+        DBPrepare.create_example_team()
+        manager = DBPrepare.create_example_manager("TestManager")
+        for i in range(5):
+            employee = DBPrepare.create_example_employee(f'TestUser{i}')
+            DBPrepare.create_example_task(employee, manager, Priority.LOW, Status.BACKLOG)
+            DBPrepare.create_example_task(employee, manager, Priority.MEDIUM, Status.IN_PROGRESS)
+            DBPrepare.create_example_task(employee, manager, Priority.HIGH, Status.DONE)
+
+    """
+    Test filtering tasks by priority
+    """
+    def test_priority_filter(self, prepare_database):
+        for priority in (Priority.LOW, Priority.HIGH, Priority.MEDIUM):
+            filtered_tasks = Task.filter_by_symbol(priority)
+            assert isinstance(filtered_tasks, QuerySet)
+            assert all(isinstance(task, Task) for task in filtered_tasks)
+            assert len(filtered_tasks) == 5
+            assert all(p == priority for p in filtered_tasks.values_list('priority', flat=True))
+
+    """
+    Test that it is impossible to filter using invalid value
+    """
+    def test_invalid_priority_filter(self, prepare_database):
+        with pytest.raises(ValueError):
+            Task.filter_by_symbol('INVALID')
+
+    """
+    Test filtering tasks by asignee id
+    """
+    def test_assignee_filter(self, prepare_database):
+        filtered_tasks = Task.filter_by_assignee(2)
+        assert isinstance(filtered_tasks, QuerySet)
+        assert all(isinstance(task, Task) for task in filtered_tasks)
+        assert len(filtered_tasks) == 3
+        assert all(assignee_id == 2 for assignee_id in filtered_tasks.values_list('assignee', flat=True))
+
+    """
+    Test that it is impossible to filter using invalid value
+    """
+    def test_invalid_assignee_filter(self):
+        with pytest.raises(ValueError):
+            Task.filter_by_assignee(-2)
+        with pytest.raises(TypeError):
+            Task.filter_by_assignee('INVALID')
+
+    """
+    Test filtering tasks by status
+    """
+    def test_status_filter(self, prepare_database):
+        for status in Status:
+            filtered_tasks = Task.filter_by_status(status)
+            assert isinstance(filtered_tasks, QuerySet)
+            assert all(isinstance(task, Task) for task in filtered_tasks)
+            assert len(filtered_tasks) == 5
+            assert all(s == status for s in filtered_tasks.values_list('status', flat=True))
+
+    """
+    Test that it is impossible to filter using invalid value
+    """
+    def test_invalid_status_filter(self, prepare_database):
+        with pytest.raises(ValueError):
+            Task.filter_by_status('INVALID')