Cutlery/calibre
1
0
Fork 0
mirror of https://github.com/kovidgoyal/calibre.git synced 2025-07-09 03:04:10 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add a test for POST with data

Browse Source
This commit is contained in:
Kovid Goyal 2024-08-17 09:51:34 +05:30
parent 23cf355130
commit b2745092c6
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C
2 changed files with 24 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/calibre/scraper/qt.py
View File

@ -167,6 +167,11 @@ class Browser:
else:
f.write(data)
cmd['data_path'] = f.name
for k, v in cmd['headers']:
if k.lower() == 'content-type':
break
else:
cmd['headers'].append(('Content-Type', 'application/x-www-form-urlencoded'))
res = FakeResponse()
self.dispatch_map[self.id_counter] = res.queue
self._send_command(cmd)

23
src/calibre/scraper/test_fetch_backend.py
View File

@ -23,6 +23,11 @@ class Handler(http.server.BaseHTTPRequestHandler):
self.test_obj = test_obj
super().__init__(*a)
def do_POST(self):
if self.test_obj.dont_send_response:
return
self.do_response()
def do_GET(self):
if self.test_obj.dont_send_response:
return
@ -35,6 +40,9 @@ class Handler(http.server.BaseHTTPRequestHandler):
self.end_headers()
self.flush_headers()
return
self.do_response()
def do_response(self):
h = {}
for k, v in self.headers.items():
h.setdefault(k, []).append(v)
@ -43,7 +51,10 @@ class Handler(http.server.BaseHTTPRequestHandler):
'path': self.path,
'headers': h,
'request_count': self.test_obj.request_count,
'method': self.command,
}
if 'Content-Length' in self.headers:
ans['data'] = self.rfile.read(int(self.headers['Content-Length'])).decode()
data = json.dumps(ans).encode()
self.send_response(http.HTTPStatus.OK)
self.send_header('Content-type', 'application/json')
@ -91,13 +102,13 @@ class TestFetchBackend(unittest.TestCase):
def u(path=''):
return f'http://localhost:{self.port}{path}'
def get(path='', headers=None, timeout=None):
def get(path='', headers=None, timeout=None, data=None):
url = u(path)
if headers:
req = Request(url, headers=headers)
else:
req = url
res = br.open(req, timeout=timeout)
res = br.open(req, data=data, timeout=timeout)
raw = res.read()
ans = json.loads(raw)
ans['final_url'] = res.geturl()
@ -118,6 +129,7 @@ class TestFetchBackend(unittest.TestCase):
try:
r = get()
self.ae(r['method'], 'GET')
self.ae(r['request_count'], 1)
self.ae(r['headers']['th'], ['1'])
self.ae(r['headers']['User-Agent'], ['test-ua'])
@ -144,17 +156,20 @@ class TestFetchBackend(unittest.TestCase):
r = get()
self.ae(r['headers']['User-Agent'], ['man in black'])
self.ae(r['headers']['Cookie'], ['sc=1; cook=ie'])
r = get(data=b'1234')
self.ae(r['method'], 'POST')
self.ae(r['data'], '1234')
finally:
br.shutdown()
def run_server(self):
import socketserver
from http.server import ThreadingHTTPServer
def create_handler(*a):
ans = Handler(self, *a)
return ans
with socketserver.TCPServer(("", 0), create_handler) as httpd:
with ThreadingHTTPServer(("", 0), create_handler) as httpd:
self.server = httpd
self.port = httpd.server_address[1]
self.server_started.set()