pythonnative/tests/test_net.py at main · pythonnative/pythonnative · GitHub

Repository navigation

History

234 lines (187 loc) · 7.44 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

"""Unit tests for pn.fetch (async HTTP client)."""

from __future__ import annotations

import asyncio

import json

import socket

import threading

from http.server import BaseHTTPRequestHandler, HTTPServer

from typing import Generator

import pytest

from pythonnative.net import HTTPError, Response, _build_request, fetch

# ======================================================================

# Mini HTTP server fixture

# ======================================================================

class _EchoHandler(BaseHTTPRequestHandler):

"""Records every request and replies with metadata or canned content.

Routes:

- ``GET /text`` → ``"hello"`` with ``Content-Type: text/plain``.

- ``GET /json`` → ``{"ok": True}``.

- ``GET /status/<code>`` → empty body with that status code.

- ``POST /echo`` → JSON object echoing method, path, body, headers.

"""

def log_message(self, format: str, *args: object) -> None: # noqa: A002

# Suppress per-request stderr noise.

return

def do_GET(self) -> None:

if self.path == "/text":

body = b"hello"

self.send_response(200)

self.send_header("Content-Type", "text/plain; charset=utf-8")

self.send_header("Content-Length", str(len(body)))

self.end_headers()

self.wfile.write(body)

return

if self.path == "/json":

body = json.dumps({"ok": True}).encode("utf-8")

self.send_response(200)

self.send_header("Content-Type", "application/json")

self.send_header("Content-Length", str(len(body)))

self.end_headers()

self.wfile.write(body)

return

if self.path.startswith("/status/"):

code = int(self.path.rsplit("/", 1)[-1])

self.send_response(code)

self.send_header("Content-Length", "0")

self.end_headers()

return

self.send_response(404)

self.send_header("Content-Length", "0")

self.end_headers()

def do_POST(self) -> None:

path_only = self.path.split("?", 1)[0]

if path_only != "/echo":

self.send_response(404)

self.send_header("Content-Length", "0")

self.end_headers()

return

length = int(self.headers.get("Content-Length", "0"))

raw = self.rfile.read(length) if length else b""

payload = {

"method": self.command,

"path": self.path,

"body": raw.decode("utf-8"),

"headers": {k: v for k, v in self.headers.items()},

}

body = json.dumps(payload).encode("utf-8")

self.send_response(200)

self.send_header("Content-Type", "application/json")

self.send_header("Content-Length", str(len(body)))

self.end_headers()

self.wfile.write(body)

@pytest.fixture(scope="module")

def echo_server() -> Generator[str, None, None]:

server = HTTPServer(("127.0.0.1", 0), _EchoHandler)

port = server.server_address[1]

thread = threading.Thread(target=server.serve_forever, daemon=True)

thread.start()

try:

yield f"http://127.0.0.1:{port}"

finally:

server.shutdown()

thread.join(timeout=2.0)

# ======================================================================

# Tests

# ======================================================================

def test_get_text(echo_server: str) -> None:

async def run() -> Response:

return await fetch(echo_server + "/text")

resp = asyncio.run(run())

assert resp.status == 200

assert resp.ok

assert resp.text() == "hello"

def test_get_json(echo_server: str) -> None:

async def run() -> Response:

return await fetch(echo_server + "/json")

resp = asyncio.run(run())

assert resp.json() == {"ok": True}

def test_repr_summarizes_body(echo_server: str) -> None:

async def run() -> Response:

return await fetch(echo_server + "/text")

resp = asyncio.run(run())

text = repr(resp)

assert "status=200" in text

assert f"{len(resp.content)} bytes" in text

assert resp.text() not in text

def test_non_2xx_does_not_raise_but_keeps_body(echo_server: str) -> None:

async def run() -> Response:

return await fetch(echo_server + "/status/418")

resp = asyncio.run(run())

assert resp.status == 418

assert resp.ok is False

def test_raise_for_status_on_4xx(echo_server: str) -> None:

async def run() -> None:

resp = await fetch(echo_server + "/status/404")

resp.raise_for_status()

with pytest.raises(HTTPError) as exc_info:

asyncio.run(run())

assert exc_info.value.status == 404

def test_post_dict_body_becomes_json(echo_server: str) -> None:

async def run() -> dict:

resp = await fetch(

echo_server + "/echo",

method="POST",

body={"name": "Alice"},

)

return resp.json()

body = asyncio.run(run())

assert body["method"] == "POST"

assert body["headers"].get("Content-Type") == "application/json"

assert json.loads(body["body"]) == {"name": "Alice"}

def test_post_string_body(echo_server: str) -> None:

async def run() -> dict:

resp = await fetch(echo_server + "/echo", method="POST", body="raw")

return resp.json()

assert asyncio.run(run())["body"] == "raw"

def test_query_params_are_appended(echo_server: str) -> None:

async def run() -> Response:

return await fetch(

echo_server + "/echo",

method="POST",

params={"q": "hi there", "limit": 5},

)

resp = asyncio.run(run())

body = resp.json()

# urlencode preserves order for sorted dicts; check both keys are present.

assert "q=hi+there" in body["path"]

assert "limit=5" in body["path"]

def test_unreachable_host_raises_oserror() -> None:

# Bind to an ephemeral port then close it, guaranteeing nothing listens.

sock = socket.socket()

sock.bind(("127.0.0.1", 0))

port = sock.getsockname()[1]

sock.close()

async def run() -> Response:

return await fetch(f"http://127.0.0.1:{port}/", timeout=2.0)

with pytest.raises(OSError):

asyncio.run(run())

@pytest.mark.parametrize(

"header_name",

["Content-Type", "content-type", "CONTENT-TYPE", "Content-type"],

)

def test_custom_content_type_is_preserved_regardless_of_casing(header_name: str) -> None:

original_headers = {header_name: "application/merge-patch+json"}

request = _build_request(

url="https://example.com/users",

method="POST",

headers=original_headers,

body={"name": "Ada"},

params=None,

)

content_types = [value for name, value in request.header_items() if name.lower() == "content-type"]

assert content_types == ["application/merge-patch+json"]

assert original_headers == {header_name: "application/merge-patch+json"}

assert isinstance(request.data, bytes)

assert json.loads(request.data.decode("utf-8")) == {"name": "Ada"}

def test_mapping_body_defaults_to_application_json() -> None:

request = _build_request(

url="https://example.com/users",

method="POST",

headers=None,

body={"name": "Ada"},

params=None,

)

content_types = [value for name, value in request.header_items() if name.lower() == "content-type"]

assert content_types == ["application/json"]

assert isinstance(request.data, bytes)

assert json.loads(request.data.decode("utf-8")) == {"name": "Ada"}

pythonnative/tests/test_net.py at main · pythonnative/pythonnative · GitHub

Repository navigation

Expand file tree

Breadcrumbs

test_net.py

Latest commit

History

Breadcrumbs

test_net.py

File metadata and controls