kaya/packages/kaya-openapi/tests/test_openapi.py

import json
import unittest
from typing import Any, Mapping, Sequence

import httpx
from pwo import async_test

from kaya.core import HttpContext, HttpMethod, KayaApp, WebSocket
from kaya.openapi import OpenAPIMixin, operation


class OpenAPITest(unittest.TestCase):
    app: KayaApp

    def setUp(self) -> None:
        self.app = KayaApp(mixins=[OpenAPIMixin(title='Test API', version='1.2.3')])

        @self.app.GET('/hello')
        async def hello(ctx: HttpContext) -> None:
            """Say hello."""
            await ctx.send_str(200, 'Hello World!')

        @self.app.GET('/users/${user_id:int}')
        @operation(summary='Get a user',
                   tags=['users'],
                   responses={
                       200: {'description': 'The user'},
                       404: {'description': 'User not found'},
                   })
        async def get_user(ctx: HttpContext, user_id: int) -> None:
            await ctx.send_str(200, str(user_id))

        @self.app.POST('/users/${name}')
        async def create_user(ctx: HttpContext, name: str) -> None:
            await ctx.send_str(201, name)

        @self.app.route('/ping')
        async def ping(ctx: HttpContext) -> None:
            await ctx.send_str(200, 'pong')

        @self.app.GET('/files/*', recursive=True)
        async def serve_file(ctx: HttpContext, path: Sequence[str]) -> None:
            await ctx.send_str(200, '/'.join(path))

        @self.app.GET('/internal/health')
        @operation(hidden=True)
        async def health(ctx: HttpContext) -> None:
            await ctx.send_str(200, 'ok')

        @self.app.websocket('/echo')
        async def echo(ws: WebSocket) -> None:
            await ws.accept()
            await ws.close()

    async def _get_spec(self) -> Mapping[str, Any]:
        transport = httpx.ASGITransport(app=self.app)
        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
            response = await client.get('/openapi.json')
            self.assertEqual(200, response.status_code)
            self.assertEqual('application/json', response.headers['Content-Type'])
            return json.loads(response.text)

    @async_test
    async def test_spec_endpoint(self) -> None:
        spec = await self._get_spec()
        self.assertEqual('3.1.0', spec['openapi'])
        self.assertEqual({'title': 'Test API', 'version': '1.2.3'}, spec['info'])
        self.assertIn('paths', spec)

    @async_test
    async def test_static_route_with_docstring(self) -> None:
        spec = await self._get_spec()
        hello = spec['paths']['/hello']['get']
        self.assertEqual('Say hello.', hello['description'])
        self.assertIn('responses', hello)

    @async_test
    async def test_int_path_parameter(self) -> None:
        spec = await self._get_spec()
        operation = spec['paths']['/users/{user_id}']['get']
        self.assertEqual('Get a user', operation['summary'])
        self.assertEqual(['users'], operation['tags'])
        self.assertEqual({
            '200': {'description': 'The user'},
            '404': {'description': 'User not found'},
        }, operation['responses'])
        self.assertEqual(
            [{'name': 'user_id', 'in': 'path', 'required': True, 'schema': {'type': 'integer'}}],
            operation['parameters'])

    @async_test
    async def test_str_path_parameter(self) -> None:
        spec = await self._get_spec()
        operation = spec['paths']['/users/{name}']['post']
        self.assertEqual(
            [{'name': 'name', 'in': 'path', 'required': True, 'schema': {'type': 'string'}}],
            operation['parameters'])

    @async_test
    async def test_method_agnostic_route(self) -> None:
        spec = await self._get_spec()
        path_item = spec['paths']['/ping']
        for method in ('get', 'put', 'post', 'delete', 'options', 'head', 'patch'):
            self.assertIn(method, path_item)

    @async_test
    async def test_excluded_routes(self) -> None:
        spec = await self._get_spec()
        paths = spec['paths']
        # wildcard routes cannot be expressed in OpenAPI
        self.assertNotIn('/files/*', paths)
        self.assertFalse(any('files' in path for path in paths))
        # websocket routes are not part of OpenAPI
        self.assertNotIn('/echo', paths)
        # hidden operations are skipped
        self.assertNotIn('/internal/health', paths)
        # the mixin's own endpoints are excluded by default
        self.assertNotIn('/openapi.json', paths)
        self.assertNotIn('/docs', paths)

    @async_test
    async def test_docs_endpoint(self) -> None:
        transport = httpx.ASGITransport(app=self.app)
        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
            response = await client.get('/docs')
            self.assertEqual(200, response.status_code)
            self.assertEqual('text/html; charset=utf-8', response.headers['Content-Type'])
            self.assertIn('swagger-ui', response.text)
            self.assertIn('/openapi.json', response.text)

    @async_test
    async def test_late_registered_routes_are_included(self) -> None:
        @self.app.GET('/late')
        async def late(ctx: HttpContext) -> None:
            await ctx.send_str(200, 'late')

        spec = await self._get_spec()
        self.assertIn('/late', spec['paths'])

    @async_test
    async def test_custom_paths_and_self_inclusion(self) -> None:
        app = KayaApp(mixins=[OpenAPIMixin(title='Custom',
                                           version='0.1.0',
                                           spec_path='/spec.json',
                                           docs_path='/swagger',
                                           include_self=True)])

        @app.route('/items/${item_id:int}', HttpMethod.DELETE)
        async def delete_item(ctx: HttpContext, item_id: int) -> None:
            await ctx.send_empty(204)

        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
            response = await client.get('/spec.json')
            self.assertEqual(200, response.status_code)
            spec = json.loads(response.text)
            self.assertIn('delete', spec['paths']['/items/{item_id}'])
            self.assertIn('/spec.json', spec['paths'])
            self.assertIn('/swagger', spec['paths'])

            response = await client.get('/swagger')
            self.assertEqual(200, response.status_code)
            self.assertIn('/spec.json', response.text)


if __name__ == '__main__':
    unittest.main()