-
Notifications
You must be signed in to change notification settings - Fork 31
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
As httpx strives for close compatibility with requests few changes needed to be made in curlify: * .lower() on headers (headers in http are case insensitive and httpx / requests use different cases. * httpx does not have a .body but contents can be accessed via .read() instead, curlify will now use either or. * httpx stores url in as URL object so we stringify it. Majority of the changes are in tests. * all tests are mocked out so there is no hitting of external services * payload tests are more relaxed. Testing for approximate existence of flags rather than exact string match. This is a workaround to deal with various discordant optional http headers.
- Loading branch information
1 parent
587be9c
commit 1c74dfa
Showing
4 changed files
with
221 additions
and
130 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,116 +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!@#$%^&*()-_=+;:' | ||
| '\'"\'"\'",[]{}<.>/?~𝘈Ḇ𝖢𝕯٤ḞԍНǏ𝙅ƘԸⲘ𝙉০Ρ𝗤Ɍ𝓢ȚЦ𝒱Ѡ𝓧ƳȤѧᖯć𝗱ễ𝑓𝙜Ⴙ𝞲𝑗𝒌ļṃʼnо𝞎𝒒ᵲꜱ𝙩ừ' | ||
| '𝗏ŵ𝒙𝒚ź"' | ||
| 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!@#$%^&*()-_=+;:\'"\'"\'",[]{}<.>/?~𝘈Ḇ𝖢𝕯٤Ḟԍ' | ||
| 'НǏ𝙅ƘԸⲘ𝙉০Ρ𝗤Ɍ𝓢ȚЦ𝒱Ѡ𝓧ƳȤѧᖯć𝗱ễ𝑓𝙜Ⴙ𝞲𝑗𝒌ļṃʼnо𝞎𝒒ᵲꜱ𝙩ừ𝗏ŵ𝒙𝒚ź"\r\n', | ||
| 'https://example.com' | ||
| ], | ||
| 'post', | ||
| 'https://example.com/', | ||
| files={'file': ('data.csv', content, 'text/csv')}, | ||
| headers={'User-agent': 'UA'} | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters