kev
/
Python-StreamedRequests
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

project released

This commit is contained in:
Kevin Froman 2019-07-08 03:01:12 -05:00
parent ddee7c4770
commit a71ac1d046
No known key found for this signature in database
GPG Key ID: 0D414D0FE405B63B
3 changed files with 43 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
Readme.md
View File

@ -1 +1,9 @@
# StreamedRequests
[![Build Status](https://travis-ci.org/beardog108/StreamedRequests.svg?branch=master)](https://travis-ci.org/beardog108/StreamedRequests)
Python module to stream HTTP requests in order to ensure content length sanity.
# Contact
https://chaoswebs.net/

6
setup.py
View File

@ -7,4 +7,10 @@ setup(name='streamedrequests',
author_email='beardog@mailbox.org',
url='https://github.com/beardog108/streamedrequests',
packages=['streamedrequests'],
install_requires=['requests'],
classifiers=[
"Programming Language :: Python :: 3",
"License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
"Operating System :: OS Independent",
],
)

36
tests/test.py
View File

@ -24,11 +24,19 @@ import streamedrequests
test_data_1 = 'test '*1000 + '\ntwo\n'
test_data = test_data_1 + 'test2'*1000
POST_PORT = 1337
class S(BaseHTTPRequestHandler):
def POST(self):
# Doesn't do anything with posted data
#https://gist.github.com/bradmontgomery/2219997
def _set_headers(self):
self.send_response(200)
self.send_header('Content-type', 'text/plain')
self.end_headers()
def do_POST(self):
self._set_headers()
self.wfile.write("<html><body><h1>POST!</h1></body></html>")
resp = "POST %s" % (test_data,)
resp = resp.encode()
self.wfile.write(resp)
def get_test_id():
return str(uuid.uuid4()) + '.dat'
@ -44,23 +52,34 @@ def run(server_class=HTTPServer, handler_class=SimpleHTTPRequestHandler):
httpd = server_class(server_address, handler_class)
httpd.serve_forever()
def run_post(server_class=S, handler_class=BaseHTTPRequestHandler):
server_address = ('127.0.0.1', 8001)
def run_post(server_class=HTTPServer, handler_class=S):
server_address = ('127.0.0.1', POST_PORT)
httpd = server_class(server_address, handler_class)
httpd.serve_forever()
def _test_callback(text):
print('got', text)
pass#print('got', text)
class TestInit(unittest.TestCase):
def test_requests(self):
if "test" not in requests.get('http://127.0.0.1:8000/').text:
raise ValueError("test not found in test data")
if "test" not in requests.post('http://127.0.0.1:%s' % (POST_PORT,)).text:
print(requests.post('http://127.0.0.1:%s' % (POST_PORT,)).text)
raise ValueError("test not found in post test data")
def test_basic(self):
streamedrequests.get('http://127.0.0.1:8000/')
def test_fail(self):
with self.assertRaises(requests.exceptions.ConnectionError):
streamedrequests.get("http://127.0.1.1:1234")
with self.assertRaises(requests.exceptions.ConnectionError):
streamedrequests.post("http://127.0.1.1:1235")
with self.assertRaises(requests.exceptions.ConnectTimeout):
streamedrequests.get('https://1.1.1.1/', connect_timeout=0.0001)
def test_callback(self):
streamedrequests.get('http://127.0.0.1:8000/', chunk_size=1, callback=_test_callback)
@ -70,9 +89,12 @@ class TestInit(unittest.TestCase):
def test_zero_chunk_size(self):
with self.assertRaises(ValueError):
streamedrequests.get('http://127.0.0.1:8000/', chunk_size=0)
with self.assertRaises(ValueError):
streamedrequests.post('http://127.0.0.1:%s/' % (POST_PORT,), post_data='test', chunk_size=0)
def test_post(self):
streamedrequests.post('http://127.0.0.1:8000/')
streamedrequests.post('http://127.0.0.1:%s/' % (POST_PORT,))
streamedrequests.post('http://127.0.0.1:%s/' % (POST_PORT,), post_data='test')
setup()
threading.Thread(target=run, daemon=True).start()