Remove legacy base methods and add tests

Author: justmobilize

.pre-commit-config.yaml

@@ -44,4 +44,4 @@ repos:
  types: [python]
  files: "^tests/"
  args:
- - --disable=consider-using-f-string,duplicate-code,missing-docstring,invalid-name,protected-access
+ - --disable=consider-using-f-string,duplicate-code,missing-docstring,invalid-name,protected-access,redefined-outer-name

adafruit_requests.py

@@ -86,7 +86,7 @@ class Response:
 
  encoding = None
 
- def __init__(self, sock: SocketType, session: Optional["Session"] = None) -> None:
+ def __init__(self, sock: SocketType, session: "Session") -> None:
  self.socket = sock
  self.encoding = "utf-8"
  self._cached = None
@@ -101,10 +101,7 @@ def __init__(self, sock: SocketType, session: Optional["Session"] = None) -> Non
 
  http = self._readto(b" ")
  if not http:
- if session:
- session._connection_manager.close_socket(self.socket)
- else:
- self.socket.close()
+ session._connection_manager.close_socket(self.socket)
  raise RuntimeError("Unable to read HTTP response.")
  self.status_code: int = int(bytes(self._readto(b" ")))
  """The status code returned by the server"""
@@ -614,67 +611,3 @@ def patch(self, url: str, **kw) -> Response:
  def delete(self, url: str, **kw) -> Response:
  """Send HTTP DELETE request"""
  return self.request("DELETE", url, **kw)
-
-
-_default_session = None # pylint: disable=invalid-name
-
-
-def create_default_session(
- socket_pool: SocketpoolModuleType,
- ssl_context: Optional[SSLContextType] = None,
-):
- """Create a default session for using globally"""
- global _default_session # pylint: disable=global-statement
- _default_session = Session(socket_pool, ssl_context)
-
-
-def request(
- method: str,
- url: str,
- data: Optional[Any] = None,
- json: Optional[Any] = None,
- headers: Optional[Dict[str, str]] = None,
- stream: bool = False,
- timeout: float = 1,
-) -> None:
- """Send HTTP request"""
- # pylint: disable=too-many-arguments
- _default_session.request(
- method,
- url,
- data=data,
- json=json,
- headers=headers,
- stream=stream,
- timeout=timeout,
- )
-
-
-def head(url: str, **kw):
- """Send HTTP HEAD request"""
- return _default_session.request("HEAD", url, **kw)
-
-
-def get(url: str, **kw):
- """Send HTTP GET request"""
- return _default_session.request("GET", url, **kw)
-
-
-def post(url: str, **kw):
- """Send HTTP POST request"""
- return _default_session.request("POST", url, **kw)
-
-
-def put(url: str, **kw):
- """Send HTTP PUT request"""
- return _default_session.request("PUT", url, **kw)
-
-
-def patch(url: str, **kw):
- """Send HTTP PATCH request"""
- return _default_session.request("PATCH", url, **kw)
-
-
-def delete(url: str, **kw):
- """Send HTTP DELETE request"""
- return _default_session.request("DELETE", url, **kw)

conftest.py

@@ -13,9 +13,25 @@
 
 IP = "1.2.3.4"
 HOST = "docs.google.com"
-PATH = (
+PATH_BASE = (
  "/spreadsheets/d/e/2PACX-1vR1WjUKz35-ek6SiR5droDfvPp51MTds4wUs57vEZNh2uDfihSTPhTaiiRo"
- "vLbNe1mkeRgurppRJ_Zy/pub?output=tsv"
+ "vLbNe1mkeRgurppRJ_Zy/"
+)
+PATH = PATH_BASE + "pub?output=tsv"
+
+FILE_REDIRECT = (
+ b"e@2PACX-1vR1WjUKz35-ek6SiR5droDfvPp51MTds4wUs57vEZNh2uDfihSTPhTai"
+ b"iRovLbNe1mkeRgurppRJ_Zy?output=tsv"
+)
+RELATIVE_RELATIVE_REDIRECT = (
+ b"370cmver1f290kjsnpar5ku2h9g/3llvt5u8njbvat22m9l19db1h4/1656191325000/109226138307867586192/*/"
+ + FILE_REDIRECT
+)
+RELATIVE_ABSOLUTE_REDIRECT = (
+ b"/pub/70cmver1f290kjsnpar5ku2h9g/" + RELATIVE_RELATIVE_REDIRECT
+)
+ABSOLUTE_ABSOLUTE_REDIRECT = (
+ b"https://doc-14-2g-sheets.googleusercontent.com" + RELATIVE_ABSOLUTE_REDIRECT
 )
 
 # response headers returned from the initial request
@@ -26,10 +42,7 @@
  b"Pragma: no-cache\r\n"
  b"Expires: Mon, 01 Jan 1990 00:00:00 GMT\r\n"
  b"Date: Sat, 25 Jun 2022 21:08:48 GMT\r\n"
- b"Location: https://doc-14-2g-sheets.googleusercontent.com/pub/70cmver1f290kjsnpar5ku2h9g/3"
- b"llvt5u8njbvat22m9l19db1h4/1656191325000"
- b"/109226138307867586192/*/e@2PACX-1vR1WjUKz35-ek6SiR5droDfvPp51MTds4wUs57vEZNh2uDfihSTPhTai"
- b"iRovLbNe1mkeRgurppRJ_Zy?output=tsv\r\n"
+ b"Location: NEW_LOCATION_HERE\r\n"
  b'P3P: CP="This is not a P3P policy! See g.co/p3phelp for more info."\r\n'
  b"X-Content-Type-Options: nosniff\r\n"
  b"X-XSS-Protection: 1; mode=block\r\n"
@@ -118,13 +131,16 @@ def _recv_into(self, buf, nbytes=0):
  return read
 
 
-def do_test_chunked_redirect():
+def test_chunked_absolute_absolute_redirect():
  pool = mocket.MocketPool()
  pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 443)),)
  chunk = _chunk(BODY_REDIRECT, len(BODY_REDIRECT))
  chunk2 = _chunk(BODY, len(BODY))
 
- sock1 = MocketRecvInto(HEADERS_REDIRECT + chunk)
+ redirect = HEADERS_REDIRECT.replace(
+ b"NEW_LOCATION_HERE", ABSOLUTE_ABSOLUTE_REDIRECT
+ )
+ sock1 = MocketRecvInto(redirect + chunk)
  sock2 = mocket.Mocket(HEADERS + chunk2)
  pool.socket.side_effect = (sock1, sock2)
 
@@ -135,9 +151,109 @@ def do_test_chunked_redirect():
  sock2.connect.assert_called_once_with(
  ("doc-14-2g-sheets.googleusercontent.com", 443)
  )
+ sock2.send.assert_has_calls([mock.call(RELATIVE_ABSOLUTE_REDIRECT[1:])])
+
+ assert response.text == str(BODY, "utf-8")
+
+
+def test_chunked_relative_absolute_redirect():
+ pool = mocket.MocketPool()
+ pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 443)),)
+ chunk = _chunk(BODY_REDIRECT, len(BODY_REDIRECT))
+ chunk2 = _chunk(BODY, len(BODY))
+
+ redirect = HEADERS_REDIRECT.replace(
+ b"NEW_LOCATION_HERE", RELATIVE_ABSOLUTE_REDIRECT
+ )
+ sock1 = MocketRecvInto(redirect + chunk)
+ sock2 = mocket.Mocket(HEADERS + chunk2)
+ pool.socket.side_effect = (sock1, sock2)
+
+ requests_session = adafruit_requests.Session(pool, mocket.SSLContext())
+ response = requests_session.get("https://" + HOST + PATH)
+
+ sock1.connect.assert_called_once_with((HOST, 443))
+ sock2.connect.assert_called_once_with(("docs.google.com", 443))
+ sock2.send.assert_has_calls([mock.call(RELATIVE_ABSOLUTE_REDIRECT[1:])])
+
+ assert response.text == str(BODY, "utf-8")
+
+
+def test_chunked_relative_relative_redirect():
+ pool = mocket.MocketPool()
+ pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 443)),)
+ chunk = _chunk(BODY_REDIRECT, len(BODY_REDIRECT))
+ chunk2 = _chunk(BODY, len(BODY))
+
+ redirect = HEADERS_REDIRECT.replace(
+ b"NEW_LOCATION_HERE", RELATIVE_RELATIVE_REDIRECT
+ )
+ sock1 = MocketRecvInto(redirect + chunk)
+ sock2 = mocket.Mocket(HEADERS + chunk2)
+ pool.socket.side_effect = (sock1, sock2)
+
+ requests_session = adafruit_requests.Session(pool, mocket.SSLContext())
+ response = requests_session.get("https://" + HOST + PATH)
+
+ sock1.connect.assert_called_once_with((HOST, 443))
+ sock2.connect.assert_called_once_with(("docs.google.com", 443))
+ sock2.send.assert_has_calls(
+ [mock.call(bytes(PATH_BASE[1:], "utf-8") + RELATIVE_RELATIVE_REDIRECT)]
+ )
 
  assert response.text == str(BODY, "utf-8")
 
 
-def test_chunked_redirect():
- do_test_chunked_redirect()
+def test_chunked_relative_relative_redirect_backstep():
+ pool = mocket.MocketPool()
+ pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 443)),)
+ chunk = _chunk(BODY_REDIRECT, len(BODY_REDIRECT))
+ chunk2 = _chunk(BODY, len(BODY))
+
+ remove_paths = 2
+ backstep = b"../" * remove_paths
+ path_base_parts = PATH_BASE.split("/")
+ # PATH_BASE starts with "/" so skip it and also remove from the count
+ path_base = (
+ "/".join(path_base_parts[1 : len(path_base_parts) - remove_paths - 1]) + "/"
+ )
+
+ redirect = HEADERS_REDIRECT.replace(
+ b"NEW_LOCATION_HERE", backstep + RELATIVE_RELATIVE_REDIRECT
+ )
+ sock1 = MocketRecvInto(redirect + chunk)
+ sock2 = mocket.Mocket(HEADERS + chunk2)
+ pool.socket.side_effect = (sock1, sock2)
+
+ requests_session = adafruit_requests.Session(pool, mocket.SSLContext())
+ response = requests_session.get("https://" + HOST + PATH)
+
+ sock1.connect.assert_called_once_with((HOST, 443))
+ sock2.connect.assert_called_once_with(("docs.google.com", 443))
+ sock2.send.assert_has_calls(
+ [mock.call(bytes(path_base, "utf-8") + RELATIVE_RELATIVE_REDIRECT)]
+ )
+
+ assert response.text == str(BODY, "utf-8")
+
+
+def test_chunked_allow_redirects_false():
+ pool = mocket.MocketPool()
+ pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 443)),)
+ chunk = _chunk(BODY_REDIRECT, len(BODY_REDIRECT))
+ chunk2 = _chunk(BODY, len(BODY))
+
+ redirect = HEADERS_REDIRECT.replace(
+ b"NEW_LOCATION_HERE", ABSOLUTE_ABSOLUTE_REDIRECT
+ )
+ sock1 = MocketRecvInto(redirect + chunk)
+ sock2 = mocket.Mocket(HEADERS + chunk2)
+ pool.socket.side_effect = (sock1, sock2)
+
+ requests_session = adafruit_requests.Session(pool, mocket.SSLContext())
+ response = requests_session.get("https://" + HOST + PATH, allow_redirects=False)
+
+ sock1.connect.assert_called_once_with((HOST, 443))
+ sock2.connect.assert_not_called()
+
+ assert response.text == str(BODY_REDIRECT, "utf-8")

tests/chunked_redirect_test.py

@@ -9,30 +9,16 @@
 
 import mocket
 
-import adafruit_requests
-
-IP = "1.2.3.4"
-HOST = "wifitest.adafruit.com"
-HOST2 = "test.adafruit.com"
-PATH = "/testwifi/index.html"
-TEXT = b"This is a test of Adafruit WiFi!\r\nIf you can read this, its working :)"
-RESPONSE = b"HTTP/1.0 200 OK\r\nContent-Length: 70\r\n\r\n" + TEXT
-
-
-def test_second_connect_fails_memoryerror():
- pool = mocket.MocketPool()
- pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),)
- sock = mocket.Mocket(RESPONSE)
- sock2 = mocket.Mocket(RESPONSE)
- sock3 = mocket.Mocket(RESPONSE)
+
+def test_second_connect_fails_memoryerror(pool, requests_ssl):
+ sock = mocket.Mocket()
+ sock2 = mocket.Mocket()
+ sock3 = mocket.Mocket()
  pool.socket.call_count = 0 # Reset call count
  pool.socket.side_effect = [sock, sock2, sock3]
  sock2.connect.side_effect = MemoryError()
 
- ssl = mocket.SSLContext()
-
- requests_session = adafruit_requests.Session(pool, ssl)
- response = requests_session.get("https://" + HOST + PATH)
+ response = requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1)
 
  sock.send.assert_has_calls(
  [
@@ -48,34 +34,29 @@ def test_second_connect_fails_memoryerror():
  mock.call(b"\r\n"),
  ]
  )
- assert response.text == str(TEXT, "utf-8")
+ assert response.text == str(mocket.MOCK_RESPONSE_TEXT, "utf-8")
 
- requests_session.get("https://" + HOST2 + PATH)
+ requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_2)
 
- sock.connect.assert_called_once_with((HOST, 443))
- sock2.connect.assert_called_once_with((HOST2, 443))
- sock3.connect.assert_called_once_with((HOST2, 443))
+ sock.connect.assert_called_once_with((mocket.MOCK_HOST_1, 443))
+ sock2.connect.assert_called_once_with((mocket.MOCK_HOST_2, 443))
+ sock3.connect.assert_called_once_with((mocket.MOCK_HOST_2, 443))
  # Make sure that the socket is closed after send fails.
  sock.close.assert_called_once()
  sock2.close.assert_called_once()
  assert sock3.close.call_count == 0
  assert pool.socket.call_count == 3
 
 
-def test_second_connect_fails_oserror():
- pool = mocket.MocketPool()
- pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),)
- sock = mocket.Mocket(RESPONSE)
- sock2 = mocket.Mocket(RESPONSE)
- sock3 = mocket.Mocket(RESPONSE)
+def test_second_connect_fails_oserror(pool, requests_ssl):
+ sock = mocket.Mocket()
+ sock2 = mocket.Mocket()
+ sock3 = mocket.Mocket()
  pool.socket.call_count = 0 # Reset call count
  pool.socket.side_effect = [sock, sock2, sock3]
  sock2.connect.side_effect = OSError(errno.ENOMEM)
 
- ssl = mocket.SSLContext()
-
- requests_session = adafruit_requests.Session(pool, ssl)
- response = requests_session.get("https://" + HOST + PATH)
+ response = requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1)
 
  sock.send.assert_has_calls(
  [
@@ -91,13 +72,13 @@ def test_second_connect_fails_oserror():
  mock.call(b"\r\n"),
  ]
  )
- assert response.text == str(TEXT, "utf-8")
+ assert response.text == str(mocket.MOCK_RESPONSE_TEXT, "utf-8")
 
- requests_session.get("https://" + HOST2 + PATH)
+ requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_2)
 
- sock.connect.assert_called_once_with((HOST, 443))
- sock2.connect.assert_called_once_with((HOST2, 443))
- sock3.connect.assert_called_once_with((HOST2, 443))
+ sock.connect.assert_called_once_with((mocket.MOCK_HOST_1, 443))
+ sock2.connect.assert_called_once_with((mocket.MOCK_HOST_2, 443))
+ sock3.connect.assert_called_once_with((mocket.MOCK_HOST_2, 443))
  # Make sure that the socket is closed after send fails.
  sock.close.assert_called_once()
  sock2.close.assert_called_once()