1  from unittest import mock
 2  import mocket
 3  import pytest
 4  import errno
 5  import adafruit_requests
 6  
 7  ip = "1.2.3.4"
 8  host = "wifitest.adafruit.com"
 9  host2 = "wifitest2.adafruit.com"
10  path = "/testwifi/index.html"
11  text = b"This is a test of Adafruit WiFi!\r\nIf you can read this, its working :)"
12  response = b"HTTP/1.0 200 OK\r\nContent-Length: 70\r\n\r\n" + text
13  
14  
15  def test_second_connect_fails_memoryerror():
16      pool = mocket.MocketPool()
17      pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),)
18      sock = mocket.Mocket(response)
19      sock2 = mocket.Mocket(response)
20      sock3 = mocket.Mocket(response)
21      pool.socket.call_count = 0  # Reset call count
22      pool.socket.side_effect = [sock, sock2, sock3]
23      sock2.connect.side_effect = MemoryError()
24  
25      ssl = mocket.SSLContext()
26  
27      s = adafruit_requests.Session(pool, ssl)
28      r = s.get("https://" + host + path)
29  
30      sock.send.assert_has_calls(
31          [mock.call(b"testwifi/index.html"),]
32      )
33  
34      sock.send.assert_has_calls(
35          [mock.call(b"Host: "), mock.call(b"wifitest.adafruit.com"), mock.call(b"\r\n"),]
36      )
37      assert r.text == str(text, "utf-8")
38  
39      host2 = "test.adafruit.com"
40      s.get("https://" + host2 + path)
41  
42      sock.connect.assert_called_once_with((host, 443))
43      sock2.connect.assert_called_once_with((host2, 443))
44      sock3.connect.assert_called_once_with((host2, 443))
45      # Make sure that the socket is closed after send fails.
46      sock.close.assert_called_once()
47      sock2.close.assert_called_once()
48      assert sock3.close.call_count == 0
49      assert pool.socket.call_count == 3
50  
51  
52  def test_second_connect_fails_oserror():
53      pool = mocket.MocketPool()
54      pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),)
55      sock = mocket.Mocket(response)
56      sock2 = mocket.Mocket(response)
57      sock3 = mocket.Mocket(response)
58      pool.socket.call_count = 0  # Reset call count
59      pool.socket.side_effect = [sock, sock2, sock3]
60      sock2.connect.side_effect = OSError(errno.ENOMEM)
61  
62      ssl = mocket.SSLContext()
63  
64      s = adafruit_requests.Session(pool, ssl)
65      r = s.get("https://" + host + path)
66  
67      sock.send.assert_has_calls(
68          [mock.call(b"testwifi/index.html"),]
69      )
70  
71      sock.send.assert_has_calls(
72          [mock.call(b"Host: "), mock.call(b"wifitest.adafruit.com"), mock.call(b"\r\n"),]
73      )
74      assert r.text == str(text, "utf-8")
75  
76      host2 = "test.adafruit.com"
77      s.get("https://" + host2 + path)
78  
79      sock.connect.assert_called_once_with((host, 443))
80      sock2.connect.assert_called_once_with((host2, 443))
81      sock3.connect.assert_called_once_with((host2, 443))
82      # Make sure that the socket is closed after send fails.
83      sock.close.assert_called_once()
84      sock2.close.assert_called_once()
85      assert sock3.close.call_count == 0
86      assert pool.socket.call_count == 3