diff --git a/.coveragerc b/.coveragerc index 9860ca9e3..0d2ac8865 100644 --- a/.coveragerc +++ b/.coveragerc @@ -1,7 +1,7 @@ [run] branch = True source = falcon -omit = falcon/tests*,falcon/cmd*,falcon/bench*,falcon/vendor/* +omit = falcon/tests*,falcon/cmd/bench.py,falcon/bench*,falcon/vendor/* parallel = True diff --git a/.travis.yml b/.travis.yml index 6b253695d..71ac8c764 100644 --- a/.travis.yml +++ b/.travis.yml @@ -53,7 +53,9 @@ matrix: env: TOXENV=check_vendored - name: Python 3.8 (Windows) - env: TOXENV=py38_nocover + env: + - TOXENV=py38_nocover + - PYTHONIOENCODING=utf8 os: windows language: bash before_install: diff --git a/docs/_newsfragments/1435.newandimproved.rst b/docs/_newsfragments/1435.newandimproved.rst new file mode 100644 index 000000000..130edd9e0 --- /dev/null +++ b/docs/_newsfragments/1435.newandimproved.rst @@ -0,0 +1,3 @@ +Added inspect module to collect information about an application regarding +the registered routes, middlewares, static routes, sinks and error handlers +(See also: :ref:`inspect`.) \ No newline at end of file diff --git a/docs/api/index.rst b/docs/api/index.rst index 09ce36d79..f3ca96c1b 100644 --- a/docs/api/index.rst +++ b/docs/api/index.rst @@ -16,5 +16,6 @@ Framework Reference cors hooks routing + inspect util testing diff --git a/docs/api/inspect.rst b/docs/api/inspect.rst new file mode 100644 index 000000000..fbeae7182 --- /dev/null +++ b/docs/api/inspect.rst @@ -0,0 +1,167 @@ +.. _inspect: + +Inspect Module +============== + +* `Using Inspect Functions`_ +* `Inspect Functions Reference`_ +* `Router Inspection`_ +* `Information Classes`_ +* `Visitor Classes`_ + +This module can be used to inspect a Falcon application to obtain information +about its registered routes, middleware objects, static routes, sinks and +error handlers. The entire application can be inspected at once using the +:func:`.inspect_app` function. Additional functions are available for +inspecting specific aspects of the app. + +A ``falcon-inspect-app`` CLI script is also available; it uses the inspect +module to print a string representation of an application, as demonstrated +below: + +.. code:: bash + + # my_module exposes the application as a variable named "app" + $ falcon-inspect-app my_module:app + + Falcon App (WSGI) + • Routes: + ⇒ /foo - MyResponder: + ├── DELETE - on_delete + ├── GET - on_get + └── POST - on_post + ⇒ /foo/{id} - MyResponder: + ├── DELETE - on_delete_id + ├── GET - on_get_id + └── POST - on_post_id + ⇒ /bar - OtherResponder: + ├── DELETE - on_delete_id + ├── GET - on_get_id + └── POST - on_post_id + • Middleware (Middleware are independent): + → MyMiddleware.process_request + → OtherMiddleware.process_request + + ↣ MyMiddleware.process_resource + ↣ OtherMiddleware.process_resource + + ├── Process route responder + + ↢ OtherMiddleware.process_response + ↢ CORSMiddleware.process_response + • Static routes: + ↦ /tests/ /path/to/tests [/path/to/test/index.html] + ↦ /falcon/ /path/to/falcon + • Sinks: + ⇥ /sink_cls SinkClass + ⇥ /sink_fn sinkFn + • Error handlers: + ⇜ RuntimeError my_runtime_handler + +The example above shows how ``falcon-inspect-app`` simply outputs the value +returned by the :meth:`.AppInfo.to_string` method. In fact, here is a simple +script that returns the same output as the ``falcon-inspect-app`` command: + +.. code:: python + + from falcon import inspect + from my_module import app + + app_info = inspect.inspect_app(app) + + # Equivalent to print(app_info.to_string()) + print(app_info) + +A more verbose description of the app can be obtained by passing +``verbose=True`` to :meth:`.AppInfo.to_string`, while the default +routes added by the framework can be included by passing ``internal=True``. The +``falcon-inspect-app`` command supports the ``--verbose`` and +``--internal`` flags to enable these options. + +Using Inspect Functions +----------------------- + +The values returned by the inspect functions are class instances that +contain the relevant information collected from the application. These +objects facilitate programmatic use of the collected data. + +To support inspection of applications that use a custom router, the +module provides a :func:`.register_router` function to register +a handler function for the custom router class. +Inspection of the default :class:`.CompiledRouter` class is +handled by the :func:`.inspect_compiled_router` +function. + +The returned information classes can be explored using the visitor +pattern. To create the string representation of the classes the +:class:`.StringVisitor` visitor is used. +This class is instantiated automatically when calling ``str()`` +on an instance or when using the ``to_string()`` method. + +Custom visitor implementations can subclass :class:`.InspectVisitor` and +use the :meth:`.InspectVisitor.process` method to visit +the classes. + +Inspect Functions Reference +--------------------------- + +This module defines the following inspect functions. + +.. autofunction:: falcon.inspect.inspect_app + +.. autofunction:: falcon.inspect.inspect_routes + +.. autofunction:: falcon.inspect.inspect_middlewares + +.. autofunction:: falcon.inspect.inspect_static_routes + +.. autofunction:: falcon.inspect.inspect_sinks + +.. autofunction:: falcon.inspect.inspect_error_handlers + +Router Inspection +----------------- + +The following functions enable route inspection. + +.. autofunction:: falcon.inspect.register_router + +.. autofunction:: falcon.inspect.inspect_compiled_router + +Information Classes +------------------- + +Information returned by the inspect functions is represented by these classes. + +.. autoclass:: falcon.inspect.AppInfo + :members: + +.. autoclass:: falcon.inspect.RouteInfo + +.. autoclass:: falcon.inspect.RouteMethodInfo + +.. autoclass:: falcon.inspect.MiddlewareInfo + +.. autoclass:: falcon.inspect.MiddlewareTreeInfo + +.. autoclass:: falcon.inspect.MiddlewareClassInfo + +.. autoclass:: falcon.inspect.MiddlewareTreeItemInfo + +.. autoclass:: falcon.inspect.MiddlewareMethodInfo + +.. autoclass:: falcon.inspect.StaticRouteInfo + +.. autoclass:: falcon.inspect.SinkInfo + +.. autoclass:: falcon.inspect.ErrorHandlerInfo + +Visitor Classes +--------------- + +The following visitors are used to traverse the information classes. + +.. autoclass:: falcon.inspect.InspectVisitor + :members: + +.. autoclass:: falcon.inspect.StringVisitor diff --git a/docs/user/quickstart.rst b/docs/user/quickstart.rst index a54d91b26..557ed458d 100644 --- a/docs/user/quickstart.rst +++ b/docs/user/quickstart.rst @@ -326,3 +326,31 @@ To test this example go to the another terminal and run: .. code:: bash $ http localhost:8000/1/things authorization:custom-token + +To visualize the application configuration the :ref:`inspect` can be used: + +.. code:: bash + + falcon-inspect-app things_advanced:app + +This would print for this example application: + +.. code:: + + Falcon App (WSGI) + • Routes: + ⇒ /{user_id}/things - ThingsResource: + ├── GET - on_get + └── POST - on_post + • Middleware (Middleware are independent): + → AuthMiddleware.process_request + → RequireJSON.process_request + → JSONTranslator.process_request + + ├── Process route responder + + ↢ JSONTranslator.process_response + • Sinks: + ⇥ /search/(?Pddg|y)\Z SinkAdapter + • Error handlers: + ⇜ StorageError handle \ No newline at end of file diff --git a/docs/user/tutorial.rst b/docs/user/tutorial.rst index 1ecd10e6d..f830a54ff 100644 --- a/docs/user/tutorial.rst +++ b/docs/user/tutorial.rst @@ -310,6 +310,22 @@ representation of the "images" resource. threaded web server, resources and their dependencies must be thread-safe. +We can use the the :ref:`inspect` to visualize the application configuration: + +.. code:: bash + + falcon-inspect-app look.app:app + +This prints the following, correctly indicating that we are handling ``GET`` +requests in the ``/images`` route: + +.. code:: + + Falcon App (WSGI) + • Routes: + ⇒ /images - Resource: + └── GET - on_get + So far we have only implemented a responder for GET. Let's see what happens when a different method is requested: @@ -1253,6 +1269,21 @@ HTTPie won't display the image, but you can see that the response headers were set correctly. Just for fun, go ahead and paste the above URI into your browser. The image should display correctly. +Inspecting the application now returns: + +.. code:: bash + + falcon-inspect-app look.app:get_app + +.. code:: + + Falcon App (WSGI) + • Routes: + ⇒ /images - Collection: + ├── GET - on_get + └── POST - on_post + ⇒ /images/{name} - Item: + └── GET - on_get .. Query Strings .. ------------- diff --git a/falcon/cmd/inspect_app.py b/falcon/cmd/inspect_app.py new file mode 100644 index 000000000..d35fcd933 --- /dev/null +++ b/falcon/cmd/inspect_app.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python +# Copyright 2013 by Rackspace Hosting, Inc. +# +# Licensed under the Apache License, Version 2.0 (the 'License'); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an 'AS IS' BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +""" +Script that prints out the routes of an App instance. +""" +import argparse +import importlib +import os +import sys + +import falcon +from falcon.inspect import inspect_app, inspect_routes, StringVisitor + +sys.path.append(os.getcwd()) + + +def make_parser(): + """Creates the parsed or the application""" + parser = argparse.ArgumentParser( + description='Example: falcon-inspect-app myprogram:app' + ) + parser.add_argument( + '-r', + '--route_only', + action='store_true', + help='Prints only the information regarding the routes', + ) + parser.add_argument( + '-v', '--verbose', action='store_true', help='More verbose output', + ) + parser.add_argument( + '-i', + '--internal', + action='store_true', + help='Print also internal falcon route methods and error handlers', + ) + parser.add_argument( + 'app_module', + help='The module and app to inspect. Example: myapp.somemodule:api', + ) + return parser + + +def load_app(parser, args): + + try: + module, instance = args.app_module.split(':', 1) + except ValueError: + parser.error('The app_module must include a colon between the module and instance') + try: + app = getattr(importlib.import_module(module), instance) + except AttributeError: + parser.error('{!r} not found in module {!r}'.format(instance, module)) + + if not isinstance(app, falcon.App): + if callable(app): + app = app() + if not isinstance(app, falcon.App): + parser.error('{} did not return a falcon.App instance'.format(args.app_module)) + else: + parser.error( + 'The instance must be of falcon.App or be ' + 'a callable without args that returns falcon.App' + ) + return app + + +def route_main(): + print('The "falcon-print-routes" command is deprecated. Please use "falcon-inspect-app"') + main() + + +def main(): + """ + Main entrypoint. + """ + parser = make_parser() + args = parser.parse_args() + app = load_app(parser, args) + if args.route_only: + routes = inspect_routes(app) + visitor = StringVisitor(args.verbose, args.internal) + for route in routes: + print(visitor.process(route)) + else: + print(inspect_app(app).to_string(args.verbose, args.internal)) + + +if __name__ == '__main__': # pragma: no cover + main() diff --git a/falcon/cmd/print_routes.py b/falcon/cmd/print_routes.py deleted file mode 100644 index ccea297f5..000000000 --- a/falcon/cmd/print_routes.py +++ /dev/null @@ -1,118 +0,0 @@ -#!/usr/bin/env python -# Copyright 2013 by Rackspace Hosting, Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -""" -Script that prints out the routes of an App instance. -""" - -from functools import partial -import inspect - -import falcon - - -def print_routes(api, verbose=False): # pragma: no cover - """ - Initial call. - - :param api: The falcon.App or callable that returns an instance to look at. - :type api: falcon.App or callable - :param verbose: If the output should be verbose. - :type verbose: bool - """ - traverse(api._router._roots, verbose=verbose) - - -def traverse(roots, parent='', verbose=False): - """ - Recursive call which also handles printing output. - - :param api: The falcon.App or callable that returns an instance to look at. - :type api: falcon.App or callable - :param parent: The parent uri path to the current iteration. - :type parent: str - :param verbose: If the output should be verbose. - :type verbose: bool - """ - for root in roots: - if root.method_map: - print('->', parent + '/' + root.raw_segment) - if verbose: - for method, func in root.method_map.items(): - # NOTE(kgriffs): Skip the default responder that the - # framework creates. - if not func.__name__.startswith('method_not_allowed'): - if isinstance(func, partial): - real_func = func.func - else: - real_func = func - - try: - source_file = inspect.getsourcefile(real_func) - source_lines = inspect.getsourcelines(real_func) - source_info = '{}:{}'.format(source_file, - source_lines[1]) - except TypeError: - # NOTE(vytas): If Falcon is cythonized, all default - # responders coming from cythonized modules will - # appear as built-in functions, and raise a - # TypeError when trying to locate the source file. - source_info = '[unknown file]' - - print('-->' + method, source_info) - - if root.children: - traverse(root.children, parent + '/' + root.raw_segment, verbose) - - -def main(): - """ - Main entrypoint. - """ - import argparse - - parser = argparse.ArgumentParser( - description='Example: print-api-routes myprogram:app') - parser.add_argument( - '-v', '--verbose', action='store_true', - help='Prints out information for each method.') - parser.add_argument( - 'api_module', - help='The module and api to inspect. Example: myapp.somemodule:api', - ) - args = parser.parse_args() - - try: - module, instance = args.api_module.split(':', 1) - except ValueError: - parser.error( - 'The api_module must include a colon between ' - 'the module and instance') - api = getattr(__import__(module, fromlist=[True]), instance) - if not isinstance(api, falcon.App): - if callable(api): - api = api() - if not isinstance(api, falcon.App): - parser.error( - '{0} did not return a falcon.App instance'.format( - args.api_module)) - else: - parser.error( - 'The instance must be of falcon.App or be ' - 'a callable without args that returns falcon.App') - print_routes(api, verbose=args.verbose) - - -if __name__ == '__main__': - main() diff --git a/falcon/inspect.py b/falcon/inspect.py new file mode 100644 index 000000000..23560fd65 --- /dev/null +++ b/falcon/inspect.py @@ -0,0 +1,786 @@ +# Copyright 2020 by Federico Caselli +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Inspect utilities for falcon applications""" +from functools import partial +import inspect +from typing import Callable, Dict, List, Optional, Type + +from falcon import App, app_helpers +from falcon.routing import CompiledRouter + + +def inspect_app(app: App) -> 'AppInfo': + """Inspects an application. + + Args: + app (falcon.App): The application to inspect. Works with both + :class:`falcon.App` and :class:`falcon.asgi.App`. + + Returns: + AppInfo: The information regarding the application. Call + :meth:`~.AppInfo.to_string` on the result to obtain a human-friendly + representation. + """ + routes = inspect_routes(app) + static = inspect_static_routes(app) + sinks = inspect_sinks(app) + error_handlers = inspect_error_handlers(app) + middleware = inspect_middlewares(app) + return AppInfo(routes, middleware, static, sinks, error_handlers, app._ASGI) + + +def inspect_routes(app: App) -> 'List[RouteInfo]': + """Inspects the routes of an application. + + Args: + app (falcon.App): The application to inspect. Works with both + :class:`falcon.App` and :class:`falcon.asgi.App`. + + Returns: + List[RouteInfo]: A list of route descriptions for the application. + """ + router = app._router + + inspect_function = _supported_routers.get(type(router)) + if inspect_function is None: + raise TypeError( + 'Unsupported router class {}. Use "register_router" ' + 'to register a function that can inspect the router ' + 'used by the provided application'.format(type(router)) + ) + return inspect_function(router) + + +def register_router(router_class): + """Register a function to inspect a particular router. + + This decorator registers a new function for a custom router + class, so that it can be inspected with the function + :func:`.inspect_routes`. + An inspection function takes the router instance used by the + application and returns a list of :class:`.RouteInfo`. Eg:: + + @register_router(MyRouterClass) + def inspect_my_router(router): + return [RouteInfo('foo', 'bar', '/path/to/foo.py:42', [])] + + Args: + router_class (Type): The router class to register. If + already registered an error will be raised. + """ + + def wraps(fn): + if router_class in _supported_routers: + raise ValueError( + 'Another function is already registered' + ' for the router {}'.format(router_class) + ) + _supported_routers[router_class] = fn + return fn + + return wraps + + +# router inspection registry +_supported_routers = {} # type: Dict[Type, Callable] + + +def inspect_static_routes(app: App) -> 'List[StaticRouteInfo]': + """Inspects the static routes of an application. + + Args: + app (falcon.App): The application to inspect. Works with both + :class:`falcon.App` and :class:`falcon.asgi.App`. + + Returns: + List[StaticRouteInfo]: A list of static routes that have + been added to the application. + """ + routes = [] + for sr in app._static_routes: + info = StaticRouteInfo(sr._prefix, sr._directory, sr._fallback_filename) + routes.append(info) + return routes + + +def inspect_sinks(app: App) -> 'List[SinkInfo]': + """Inspects the sinks of an application. + + Args: + app (falcon.App): The application to inspect. Works with both + :class:`falcon.App` and :class:`falcon.asgi.App`. + + Returns: + List[SinkInfo]: A list of sinks used by the application. + """ + sinks = [] + for prefix, sink in app._sinks: + source_info, name = _get_source_info_and_name(sink) + info = SinkInfo(prefix.pattern, name, source_info) + sinks.append(info) + return sinks + + +def inspect_error_handlers(app: App) -> 'List[ErrorHandlerInfo]': + """Inspects the error handlers of an application. + + Args: + app (falcon.App): The application to inspect. Works with both + :class:`falcon.App` and :class:`falcon.asgi.App`. + + Returns: + List[ErrorHandlerInfo]: A list of error handlers used by the + application. + """ + errors = [] + for exc, fn in app._error_handlers.items(): + source_info, name = _get_source_info_and_name(fn) + info = ErrorHandlerInfo(exc.__name__, name, source_info, _is_internal(fn)) + errors.append(info) + return errors + + +def inspect_middlewares(app: App) -> 'MiddlewareInfo': + """Inspects the middleware components of an application. + + Args: + app (falcon.App): The application to inspect. Works with both + :class:`falcon.App` and :class:`falcon.asgi.App`. + + Returns: + MiddlewareInfo: Information about the app's middleware components. + """ + types_ = app_helpers.prepare_middleware(app._unprepared_middleware, True, app._ASGI) + + type_infos = [] + for stack in types_: + current = [] + for method in stack: + _, name = _get_source_info_and_name(method) + cls = type(method.__self__) + _, cls_name = _get_source_info_and_name(cls) + current.append(MiddlewareTreeItemInfo(name, cls_name)) + type_infos.append(current) + middlewareTree = MiddlewareTreeInfo(*type_infos) + + middlewareClasses = [] + names = 'Process request', 'Process resource', 'Process response' + for m in app._unprepared_middleware: + fns = app_helpers.prepare_middleware([m], True, app._ASGI) + class_source_info, cls_name = _get_source_info_and_name(type(m)) + methods = [] + for method, name in zip(fns, names): + if method: + real_func = method[0] + source_info = _get_source_info(real_func) + methods.append(MiddlewareMethodInfo(real_func.__name__, source_info)) + m_info = MiddlewareClassInfo(cls_name, class_source_info, methods) + middlewareClasses.append(m_info) + + return MiddlewareInfo( + middlewareTree, middlewareClasses, app._independent_middleware + ) + + +@register_router(CompiledRouter) +def inspect_compiled_router(router: CompiledRouter) -> 'List[RouteInfo]': + """Default route inspector for CompiledRouter. + + Walks an instance of :class:`~.CompiledRouter` and returns a list of + defined routes. + + Args: + router (CompiledRouter): The router to inspect. + + Returns: + List[RouteInfo]: A list of :class:`~.RouteInfo`. + """ + + def _traverse(roots, parent): + for root in roots: + path = parent + '/' + root.raw_segment + if root.resource is not None: + methods = [] + if root.method_map: + for method, func in root.method_map.items(): + if isinstance(func, partial): + real_func = func.func + else: + real_func = func + + source_info = _get_source_info(real_func) + internal = _is_internal(real_func) + + method_info = RouteMethodInfo( + method, source_info, real_func.__name__, internal + ) + methods.append(method_info) + source_info, class_name = _get_source_info_and_name(root.resource) + + route_info = RouteInfo(path, class_name, source_info, methods) + routes.append(route_info) + + if root.children: + _traverse(root.children, path) + + routes = [] # type: List[RouteInfo] + _traverse(router._roots, '') + return routes + + +# ------------------------------------------------------------------------ +# Inspection classes +# ------------------------------------------------------------------------ + + +class _Traversable: + __visit_name__ = 'N/A' + + def to_string(self, verbose=False, internal=False) -> str: + """Returns a string representation of this class. + + Args: + verbose (bool, optional): Adds more information. Defaults to False. + internal (bool, optional): Also include internal route methods + and error handlers added by the framework. Defaults to + ``False``. + + Returns: + str: string representation of this class. + """ + return StringVisitor(verbose, internal).process(self) + + def __repr__(self): + return self.to_string() + + +class RouteMethodInfo(_Traversable): + """Describes a responder method. + + Args: + method (str): The HTTP method of this responder. + source_info (str): The source path of this function. + function_name (str): Name of the function. + internal (bool): Whether or not this was a default responder added + by the framework. + + Attributes: + suffix (str): The suffix of this route function. This is set to an empty + string when the function has no suffix. + """ + + __visit_name__ = 'route_method' + + def __init__( + self, method: str, source_info: str, function_name: str, internal: bool + ): + self.method = method + self.source_info = source_info + self.function_name = function_name + self.internal = internal + # NOTE(CaselIT): internal falcon names do not start with on and do not have suffix + if function_name.startswith('on'): + self.suffix = '_'.join(function_name.split('_')[2:]) + else: + self.suffix = '' + + +class RouteInfo(_Traversable): + """Describes a route. + + Args: + path (str): The path of this route. + class_name (str): The class name of the responder of this route. + source_info (str): The source path where this responder was defined. + methods (List[MethodInfo]): List of methods defined in the route. + """ + + __visit_name__ = 'route' + + def __init__( + self, + path: str, + class_name: str, + source_info: str, + methods: List[RouteMethodInfo], + ): + self.path = path + self.class_name = class_name + self.source_info = source_info + self.methods = methods + + +class StaticRouteInfo(_Traversable): + """Describes a static route. + + Args: + path (str): The prefix of the static route. + directory (str): The directory for the static route. + fallback_filename (str or None): Fallback filename to serve. + """ + + __visit_name__ = 'static_route' + + def __init__(self, prefix: str, directory: str, fallback_filename: Optional[str]): + self.prefix = prefix + self.directory = directory + self.fallback_filename = fallback_filename + + +class SinkInfo(_Traversable): + """Describes a sink. + + Args: + prefix (str): The prefix of the sink. + name (str): The name of the sink function or class. + source_info (str): The source path where this sink was defined. + """ + + __visit_name__ = 'sink' + + def __init__(self, prefix: str, name: str, source_info: str): + self.prefix = prefix + self.name = name + self.source_info = source_info + + +class ErrorHandlerInfo(_Traversable): + """Desribes an error handler. + + Args: + error (name): The name of the error type. + name (str): The name of the handler. + source_info (str): The source path where this error handler was defined. + internal (bool): Whether or not this is a default error handler added by + the framework. + """ + + __visit_name__ = 'error_handler' + + def __init__(self, error: str, name: str, source_info: str, internal: bool): + self.error = error + self.name = name + self.source_info = source_info + self.internal = internal + + +class MiddlewareMethodInfo(_Traversable): + """Describes a middleware method. + + Args: + function_name (str): Name of the method. + source_info (str): The source path of the method. + """ + + __visit_name__ = 'middleware_method' + + def __init__(self, function_name: str, source_info: str): + self.function_name = function_name + self.source_info = source_info + self.internal = False # added for compatibility with RouteMethodInfo + + +class MiddlewareClassInfo(_Traversable): + """Describes a middleware class. + + Args: + name (str): The name of the middleware class. + source_info (str): The source path where the middleware was defined. + methods (List[MiddlewareMethodInfo]): List of method defined by the middleware class. + """ + + __visit_name__ = 'middleware_class' + + def __init__( + self, name: str, source_info: str, methods: List[MiddlewareMethodInfo] + ): + self.name = name + self.source_info = source_info + self.methods = methods + + +class MiddlewareTreeItemInfo(_Traversable): + """Describes a middleware tree entry. + + Args: + name (str): The name of the method. + class_name (str): The class name of the method. + """ + + __visit_name__ = 'middleware_tree_item' + + _symbols = { + 'process_request': '→', + 'process_resource': '↣', + 'process_response': '↢', + } + + def __init__(self, name: str, class_name: str): + self.name = name + self.class_name = class_name + + +class MiddlewareTreeInfo(_Traversable): + """Describes the middleware methods used by the app. + + Args: + request (List[MiddlewareTreeItemInfo]): The `process_request` methods. + resource (List[MiddlewareTreeItemInfo]): The `process_resource` methods. + response (List[MiddlewareTreeItemInfo]): The `process_response` methods. + """ + + __visit_name__ = 'middleware_tree' + + def __init__( + self, + request: List[MiddlewareTreeItemInfo], + resource: List[MiddlewareTreeItemInfo], + response: List[MiddlewareTreeItemInfo], + ): + self.request = request + self.resource = resource + self.response = response + + +class MiddlewareInfo(_Traversable): + """Describes the middleware of the app. + + Args: + middlewareTree (MiddlewareTreeInfo): The middleware tree of the app. + middlewareClasses (List[MiddlewareClassInfo]): The middleware classes of the app. + independent (bool): Whether or not the middleware components are executed + independently. + + Attributes: + independent_text (str): Text created from the `independent` arg. + """ + + __visit_name__ = 'middleware' + + def __init__( + self, + middleware_tree: MiddlewareTreeInfo, + middleware_classes: List[MiddlewareClassInfo], + independent: bool, + ): + self.middleware_tree = middleware_tree + self.middleware_classes = middleware_classes + self.independent = independent + + if independent: + self.independent_text = 'Middleware are independent' + else: + self.independent_text = 'Middleware are dependent' + + +class AppInfo(_Traversable): + """Describes an application. + + Args: + routes (List[RouteInfo]): The routes of the application. + middleware (MiddlewareInfo): The middleware information in the application. + static_routes (List[StaticRouteInfo]): The static routes of this application. + sinks (List[SinkInfo]): The sinks of this application. + error_handlers (List[ErrorHandlerInfo]): The error handlers of this application. + asgi (bool): Whether or not this is an ASGI application. + """ + + __visit_name__ = 'app' + + def __init__( + self, + routes: List[RouteInfo], + middleware: MiddlewareInfo, + static_routes: List[StaticRouteInfo], + sinks: List[SinkInfo], + error_handlers: List[ErrorHandlerInfo], + asgi: bool, + ): + self.routes = routes + self.middleware = middleware + self.static_routes = static_routes + self.sinks = sinks + self.error_handlers = error_handlers + self.asgi = asgi + + def to_string(self, verbose=False, internal=False, name='') -> str: + """Returns a string representation of this class. + + Args: + verbose (bool, optional): Adds more information. Defaults to False. + internal (bool, optional): Also include internal falcon route methods + and error handlers. Defaults to ``False``. + name (str, optional): The name of the application, to be output at the + beginning of the text. Defaults to ``'Falcon App'``. + Returns: + str: A string representation of the application. + """ + return StringVisitor(verbose, internal, name).process(self) + + +# ------------------------------------------------------------------------ +# Visitor classes +# ------------------------------------------------------------------------ + + +class InspectVisitor: + """Base visitor class that implements the `process` method. + + Subclasses must implement ``visit_`` methods for each supported class. + """ + + def process(self, instance: _Traversable): + """Process the instance, by calling the appropriate visit method. + Uses the `__visit_name__` attribute of the `instance` to obtain the method to use. + + Args: + instance (_Traversable): The instance to process. + """ + try: + return getattr(self, 'visit_{}'.format(instance.__visit_name__))(instance) + except AttributeError as e: + raise RuntimeError( + 'This visitor does not support {}'.format(type(instance)) + ) from e + + +class StringVisitor(InspectVisitor): + """Visitor implementation that returns a string representation of the info class. + + This is used automatically by calling ``to_string()`` on the info class. + It can also be used directly by calling ``StringVisitor.process(info_instance)``. + + Args: + verbose (bool, optional): Adds more information. Defaults to ``False``. + internal (bool, optional): Also include internal route methods + and error handlers added by the framework. Defaults to ``False``. + name (str, optional): The name of the application, to be output at the + beginning of the text. Defaults to ``'Falcon App'``. + """ + + def __init__(self, verbose=False, internal=False, name=''): + self.verbose = verbose + self.internal = internal + self.name = name + self.indent = 0 + + @property + def tab(self): + """Gets the current tabulation""" + return ' ' * self.indent + + def visit_route_method(self, route_method: RouteMethodInfo) -> str: + """Visit a RouteMethodInfo instance. Usually called by `process`""" + text = '{0.method} - {0.function_name}'.format(route_method) + if self.verbose: + text += ' ({0.source_info})'.format(route_method) + return text + + def _methods_to_string(self, methods: List): + """Returns a string from the list of methods""" + tab = self.tab + ' ' * 3 + methods = _filter_internal(methods, self.internal) + if not methods: + return '' + text_list = [self.process(m) for m in methods] + method_text = ['{}├── {}'.format(tab, m) for m in text_list[:-1]] + method_text += ['{}└── {}'.format(tab, m) for m in text_list[-1:]] + return '\n'.join(method_text) + + def visit_route(self, route: RouteInfo) -> str: + """Visit a RouteInfo instance. Usually called by `process`""" + text = '{0}⇒ {1.path} - {1.class_name}'.format(self.tab, route) + if self.verbose: + text += ' ({0.source_info})'.format(route) + + method_text = self._methods_to_string(route.methods) + if not method_text: + return text + + return '{}:\n{}'.format(text, method_text) + + def visit_static_route(self, static_route: StaticRouteInfo) -> str: + """Visit a StaticRouteInfo instance. Usually called by `process`""" + text = '{0}↦ {1.prefix} {1.directory}'.format(self.tab, static_route) + if static_route.fallback_filename: + text += ' [{0.fallback_filename}]'.format(static_route) + return text + + def visit_sink(self, sink: SinkInfo) -> str: + """Visit a SinkInfo instance. Usually called by `process`""" + text = '{0}⇥ {1.prefix} {1.name}'.format(self.tab, sink) + if self.verbose: + text += ' ({0.source_info})'.format(sink) + return text + + def visit_error_handler(self, error_handler: ErrorHandlerInfo) -> str: + """Visit a ErrorHandlerInfo instance. Usually called by `process`""" + text = '{0}⇜ {1.error} {1.name}'.format(self.tab, error_handler) + if self.verbose: + text += ' ({0.source_info})'.format(error_handler) + return text + + def visit_middleware_method(self, middleware_method: MiddlewareMethodInfo) -> str: + """Visit a MiddlewareMethodInfo instance. Usually called by `process`""" + text = '{0.function_name}'.format(middleware_method) + if self.verbose: + text += ' ({0.source_info})'.format(middleware_method) + return text + + def visit_middleware_class(self, middleware_class: MiddlewareClassInfo) -> str: + """Visit a ErrorHandlerInfo instance. Usually called by `process`""" + text = '{0}↣ {1.name}'.format(self.tab, middleware_class) + if self.verbose: + text += ' ({0.source_info})'.format(middleware_class) + + method_text = self._methods_to_string(middleware_class.methods) + if not method_text: + return text + + return '{}:\n{}'.format(text, method_text) + + def visit_middleware_tree_item(self, mti: MiddlewareTreeItemInfo) -> str: + """Visit a MiddlewareTreeItemInfo instance. Usually called by `process`""" + symbol = mti._symbols.get(mti.name, '→') + return '{0}{1} {2.class_name}.{2.name}'.format(self.tab, symbol, mti) + + def visit_middleware_tree(self, m_tree: MiddlewareTreeInfo) -> str: + """Visit a MiddlewareTreeInfo instance. Usually called by `process`""" + before = len(m_tree.request) + len(m_tree.resource) + after = len(m_tree.response) + + if before + after == 0: + return '' + + each = 2 + initial = self.indent + if after > before: + self.indent += each * (after - before) + + text = [] + for r in m_tree.request: + text.append(self.process(r)) + self.indent += each + if text: + text.append('') + for r in m_tree.resource: + text.append(self.process(r)) + self.indent += each + + if m_tree.resource or not text: + text.append('') + self.indent += each + text.append('{}├── Process route responder'.format(self.tab)) + self.indent -= each + if m_tree.response: + text.append('') + + for r in m_tree.response: + self.indent -= each + text.append(self.process(r)) + + self.indent = initial + return '\n'.join(text) + + def visit_middleware(self, middleware: MiddlewareInfo) -> str: + """Visit a MiddlewareInfo instance. Usually called by `process`""" + text = self.process(middleware.middleware_tree) + if self.verbose: + self.indent += 4 + m_text = '\n'.join(self.process(m) for m in middleware.middleware_classes) + self.indent -= 4 + if m_text: + text += '\n{}- Middlewares classes:\n{}'.format(self.tab, m_text) + + return text + + def visit_app(self, app: AppInfo) -> str: + """Visit a AppInfo instance. Usually called by `process`""" + + type_ = 'ASGI' if app.asgi else 'WSGI' + self.indent = 4 + text = '{} ({})'.format(self.name or 'Falcon App', type_) + + if app.routes: + routes = '\n'.join(self.process(r) for r in app.routes) + text += '\n• Routes:\n{}'.format(routes) + + middleware_text = self.process(app.middleware) + if middleware_text: + text += '\n• Middleware ({}):\n{}'.format( + app.middleware.independent_text, middleware_text + ) + + if app.static_routes: + static_routes = '\n'.join(self.process(sr) for sr in app.static_routes) + text += '\n• Static routes:\n{}'.format(static_routes) + + if app.sinks: + sinks = '\n'.join(self.process(s) for s in app.sinks) + text += '\n• Sinks:\n{}'.format(sinks) + + errors = _filter_internal(app.error_handlers, self.internal) + if errors: + errs = '\n'.join(self.process(e) for e in errors) + text += '\n• Error handlers:\n{}'.format(errs) + + return text + + +# ------------------------------------------------------------------------ +# Helpers functions +# ------------------------------------------------------------------------ + + +def _get_source_info(obj, default='[unknown file]'): + """Tries to get the definition file and line of obj. Returns default on error""" + try: + source_file = inspect.getsourcefile(obj) + source_lines = inspect.findsource(obj) + source_info = '{}:{}'.format(source_file, source_lines[1]) + except Exception: + # NOTE(vytas): If Falcon is cythonized, all default + # responders coming from cythonized modules will + # appear as built-in functions, and raise a + # TypeError when trying to locate the source file. + source_info = default + return source_info + + +def _get_source_info_and_name(obj): + """Tries to get the definition file and line of obj and its name""" + source_info = _get_source_info(obj, None) + if source_info is None: + # NOTE(caselit): a class instances return None. Try the type + source_info = _get_source_info(type(obj)) + name = getattr(obj, '__name__', None) + if name is None: + name = getattr(type(obj), '__name__', '[unknown]') + return source_info, name + + +def _is_internal(obj): + """Checks if the module of the object is a falcon module""" + module = inspect.getmodule(obj) + if module: + return module.__name__.startswith('falcon.') + return False + + +def _filter_internal(iterable, return_internal): + """Filters the internal elements of an iterable""" + if return_internal: + return iterable + return [el for el in iterable if not el.internal] diff --git a/setup.py b/setup.py index 641cc861d..62c8c1172 100644 --- a/setup.py +++ b/setup.py @@ -174,7 +174,8 @@ def load_description(): entry_points={ 'console_scripts': [ 'falcon-bench = falcon.cmd.bench:main', - 'falcon-print-routes = falcon.cmd.print_routes:main' + 'falcon-inspect-app = falcon.cmd.inspect_app:main', + 'falcon-print-routes = falcon.cmd.inspect_app:route_main', ] } ) diff --git a/tests/_inspect_fixture.py b/tests/_inspect_fixture.py new file mode 100644 index 000000000..5d6ec79f2 --- /dev/null +++ b/tests/_inspect_fixture.py @@ -0,0 +1,110 @@ +from falcon.routing import CompiledRouter + + +class MyRouter(CompiledRouter): + pass + + +class MyResponder: + def on_get(self, req, res): + pass + + def on_post(self, req, res): + pass + + def on_delete(self, req, res): + pass + + def on_get_id(self, req, res, id): + pass + + def on_put_id(self, req, res, id): + pass + + def on_delete_id(self, req, res, id): + pass + + +class MyResponderAsync: + async def on_get(self, req, res): + pass + + async def on_post(self, req, res): + pass + + async def on_delete(self, req, res): + pass + + async def on_get_id(self, req, res, id): + pass + + async def on_put_id(self, req, res, id): + pass + + async def on_delete_id(self, req, res, id): + pass + + +class OtherResponder: + def on_post_id(self, *args): + pass + + +class OtherResponderAsync: + async def on_post_id(self, *args): + pass + + +def sinkFn(*args): + pass + + +class SinkClass: + def __call__(self, *args): + pass + + +def my_error_handler(req, resp, ex, params): + pass + + +async def my_error_handler_async(req, resp, ex, params): + pass + + +class MyMiddleware: + def process_request(self, *args): + pass + + def process_resource(self, *args): + pass + + def process_response(self, *args): + pass + + +class OtherMiddleware: + def process_request(self, *args): + pass + + def process_response(self, *args): + pass + + +class MyMiddlewareAsync: + async def process_request(self, *args): + pass + + async def process_resource(self, *args): + pass + + async def process_response(self, *args): + pass + + +class OtherMiddlewareAsync: + async def process_request(self, *args): + pass + + async def process_response(self, *args): + pass diff --git a/tests/test_cmd_inspect_app.py b/tests/test_cmd_inspect_app.py new file mode 100644 index 000000000..e4f81d952 --- /dev/null +++ b/tests/test_cmd_inspect_app.py @@ -0,0 +1,169 @@ +from argparse import Namespace +import io +import sys + +import pytest + +from falcon import App, inspect +from falcon.cmd import inspect_app +from falcon.testing import redirected + +from _util import create_app # NOQA + +_WIN32 = sys.platform.startswith('win') +_MODULE = 'tests.test_cmd_inspect_app' + + +class DummyResource: + + def on_get(self, req, resp): + resp.body = 'Test\n' + resp.status = '200 OK' + + +class DummyResourceAsync: + + async def on_get(self, req, resp): + resp.body = 'Test\n' + resp.status = '200 OK' + + +def make_app(asgi=False): + app = create_app(asgi) + app.add_route('/test', DummyResourceAsync() if asgi else DummyResource()) + + return app + + +_APP = make_app() + + +@pytest.fixture +def app(asgi): + return make_app(asgi) + + +class TestMakeParser: + @pytest.mark.parametrize('args, exp', ( + (['foo'], Namespace(app_module='foo', route_only=False, verbose=False, internal=False)), + ( + ['foo', '-r'], + Namespace(app_module='foo', route_only=True, verbose=False, internal=False) + ), + ( + ['foo', '--route_only'], + Namespace(app_module='foo', route_only=True, verbose=False, internal=False) + ), + ( + ['foo', '-v'], + Namespace(app_module='foo', route_only=False, verbose=True, internal=False) + ), + ( + ['foo', '--verbose'], + Namespace(app_module='foo', route_only=False, verbose=True, internal=False) + ), + ( + ['foo', '-i'], + Namespace(app_module='foo', route_only=False, verbose=False, internal=True) + ), + ( + ['foo', '--internal'], + Namespace(app_module='foo', route_only=False, verbose=False, internal=True) + ), + ( + ['foo', '-r', '-v', '-i'], + Namespace(app_module='foo', route_only=True, verbose=True, internal=True) + ), + )) + def test_make_parser(self, args, exp): + parser = inspect_app.make_parser() + actual = parser.parse_args(args) + assert actual == exp + + def test_make_parser_error(self): + parser = inspect_app.make_parser() + with pytest.raises(SystemExit): + parser.parse_args([]) + + +class TestLoadApp: + @pytest.mark.parametrize('name', ('_APP', 'make_app')) + def test_load_app(self, name): + parser = inspect_app.make_parser() + args = Namespace(app_module='{}:{}'.format(_MODULE, name), route_only=False, verbose=False) + app = inspect_app.load_app(parser, args) + assert isinstance(app, App) + assert app._router.find('/test') is not None + + @pytest.mark.parametrize('name', ( + 'foo', # not exists + '_MODULE', # not callable and not app + 'DummyResource', # callable and not app + )) + def test_load_app_error(self, name): + parser = inspect_app.make_parser() + args = Namespace(app_module='{}:{}'.format(_MODULE, name), route_only=False, verbose=False) + with pytest.raises(SystemExit): + inspect_app.load_app(parser, args) + + def test_load_app_module_error(self): + parser = inspect_app.make_parser() + args = Namespace(app_module='foo', route_only=False, verbose=False) + with pytest.raises(SystemExit): + inspect_app.load_app(parser, args) + + +@pytest.mark.skipif(sys.version_info < (3, 6), reason='dict order is not stable') +@pytest.mark.parametrize('verbose', (True, False), ids=['verbose', 'not-verbose']) +@pytest.mark.parametrize('internal', (True, False), ids=['internal', 'not-internal']) +class TestMain: + def check(self, actual, expect): + if _WIN32: + # windows randomly returns the driver name as lowercase + assert actual.casefold() == expect.casefold() + else: + assert actual == expect + + def test_routes_only(self, verbose, internal, monkeypatch): + args = ['some-file.py', '{}:{}'.format(_MODULE, '_APP'), '-r'] + if verbose: + args.append('-v') + if internal: + args.append('-i') + monkeypatch.setattr('sys.argv', args) + output = io.StringIO() + with redirected(stdout=output): + inspect_app.main() + routes = inspect.inspect_routes(_APP) + sv = inspect.StringVisitor(verbose, internal) + expect = '\n'.join([sv.process(r) for r in routes]) + self.check(output.getvalue().strip(), expect) + + def test_inspect(self, verbose, internal, monkeypatch): + args = ['some-file.py', '{}:{}'.format(_MODULE, '_APP')] + if verbose: + args.append('-v') + if internal: + args.append('-i') + monkeypatch.setattr('sys.argv', args) + output = io.StringIO() + with redirected(stdout=output): + inspect_app.main() + ins = inspect.inspect_app(_APP) + self.check(output.getvalue().strip(), ins.to_string(verbose, internal)) + + +def test_route_main(monkeypatch): + called = False + + def mock(): + nonlocal called + called = True + + monkeypatch.setattr(inspect_app, 'main', mock) + output = io.StringIO() + with redirected(stdout=output): + inspect_app.route_main() + + assert 'deprecated' in output.getvalue() + assert called diff --git a/tests/test_cmd_print_api.py b/tests/test_cmd_print_api.py deleted file mode 100644 index beeb004aa..000000000 --- a/tests/test_cmd_print_api.py +++ /dev/null @@ -1,73 +0,0 @@ -import io -from os.path import normpath - -import pytest - -from falcon.cmd import print_routes -from falcon.testing import redirected - -from _util import create_app # NOQA - - -class DummyResource: - - def on_get(self, req, resp): - resp.body = 'Test\n' - resp.status = '200 OK' - - -class DummyResourceAsync: - - async def on_get(self, req, resp): - resp.body = 'Test\n' - resp.status = '200 OK' - - -@pytest.fixture -def app(asgi): - app = create_app(asgi) - app.add_route('/test', DummyResourceAsync() if asgi else DummyResource()) - - return app - - -def test_traverse_with_verbose(app): - """Ensure traverse() finds the proper routes and outputs verbose info.""" - - output = io.StringIO() - with redirected(stdout=output): - print_routes.traverse(app._router._roots, verbose=True) - - route, get_info, options_info = output.getvalue().strip().split('\n') - assert '-> /test' == route - - # NOTE(kgriffs) We might receive these in either order, since the - # method map is not ordered, so check and swap if necessary. - if options_info.startswith('-->GET'): - get_info, options_info = options_info, get_info - - assert options_info.startswith('-->OPTIONS') - assert '{}:'.format(normpath('falcon/responders.py')) in options_info - - assert get_info.startswith('-->GET') - - # NOTE(vytas): This builds upon the fact that on_get is defined on line - # 18 or 25 (in the case of DummyResourceAsync) in the present file. - # Adjust the test if the said responder is relocated, or just check for - # any number if this becomes too painful to maintain. - path = normpath('tests/test_cmd_print_api.py') - - assert ( - get_info.endswith('{}:14'.format(path)) or - get_info.endswith('{}:21'.format(path)) - ) - - -def test_traverse(app): - """Ensure traverse() finds the proper routes.""" - output = io.StringIO() - with redirected(stdout=output): - print_routes.traverse(app._router._roots, verbose=False) - - route = output.getvalue().strip() - assert '-> /test' == route diff --git a/tests/test_http_custom_method_routing.py b/tests/test_http_custom_method_routing.py index 06483d71f..1c67ae45b 100644 --- a/tests/test_http_custom_method_routing.py +++ b/tests/test_http_custom_method_routing.py @@ -29,10 +29,9 @@ def cleanup_constants(): # Forcing reload to make sure we used a module import and didn't import # the list directly. importlib.reload(falcon.constants) + orig = list(falcon.constants.COMBINED_METHODS) yield - falcon.constants.COMBINED_METHODS = list( - set(falcon.constants.COMBINED_METHODS) - set(FALCON_CUSTOM_HTTP_METHODS) - ) + falcon.constants.COMBINED_METHODS = orig if 'FALCON_CUSTOM_HTTP_METHODS' in os.environ: del os.environ['FALCON_CUSTOM_HTTP_METHODS'] diff --git a/tests/test_inspect.py b/tests/test_inspect.py new file mode 100644 index 000000000..18c7f270f --- /dev/null +++ b/tests/test_inspect.py @@ -0,0 +1,654 @@ +from functools import partial +import os +import sys + +import _inspect_fixture as i_f +import pytest + +from falcon import inspect, routing + + +def get_app(asgi, cors=True, **kw): + if asgi: + from falcon.asgi import App as AsyncApp + return AsyncApp(cors_enable=cors, **kw) + else: + from falcon import App + return App(cors_enable=cors, **kw) + + +def make_app(): + app = get_app(False, cors=True) + app.add_middleware(i_f.MyMiddleware()) + app.add_middleware(i_f.OtherMiddleware()) + + app.add_sink(i_f.sinkFn, '/sink_fn') + app.add_sink(i_f.SinkClass(), '/sink_cls') + + app.add_error_handler(RuntimeError, i_f.my_error_handler) + + app.add_route('/foo', i_f.MyResponder()) + app.add_route('/foo/{id}', i_f.MyResponder(), suffix='id') + app.add_route('/bar', i_f.OtherResponder(), suffix='id') + + app.add_static_route('/fal', os.path.abspath('falcon')) + app.add_static_route('/tes', os.path.abspath('tests'), fallback_filename='conftest.py') + return app + + +def make_app_async(): + app = get_app(True, cors=True) + app.add_middleware(i_f.MyMiddlewareAsync()) + app.add_middleware(i_f.OtherMiddlewareAsync()) + + app.add_sink(i_f.sinkFn, '/sink_fn') + app.add_sink(i_f.SinkClass(), '/sink_cls') + + app.add_error_handler(RuntimeError, i_f.my_error_handler_async) + + app.add_route('/foo', i_f.MyResponderAsync()) + app.add_route('/foo/{id}', i_f.MyResponderAsync(), suffix='id') + app.add_route('/bar', i_f.OtherResponderAsync(), suffix='id') + + app.add_static_route('/fal', os.path.abspath('falcon')) + app.add_static_route('/tes', os.path.abspath('tests'), fallback_filename='conftest.py') + return app + + +class TestInspectApp: + def test_empty_app(self, asgi): + ai = inspect.inspect_app(get_app(asgi, False)) + + assert ai.routes == [] + assert ai.middleware.middleware_tree.request == [] + assert ai.middleware.middleware_tree.resource == [] + assert ai.middleware.middleware_tree.response == [] + assert ai.middleware.middleware_classes == [] + assert ai.middleware.independent is True + assert ai.static_routes == [] + assert ai.sinks == [] + assert len(ai.error_handlers) == 3 + assert ai.asgi is asgi + + def test_dependent_middlewares(self, asgi): + app = get_app(asgi, cors=False, independent_middleware=False) + ai = inspect.inspect_app(app) + assert ai.middleware.independent is False + + def test_app(self, asgi): + ai = inspect.inspect_app(make_app_async() if asgi else make_app()) + + assert len(ai.routes) == 3 + assert len(ai.middleware.middleware_tree.request) == 2 + assert len(ai.middleware.middleware_tree.resource) == 1 + assert len(ai.middleware.middleware_tree.response) == 3 + assert len(ai.middleware.middleware_classes) == 3 + assert len(ai.static_routes) == 2 + assert len(ai.sinks) == 2 + assert len(ai.error_handlers) == 4 + assert ai.asgi is asgi + + def check_route(self, asgi, r, p, cn, ml, fnt): + assert isinstance(r, inspect.RouteInfo) + assert r.path == p + if asgi: + cn += 'Async' + assert r.class_name == cn + assert '_inspect_fixture.py' in r.source_info + + for m in r.methods: + assert isinstance(m, inspect.RouteMethodInfo) + internal = '_inspect_fixture.py' not in m.source_info + assert m.internal is internal + if not internal: + assert m.method in ml + assert '_inspect_fixture.py' in m.source_info + assert m.function_name == fnt.format(m.method).lower() + + def test_routes(self, asgi): + routes = inspect.inspect_routes(make_app_async() if asgi else make_app()) + + self.check_route( + asgi, routes[0], '/foo', 'MyResponder', ['GET', 'POST', 'DELETE'], 'on_{}' + ) + self.check_route( + asgi, routes[1], '/foo/{id}', 'MyResponder', ['GET', 'PUT', 'DELETE'], 'on_{}_id' + ) + self.check_route(asgi, routes[2], '/bar', 'OtherResponder', ['POST'], 'on_{}_id') + + def test_routes_empty_paths(self, asgi): + app = get_app(asgi) + r = i_f.MyResponderAsync() if asgi else i_f.MyResponder() + app.add_route('/foo/bar/baz', r) + + routes = inspect.inspect_routes(app) + + assert len(routes) == 1 + + self.check_route( + asgi, routes[0], '/foo/bar/baz', 'MyResponder', ['GET', 'POST', 'DELETE'], 'on_{}' + ) + + def test_static_routes(self, asgi): + routes = inspect.inspect_static_routes(make_app_async() if asgi else make_app()) + + assert all(isinstance(sr, inspect.StaticRouteInfo) for sr in routes) + assert routes[-1].prefix == '/fal/' + assert routes[-1].directory == os.path.abspath('falcon') + assert routes[-1].fallback_filename is None + assert routes[-2].prefix == '/tes/' + assert routes[-2].directory == os.path.abspath('tests') + assert routes[-2].fallback_filename.endswith('conftest.py') + + def test_sync(self, asgi): + sinks = inspect.inspect_sinks(make_app_async() if asgi else make_app()) + + assert all(isinstance(s, inspect.SinkInfo) for s in sinks) + assert sinks[-1].prefix == '/sink_fn' + assert sinks[-1].name == 'sinkFn' + assert '_inspect_fixture.py' in sinks[-1].source_info + assert sinks[-2].prefix == '/sink_cls' + assert sinks[-2].name == 'SinkClass' + assert '_inspect_fixture.py' in sinks[-2].source_info + + @pytest.mark.skipif(sys.version_info < (3, 6), reason='dict order is not stable') + def test_error_handler(self, asgi): + errors = inspect.inspect_error_handlers(make_app_async() if asgi else make_app()) + + assert all(isinstance(e, inspect.ErrorHandlerInfo) for e in errors) + assert errors[-1].error == 'RuntimeError' + assert errors[-1].name == 'my_error_handler_async' if asgi else 'my_error_handler' + assert '_inspect_fixture.py' in errors[-1].source_info + assert errors[-1].internal is False + for eh in errors[:-1]: + assert eh.internal + assert eh.error in ('Exception', 'HTTPStatus', 'HTTPError') + + def test_middleware(self, asgi): + mi = inspect.inspect_middlewares(make_app_async() if asgi else make_app()) + + def test(m, cn, ml, inte): + assert isinstance(m, inspect.MiddlewareClassInfo) + assert m.name == cn + if inte: + assert '_inspect_fixture.py' not in m.source_info + else: + assert '_inspect_fixture.py' in m.source_info + + for mm in m.methods: + assert isinstance(mm, inspect.MiddlewareMethodInfo) + if inte: + assert '_inspect_fixture.py' not in mm.source_info + else: + assert '_inspect_fixture.py' in mm.source_info + assert mm.function_name in ml + + test( + mi.middleware_classes[0], + 'CORSMiddleware', + ['process_response_async'] if asgi else ['process_response'], + True, + ) + test( + mi.middleware_classes[1], + 'MyMiddlewareAsync' if asgi else 'MyMiddleware', + ['process_request', 'process_resource', 'process_response'], + False, + ) + test( + mi.middleware_classes[2], + 'OtherMiddlewareAsync' if asgi else 'OtherMiddleware', + ['process_request', 'process_resource', 'process_response'], + False, + ) + + def test_middleware_tree(self, asgi): + mi = inspect.inspect_middlewares(make_app_async() if asgi else make_app()) + + def test(tl, names, cls): + for (t, n, c) in zip(tl, names, cls): + assert isinstance(t, inspect.MiddlewareTreeItemInfo) + assert t.name == n + assert t.class_name == c + + assert isinstance(mi.middleware_tree, inspect.MiddlewareTreeInfo) + + test( + mi.middleware_tree.request, + ['process_request'] * 2, + [n + 'Async' if asgi else n for n in ['MyMiddleware', 'OtherMiddleware']], + ) + test( + mi.middleware_tree.resource, + ['process_resource'], + ['MyMiddlewareAsync' if asgi else 'MyMiddleware'], + ) + test( + mi.middleware_tree.response, + [ + 'process_response', + 'process_response', + 'process_response_async' if asgi else 'process_response', + ], + [ + 'OtherMiddlewareAsync' if asgi else 'OtherMiddleware', + 'MyMiddlewareAsync' if asgi else 'MyMiddleware', + 'CORSMiddleware', + ], + ) + + +def test_route_method_info_suffix(): + ri = inspect.RouteMethodInfo('foo', '', 'on_get', False) + assert ri.suffix == '' + + ri = inspect.RouteMethodInfo('foo', '', 'on_get_suffix', False) + assert ri.suffix == 'suffix' + + ri = inspect.RouteMethodInfo('foo', '', 'on_get_multiple_underscores_suffix', False) + assert ri.suffix == 'multiple_underscores_suffix' + + ri = inspect.RouteMethodInfo('foo', '', 'some_other_fn_name', False) + assert ri.suffix == '' + + +class TestRouter: + def test_compiled_partial(self): + r = routing.CompiledRouter() + r.add_route('/foo', i_f.MyResponder()) + # override a method with a partial + r._roots[0].method_map['GET'] = partial(r._roots[0].method_map['GET']) + ri = inspect.inspect_compiled_router(r) + + for m in ri[0].methods: + if m.method == 'GET': + assert '_inspect_fixture' in m.source_info + + def test_compiled_no_method_map(self): + r = routing.CompiledRouter() + r.add_route('/foo', i_f.MyResponder()) + # clear the method map + r._roots[0].method_map.clear() + ri = inspect.inspect_compiled_router(r) + + assert ri[0].path == '/foo' + assert ri[0].class_name == 'MyResponder' + assert ri[0].methods == [] + + def test_register_router_not_found(self, monkeypatch): + monkeypatch.setattr(inspect, '_supported_routers', {}) + + app = get_app(False) + with pytest.raises(TypeError, match='Unsupported router class'): + inspect.inspect_routes(app) + + def test_register_other_router(self, monkeypatch): + monkeypatch.setattr(inspect, '_supported_routers', {}) + + app = get_app(False) + app._router = i_f.MyRouter() + + @inspect.register_router(i_f.MyRouter) + def print_routes(r): + assert r is app._router + return [inspect.RouteInfo('foo', 'bar', '', [])] + + ri = inspect.inspect_routes(app) + + assert ri[0].source_info == '' + assert ri[0].path == 'foo' + assert ri[0].class_name == 'bar' + assert ri[0].methods == [] + + def test_register_router_multiple_time(self, monkeypatch): + monkeypatch.setattr(inspect, '_supported_routers', {}) + + @inspect.register_router(i_f.MyRouter) + def print_routes(r): + return [] + + with pytest.raises(ValueError, match='Another function is already registered'): + @inspect.register_router(i_f.MyRouter) + def print_routes2(r): + return [] + + +def test_info_class_repr_to_string(): + ai = inspect.inspect_app(make_app()) + + assert str(ai) == ai.to_string() + assert str(ai.routes[0]) == ai.routes[0].to_string() + assert str(ai.routes[0].methods[0]) == ai.routes[0].methods[0].to_string() + assert str(ai.middleware) == ai.middleware.to_string() + s = str(ai.middleware.middleware_classes[0]) + assert s == ai.middleware.middleware_classes[0].to_string() + s = str(ai.middleware.middleware_tree.request[0]) + assert s == ai.middleware.middleware_tree.request[0].to_string() + assert str(ai.static_routes[0]) == ai.static_routes[0].to_string() + assert str(ai.sinks[0]) == ai.sinks[0].to_string() + assert str(ai.error_handlers[0]) == ai.error_handlers[0].to_string() + + +class TestInspectVisitor: + def test_inspect_visitor(self): + iv = inspect.InspectVisitor() + with pytest.raises(RuntimeError, match='This visitor does not support'): + iv.process(123) + with pytest.raises(RuntimeError, match='This visitor does not support'): + iv.process(inspect.RouteInfo('f', 'o', 'o', [])) + + def test_process(self): + class FooVisitor(inspect.InspectVisitor): + def visit_route(self, route): + return 'foo' + + assert FooVisitor().process(inspect.RouteInfo('f', 'o', 'o', [])) == 'foo' + + +def test_string_visitor_class(): + assert issubclass(inspect.StringVisitor, inspect.InspectVisitor) + + sv = inspect.StringVisitor() + assert sv.verbose is False + assert sv.internal is False + assert sv.name == '' + + +@pytest.mark.parametrize('internal', (True, False)) +class TestStringVisitor: + + def test_route_method(self, internal): + sv = inspect.StringVisitor(False, internal) + rm = inspect.inspect_routes(make_app())[0].methods[0] + + assert sv.process(rm) == '{0.method} - {0.function_name}'.format(rm) + + def test_route_method_verbose(self, internal): + sv = inspect.StringVisitor(True, internal) + rm = inspect.inspect_routes(make_app())[0].methods[0] + + assert sv.process(rm) == '{0.method} - {0.function_name} ({0.source_info})'.format(rm) + + def test_route(self, internal): + sv = inspect.StringVisitor(False, internal) + r = inspect.inspect_routes(make_app())[0] + + ml = [' ├── {}'.format(sv.process(m)) + for m in r.methods if not m.internal or internal][:-1] + ml += [' └── {}'.format(sv.process(m)) + for m in r.methods if not m.internal or internal][-1:] + + exp = '⇒ {0.path} - {0.class_name}:\n{1}'.format(r, '\n'.join(ml)) + assert sv.process(r) == exp + + def test_route_verbose(self, internal): + sv = inspect.StringVisitor(True, internal) + r = inspect.inspect_routes(make_app())[0] + + ml = [' ├── {}'.format(sv.process(m)) + for m in r.methods if not m.internal or internal][:-1] + ml += [' └── {}'.format(sv.process(m)) + for m in r.methods if not m.internal or internal][-1:] + + exp = '⇒ {0.path} - {0.class_name} ({0.source_info}):\n{1}'.format(r, '\n'.join(ml)) + assert sv.process(r) == exp + + def test_route_no_methods(self, internal): + sv = inspect.StringVisitor(False, internal) + r = inspect.inspect_routes(make_app())[0] + r.methods.clear() + exp = '⇒ {0.path} - {0.class_name}'.format(r) + assert sv.process(r) == exp + + @pytest.mark.parametrize('verbose', (True, False)) + def test_static_route(self, verbose, internal): + sv = inspect.StringVisitor(verbose, internal) + sr = inspect.inspect_static_routes(make_app()) + no_file = sr[1] + assert sv.process(no_file) == '↦ {0.prefix} {0.directory}'.format(no_file) + with_file = sr[0] + exp = '↦ {0.prefix} {0.directory} [{0.fallback_filename}]'.format(with_file) + assert sv.process(with_file) == exp + + def test_sink(self, internal): + sv = inspect.StringVisitor(False, internal) + s = inspect.inspect_sinks(make_app())[0] + + assert sv.process(s) == '⇥ {0.prefix} {0.name}'.format(s) + + def test_sink_verbose(self, internal): + sv = inspect.StringVisitor(True, internal) + s = inspect.inspect_sinks(make_app())[0] + + assert sv.process(s) == '⇥ {0.prefix} {0.name} ({0.source_info})'.format(s) + + def test_error_handler(self, internal): + sv = inspect.StringVisitor(False, internal) + e = inspect.inspect_error_handlers(make_app())[0] + + assert sv.process(e) == '⇜ {0.error} {0.name}'.format(e) + + def test_error_handler_verbose(self, internal): + sv = inspect.StringVisitor(True, internal) + e = inspect.inspect_error_handlers(make_app())[0] + + assert sv.process(e) == '⇜ {0.error} {0.name} ({0.source_info})'.format(e) + + def test_middleware_method(self, internal): + sv = inspect.StringVisitor(False, internal) + mm = inspect.inspect_middlewares(make_app()).middleware_classes[0].methods[0] + + assert sv.process(mm) == '{0.function_name}'.format(mm) + + def test_middleware_method_verbose(self, internal): + sv = inspect.StringVisitor(True, internal) + mm = inspect.inspect_middlewares(make_app()).middleware_classes[0].methods[0] + + assert sv.process(mm) == '{0.function_name} ({0.source_info})'.format(mm) + + def test_middleware_class(self, internal): + sv = inspect.StringVisitor(False, internal) + mc = inspect.inspect_middlewares(make_app()).middleware_classes[0] + + mml = [' ├── {}'.format(sv.process(m)) for m in mc.methods][:-1] + mml += [' └── {}'.format(sv.process(m)) for m in mc.methods][-1:] + + exp = '↣ {0.name}:\n{1}'.format(mc, '\n'.join(mml)) + assert sv.process(mc) == exp + + def test_middleware_class_verbose(self, internal): + sv = inspect.StringVisitor(True, internal) + mc = inspect.inspect_middlewares(make_app()).middleware_classes[0] + + mml = [' ├── {}'.format(sv.process(m)) for m in mc.methods][:-1] + mml += [' └── {}'.format(sv.process(m)) for m in mc.methods][-1:] + + exp = '↣ {0.name} ({0.source_info}):\n{1}'.format(mc, '\n'.join(mml)) + assert sv.process(mc) == exp + + def test_middleware_class_no_methods(self, internal): + sv = inspect.StringVisitor(False, internal) + mc = inspect.inspect_middlewares(make_app()).middleware_classes[0] + mc.methods.clear() + exp = '↣ {0.name}'.format(mc) + assert sv.process(mc) == exp + + @pytest.mark.parametrize('verbose', (True, False)) + def test_middleware_tree_item(self, verbose, internal): + sv = inspect.StringVisitor(verbose, internal) + mt = inspect.inspect_middlewares(make_app()).middleware_tree + for r, s in ((mt.request[0], '→'), (mt.resource[0], '↣'), (mt.response[0], '↢')): + assert sv.process(r) == '{0} {1.class_name}.{1.name}'.format(s, r) + + @pytest.mark.parametrize('verbose', (True, False)) + def test_middleware_tree(self, verbose, internal): + sv = inspect.StringVisitor(verbose, internal) + mt = inspect.inspect_middlewares(make_app()).middleware_tree + lines = [] + space = '' + for r in mt.request: + lines.append(space + sv.process(r)) + space += ' ' + lines.append('') + for r in mt.resource: + lines.append(space + sv.process(r)) + space += ' ' + lines.append('') + lines.append(space + ' ├── Process route responder') + lines.append('') + for r in mt.response: + space = space[:-2] + lines.append(space + sv.process(r)) + + assert sv.process(mt) == '\n'.join(lines) + + def test_middleware_tree_response_only(self, internal): + sv = inspect.StringVisitor(False, internal) + mt = inspect.inspect_middlewares(make_app()).middleware_tree + mt.request.clear() + mt.resource.clear() + lines = [] + space = ' ' * len(mt.response) + lines.append('') + lines.append(space + ' ├── Process route responder') + lines.append('') + for r in mt.response: + space = space[:-2] + lines.append(space + sv.process(r)) + + assert sv.process(mt) == '\n'.join(lines) + + def test_middleware_tree_no_response(self, internal): + sv = inspect.StringVisitor(False, internal) + mt = inspect.inspect_middlewares(make_app()).middleware_tree + mt.response.clear() + lines = [] + space = '' + for r in mt.request: + lines.append(space + sv.process(r)) + space += ' ' + lines.append('') + for r in mt.resource: + lines.append(space + sv.process(r)) + space += ' ' + lines.append('') + lines.append(space + ' ├── Process route responder') + + assert sv.process(mt) == '\n'.join(lines) + + def test_middleware_tree_no_resource(self, internal): + sv = inspect.StringVisitor(False, internal) + mt = inspect.inspect_middlewares(make_app()).middleware_tree + mt.resource.clear() + lines = [] + space = ' ' + for r in mt.request: + lines.append(space + sv.process(r)) + space += ' ' + lines.append('') + lines.append(space + ' ├── Process route responder') + lines.append('') + for r in mt.response: + space = space[:-2] + lines.append(space + sv.process(r)) + + assert sv.process(mt) == '\n'.join(lines) + + def test_middleware(self, internal): + sv = inspect.StringVisitor(False, internal) + m = inspect.inspect_middlewares(make_app()) + + assert sv.process(m) == sv.process(m.middleware_tree) + + def test_middleware_verbose(self, internal): + sv = inspect.StringVisitor(True, internal) + m = inspect.inspect_middlewares(make_app()) + + mt = sv.process(m.middleware_tree) + sv.indent += 4 + mc = '\n'.join(sv.process(cls) for cls in m.middleware_classes) + exp = '{}\n- Middlewares classes:\n{}'.format(mt, mc) + assert inspect.StringVisitor(True).process(m) == exp + + def make(self, sv, app, v, i, r=True, m=True, sr=True, s=True, e=True): + text = 'Falcon App (WSGI)' + sv.indent = 4 + if r: + text += '\n• Routes:\n{}'.format('\n'.join(sv.process(r) for r in app.routes)) + if m: + mt = sv.process(app.middleware) + text += '\n• Middleware ({}):\n{}'.format(app.middleware.independent_text, mt) + if sr: + sr = '\n'.join(sv.process(sr) for sr in app.static_routes) + text += '\n• Static routes:\n{}'.format(sr) + if s: + text += '\n• Sinks:\n{}'.format('\n'.join(sv.process(s) for s in app.sinks)) + if e: + err = '\n'.join(sv.process(e) for e in app.error_handlers if not e.internal or i) + text += '\n• Error handlers:\n{}'.format(err) + return text + + @pytest.mark.parametrize('verbose', (True, False)) + def test_app(self, verbose, internal): + sv = inspect.StringVisitor(verbose, internal) + app = inspect.inspect_app(make_app()) + + assert inspect.StringVisitor(verbose, internal).process(app) == self.make( + sv, app, verbose, internal) + + @pytest.mark.parametrize('verbose', (True, False)) + def test_app_no_routes(self, verbose, internal): + sv = inspect.StringVisitor(verbose, internal) + app = inspect.inspect_app(make_app()) + app.routes.clear() + assert inspect.StringVisitor(verbose, internal).process(app) == self.make( + sv, app, verbose, internal, r=False) + + @pytest.mark.parametrize('verbose', (True, False)) + def test_app_no_middleware(self, verbose, internal): + sv = inspect.StringVisitor(verbose, internal) + app = inspect.inspect_app(make_app()) + app.middleware.middleware_classes.clear() + app.middleware.middleware_tree.request.clear() + app.middleware.middleware_tree.resource.clear() + app.middleware.middleware_tree.response.clear() + assert inspect.StringVisitor(verbose, internal).process(app) == self.make( + sv, app, verbose, internal, m=False) + + @pytest.mark.parametrize('verbose', (True, False)) + def test_app_static_routes(self, verbose, internal): + sv = inspect.StringVisitor(verbose, internal) + app = inspect.inspect_app(make_app()) + app.static_routes.clear() + assert inspect.StringVisitor(verbose, internal).process(app) == self.make( + sv, app, verbose, internal, sr=False) + + @pytest.mark.parametrize('verbose', (True, False)) + def test_app_no_sink(self, verbose, internal): + sv = inspect.StringVisitor(verbose, internal) + app = inspect.inspect_app(make_app()) + app.sinks.clear() + assert inspect.StringVisitor(verbose, internal).process(app) == self.make( + sv, app, verbose, internal, s=False) + + @pytest.mark.parametrize('verbose', (True, False)) + def test_app_no_errors(self, verbose, internal): + sv = inspect.StringVisitor(verbose, internal) + app = inspect.inspect_app(make_app()) + app.error_handlers.clear() + assert inspect.StringVisitor(verbose, internal).process(app) == self.make( + sv, app, verbose, internal, e=False) + + def test_app_name(self, internal): + sv = inspect.StringVisitor(False, internal, name='foo') + app = inspect.inspect_app(make_app()) + + s = sv.process(app).splitlines()[0] + assert s == 'foo (WSGI)' + assert app.to_string(name='bar').splitlines()[0] == 'bar (WSGI)' + + +def test_is_internal(): + assert inspect._is_internal(1) is False + assert inspect._is_internal(dict) is False + assert inspect._is_internal(inspect) is True