Some checks failed
CI/CD Pipeline / Code Quality Checks (pull_request) Failing after 9s
CI/CD Pipeline / Run Tests (pull_request) Failing after 48s
CI/CD Pipeline / Build Docker Images (pull_request) Has been skipped
CI/CD Pipeline / Push to Gitea Registry (pull_request) Has been skipped
228 lines
8.7 KiB
Python
228 lines
8.7 KiB
Python
"""Tests for core OpenAPI utilities"""
|
||
|
||
import json
|
||
|
||
from apps.core.openapi import (
|
||
CommonParameters,
|
||
CommonResponses,
|
||
_get_status_description,
|
||
api_docs,
|
||
paginated_response,
|
||
)
|
||
from django.test import TestCase
|
||
from drf_yasg import openapi
|
||
from rest_framework import serializers
|
||
|
||
|
||
class DummySerializer(serializers.Serializer):
|
||
"""Dummy serializer for testing"""
|
||
|
||
id = serializers.IntegerField()
|
||
name = serializers.CharField()
|
||
|
||
|
||
class ApiDocsDecoratorTest(TestCase):
|
||
"""Tests for @api_docs decorator"""
|
||
|
||
def test_decorator_returns_function(self):
|
||
"""Test decorator returns wrapped function"""
|
||
|
||
@api_docs(summary="Test endpoint")
|
||
def my_view(request):
|
||
pass
|
||
|
||
self.assertTrue(callable(my_view))
|
||
|
||
def test_decorator_preserves_function_name(self):
|
||
"""Test decorator preserves original function name"""
|
||
|
||
@api_docs(summary="Test endpoint")
|
||
def my_view(request):
|
||
pass
|
||
|
||
self.assertEqual(my_view.__name__, "my_view")
|
||
|
||
|
||
class OpenApiSchemaViewTest(TestCase):
|
||
"""Tests for generated Swagger schema."""
|
||
|
||
def _operation_tags(self, paths: dict) -> list[str]:
|
||
tags = []
|
||
for operations in paths.values():
|
||
for method, operation in operations.items():
|
||
if method == "parameters":
|
||
continue
|
||
tags.extend(operation.get("tags", []))
|
||
return tags
|
||
|
||
def test_schema_exposes_parser_and_job_tags(self):
|
||
"""Test dashboard/parser APIs are visible as separate Swagger groups."""
|
||
response = self.client.get("/?format=openapi")
|
||
|
||
self.assertEqual(response.status_code, 200)
|
||
schema = json.loads(response.content)
|
||
paths = schema["paths"]
|
||
self.assertIn("/api/v1/parsers/sources/", paths)
|
||
self.assertIn("/api/v1/parsers/records/", paths)
|
||
self.assertIn("/api/v1/parsers/run/{source_key}/", paths)
|
||
self.assertIn("/api/v1/parsers/upload/{source_key}/", paths)
|
||
self.assertIn("/api/v1/users/register/", paths)
|
||
self.assertIn("/api/v1/users/login/", paths)
|
||
self.assertIn("/api/v1/users/token/refresh/", paths)
|
||
self.assertIn("/api/v1/users/token/verify/", paths)
|
||
self.assertIn("/api/v1/users/me/", paths)
|
||
self.assertIn("/api/v1/fns/reports/", paths)
|
||
self.assertIn("/api/v1/fns/reports/{id}/", paths)
|
||
self.assertIn("/api/v1/fns/upload/", paths)
|
||
self.assertIn("/api/v1/minpromtorg/manufacturers/", paths)
|
||
self.assertIn("/api/v1/zakupki/", paths)
|
||
self.assertIn("/api/v1/zakupki/{id}/", paths)
|
||
self.assertIn("/api/v1/jobs/", paths)
|
||
self.assertNotIn("/api/v1/zakupki/procurements-44fz/", paths)
|
||
self.assertNotIn("/api/v1/zakupki/procurements-223fz/", paths)
|
||
self.assertNotIn("/api/v1/zakupki/contracts/", paths)
|
||
self.assertNotIn("/api/v1/zakupki/upload/", paths)
|
||
self.assertNotIn("/api/v2/fns/reports/", paths)
|
||
self.assertEqual(
|
||
paths["/api/v1/parsers/sources/"]["get"]["tags"],
|
||
["Parser Management"],
|
||
)
|
||
self.assertEqual(
|
||
paths["/api/v1/parsers/upload/{source_key}/"]["post"]["tags"],
|
||
["Parser Management"],
|
||
)
|
||
self.assertEqual(
|
||
paths["/api/v1/fns/reports/"]["get"]["tags"],
|
||
["FNS"],
|
||
)
|
||
self.assertEqual(
|
||
paths["/api/v1/fns/reports/{id}/"]["get"]["tags"],
|
||
["FNS"],
|
||
)
|
||
self.assertEqual(
|
||
paths["/api/v1/fns/upload/"]["post"]["tags"],
|
||
["FNS"],
|
||
)
|
||
self.assertEqual(
|
||
paths["/api/v1/minpromtorg/manufacturers/"]["get"]["tags"],
|
||
["Minpromtorg"],
|
||
)
|
||
self.assertEqual(
|
||
paths["/api/v1/zakupki/"]["get"]["tags"],
|
||
["EIS Zakupki"],
|
||
)
|
||
for path in (
|
||
"/api/v1/users/register/",
|
||
"/api/v1/users/login/",
|
||
"/api/v1/users/token/refresh/",
|
||
"/api/v1/users/token/verify/",
|
||
"/api/v1/users/me/",
|
||
):
|
||
operation = (
|
||
paths[path]["post"] if "post" in paths[path] else paths[path]["get"]
|
||
)
|
||
self.assertEqual(operation["tags"], ["Users"])
|
||
for path in (
|
||
"/api/v1/users/register/",
|
||
"/api/v1/users/login/",
|
||
"/api/v1/users/token/refresh/",
|
||
"/api/v1/users/token/verify/",
|
||
):
|
||
self.assertEqual(paths[path]["post"]["security"], [])
|
||
self.assertEqual(paths["/api/v1/jobs/"]["get"]["tags"], ["Jobs"])
|
||
for tag in self._operation_tags(paths):
|
||
self.assertFalse(any("\u0400" <= char <= "\u04ff" for char in tag), tag)
|
||
self.assertIn("Bearer", schema["securityDefinitions"])
|
||
|
||
|
||
class GetStatusDescriptionTest(TestCase):
|
||
"""Tests for _get_status_description function"""
|
||
|
||
def test_known_status_codes(self):
|
||
"""Test known status codes return Russian descriptions"""
|
||
self.assertEqual(_get_status_description(200), "Успешный запрос")
|
||
self.assertEqual(_get_status_description(201), "Ресурс создан")
|
||
self.assertEqual(_get_status_description(400), "Некорректный запрос")
|
||
self.assertEqual(_get_status_description(401), "Не авторизован")
|
||
self.assertEqual(_get_status_description(403), "Доступ запрещён")
|
||
self.assertEqual(_get_status_description(404), "Ресурс не найден")
|
||
self.assertEqual(_get_status_description(500), "Внутренняя ошибка сервера")
|
||
|
||
def test_unknown_status_code(self):
|
||
"""Test unknown status code returns generic description"""
|
||
result = _get_status_description(418)
|
||
self.assertEqual(result, "HTTP 418")
|
||
|
||
|
||
class CommonResponsesTest(TestCase):
|
||
"""Tests for CommonResponses class"""
|
||
|
||
def test_success_response_type(self):
|
||
"""Test SUCCESS is an openapi.Response"""
|
||
self.assertIsInstance(CommonResponses.SUCCESS, openapi.Response)
|
||
|
||
def test_created_response_type(self):
|
||
"""Test CREATED is an openapi.Response"""
|
||
self.assertIsInstance(CommonResponses.CREATED, openapi.Response)
|
||
|
||
def test_not_found_response_type(self):
|
||
"""Test NOT_FOUND is an openapi.Response"""
|
||
self.assertIsInstance(CommonResponses.NOT_FOUND, openapi.Response)
|
||
|
||
def test_unauthorized_response_type(self):
|
||
"""Test UNAUTHORIZED is an openapi.Response"""
|
||
self.assertIsInstance(CommonResponses.UNAUTHORIZED, openapi.Response)
|
||
|
||
def test_validation_error_response_type(self):
|
||
"""Test VALIDATION_ERROR is an openapi.Response"""
|
||
self.assertIsInstance(CommonResponses.VALIDATION_ERROR, openapi.Response)
|
||
|
||
def test_server_error_response_type(self):
|
||
"""Test SERVER_ERROR is an openapi.Response"""
|
||
self.assertIsInstance(CommonResponses.SERVER_ERROR, openapi.Response)
|
||
|
||
|
||
class CommonParametersTest(TestCase):
|
||
"""Tests for CommonParameters class"""
|
||
|
||
def test_page_parameter(self):
|
||
"""Test PAGE parameter configuration"""
|
||
self.assertEqual(CommonParameters.PAGE.name, "page")
|
||
self.assertEqual(CommonParameters.PAGE.in_, openapi.IN_QUERY)
|
||
self.assertEqual(CommonParameters.PAGE.type, openapi.TYPE_INTEGER)
|
||
|
||
def test_page_size_parameter(self):
|
||
"""Test PAGE_SIZE parameter configuration"""
|
||
self.assertEqual(CommonParameters.PAGE_SIZE.name, "page_size")
|
||
self.assertEqual(CommonParameters.PAGE_SIZE.in_, openapi.IN_QUERY)
|
||
|
||
def test_search_parameter(self):
|
||
"""Test SEARCH parameter configuration"""
|
||
self.assertEqual(CommonParameters.SEARCH.name, "search")
|
||
self.assertEqual(CommonParameters.SEARCH.type, openapi.TYPE_STRING)
|
||
|
||
def test_ordering_parameter(self):
|
||
"""Test ORDERING parameter configuration"""
|
||
self.assertEqual(CommonParameters.ORDERING.name, "ordering")
|
||
self.assertEqual(CommonParameters.ORDERING.type, openapi.TYPE_STRING)
|
||
|
||
def test_id_parameter(self):
|
||
"""Test ID parameter configuration"""
|
||
self.assertEqual(CommonParameters.ID.name, "id")
|
||
self.assertEqual(CommonParameters.ID.in_, openapi.IN_PATH)
|
||
self.assertTrue(CommonParameters.ID.required)
|
||
|
||
|
||
class PaginatedResponseTest(TestCase):
|
||
"""Tests for paginated_response function"""
|
||
|
||
def test_returns_response_object(self):
|
||
"""Test function returns openapi.Response"""
|
||
result = paginated_response(DummySerializer)
|
||
self.assertIsInstance(result, openapi.Response)
|
||
|
||
def test_response_has_description(self):
|
||
"""Test response has description"""
|
||
result = paginated_response(DummySerializer)
|
||
self.assertEqual(result.description, "Пагинированный список")
|