diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..524eb31 --- /dev/null +++ b/Makefile @@ -0,0 +1,18 @@ +clean-test: ## remove test and coverage artifacts + rm -fr .tox/ + rm -f .coverage + rm -fr htmlcov/ + +lint: ## check style + flake8 curlify.py curlify_test.py + +test: + py.test --cov=curlify --cov-report term-missing --cov-fail-under=100 --cov-branch + +test-all: lint test + +install: + pip install . --upgrade + +install-dev: + pip install -e '.[testing]' --upgrade diff --git a/curlify.py b/curlify.py index 6b5d902..3e7c2f9 100644 --- a/curlify.py +++ b/curlify.py @@ -3,7 +3,7 @@ if sys.version_info.major >= 3: from shlex import quote -else: +else: # pragma: no cover, 2.7 flavor is not tested from pipes import quote @@ -22,13 +22,13 @@ def to_curl(request, compressed=False, verify=True): ] for k, v in sorted(request.headers.items()): - parts += [('-H', '{0}: {1}'.format(k, v))] + parts += [('-H', '{0}: {1}'.format(k.lower(), v))] - if request.body: - body = request.body - if isinstance(body, bytes): - body = body.decode('utf-8') - parts += [('-d', body)] + body_content = request.body if hasattr(request, "body") else request.read() + if body_content: + if isinstance(body_content, bytes): + body_content = body_content.decode('utf-8') + parts += [('-d', body_content)] if compressed: parts += [('--compressed', None)] @@ -36,7 +36,7 @@ def to_curl(request, compressed=False, verify=True): if not verify: parts += [('--insecure', None)] - parts += [(None, request.url)] + parts += [(None, str(request.url))] flat_parts = [] for k, v in parts: diff --git a/curlify_test.py b/curlify_test.py index 16e71b9..e6ee29c 100644 --- a/curlify_test.py +++ b/curlify_test.py @@ -1,113 +1,205 @@ # coding: utf-8 -import curlify -import re +import unittest + +import httpx import requests +import responses +import respx + +import curlify -def test_empty_data(): - r = requests.post( - "http://google.ru", - headers={"user-agent": "mytest"}, - ) - assert curlify.to_curl(r.request) == ( - "curl -X POST " - "-H 'Accept: */*' " - "-H 'Accept-Encoding: gzip, deflate' " - "-H 'Connection: keep-alive' " - "-H 'Content-Length: 0' " - "-H 'user-agent: mytest' " - "http://google.ru/" - ) - - -def test_ok(): - r = requests.get( - "http://google.ru", - data={"a": "b"}, - cookies={"foo": "bar"}, - headers={"user-agent": "mytest"}, - ) - assert curlify.to_curl(r.request) == ( - "curl -X GET " - "-H 'Accept: */*' " - "-H 'Accept-Encoding: gzip, deflate' " - "-H 'Connection: keep-alive' " - "-H 'Content-Length: 3' " - "-H 'Content-Type: application/x-www-form-urlencoded' " - "-H 'Cookie: foo=bar' " - "-H 'user-agent: mytest' " - "-d a=b http://google.ru/" - ) - - -def test_prepare_request(): - request = requests.Request( - 'GET', "http://google.ru", - headers={"user-agent": "UA"}, - ) - - assert curlify.to_curl(request.prepare()) == ( - "curl -X GET " - "-H 'user-agent: UA' " - "http://google.ru/" - ) - - -def test_compressed(): - request = requests.Request( - 'GET', "http://google.ru", - headers={"user-agent": "UA"}, - ) - assert curlify.to_curl(request.prepare(), compressed=True) == ( - "curl -X GET -H 'user-agent: UA' --compressed http://google.ru/" - ) - - -def test_verify(): - request = requests.Request( - 'GET', "http://google.ru", - headers={"user-agent": "UA"}, - ) - assert curlify.to_curl(request.prepare(), verify=False) == ( - "curl -X GET -H 'user-agent: UA' --insecure http://google.ru/" - ) - - -def test_post_json(): - data = {'foo': 'bar'} - url = 'https://httpbin.org/post' - - r = requests.Request('POST', url, json=data) - curlified = curlify.to_curl(r.prepare()) - - assert curlified == ( - "curl -X POST -H 'Content-Length: 14' " - "-H 'Content-Type: application/json' " - "-d '{\"foo\": \"bar\"}' https://httpbin.org/post" - ) - - -def test_post_csv_file(): - r = requests.Request( - method='POST', - url='https://httpbin.org/post', - files={'file': open('data.csv', 'r')}, - headers={'User-agent': 'UA'} - ) - - curlified = curlify.to_curl(r.prepare()) - boundary = re.search(r'boundary=(\w+)', curlified).group(1) - - expected = ( - 'curl -X POST -H \'Content-Length: 519\'' - f' -H \'Content-Type: multipart/form-data; boundary={boundary}\'' - ' -H \'User-agent: UA\'' - f' -d \'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="data.csv"\r\n\r\n' - '"Id";"Title";"Content"\n' - '1;"Simple Test";"Ici un test d\'"\'"\'รฉchappement de simple quote"\n' - '2;"UTF-8 Test";"ฤƒัฃ๐” ีฎแปลฟฤฃศŸแŽฅ๐’‹วฉฤพแธฟ๊ž‘ศฏ๐˜ฑ๐‘ž๐—‹๐˜ดศถ๐ž„๐œˆฯˆ๐’™๐˜†๐šฃ1234567890!@#$%^&*()-_=+;:\'"\'"\'",[]{}<.>/?~๐˜ˆแธ†๐–ข๐•ฏูคแธžิะว๐™…ฦ˜ิธโฒ˜๐™‰เงฆฮก๐—คษŒ๐“ขศšะฆ๐’ฑั ๐“งฦณศคังแ–ฏฤ‡๐—ฑแป…๐‘“๐™œแ‚น๐žฒ๐‘—๐’Œฤผแนƒล‰ะพ๐žŽ๐’’แตฒ๊œฑ๐™ฉแปซ๐—ลต๐’™๐’šลบ"' - f'\r\n--{boundary}--\r\n\'' - ' https://httpbin.org/post' - ) - - assert curlified == expected +class TestCurlify(unittest.TestCase): + def mock_add_get(self): + self.request_reponses.add( + responses.GET, + 'https://example.com/', + body='fake_example.com', + status=200 + ) + self.httpx_respx.get( + "/", + content="fake_example.com", + status_code=200 + ) + + def mock_add_post(self): + self.request_reponses.add( + responses.POST, + 'https://example.com/', + body='fake_example.com', + status=200 + ) + self.httpx_respx.post( + "/", + content="fake_example.com", + status_code=201 + ) + + def setUp(self): + self.request_reponses = responses.RequestsMock() + self.request_reponses.start() + + self.httpx_respx = respx.mock(base_url="https://example.com") + self.httpx_respx.start() + + def tearDown(self): + self.request_reponses.stop() + self.httpx_respx.stop() + + def test_mocks(self): + self.mock_add_get() + self.mock_add_post() + + r = requests.get("https://example.com/") + assert r.text == "fake_example.com" + r = httpx.get("https://example.com/") + assert r.text == "fake_example.com" + r = requests.post("https://example.com/") + assert r.text == "fake_example.com" + r = httpx.post("https://example.com/") + assert r.text == "fake_example.com" + + def assert_approx_curl(self, curl_equivalent, curlify_string): + """ + Check that all elements of curl_equivalent are in result of to_curl + The approximation allows for to_curl to include other flags and + shuffle order. + + equivalency is not done as + * requests does not set host explicitly and relies http lib to do + it on its behalf + * httpx does not set on empty bodycontent-length + """ + for arg in curl_equivalent: + assert arg in curlify_string + + def assert_all(self, curl_equivalent, method, *args, **kwargs): + for module in requests, httpx: + response = getattr(module, method)(*args, **kwargs) + self.assert_approx_curl( + curl_equivalent, + curlify.to_curl(response.request) + ) + + def test_post_empty_data(self): + self.mock_add_post() + self.assert_all( + [ + "curl -X POST ", + "-H 'accept: */*' ", + "-H 'accept-encoding: gzip, deflate' ", + "-H 'connection: keep-alive' ", + "-H 'user-agent: mytest' ", + "https://example.com/", + ], + "post", + "https://example.com/", + headers={ + "user-agent": "mytest", + }, + ) + + def test_post(self): + self.mock_add_post() + self.assert_all( + [ + "curl -X POST ", + "-H 'accept: */*' ", + "-H 'accept-encoding: gzip, deflate' ", + "-H 'connection: keep-alive' ", + "-H 'content-length: 3' ", + "-H 'content-type: application/x-www-form-urlencoded' ", + "-H 'cookie: foo=bar' ", + "-H 'user-agent: mytest' ", + "-d a=b ", + "https://example.com/", + ], + "post", + "https://example.com/", + data={"a": "b"}, + cookies={"foo": "bar"}, + headers={"user-agent": "mytest"}, + ) + + def test_prepare_request(self): + request = requests.Request( + 'GET', "https://example.com/", + headers={"user-agent": "UA"}, + ) + assert curlify.to_curl(request.prepare()) == ( + "curl -X GET " + "-H 'user-agent: UA' " + "https://example.com/" + ) + + def test_httpx_request(self): + request = httpx.Request( + 'GET', "https://example.com", + headers={"user-agent": "UA"}, + ) + curl_equivalent = curlify.to_curl(request) + for substring in [ + "curl -X GET ", + "-H 'user-agent: UA' ", + "https://example.com", + ]: + assert substring in curl_equivalent + + def test_compressed_request(self): + request = requests.Request( + 'GET', "https://example.com/", + headers={"user-agent": "UA"}, + ) + assert curlify.to_curl(request.prepare(), compressed=True) == ( + "curl -X GET -H 'user-agent: UA' --compressed https://example.com/" + ) + + def test_verify(self): + request = requests.Request( + 'GET', "https://example.com/", + headers={"user-agent": "UA"}, + ) + assert curlify.to_curl(request.prepare(), verify=False) == ( + "curl -X GET -H 'user-agent: UA' --insecure https://example.com/" + ) + + def test_post_json(self): + self.mock_add_post() + self.assert_all( + [ + "curl -X POST ", + "-H 'content-length: 14' ", + "-H 'content-type: application/json' ", + "-d '{\"foo\": \"bar\"}' ", + "https://example.com/", + ], + "post", + 'https://example.com/', + json={'foo': 'bar'}, + ) + + def test_post_csv_file(self): + self.mock_add_post() + with open('data.csv', 'r') as fd: + content = fd.read() + self.assert_all( + [ + 'curl -X POST ', + '-H \'content-length: 543\' ', + '-H \'content-type: multipart/form-data; boundary=', + '-H \'user-agent: UA\'', + '-d \'--', + 'Content-Disposition: form-data; name="file"; filename="da' + 'ta.csv"\r\nContent-Type: text/csv\r\n\r\n"Id";"Title";"Co' + 'ntent"\n1;"Simple Test";"Ici un test d\'"\'"\'รฉchappement' + ' de simple quote"\n2;"UTF-8 Test";"ฤƒัฃ๐” ีฎแปลฟฤฃศŸแŽฅ๐’‹วฉฤพแธฟ๊ž‘ศฏ๐˜ฑ๐‘ž๐—‹๐˜ดศถ๐ž„๐œˆ' + 'ฯˆ๐’™๐˜†๐šฃ1234567890!@#$%^&*()-_=+;:\'"\'"\'",[]{}<.>/?~๐˜ˆแธ†๐–ข๐•ฏูคแธžิ' + 'ะว๐™…ฦ˜ิธโฒ˜๐™‰เงฆฮก๐—คษŒ๐“ขศšะฆ๐’ฑั ๐“งฦณศคังแ–ฏฤ‡๐—ฑแป…๐‘“๐™œแ‚น๐žฒ๐‘—๐’Œฤผแนƒล‰ะพ๐žŽ๐’’แตฒ๊œฑ๐™ฉแปซ๐—ลต๐’™๐’šลบ"\r\n', + 'https://example.com' + ], + 'post', + 'https://example.com/', + files={'file': ('data.csv', content, 'text/csv')}, + headers={'User-agent': 'UA'} + ) diff --git a/setup.py b/setup.py index 16da08f..b71552b 100644 --- a/setup.py +++ b/setup.py @@ -30,10 +30,21 @@ def run(self): ], include_package_data=True, install_requires=[ - 'requests', ], + extras_require={ + 'testing': [ + 'flake8', + 'httpx', + 'pyflakes', + 'pytest', + 'pytest-cov', + 'requests', + 'responses', + 'respx', + ], + }, license='MIT License', - description='Library to convert python requests object to curl command.', + description='Library to convert python requests / httpx object to curl command.', author='Egor Orlov', author_email='oeegor@gmail.com', platforms='any',