1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
import io
import json
import pytest
from cross_web import AsyncFlaskHTTPRequestAdapter, FlaskHTTPRequestAdapter
pytestmark = [pytest.mark.flask]
def test_sync_flask_adapter() -> None:
from flask import Flask
app = Flask(__name__)
# Create test context directly
with app.test_request_context(
"/test?query=test",
method="POST",
data={"form": "data", "file": (io.BytesIO(b"upload"), "test.txt")},
headers={"Cookie": "session=123"},
):
from flask import request
adapter = FlaskHTTPRequestAdapter(request)
assert adapter.query_params == {"query": "test"}
assert adapter.method == "POST"
assert adapter.headers["Content-Type"].startswith("multipart/form-data")
assert adapter.post_data["form"] == "data"
assert "file" in adapter.files
assert adapter.content_type is not None and adapter.content_type.startswith(
"multipart/form-data"
)
assert "/test" in adapter.url
assert dict(adapter.cookies) == {"session": "123"}
form_data = adapter.get_form_data()
assert "file" in form_data.files
assert form_data.form["form"] == "data"
def test_sync_flask_adapter_json() -> None:
from flask import Flask
app = Flask(__name__)
with app.test_request_context(
"/test?query=test",
method="POST",
data=json.dumps({"key": "value"}),
content_type="application/json",
headers={"Cookie": "session=123"},
):
from flask import request
adapter = FlaskHTTPRequestAdapter(request)
assert adapter.query_params == {"query": "test"}
assert adapter.body == '{"key": "value"}'
assert adapter.method == "POST"
assert adapter.headers["Content-Type"] == "application/json"
assert adapter.content_type == "application/json"
assert "/test" in adapter.url
assert dict(adapter.cookies) == {"session": "123"}
@pytest.mark.asyncio
async def test_async_flask_adapter() -> None:
from flask import Flask
app = Flask(__name__)
with app.test_request_context(
"/test?query=test",
method="POST",
data=json.dumps({"key": "value"}),
content_type="application/json",
headers={"Cookie": "session=123"},
):
from flask import request
adapter = AsyncFlaskHTTPRequestAdapter(request)
assert adapter.query_params == {"query": "test"}
body = await adapter.get_body()
assert body == b'{"key": "value"}'
assert adapter.method == "POST"
assert adapter.headers["Content-Type"] == "application/json"
assert adapter.content_type == "application/json"
assert "/test" in adapter.url
assert dict(adapter.cookies) == {"session": "123"}
@pytest.mark.asyncio
async def test_async_flask_adapter_multipart() -> None:
from flask import Flask
app = Flask(__name__)
# Create test context with multipart data
with app.test_request_context(
"/test?query=test",
method="POST",
data={"form": "data", "file": (io.BytesIO(b"upload"), "test.txt")},
headers={"Cookie": "session=123"},
):
from flask import request
adapter = AsyncFlaskHTTPRequestAdapter(request)
assert adapter.query_params == {"query": "test"}
assert adapter.method == "POST"
assert adapter.headers["Content-Type"].startswith("multipart/form-data")
assert adapter.content_type is not None and adapter.content_type.startswith(
"multipart/form-data"
)
assert "/test" in adapter.url
assert dict(adapter.cookies) == {"session": "123"}
form_data = await adapter.get_form_data()
assert "file" in form_data.files
assert form_data.form["form"] == "data"
|
