From adda0a0ef7e74773d2956319756b4fec3976992e Mon Sep 17 00:00:00 2001 From: rikasai233 Date: Sun, 21 Apr 2024 17:52:00 +0800 Subject: [PATCH] test: add more unittest, coverage up to 96% --- .coveragerc | 11 +++++++++ Makefile | 26 +++++++++++++++++++++ apidemo/urls.py | 4 ++-- core/middleware.py | 2 +- core/router.py | 2 +- core/schemas.py | 2 +- core/service.py | 57 +++++++++++++++++++++++++++------------------- core/tests.py | 45 ++++++++++++++++++++++++++++++++++-- employee/tests.py | 19 ++++++++++++---- manage.py | 2 +- 10 files changed, 135 insertions(+), 35 deletions(-) create mode 100644 .coveragerc create mode 100644 Makefile diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..ab84917 --- /dev/null +++ b/.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 \ No newline at end of file diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..9de46ad --- /dev/null +++ b/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 \ No newline at end of file diff --git a/apidemo/urls.py b/apidemo/urls.py index a31e2c2..ce02c05 100644 --- a/apidemo/urls.py +++ b/apidemo/urls.py @@ -36,7 +36,7 @@ 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) @@ -44,7 +44,7 @@ def obtain_token_exception_handler(request, exc): return response -def auth_error_exception_handler(request, exc): +def auth_error_exception_handler(request, exc): # pragma: no cover data = { "message": "AuthenticationError", "success": False, diff --git a/core/middleware.py b/core/middleware.py index bc28817..49c704b 100644 --- a/core/middleware.py +++ b/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 diff --git a/core/router.py b/core/router.py index 5625402..7a352fb 100644 --- a/core/router.py +++ b/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): diff --git a/core/schemas.py b/core/schemas.py index db8135f..96052e1 100644 --- a/core/schemas.py +++ b/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, diff --git a/core/service.py b/core/service.py index bb28ba7..245397c 100644 --- a/core/service.py +++ b/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) diff --git a/core/tests.py b/core/tests.py index 33ce912..e16bd16 100644 --- a/core/tests.py +++ b/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) \ No newline at end of file diff --git a/employee/tests.py b/employee/tests.py index f1a3b86..c9127f1 100644 --- a/employee/tests.py +++ b/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) + diff --git a/manage.py b/manage.py index 871903f..c51dd1a 100755 --- a/manage.py +++ b/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 "