Update tests

src/httpx/_network.py

@@ -68,16 +68,23 @@ def write(self, buffer: bytes) -> None:
 
     def start_tls(self, ctx: ssl.SSLContext, hostname: str | None = None) -> None:
         self._socket = ctx.wrap_socket(self._socket, server_hostname=hostname)
+        self._socket.do_handshake()
         self._is_tls = True
 
     def close(self) -> None:
-        timeout = get_current_timeout()
-        self._socket.settimeout(timeout)
-        self._socket.close()
-        self._is_closed = True
+        if not self._is_closed:
+            timeout = get_current_timeout()
+            self._socket.settimeout(timeout)
+            if self._is_tls:
+                self._socket.shutdown(socket.SHUT_RDWR)
+            self._socket.close()
+            self._is_tls = False
+            self._is_closed = True
 
     def __repr__(self):
-        description = " CLOSED" if self._is_closed else ""
+        description = (
+            " CLOSED" if self._is_closed else " TLS" if self._is_tls else ""
+        )
         return f"<NetworkStream [{self._address!r}{description}]>"
 
     # Context managed usage...

src/httpx/_urls.py

@@ -216,7 +216,7 @@ def path(self) -> str:
         return unquote(path)
 
     @property
-    def query(self) -> bytes:
+    def query(self) -> str:
         """
         The URL query string, as raw bytes, excluding the leading b"?".
 
@@ -229,8 +229,7 @@ def query(self) -> bytes:
         url = httpx.URL("https://example.com/?filter=some%20search%20terms")
         assert url.query == b"filter=some%20search%20terms"
         """
-        query = self._uri_reference.query or ""
-        return query.encode("ascii")
+        return self._uri_reference.query or ""
 
     @property
     def params(self) -> QueryParams:

tests/test_01_clients.py

@@ -49,3 +49,11 @@ def test_delete(httpbin, cli):
     r = cli.delete(url)
     assert r.code == 200
     assert r.json()["args"] == {}
+
+
+def test_stream(httpbin, cli):
+    url = f'{httpbin.url}/get'
+    with cli.stream("GET", url) as r:
+        r.read()
+        assert r.code == 200
+        assert r.json()["args"] == {}

tests/test_02_components.py

@@ -4,31 +4,70 @@
 def test_url():
     url = httpx.URL('https://www.example.com/')
     assert repr(url) == "<URL 'https://www.example.com/'>"
+    assert str(url) == "https://www.example.com/"
 
-# >>> url = httpx.URL('https://www.EXAMPLE.com:443/path/../main')
-# >>> url
-# <URL 'https://www.example.com/main'>
 
-# >>> url = httpx.URL('/README.md')
-# >>> url
-# <URL '/README.md'>
+def test_url_components():
+    url = httpx.URL("HTTPS://jo%40email.com:a%20secret@müller.de:1234/pa%20th?search=ab#anchorlink")
 
-# >>> httpx.URL('https://example.com/path to here?search=🦋')
-# <URL 'https://example.com/path%20to%20here?search=%F0%9F%A6%8B'>
+    assert url.scheme == "https"
+    assert url.username == "jo@email.com"
+    assert url.password == "a secret"
+    assert url.userinfo == "jo%40email.com:a%20secret"
+    assert url.host == "xn--mller-kva.de"
+    assert url.port == 1234
+    assert url.netloc == "xn--mller-kva.de:1234"
+    assert url.path == "/pa th"
+    assert url.query == "?search=ab"
+    assert url.fragment == "anchorlink"
 
-# >>> httpx.URL(scheme="https", host="example.com", path="/")
-# <URL 'https://example.com/'>
 
-# >>> httpx.URL("https://example.com/", params={"search": "some text"})
-# <URL 'https://example.com/?search=some+text'>
+def test_url_normalisation():
+    url = httpx.URL('https://www.EXAMPLE.com:443/path/../main')
+    assert repr(url) == "<URL 'https://www.example.com/main'>"
+    assert str(url) == "https://www.example.com/main"
 
-# >>> params = httpx.QueryParams({"color": "black", "size": "medium"})
-# >>> params
-# <QueryParams 'color=black&size=medium'>
 
-# >>> params = httpx.QueryParams({"filter": ["60GHz", "75GHz", "100GHz"]})
-# >>> params
-# <QueryParams 'filter=60GHz&filter=75GHz&filter=100GHz'>
+def test_relative_url():
+    url = httpx.URL('/README.md')
+    assert repr(url) == "<URL '/README.md'>"
+    assert str(url) == "/README.md"
+
+
+def test_url_escape():
+    url = httpx.URL('https://example.com/path to here?search=🦋')
+    assert repr(url) == "<URL 'https://example.com/path%20to%20here?search=%F0%9F%A6%8B'>"
+    assert str(url) == "https://example.com/path%20to%20here?search=%F0%9F%A6%8B"
+
+
+def test_url_components():
+    url = httpx.URL(scheme="https", host="example.com", path="/")
+    assert repr(url) == "<URL 'https://example.com/'>"
+    assert str(url) == "https://example.com/"
+
+
+def test_url_params():
+    url = httpx.URL("https://example.com/", params={"search": "some text"})
+    assert repr(url) == "<URL 'https://example.com/?search=some+text'>"
+    assert str(url) == "https://example.com/?search=some+text"
+
+
+def test_queryparams():
+    params = httpx.QueryParams({"color": "black", "size": "medium"})
+    assert repr(params) == "<QueryParams 'color=black&size=medium'>"
+    assert str(params) == "color=black&size=medium"
+
+
+def test_queryparams_multi():
+    params = httpx.QueryParams({"filter": ["60GHz", "75GHz", "100GHz"]})
+    assert repr(params) == "<QueryParams 'filter=60GHz&filter=75GHz&filter=100GHz'>"
+    assert str(params) == "filter=60GHz&filter=75GHz&filter=100GHz"
+
+
+def test_queryparams_keys():
+    params = httpx.QueryParams("a=123&a=456&b=789")
+    assert list(params.keys()) == ["a", "b"]
+
 
 # >>> params = httpx.QueryParams("color=black&size=medium")
 # >>> params
@@ -47,4 +86,3 @@ def test_url():
 # <Headers {"Accept": "*/*"}>
 # >>> headers['accept']
 # '*/*'
-

tests/test_05_networking.py

@@ -1,4 +1,10 @@
 import httpx
+import os
+import pytest_httpbin
+import ssl
+
+
+CERTS_DIR = os.path.join(os.path.dirname(pytest_httpbin.__file__), "certs")
 
 
 def test_network_backend():
@@ -15,10 +21,26 @@ def test_network_backend_connect(httpbin):
 def test_network_backend_context_managed(httpbin):
     net = httpx.NetworkBackend()
     with net.connect(httpbin.host, httpbin.port) as stream:
-        ...
+        assert repr(stream) == f"<NetworkStream ['{httpbin.host}:{httpbin.port}']>"
+        stream.write(b"GET / HTTP/1.1\r\nConnetion: close\r\n\r\n")
+        resp = stream.read()
+        assert isinstance(resp, bytes)
     assert repr(stream) == f"<NetworkStream ['{httpbin.host}:{httpbin.port}' CLOSED]>"
 
 
+def test_network_backend_context_managed_ssl(httpbin_secure):
+    net = httpx.NetworkBackend()
+    with net.connect(httpbin_secure.host, httpbin_secure.port) as stream:
+        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        context.load_verify_locations(pytest_httpbin.certs.where())
+        stream.start_tls(context, hostname=httpbin_secure.host)
+        assert repr(stream) == f"<NetworkStream ['{httpbin_secure.host}:{httpbin_secure.port}' TLS]>"
+        stream.write(b"GET / HTTP/1.1\r\nConnetion: close\r\n\r\n")
+        resp = stream.read()
+        assert isinstance(resp, bytes)
+    assert repr(stream) == f"<NetworkStream ['{httpbin_secure.host}:{httpbin_secure.port}' CLOSED]>"
+
+
 # >>> net = httpx.NetworkBackend()
 # >>> stream = net.connect("dev.encode.io", 80)
 # >>> try:
@@ -56,4 +78,3 @@ def test_network_backend_context_managed(httpbin):
 #             while part := stream.read():
 #                 buffer.append(part)
 #             resp = b''.join(buffer)
-

tests/test_url.py

@@ -0,0 +1,91 @@
+# The WHATWG have various tests that can be used to validate the URL parsing.
+#
+# https://url.spec.whatwg.org/
+
+import json
+import os
+import pytest
+import httpx
+
+
+suite_path = os.path.join("tests", "urltestdata.json")
+
+# URL test cases from...
+# https://github.com/web-platform-tests/wpt/blob/master/url/resources/urltestdata.json
+with open(suite_path, "r", encoding="utf-8") as input:
+    test_cases = json.load(input)
+    test_cases = [
+        item
+        for item in test_cases
+        if not isinstance(item, str) and not item.get("failure")
+        # and not "\t" in item["input"] and not "\n" in item["input"]
+    ]
+
+
+@pytest.mark.parametrize("test_case", test_cases)
+def test_urlparse(test_case):
+    # base = httpx.URL(test_case["base"] or "")
+    # input = httpx.URL(test_case["input"])
+
+    href = httpx.URL(test_case["href"])
+
+    # The following properties are included in the test cases...
+    #
+    # "origin": "http://example.org"
+    # "protocol": "http:"
+    # "username": ""
+    # "password": ""
+    # "host": "example.org"
+    # "hostname": "example.org"
+    # "port": ""
+    # "pathname": "/foo/bar"
+    # "search": ""
+    # "hash": "#%CE%B2"
+
+    # origin
+    protocol = f"{href.scheme}:"
+    # username
+    # password
+    # host
+    # hostname
+    port = "" if href.port is None else str(href.port)
+    # pathname = href.path
+    search = f"?{href.query}" if href.query else ""
+    # hash = f"#{href.fragment}" if href.fragment else ""
+
+    assert protocol == test_case["protocol"]
+    # assert href.username == test_case["username"]
+    # assert href.password == test_case["password"]
+    # assert href.host == test_case["host"]
+    # assert href.hostname == test_case["hostname"]
+    assert port == test_case["port"]
+    # assert pathname == test_case["pathname"]
+    assert search == test_case["search"]
+    # assert hash == test_case["hash"]
+
+
+    # -----------
+
+    # p = urlparse(test_case["href"])
+
+    # # Test cases include the protocol with the trailing ":"
+    # protocol = p.scheme + ":"
+    # # Include the square brackets for IPv6 addresses.
+    # hostname = f"[{p.host}]" if ":" in p.host else p.host
+    # # The test cases use a string representation of the port.
+    # port = "" if p.port is None else str(p.port)
+    # # I have nothing to say about this one.
+    # path = p.path
+    # # The 'search' and 'hash' components in the whatwg tests are semantic, not literal.
+    # # Our parsing differentiates between no query/hash and empty-string query/hash.
+    # search = "" if p.query in (None, "") else "?" + str(p.query)
+    # hash = "" if p.fragment in (None, "") else "#" + str(p.fragment)
+
+    # # URL hostnames are case-insensitive.
+    # # We normalize these, unlike the WHATWG test cases.
+    # assert protocol == test_case["protocol"]
+    # assert hostname.lower() == test_case["hostname"].lower()
+    # assert port == test_case["port"]
+    # assert path == test_case["pathname"]
+    # assert search == test_case["search"]
+    # assert hash == test_case["hash"]