14284555682

Committed 05 Apr 2025 06:17PM UTC coverage: 88.306% (+0.09%) from 88.221%

Build # 14284555682

Build Type

Pull #187

github

Committed by

cle-b

Commit Message

xfail for windows

Pull Request Pull Request #187: add fastapi support

Run Details

215 of 235 new or added lines in 10 files covered. (91.49%)

3 existing lines in 1 file now uncovered.

2054 of 2326 relevant lines covered (88.31%)

0.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.67

/httpdbg/hooks/fastapi.py

# -*- coding: utf-8 -*-
from collections.abc import Callable
from contextlib import contextmanager
import functools
from functools import wraps
from typing import Dict
from typing import Generator
from typing import Union

import datetime

from httpdbg.hooks.utils import getcallargs
from httpdbg.hooks.utils import decorate
from httpdbg.hooks.utils import undecorate
from httpdbg.initiator import httpdbg_endpoint
from httpdbg.records import HTTPRecords


def set_hook_fastapi_endpoint(records: HTTPRecords, method: Callable):

    @wraps(method)
    def hook(*args, **kwargs):

        with httpdbg_endpoint(
            records,
            method,
            *args,
            **kwargs,
        ):
            ret = method(*args, **kwargs)
        return ret

    return hook


def set_hook_fastapi_apirouter_add_api_route(
    records: HTTPRecords,
    method: Callable,
    already_mapped: Union[Dict[Callable, Callable], None] = None,
):

    @wraps(method)
    def hook(*args, **kwargs):
        callargs = getcallargs(method, *args, **kwargs)

        param = "endpoint"

        if param in callargs:
            original_view_func = callargs[param]
            if original_view_func not in already_mapped:
                callargs[param] = decorate(
                    records, callargs[param], set_hook_fastapi_endpoint
                )
                already_mapped[original_view_func] = callargs[param]
            else:
                callargs[param] = already_mapped[original_view_func]

            args = [callargs[param] if x == original_view_func else x for x in args]
            if param in kwargs:
                kwargs[param] = callargs[param]

        return method(*args, **kwargs)

    return hook


def set_hook_fastapi_app(records: HTTPRecords, method: Callable):

    @wraps(method)
    async def hook(*args, **kwargs):

        socketdata = None

        callargs = getcallargs(method, *args, **kwargs)

        if "request" in callargs:
            if hasattr(callargs["request"], "__httpdbg_socketdata"):
                socketdata = callargs["request"].__httpdbg_socketdata

        with httpdbg_endpoint(
            records,
            method,
            *args,
            **kwargs,
        ) as group:
            if group and (group.id in records.groups):
                if socketdata and (socketdata in records._sockets):
                    records.groups[group.id].tbegin = records._sockets[
                        socketdata
                    ].record.tbegin - datetime.timedelta(milliseconds=1)
                    records._sockets[socketdata].record.group_id = group.id

            ret = await method(*args, **kwargs)

        return ret

    return hook


def set_hook_fastapi_routing_get_request_handler(
    records: HTTPRecords, method: Callable
):

    @wraps(method)
    def hook(*args, **kwargs):

        app = method(*args, **kwargs)

        app = decorate(records, app, set_hook_fastapi_app)

        return app

    return hook


@contextmanager
def hook_fastapi(records: HTTPRecords) -> Generator[None, None, None]:
    hooks = False
    try:
        import fastapi.routing

        already_mapped: Dict[Callable, Callable] = {}

        set_hook_fastapi_apirouter_add_api_route_with_already_mapped = (
            functools.partial(
                set_hook_fastapi_apirouter_add_api_route, already_mapped=already_mapped
            )
        )

        fastapi.routing.APIRouter.add_api_route = decorate(
            records,
            fastapi.routing.APIRouter.add_api_route,
            set_hook_fastapi_apirouter_add_api_route_with_already_mapped,
        )

        fastapi.routing.get_request_handler = decorate(
            records,
            fastapi.routing.get_request_handler,
            set_hook_fastapi_routing_get_request_handler,
        )

        hooks = True
    except ImportError:
        pass

    yield

    if hooks:
        fastapi.routing.APIRouter.add_api_route = undecorate(
            fastapi.routing.APIRouter.add_api_route
        )
        fastapi.routing.get_request_handler = undecorate(
            fastapi.routing.get_request_handler
        )

1	# -- coding: utf-8 --
2	from collections.abc import Callable	1✔
3	from contextlib import contextmanager	1✔
4	import functools	1✔
5	from functools import wraps	1✔
6	from typing import Dict	1✔
7	from typing import Generator	1✔
8	from typing import Union	1✔
9
10	import datetime	1✔
11
12	from httpdbg.hooks.utils import getcallargs	1✔
13	from httpdbg.hooks.utils import decorate	1✔
14	from httpdbg.hooks.utils import undecorate	1✔
15	from httpdbg.initiator import httpdbg_endpoint	1✔
16	from httpdbg.records import HTTPRecords	1✔
17
18
19	def set_hook_fastapi_endpoint(records: HTTPRecords, method: Callable):	1✔
20
21	@wraps(method)	1✔
22	def hook(args, *kwargs):	1✔
23
24	with httpdbg_endpoint(	1✔
25	records,
26	method,
27	*args,
28	**kwargs,
29	):
30	ret = method(args, *kwargs)	1✔
31	return ret	1✔
32
33	return hook	1✔
34
35
36	def set_hook_fastapi_apirouter_add_api_route(	1✔
37	records: HTTPRecords,
38	method: Callable,
39	already_mapped: Union[Dict[Callable, Callable], None] = None,
40	):
41
42	@wraps(method)	1✔
43	def hook(args, *kwargs):	1✔
44	callargs = getcallargs(method, args, *kwargs)	1✔
45
46	param = "endpoint"	1✔
47
48	if param in callargs:	1✔
49	original_view_func = callargs[param]	1✔
50	if original_view_func not in already_mapped:	1✔
51	callargs[param] = decorate(	1✔
52	records, callargs[param], set_hook_fastapi_endpoint
53	)
54	already_mapped[original_view_func] = callargs[param]	1✔
55	else:
NEW 56	callargs[param] = already_mapped[original_view_func]	×
57
58	args = [callargs[param] if x == original_view_func else x for x in args]	1✔
59	if param in kwargs:	1✔
NEW 60	kwargs[param] = callargs[param]	×
61
62	return method(args, *kwargs)	1✔
63
64	return hook	1✔
65
66
67	def set_hook_fastapi_app(records: HTTPRecords, method: Callable):	1✔
68
69	@wraps(method)	1✔
70	async def hook(args, *kwargs):	1✔
71
72	socketdata = None	1✔
73
74	callargs = getcallargs(method, args, *kwargs)	1✔
75
76	if "request" in callargs:	1✔
77	if hasattr(callargs["request"], "__httpdbg_socketdata"):	1✔
78	socketdata = callargs["request"].__httpdbg_socketdata	1✔
79
80	with httpdbg_endpoint(	1✔
81	records,
82	method,
83	*args,
84	**kwargs,
85	) as group:
86	if group and (group.id in records.groups):	1✔
87	if socketdata and (socketdata in records._sockets):	1✔
88	records.groups[group.id].tbegin = records._sockets[	1✔
89	socketdata
90	].record.tbegin - datetime.timedelta(milliseconds=1)
91	records._sockets[socketdata].record.group_id = group.id	1✔
92
93	ret = await method(args, *kwargs)	1✔
94
95	return ret	1✔
96
97	return hook	1✔
98
99
100	def set_hook_fastapi_routing_get_request_handler(	1✔
101	records: HTTPRecords, method: Callable
102	):
103
104	@wraps(method)	1✔
105	def hook(args, *kwargs):	1✔
106
107	app = method(args, *kwargs)	1✔
108
109	app = decorate(records, app, set_hook_fastapi_app)	1✔
110
111	return app	1✔
112
113	return hook	1✔
114
115
116	@contextmanager	1✔
117	def hook_fastapi(records: HTTPRecords) -> Generator[None, None, None]:	1✔
118	hooks = False	1✔
119	try:	1✔
120	import fastapi.routing	1✔
121
122	already_mapped: Dict[Callable, Callable] = {}	1✔
123
124	set_hook_fastapi_apirouter_add_api_route_with_already_mapped = (	1✔
125	functools.partial(
126	set_hook_fastapi_apirouter_add_api_route, already_mapped=already_mapped
127	)
128	)
129
130	fastapi.routing.APIRouter.add_api_route = decorate(	1✔
131	records,
132	fastapi.routing.APIRouter.add_api_route,
133	set_hook_fastapi_apirouter_add_api_route_with_already_mapped,
134	)
135
136	fastapi.routing.get_request_handler = decorate(	1✔
137	records,
138	fastapi.routing.get_request_handler,
139	set_hook_fastapi_routing_get_request_handler,
140	)
141
142	hooks = True	1✔
NEW 143	except ImportError:	×
NEW 144	pass	×
145
146	yield	1✔
147
148	if hooks:	1✔
149	fastapi.routing.APIRouter.add_api_route = undecorate(	1✔
150	fastapi.routing.APIRouter.add_api_route
151	)
152	fastapi.routing.get_request_handler = undecorate(	1✔
153	fastapi.routing.get_request_handler
154	)

cle-b / httpdbg / 14284555682

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous