create tests for validators (TDD)

Author: mahenzon

tests/fixtures/app.py

@@ -2,6 +2,7 @@
 from typing import Type
 
 import pytest
+import uvicorn
 from fastapi import APIRouter, FastAPI
 
 from fastapi_jsonapi import RoutersJSONAPI, init
@@ -19,6 +20,7 @@
     ParentToChildAssociation,
     Post,
     PostComment,
+    Task,
     User,
     UserBio,
 )
@@ -36,6 +38,9 @@
     PostInSchema,
     PostPatchSchema,
     PostSchema,
+    TaskInSchema,
+    TaskPatchSchema,
+    TaskSchema,
     UserBioSchema,
     UserInSchema,
     UserPatchSchema,
@@ -161,6 +166,19 @@ def add_routers(app_plain: FastAPI):
         schema_in_post=ComputerInSchema,
     )
 
+    RoutersJSONAPI(
+        router=router,
+        path="/tasks",
+        tags=["Task"],
+        class_detail=DetailViewBaseGeneric,
+        class_list=ListViewBaseGeneric,
+        model=Task,
+        schema=TaskSchema,
+        resource_type="task",
+        schema_in_patch=TaskPatchSchema,
+        schema_in_post=TaskInSchema,
+    )
+
     atomic = AtomicOperations()
 
     app_plain.include_router(router, prefix="")

tests/models.py

@@ -222,6 +222,15 @@ def __repr__(self):
         return f"{self.__class__.__name__}(id={self.id}, name={self.name!r}, user_id={self.user_id})"
 
 
+class Task(Base):
+    __tablename__ = "tasks"
+    id = Column(Integer, primary_key=True)
+    task_ids = Column(JSON, nullable=True, unique=False)
+
+
+# uuid below
+
+
 class CustomUUIDType(TypeDecorator):
     cache_ok = True
 

tests/schemas.py

@@ -1,6 +1,8 @@
 from typing import Dict, List, Optional
 from uuid import UUID
 
+from pydantic import validator
+
 from fastapi_jsonapi.schema_base import BaseModel, Field, RelationshipInfo
 
 
@@ -354,6 +356,40 @@ class Config:
     id: int
 
 
+# task
+class TaskBaseSchema(BaseModel):
+    class Config:
+        orm_mode = True
+
+    task_ids: Optional[list[str]] = None
+
+    # noinspection PyMethodParameters
+    @validator("task_ids", pre=True)
+    def task_ids_validator(cls, value: Optional[list[str]]):
+        """
+        return `[]`, if value is None
+        both on get and on create
+        """
+        return value or []
+
+
+class TaskPatchSchema(TaskBaseSchema):
+    """Task PATCH schema."""
+
+
+class TaskInSchema(TaskBaseSchema):
+    """Task create schema."""
+
+
+class TaskSchema(TaskBaseSchema):
+    """Task item schema."""
+
+    id: int
+
+
+# uuid below
+
+
 class IdCastSchema(BaseModel):
     id: UUID = Field(client_can_set_id=True)
 

tests/test_api/test_validators.py

@@ -0,0 +1,129 @@
+import pytest
+from fastapi import FastAPI, status
+from httpx import AsyncClient
+from pytest_asyncio import fixture
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from tests.models import Task
+from tests.schemas import TaskBaseSchema
+
+pytestmark = pytest.mark.asyncio
+
+
+@fixture()
+async def task_with_none_ids(
+    async_session: AsyncSession,
+) -> Task:
+    task = Task(task_ids=None)
+    async_session.add(task)
+    await async_session.commit()
+
+    yield task
+
+    await async_session.delete(task)
+    await async_session.commit()
+
+
+@pytest.fixture()
+def resource_type():
+    return "task"
+
+
+class TestTaskValidators:
+    async def test_base_model_root_validator_get_one(
+        self,
+        app: FastAPI,
+        client: AsyncClient,
+        resource_type: str,
+        task_with_none_ids: Task,
+    ):
+        url = app.url_path_for(f"get_{resource_type}_detail", obj_id=task_with_none_ids.id)
+        res = await client.get(url)
+        assert res.status_code == status.HTTP_200_OK, res.text
+        response_data = res.json()
+        attributes = response_data["data"].pop("attributes")
+        assert response_data == {
+            "data": {
+                "id": str(task_with_none_ids.id),
+                "type": resource_type,
+            },
+            "jsonapi": {"version": "1.0"},
+            "meta": None,
+        }
+        assert attributes == {
+            # not `None`! schema validator returns empty list `[]`
+            # "task_ids": None,
+            "task_ids": [],
+        }
+        assert attributes == TaskBaseSchema.from_orm(task_with_none_ids)
+
+    async def test_base_model_root_validator_get_list(
+        self,
+        app: FastAPI,
+        client: AsyncClient,
+        resource_type: str,
+        task_with_none_ids: Task,
+    ):
+        url = app.url_path_for(f"get_{resource_type}_list")
+        res = await client.get(url)
+        assert res.status_code == status.HTTP_200_OK, res.text
+        response_data = res.json()
+        assert response_data == {
+            "data": [
+                {
+                    "id": str(task_with_none_ids.id),
+                    "type": resource_type,
+                    "attributes": {
+                        # not `None`! schema validator returns empty list `[]`
+                        # "task_ids": None,
+                        "task_ids": [],
+                    },
+                },
+            ],
+            "jsonapi": {
+                "version": "1.0",
+            },
+            "meta": {
+                "count": 1,
+                "totalPages": 1,
+            },
+        }
+
+    async def test_base_model_root_validator_create(
+        self,
+        app: FastAPI,
+        client: AsyncClient,
+        resource_type: str,
+        async_session: AsyncSession,
+    ):
+        task_data = {
+            # should be converted to [] by schema on create
+            "task_ids": None,
+        }
+        data_create = {
+            "data": {
+                "type": resource_type,
+                "attributes": task_data,
+            },
+        }
+        url = app.url_path_for(f"create_{resource_type}_list")
+        res = await client.post(url, json=data_create)
+        assert res.status_code == status.HTTP_201_CREATED, res.text
+        response_data: dict = res.json()
+        task_id = response_data["data"].pop("id")
+        task = await async_session.get(Task, int(task_id))
+        assert isinstance(task, Task)
+        # we sent request with `None`, but value in db is `[]`
+        # because validator converted data before object creation
+        assert task.task_ids == []
+        assert response_data == {
+            "data": {
+                "type": resource_type,
+                "attributes": {
+                    # should be empty list
+                    "task_ids": [],
+                },
+            },
+            "jsonapi": {"version": "1.0"},
+            "meta": None,
+        }

tests/test_atomic/test_update_objects.py

@@ -2,15 +2,14 @@
 
 import pytest
 from httpx import AsyncClient
-from pytest import mark  # noqa
 from sqlalchemy.ext.asyncio import AsyncSession
 from starlette import status
 
 from tests.misc.utils import fake
 from tests.models import Computer, User, UserBio
 from tests.schemas import UserAttributesBaseSchema, UserBioAttributesBaseSchema
 
-pytestmark = mark.asyncio
+pytestmark = pytest.mark.asyncio
 
 logging.basicConfig(level=logging.DEBUG)