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, generate_spec, operation, schema_ref


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)


class ComponentsTest(unittest.TestCase):
    """Reusable schemas through ``components`` and ``schema_ref``."""

    async def _spec_of(self, app: KayaApp) -> Mapping[str, Any]:
        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('/openapi.json')
            self.assertEqual(200, response.status_code)
            return json.loads(response.text)

    def test_schema_ref_helper(self) -> None:
        self.assertEqual({'$ref': '#/components/schemas/User'}, schema_ref('User'))

    @async_test
    async def test_components_merged_and_refs_preserved(self) -> None:
        user = {'type': 'object', 'properties': {'id': {'type': 'integer'}}}
        page = {'type': 'object',
                'properties': {'results': {'type': 'array', 'items': schema_ref('User')}}}
        app = KayaApp(mixins=[OpenAPIMixin(
            title='Components API',
            version='1.0.0',
            components={'schemas': {'User': user, 'UserPage': page}},
        )])

        @app.GET('/users')
        @operation(responses={200: {
            'description': 'A page of users',
            'content': {'application/json': {'schema': schema_ref('UserPage')}},
        }})
        async def list_users(ctx: HttpContext) -> None:
            await ctx.send_str(200, '[]')

        spec = await self._spec_of(app)
        self.assertEqual({'schemas': {'User': user, 'UserPage': page}}, spec['components'])
        schema = spec['paths']['/users']['get']['responses']['200']['content']['application/json']['schema']
        self.assertEqual(schema_ref('UserPage'), schema)

    @async_test
    async def test_request_body_ref_preserved(self) -> None:
        app = KayaApp(mixins=[OpenAPIMixin(
            title='Components API',
            version='1.0.0',
            components={'schemas': {'User': {'type': 'object'}}},
        )])

        @app.POST('/users')
        @operation(request_body={'required': True,
                                 'content': {'application/json': {'schema': schema_ref('User')}}},
                   responses={201: {'description': 'Created'}})
        async def create_user(ctx: HttpContext) -> None:
            await ctx.send_str(201, '{}')

        spec = await self._spec_of(app)
        schema = spec['paths']['/users']['post']['requestBody']['content']['application/json']['schema']
        self.assertEqual(schema_ref('User'), schema)

    @async_test
    async def test_default_spec_has_no_components(self) -> None:
        app = KayaApp(mixins=[OpenAPIMixin(title='Plain API', version='1.0.0')])

        @app.GET('/plain')
        async def plain(ctx: HttpContext) -> None:
            await ctx.send_str(200, 'ok')

        spec = await self._spec_of(app)
        self.assertNotIn('components', spec)

    def test_unknown_schema_ref_raises(self) -> None:
        app = KayaApp()

        @app.GET('/broken')
        @operation(responses={200: {
            'description': 'x',
            'content': {'application/json': {'schema': schema_ref('Missing')}},
        }})
        async def broken(ctx: HttpContext) -> None:
            await ctx.send_str(200, 'x')

        with self.assertRaises(ValueError) as caught:
            generate_spec(app, title='Broken', version='1.0.0',
                          components={'schemas': {'User': {}}})
        message = str(caught.exception)
        self.assertIn('#/components/schemas/Missing', message)
        self.assertIn('GET /broken', message)

    def test_component_to_component_ref_validates(self) -> None:
        app = KayaApp()
        spec = generate_spec(app, title='Nested', version='1.0.0', components={'schemas': {
            'User': {'type': 'object'},
            'Page': {'type': 'object',
                     'properties': {'results': {'type': 'array', 'items': schema_ref('User')}}},
        }})
        self.assertIn('components', spec)

    def test_external_refs_are_ignored(self) -> None:
        app = KayaApp()

        @app.GET('/external')
        @operation(responses={200: {
            'description': 'x',
            'content': {'application/json': {'schema': {'$ref': './other.json#/Thing'}}},
        }})
        async def external(ctx: HttpContext) -> None:
            await ctx.send_str(200, 'x')

        spec = generate_spec(app, title='External', version='1.0.0')
        self.assertNotIn('components', spec)

    def test_generic_component_section_validated(self) -> None:
        app = KayaApp()

        @app.GET('/p')
        @operation(parameters=[{'name': 'q', 'in': 'query',
                                'schema': {'$ref': '#/components/parameters/Missing'}}])
        async def p(ctx: HttpContext) -> None:
            await ctx.send_str(200, 'x')

        with self.assertRaises(ValueError) as caught:
            generate_spec(app, title='P', version='1.0.0', components={'parameters': {}})
        self.assertIn('#/components/parameters/Missing', str(caught.exception))


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