Function app

tests/unit/server.py:95–142

Main ASGI application handler that routes requests to specific handlers. Args: scope: The ASGI connection scope. receive: The ASGI receive function. send: The ASGI send function.

(scope: dict[str, Any], receive: Receive, send: Send)

async def app(scope: dict[str, Any], receive: Receive, send: Send) -> None:
    """Main ASGI application handler that routes requests to specific handlers.

    Args:
        scope: The ASGI connection scope.
        receive: The ASGI receive function.
        send: The ASGI send function.
    """
    assert scope['type'] == 'http'
    paths: dict[str, PathHandler] = {
        'start_enqueue': start_enqueue_endpoint,
        'start_enqueue_non_href': start_enqueue_non_href_endpoint,
        'sub_index': secondary_index_endpoint,
        'incapsula': incapsula_endpoint,
        'page_1': generic_response_endpoint,
        'page_2': generic_response_endpoint,
        'page_3': generic_response_endpoint,
        'base_page': base_index_endpoint,
        'problematic_links': problematic_links_endpoint,
        'non_href_links': non_href_links_endpoint,
        'set_cookies': set_cookies,
        'set_complex_cookies': set_complex_cookies,
        'cookies': get_cookies,
        'status': echo_status,
        'headers': echo_headers,
        'user-agent': echo_user_agent,
        'echo_content': echo_content,
        'sitemap.txt': echo_content,
        'sitemap.xml': echo_content,
        'sitemap.xml.gz': echo_content,
        'get': get_echo,
        'post': post_echo,
        'redirect': redirect_to_url,
        'json': hello_world_json,
        'xml': hello_world_xml,
        'robots.txt': robots_txt,
        'get_compressed': get_compressed,
        'slow': slow_response,
        'infinite_scroll': infinite_scroll_endpoint,
        'resource_loading_page': resource_loading_endpoint,
    }
    path = URL(scope['path']).parts[1]
    # Route requests to appropriate handlers
    if path in paths:
        path_func = paths[path]
        await path_func(scope, receive, send)
    else:
        await hello_world(scope, receive, send)


async def get_cookies(scope: dict[str, Any], _receive: Receive, send: Send) -> None: