Httpx support

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

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,21 +22,21 @@ 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)]
 
     if not verify:
         parts += [('--insecure', None)]
 
-    parts += [(None, request.url)]
+    parts += [(None, str(request.url))]
 
     flat_parts = []
     for k, v in parts:

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!@#$%^&*()-_=+;:\'"\'"\'",[]{}<.>/?~𝘈Ḇ𝖢𝕯٤ḞԍНǏ𝙅ƘԸⲘ𝙉০Ρ𝗤Ɍ𝓢ȚЦ𝒱Ѡ𝓧ƳȤѧᖯć𝗱ễ𝑓𝙜Ⴙ𝞲𝑗𝒌ļṃʼ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'}
+        )

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',