Merge pull request #12 from lihuacai168/feat/crud-router

test: add more unittest, coverage up to 96%

.coveragerc

@@ -0,0 +1,11 @@
+[run]
+omit =
+    */migrations/*
+    */tests/*
+    apidemo/asgi.py
+    apidemo/settings_dev.py
+    apidemo/settings_docker.py
+    apidemo/wsgi.py
+    apidemo/wsgi_docker.py
+    core/cache.py
+    apidemo/celery_config.py

Makefile

@@ -0,0 +1,26 @@
+# Makefile
+
+# Defining the directory variables
+PROJECT_DIR = .
+COVERAGE_DIR = htmlcov
+
+# Run Django tests and collect coverage data
+test:
+	@echo "Running tests with coverage..."
+	@coverage run --source='$(PROJECT_DIR)' manage.py test
+
+# Generate an HTML coverage report
+html-report:
+	@echo "Generating HTML coverage report..."
+	@coverage html
+
+# Open the HTML coverage report in the browser
+open-report:
+	@echo "Opening HTML coverage report..."
+	@open $(COVERAGE_DIR)/index.html
+
+# Combine commands: run tests, generate and open HTML coverage report
+coverage: test html-report open-report
+
+# Marking targets as phony, they are not files
+.PHONY: test html-report open-report coverage

apidemo/urls.py

@@ -36,15 +36,15 @@ def obtain_token_exception_handler(request, exc):
             "success": False,
             "data": None,
         }
-    else:
+    else:  # pragma: no cover
         data = {"message": exc.detail, "success": False, "data": None}
 
     response = api_v1.create_response(request, data, status=exc.status_code)
 
     return response
 
 
-def auth_error_exception_handler(request, exc):
+def auth_error_exception_handler(request, exc): # pragma: no cover
     data = {
         "message": "AuthenticationError",
         "success": False,

core/middleware.py

@@ -29,7 +29,7 @@ def __call__(self, request):
                 _response.status_code = response.status_code
                 response = _response
 
-            except Exception:
+            except Exception:  # pragma: no cover
                 ...
 
         return response

core/router.py

@@ -52,7 +52,7 @@ def list_obj(request, filters: self.filters_class = Query(...)):
         # full update obj
         @self.put(
             f"{self.path}/{{id}}",
-            response=StandResponse[Union[DictId, None]],
+            response=StandResponse[Union[DictId, dict]],
             description="full obj update",
         )
         def update_obj(request, id: int, payload: self.in_schema):

core/schemas.py

@@ -37,7 +37,7 @@ class PageFilter(Schema):
     def page_index_check(cls, page_index):
         if page_index <= 1:
             return 1
-        return page_index
+        return page_index  # pragma: no cover
 
     def dict(
         self,

core/service.py

@@ -2,65 +2,68 @@
 from abc import ABC, abstractmethod
 from typing import Any, Union
 
-from django.db.models import Model
-from django.http import Http404
-from django.shortcuts import get_object_or_404
-
 from core import cache, response
 from core.model import CoreModelSoftDelete
 from core.schemas import DictId, PageFilter, PageSchema, StandResponse
+from django.db.models import Model
+from django.http import Http404
+from django.shortcuts import get_object_or_404
 from utils import model_opertion
 from utils.model_opertion import GenericPayload
 
 
 class BaseCURD(ABC):
     @abstractmethod
-    def create_obj(self, payload: GenericPayload) -> DictId:
+    def create_obj(self, payload: GenericPayload) -> DictId:  # pragma: no cover
         ...
 
     @abstractmethod
-    def get_obj(self, id: int) -> Union[Model, None]:
+    def get_obj(self, id: int) -> Union[Model, None]:  # pragma: no cover
         ...
 
     @abstractmethod
-    def list_obj(self, page_filter: PageFilter) -> PageSchema:
+    def list_obj(self, page_filter: PageFilter) -> PageSchema:  # pragma: no cover
         ...
 
     @abstractmethod
     def update_obj(
         self, id: int, payload: GenericPayload, user_email: str
-    ) -> Union[DictId, dict]:
+    ) -> Union[DictId, dict]:  # pragma: no cover
         ...
 
-    def delete_obj(self, id: int) -> bool:
+    def delete_obj(self, id: int) -> bool:  # pragma: no cover
         ...
 
     def partial_update(
         self, id: int, user_email: str, **fields_kv
-    ) -> Union[DictId, dict]:
+    ) -> Union[DictId, dict]:  # pragma: no cover
         ...
 
     @staticmethod
-    def query_or_cache(ttl: int, alias: str, key: str, func, *args, **kwargs):
+    def query_or_cache(
+        ttl: int, alias: str, key: str, func, *args, **kwargs
+    ):  # pragma: no cover
         return cache.query_or_cache(ttl, alias, key, func, *args, **kwargs)
 
     @staticmethod
-    def query_or_cache_default(func, key: str, *args, **kwargs):
+    def query_or_cache_default(func, key: str, *args, **kwargs):  # pragma: no cover
         return cache.query_or_cache_default_10min(func, key, *args, **kwargs)
 
     @staticmethod
-    def query_or_cache_ttl(func, key: str, ttl: int, *args, **kwargs):
+    def query_or_cache_ttl(
+        func, key: str, ttl: int, *args, **kwargs
+    ):  # pragma: no cover
         return cache.query_or_cache_default(func, key, ttl, *args, **kwargs)
 
     @staticmethod
-    def execute_sql(sql: str, db_conn):
+    def execute_sql(sql: str, db_conn):  # pragma: no cover
         with db_conn.cursor() as cursor:
             cursor.execute(sql)
             result = cursor.fetchall()
             return result
 
 
-class GenericCURD(BaseCURD):
+class GenericCURD(BaseCURD):  # pragma: no cover
     def __init__(self, model):
         self.model = model
 
@@ -74,7 +77,9 @@ def get_obj(self, id: int) -> StandResponse:
 
     def list_obj(self, page_filter: PageFilter, page_schema: PageSchema) -> PageSchema:
         qs = self.model.objects.filter(**page_filter.dict())
-        return response.get_page(queryset=qs, pager_filter=page_filter, generic_result_type=page_schema)
+        return response.get_page(
+            queryset=qs, pager_filter=page_filter, generic_result_type=page_schema
+        )
 
     def update_obj(
         self, id: int, payload, user_email
@@ -108,7 +113,7 @@ def create_obj_with_validate_unique(
         payload,
         user_email: str,
         exclude: Any = None,
-    ) -> StandResponse[Union[DictId, dict]]:
+    ) -> StandResponse[Union[DictId, dict]]:  # pragma: no cover
         return model_opertion.create_obj_with_validate_unique(
             model=self.model,
             payload=payload,
@@ -128,29 +133,35 @@ def get_obj(self, id: int) -> StandResponse:
 
     def list_obj(self, page_filter: PageFilter, page_schema: PageSchema) -> PageSchema:
         qs = self.model.objects.filter(**page_filter.dict(), is_deleted=0)
-        return response.get_page(queryset=qs, pager_filter=page_filter, generic_result_type=page_schema)
+        return response.get_page(
+            queryset=qs, pager_filter=page_filter, generic_result_type=page_schema
+        )
 
     def update_obj(
         self, id: int, payload: GenericPayload, user_email: str
-    ) -> StandResponse[Union[DictId, None]]:
+    ) -> StandResponse[Union[DictId, dict]]:
         obj = self._get_obj_by_id(id=id)
         if obj is None:
-            return StandResponse[dict](message=f"{id=} not exist", success=False, data={})
+            return StandResponse[dict](
+                message=f"{id=} not exist", success=False, data={}
+            )
         return model_opertion.update_by_obj(
             updater=user_email, obj=obj, **payload.dict()
         )
 
-    def partial_update(self, id: int, user_email: str, **fields_kv) -> StandResponse[Union[DictId, None]]:
+    def partial_update(
+        self, id: int, user_email: str, **fields_kv
+    ) -> StandResponse[Union[DictId, dict]]:
         obj = self._get_obj_by_id(id=id)
         return model_opertion.update_by_obj(updater=user_email, obj=obj, **fields_kv)
 
     def delete_obj(self, id: int) -> StandResponse[bool]:
         obj = self._get_obj_by_id(id=id)
         if obj is None:
-            return StandResponse[bool](data=False, message=f"{id=} not exist")
+            return StandResponse[bool](data=False, message=f"{id=} not exist", success=False)
         obj.is_deleted = str(uuid.uuid1())
         obj.save()
         return StandResponse[bool](data=True)
 
-    def bulk_create(self, objs: list) -> list:
+    def bulk_create(self, objs: list) -> list:  # pragma: no cover
         return self.model.objects.bulk_create(objs)

core/tests.py

@@ -1,13 +1,15 @@
 from django.test import TestCase, Client
 from django.contrib.auth.models import User
 
+from core.service import GenericCURD
+
 
 class AuthTokenTestCase(TestCase):
     def setUp(self):
         self.client = Client()
         self.user = User.objects.create_user(username="testuser", password="12345")
 
-    def test_auth_token_success(self):
+    def test_get_token_fresh_success(self):
         """Test obtain pair token
         """
         response = self.client.post(
@@ -28,7 +30,14 @@ def test_auth_token_success(self):
 
         self.assertEqual(response_data["success"], True)
 
-    def test_auth_token_failure(self):
+        response = self.client.post(
+            "/api/token/refresh",
+            data={"refresh": response_data["data"]["refresh"]},
+            content_type="application/json",
+        ).json()
+        self.assertEqual(response['data']['refresh'], response_data['data']['refresh'])
+
+    def test_auth_token_failure_wrong_user_and_password(self):
         """Test the failure of authentication token generation.
         """
         response = self.client.post(
@@ -43,3 +52,35 @@ def test_auth_token_failure(self):
             "No active account found with the given credentials",
             response.json()["message"],
         )
+
+    def test_auth_token_failure_right_user_and_wrong_password(self):
+        """Test the failure of authentication token generation.
+        """
+        response = self.client.post(
+            "/api/token/pair",
+            {"username": "testuser", "password": "wrongpassword"},
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, 401)
+        self.assertEqual(response.json()["success"], False)
+        self.assertIn("message", response.json())
+        self.assertIn(
+            "No active account found with the given credentials",
+            response.json()["message"],
+        )
+
+    def test_refresh_token_failure(self):
+        response = self.client.post(
+            "/api/token/refresh",
+            data={"refresh": "12345"},
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, 401)
+
+    def test_access_token_failure(self):
+        response = self.client.post(
+            "/api/token/refresh",
+            data={"refresh": "12345"},
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, 401)

employee/tests.py

@@ -35,7 +35,7 @@ def request(
         return self._call(func, request, kwargs)  # type: ignore
 
 
-class HelloTest(TestCase):
+class FastCrudRouterTest(TestCase):
     def setUp(self):
         self.employee_in = EmployeeIn(
             first_name="John",
@@ -75,7 +75,7 @@ def test_create_obj(self):
         self.assertEqual(response.json()["data"]["id"], 2)
 
     def test_list_obj(self):
-        response = self.token_client.get("/employees")
+        response = self.token_client.get("/employees?page_index=-1")
         self.assertEqual(response.status_code, 200)
         # Check if response contains list of employees
         self.assertIsInstance(response.json()["data"]["details"], list)
@@ -100,6 +100,12 @@ def test_update_obj(self):
             update_data_response.json()["data"]["first_name"], update_data["first_name"]
         )
 
+        # no exist id
+        response = self.token_client.put("/employees/2", json=update_data).json()
+        self.assertEqual(response["success"], False)
+        #
+
+
     def test_partial_update_obj(self):
         partial_update_data = {"first_name": "Partially Updated"}
         response = self.token_client.patch("/employees/1", json=partial_update_data)
@@ -115,8 +121,13 @@ def test_partial_update_obj(self):
         )
 
     def test_delete_obj(self):
-        response = self.token_client.delete("/employees/1")
-        self.assertEqual(response.status_code, 200)
+        response1 = self.token_client.delete("/employees/1").json()
+        self.assertEqual(response1["success"], True)
+
         # Check if object does not exist anymore
         with self.assertRaises(ObjectDoesNotExist):
             Employee.objects.get(pk=1, is_deleted=False)
+
+        response2 = self.token_client.delete("/employees/1").json()
+        self.assertEqual(response2["success"], False)
+

manage.py

@@ -9,7 +9,7 @@ def main():
     os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'apidemo.settings')
     try:
         from django.core.management import execute_from_command_line
-    except ImportError as exc:
+    except ImportError as exc:  # pragma: no cover
         raise ImportError(
             "Couldn't import Django. Are you sure it's installed and "
             "available on your PYTHONPATH environment variable? Did you "