1  from unittest import mock
  2  import mocket
  3  import adafruit_requests
  4  
  5  ip = "1.2.3.4"
  6  host = "wifitest.adafruit.com"
  7  path = "/testwifi/index.html"
  8  text = b"This is a test of Adafruit WiFi!\r\nIf you can read this, its working :)"
  9  headers = b"HTTP/1.0 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n"
 10  
 11  
 12  def _chunk(response, split, extra=b""):
 13      i = 0
 14      chunked = b""
 15      while i < len(response):
 16          remaining = len(response) - i
 17          chunk_size = split
 18          if remaining < chunk_size:
 19              chunk_size = remaining
 20          new_i = i + chunk_size
 21          chunked += (
 22              hex(chunk_size)[2:].encode("ascii")
 23              + extra
 24              + b"\r\n"
 25              + response[i:new_i]
 26              + b"\r\n"
 27          )
 28          i = new_i
 29      # The final chunk is zero length.
 30      chunked += b"0\r\n\r\n"
 31      return chunked
 32  
 33  
 34  def do_test_get_text(extra=b""):
 35      pool = mocket.MocketPool()
 36      pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),)
 37      c = _chunk(text, 33, extra)
 38      print(c)
 39      sock = mocket.Mocket(headers + c)
 40      pool.socket.return_value = sock
 41  
 42      s = adafruit_requests.Session(pool)
 43      r = s.get("http://" + host + path)
 44  
 45      sock.connect.assert_called_once_with((ip, 80))
 46  
 47      sock.send.assert_has_calls(
 48          [
 49              mock.call(b"GET"),
 50              mock.call(b" /"),
 51              mock.call(b"testwifi/index.html"),
 52              mock.call(b" HTTP/1.1\r\n"),
 53          ]
 54      )
 55      sock.send.assert_has_calls(
 56          [mock.call(b"Host: "), mock.call(b"wifitest.adafruit.com"),]
 57      )
 58      assert r.text == str(text, "utf-8")
 59  
 60  
 61  def test_get_text():
 62      do_test_get_text()
 63  
 64  
 65  def test_get_text_extra():
 66      do_test_get_text(b";blahblah; blah")
 67  
 68  
 69  def do_test_close_flush(extra=b""):
 70      """Test that a chunked response can be closed even when the request contents were not accessed."""
 71      pool = mocket.MocketPool()
 72      pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),)
 73      c = _chunk(text, 33, extra)
 74      print(c)
 75      sock = mocket.Mocket(headers + c)
 76      pool.socket.return_value = sock
 77  
 78      s = adafruit_requests.Session(pool)
 79      r = s.get("http://" + host + path)
 80  
 81      sock.connect.assert_called_once_with((ip, 80))
 82  
 83      sock.send.assert_has_calls(
 84          [
 85              mock.call(b"GET"),
 86              mock.call(b" /"),
 87              mock.call(b"testwifi/index.html"),
 88              mock.call(b" HTTP/1.1\r\n"),
 89          ]
 90      )
 91      sock.send.assert_has_calls(
 92          [mock.call(b"Host: "), mock.call(b"wifitest.adafruit.com"),]
 93      )
 94  
 95      r.close()
 96  
 97  
 98  def test_close_flush():
 99      do_test_close_flush()
100  
101  
102  def test_close_flush_extra():
103      do_test_close_flush(b";blahblah; blah")