1  from unittest import mock
  2  import mocket
  3  import pytest
  4  import adafruit_requests
  5  
  6  ip = "1.2.3.4"
  7  host = "wifitest.adafruit.com"
  8  host2 = "wifitest2.adafruit.com"
  9  path = "/testwifi/index.html"
 10  text = b"This is a test of Adafruit WiFi!\r\nIf you can read this, its working :)"
 11  response = b"HTTP/1.0 200 OK\r\nContent-Length: 70\r\n\r\n" + text
 12  
 13  
 14  def test_get_twice():
 15      pool = mocket.MocketPool()
 16      pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),)
 17      sock = mocket.Mocket(response + response)
 18      pool.socket.return_value = sock
 19      ssl = mocket.SSLContext()
 20  
 21      s = adafruit_requests.Session(pool, ssl)
 22      r = s.get("https://" + host + path)
 23  
 24      sock.send.assert_has_calls(
 25          [
 26              mock.call(b"GET"),
 27              mock.call(b" /"),
 28              mock.call(b"testwifi/index.html"),
 29              mock.call(b" HTTP/1.1\r\n"),
 30          ]
 31      )
 32      sock.send.assert_has_calls(
 33          [mock.call(b"Host: "), mock.call(b"wifitest.adafruit.com"),]
 34      )
 35      assert r.text == str(text, "utf-8")
 36  
 37      r = s.get("https://" + host + path + "2")
 38  
 39      sock.send.assert_has_calls(
 40          [
 41              mock.call(b"GET"),
 42              mock.call(b" /"),
 43              mock.call(b"testwifi/index.html2"),
 44              mock.call(b" HTTP/1.1\r\n"),
 45          ]
 46      )
 47      sock.send.assert_has_calls(
 48          [mock.call(b"Host: "), mock.call(b"wifitest.adafruit.com"),]
 49      )
 50  
 51      assert r.text == str(text, "utf-8")
 52      sock.connect.assert_called_once_with((host, 443))
 53      pool.socket.assert_called_once()
 54  
 55  
 56  def test_get_twice_after_second():
 57      pool = mocket.MocketPool()
 58      pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),)
 59      sock = mocket.Mocket(response + response)
 60      pool.socket.return_value = sock
 61      ssl = mocket.SSLContext()
 62  
 63      s = adafruit_requests.Session(pool, ssl)
 64      r = s.get("https://" + host + path)
 65  
 66      sock.send.assert_has_calls(
 67          [
 68              mock.call(b"GET"),
 69              mock.call(b" /"),
 70              mock.call(b"testwifi/index.html"),
 71              mock.call(b" HTTP/1.1\r\n"),
 72          ]
 73      )
 74      sock.send.assert_has_calls(
 75          [mock.call(b"Host: "), mock.call(b"wifitest.adafruit.com"),]
 76      )
 77  
 78      r2 = s.get("https://" + host + path + "2")
 79  
 80      sock.send.assert_has_calls(
 81          [
 82              mock.call(b"GET"),
 83              mock.call(b" /"),
 84              mock.call(b"testwifi/index.html2"),
 85              mock.call(b" HTTP/1.1\r\n"),
 86          ]
 87      )
 88      sock.send.assert_has_calls(
 89          [mock.call(b"Host: "), mock.call(b"wifitest.adafruit.com"),]
 90      )
 91      sock.connect.assert_called_once_with((host, 443))
 92      pool.socket.assert_called_once()
 93  
 94      with pytest.raises(RuntimeError):
 95          r.text == str(text, "utf-8")
 96  
 97  
 98  def test_connect_out_of_memory():
 99      pool = mocket.MocketPool()
100      pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),)
101      sock = mocket.Mocket(response)
102      sock2 = mocket.Mocket(response)
103      sock3 = mocket.Mocket(response)
104      pool.socket.side_effect = [sock, sock2, sock3]
105      sock2.connect.side_effect = MemoryError()
106      ssl = mocket.SSLContext()
107  
108      s = adafruit_requests.Session(pool, ssl)
109      r = s.get("https://" + host + path)
110  
111      sock.send.assert_has_calls(
112          [
113              mock.call(b"GET"),
114              mock.call(b" /"),
115              mock.call(b"testwifi/index.html"),
116              mock.call(b" HTTP/1.1\r\n"),
117          ]
118      )
119      sock.send.assert_has_calls(
120          [mock.call(b"Host: "), mock.call(b"wifitest.adafruit.com"),]
121      )
122      assert r.text == str(text, "utf-8")
123  
124      r = s.get("https://" + host2 + path)
125      sock3.send.assert_has_calls(
126          [
127              mock.call(b"GET"),
128              mock.call(b" /"),
129              mock.call(b"testwifi/index.html"),
130              mock.call(b" HTTP/1.1\r\n"),
131          ]
132      )
133      sock3.send.assert_has_calls(
134          [mock.call(b"Host: "), mock.call(b"wifitest2.adafruit.com"),]
135      )
136  
137      assert r.text == str(text, "utf-8")
138      sock.close.assert_called_once()
139      sock.connect.assert_called_once_with((host, 443))
140      sock3.connect.assert_called_once_with((host2, 443))
141  
142  
143  def test_second_send_fails():
144      pool = mocket.MocketPool()
145      pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),)
146      sock = mocket.Mocket(response)
147      sock2 = mocket.Mocket(response)
148      pool.socket.side_effect = [sock, sock2]
149  
150      ssl = mocket.SSLContext()
151  
152      s = adafruit_requests.Session(pool, ssl)
153      r = s.get("https://" + host + path)
154  
155      sock.send.assert_has_calls(
156          [mock.call(b"testwifi/index.html"),]
157      )
158  
159      sock.send.assert_has_calls(
160          [mock.call(b"Host: "), mock.call(b"wifitest.adafruit.com"), mock.call(b"\r\n"),]
161      )
162      assert r.text == str(text, "utf-8")
163  
164      sock.fail_next_send = True
165      s.get("https://" + host + path + "2")
166  
167      sock.connect.assert_called_once_with((host, 443))
168      sock2.connect.assert_called_once_with((host, 443))
169      # Make sure that the socket is closed after send fails.
170      sock.close.assert_called_once()
171      assert sock2.close.call_count == 0
172      assert pool.socket.call_count == 2