jiigen666/libtorrent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rename s -> self in web_server.py test facility

Browse Source
This commit is contained in:
arvidn
2021-08-26 10:27:19 +02:00
committed by Arvid Norberg
parent 8603514d3c
commit b905dba322
1 changed files with 66 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
test/web_server.py
View File

@ -42,67 +42,67 @@ class http_handler(BaseHTTPRequestHandler):
traceback.print_exc(file=sys.stdout)
sys.stdout.flush()
def inner_do_GET(s):
def inner_do_GET(self):
print('INCOMING-REQUEST [from: {}]: {}'.format(s.request.getsockname(), s.requestline))
print(s.headers)
print('INCOMING-REQUEST [from: {}]: {}'.format(self.request.getsockname(), self.requestline))
print(self.headers)
sys.stdout.flush()
global chunked_encoding
global keepalive
# if the request contains the hostname and port. strip it
if s.path.startswith('http://') or s.path.startswith('https://'):
s.path = s.path[8:]
s.path = s.path[s.path.find('/'):]
if self.path.startswith('http://') or self.path.startswith('https://'):
self.path = self.path[8:]
self.path = self.path[self.path.find('/'):]
file_path = os.path.normpath(s.path)
file_path = os.path.normpath(self.path)
sys.stdout.flush()
if s.path == '/password_protected':
if self.path == '/password_protected':
passed = False
if 'Authorization' in s.headers:
auth = s.headers['Authorization']
if 'Authorization' in self.headers:
auth = self.headers['Authorization']
passed = auth == 'Basic %s' % base64.b64encode(b'testuser:testpass').decode()
if not passed:
s.send_response(401)
s.send_header("Connection", "close")
s.end_headers()
self.send_response(401)
self.send_header("Connection", "close")
self.end_headers()
return
s.path = '/test_file'
self.path = '/test_file'
file_path = os.path.normpath('/test_file')
if s.path == '/redirect':
s.send_response(301)
s.send_header("Location", "/test_file")
s.send_header("Connection", "close")
s.end_headers()
elif s.path == '/infinite_redirect':
s.send_response(301)
s.send_header("Location", "/infinite_redirect")
s.send_header("Connection", "close")
s.end_headers()
elif s.path == '/relative/redirect':
s.send_response(301)
s.send_header("Location", "../test_file")
s.send_header("Connection", "close")
s.end_headers()
elif s.path.startswith('/announce'):
s.send_response(200)
if self.path == '/redirect':
self.send_response(301)
self.send_header("Location", "/test_file")
self.send_header("Connection", "close")
self.end_headers()
elif self.path == '/infinite_redirect':
self.send_response(301)
self.send_header("Location", "/infinite_redirect")
self.send_header("Connection", "close")
self.end_headers()
elif self.path == '/relative/redirect':
self.send_response(301)
self.send_header("Location", "../test_file")
self.send_header("Connection", "close")
self.end_headers()
elif self.path.startswith('/announce'):
self.send_response(200)
response = b'd8:intervali1800e8:completei1e10:incompletei1e' + \
b'12:min intervali' + min_interval.encode() + b'e' + \
b'5:peers12:AAAABBCCCCDD' + \
b'6:peers618:EEEEEEEEEEEEEEEEFF' + \
b'e'
s.send_header("Content-Length", "%d" % len(response))
s.send_header("Connection", "close")
s.end_headers()
s.wfile.write(response)
s.request.close()
elif os.path.split(s.path)[1].startswith('seed?'):
query = s.path[6:]
self.send_header("Content-Length", "%d" % len(response))
self.send_header("Connection", "close")
self.end_headers()
self.wfile.write(response)
self.request.close()
elif os.path.split(self.path)[1].startswith('seed?'):
query = self.path[6:]
args_raw = query.split('&')
args = {}
for a in args_raw:
@ -113,7 +113,7 @@ class http_handler(BaseHTTPRequestHandler):
filename = ''
try:
filename = os.path.normpath(s.path[1:s.path.find('seed?') + 4])
filename = os.path.normpath(self.path[1:self.path.find('seed?') + 4])
print('filename = %s' % filename)
sys.stdout.flush()
f = open(filename, 'rb')
@ -121,20 +121,20 @@ class http_handler(BaseHTTPRequestHandler):
data = f.read(int(ranges[1]) - int(ranges[0]) + 1)
f.close()
s.send_response(200)
self.send_response(200)
print('sending %d bytes' % len(data))
sys.stdout.flush()
s.send_header("Content-Length", "%d" % len(data))
s.end_headers()
s.wfile.write(data)
self.send_header("Content-Length", "%d" % len(data))
self.end_headers()
self.wfile.write(data)
except Exception as e:
print('FILE ERROR: ', filename, e)
traceback.print_exc(file=sys.stdout)
sys.stdout.flush()
s.send_response(404)
s.send_header("Content-Length", "0")
self.send_response(404)
self.send_header("Content-Length", "0")
try:
s.end_headers()
self.end_headers()
except Exception:
pass
else:
@ -146,9 +146,9 @@ class http_handler(BaseHTTPRequestHandler):
size = int(os.stat(filename).st_size)
start_range = 0
end_range = size
if 'Range' in s.headers:
s.send_response(206)
st, e = s.headers['range'][6:].split('-', 1)
if 'Range' in self.headers:
self.send_response(206)
st, e = self.headers['range'][6:].split('-', 1)
sl = len(st)
el = len(e)
if sl > 0:
@ -159,54 +159,54 @@ class http_handler(BaseHTTPRequestHandler):
ei = int(e)
if ei < size:
start_range = size - ei
s.send_header('Content-Range', 'bytes ' + str(start_range)
self.send_header('Content-Range', 'bytes ' + str(start_range)
+ '-' + str(end_range - 1) + '/' + str(size))
else:
s.send_response(200)
s.send_header('Accept-Ranges', 'bytes')
self.send_response(200)
self.send_header('Accept-Ranges', 'bytes')
if chunked_encoding:
s.send_header('Transfer-Encoding', 'chunked')
s.send_header('Content-Length', end_range - start_range)
self.send_header('Transfer-Encoding', 'chunked')
self.send_header('Content-Length', end_range - start_range)
if filename.endswith('.gz'):
s.send_header('Content-Encoding', 'gzip')
self.send_header('Content-Encoding', 'gzip')
if not keepalive:
s.send_header("Connection", "close")
self.send_header("Connection", "close")
if not use_ssl:
s.request.shutdown(socket.SHUT_RD)
self.request.shutdown(socket.SHUT_RD)
s.end_headers()
self.end_headers()
f.seek(start_range)
length = end_range - start_range
while length > 0:
to_send = min(length, 0x900)
if chunked_encoding:
s.wfile.write(b'%x\r\n' % to_send)
self.wfile.write(b'%x\r\n' % to_send)
data = f.read(to_send)
print('read %d bytes' % to_send)
sys.stdout.flush()
s.wfile.write(data)
self.wfile.write(data)
if chunked_encoding:
s.wfile.write(b'\r\n')
self.wfile.write(b'\r\n')
length -= to_send
print('sent %d bytes (%d bytes left)' % (len(data), length))
sys.stdout.flush()
if chunked_encoding:
s.wfile.write(b'0\r\n\r\n')
self.wfile.write(b'0\r\n\r\n')
except Exception as e:
print('FILE ERROR: ', filename, e)
traceback.print_exc(file=sys.stdout)
sys.stdout.flush()
s.send_response(404)
s.send_header("Content-Length", "0")
self.send_response(404)
self.send_header("Content-Length", "0")
try:
s.end_headers()
self.end_headers()
except Exception:
pass
print("...DONE")
sys.stdout.flush()
s.wfile.flush()
self.wfile.flush()
if __name__ == '__main__':