legacy_test.py
1 from unittest import mock 2 import legacy_mocket as mocket 3 import json 4 import pytest 5 import adafruit_requests 6 7 ip = "1.2.3.4" 8 host = "httpbin.org" 9 response = {"Date": "July 25, 2019"} 10 encoded = json.dumps(response).encode("utf-8") 11 headers = "HTTP/1.0 200 OK\r\nContent-Length: {}\r\n\r\n".format(len(encoded)).encode( 12 "utf-8" 13 ) 14 15 16 def test_get_json(): 17 mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) 18 sock = mocket.Mocket(headers + encoded) 19 mocket.socket.return_value = sock 20 21 adafruit_requests.set_socket(mocket, mocket.interface) 22 r = adafruit_requests.get("http://" + host + "/get") 23 24 sock.connect.assert_called_once_with((ip, 80)) 25 assert r.json() == response 26 r.close() 27 28 29 def test_tls_mode(): 30 mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) 31 sock = mocket.Mocket(headers + encoded) 32 mocket.socket.return_value = sock 33 34 adafruit_requests.set_socket(mocket, mocket.interface) 35 r = adafruit_requests.get("https://" + host + "/get") 36 37 sock.connect.assert_called_once_with((host, 443), mocket.interface.TLS_MODE) 38 assert r.json() == response 39 r.close() 40 41 42 def test_post_string(): 43 mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) 44 sock = mocket.Mocket(headers + encoded) 45 mocket.socket.return_value = sock 46 47 adafruit_requests.set_socket(mocket, mocket.interface) 48 data = "31F" 49 r = adafruit_requests.post("http://" + host + "/post", data=data) 50 sock.connect.assert_called_once_with((ip, 80)) 51 sock.send.assert_called_with(b"31F") 52 r.close() 53 54 55 def test_second_tls_send_fails(): 56 mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) 57 sock = mocket.Mocket(headers + encoded) 58 sock2 = mocket.Mocket(headers + encoded) 59 mocket.socket.call_count = 0 # Reset call count 60 mocket.socket.side_effect = [sock, sock2] 61 62 adafruit_requests.set_socket(mocket, mocket.interface) 63 r = adafruit_requests.get("https://" + host + "/testwifi/index.html") 64 65 sock.send.assert_has_calls( 66 [mock.call(b"testwifi/index.html"),] 67 ) 68 69 sock.send.assert_has_calls( 70 [mock.call(b"Host: "), mock.call(host.encode("utf-8")), mock.call(b"\r\n"),] 71 ) 72 assert r.text == str(encoded, "utf-8") 73 74 sock.fail_next_send = True 75 adafruit_requests.get("https://" + host + "/get2") 76 77 sock.connect.assert_called_once_with((host, 443), mocket.interface.TLS_MODE) 78 sock2.connect.assert_called_once_with((host, 443), mocket.interface.TLS_MODE) 79 # Make sure that the socket is closed after send fails. 80 sock.close.assert_called_once() 81 assert sock2.close.call_count == 0 82 assert mocket.socket.call_count == 2 83 84 85 def test_second_send_fails(): 86 mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) 87 sock = mocket.Mocket(headers + encoded) 88 sock2 = mocket.Mocket(headers + encoded) 89 mocket.socket.call_count = 0 # Reset call count 90 mocket.socket.side_effect = [sock, sock2] 91 92 adafruit_requests.set_socket(mocket, mocket.interface) 93 r = adafruit_requests.get("http://" + host + "/testwifi/index.html") 94 95 sock.send.assert_has_calls( 96 [mock.call(b"testwifi/index.html"),] 97 ) 98 99 sock.send.assert_has_calls( 100 [mock.call(b"Host: "), mock.call(host.encode("utf-8")), mock.call(b"\r\n"),] 101 ) 102 assert r.text == str(encoded, "utf-8") 103 104 sock.fail_next_send = True 105 adafruit_requests.get("http://" + host + "/get2") 106 107 sock.connect.assert_called_once_with((ip, 80)) 108 sock2.connect.assert_called_once_with((ip, 80)) 109 # Make sure that the socket is closed after send fails. 110 sock.close.assert_called_once() 111 assert sock2.close.call_count == 0 112 assert mocket.socket.call_count == 2 113 114 115 def test_first_read_fails(): 116 mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) 117 sock = mocket.Mocket(b"") 118 mocket.socket.call_count = 0 # Reset call count 119 mocket.socket.side_effect = [sock] 120 121 adafruit_requests.set_socket(mocket, mocket.interface) 122 123 with pytest.raises(RuntimeError): 124 r = adafruit_requests.get("http://" + host + "/testwifi/index.html") 125 126 sock.send.assert_has_calls( 127 [mock.call(b"testwifi/index.html"),] 128 ) 129 130 sock.send.assert_has_calls( 131 [mock.call(b"Host: "), mock.call(host.encode("utf-8")), mock.call(b"\r\n"),] 132 ) 133 134 sock.connect.assert_called_once_with((ip, 80)) 135 # Make sure that the socket is closed after the first receive fails. 136 sock.close.assert_called_once() 137 assert mocket.socket.call_count == 1 138 139 140 def test_second_tls_connect_fails(): 141 mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) 142 sock = mocket.Mocket(headers + encoded) 143 sock2 = mocket.Mocket(headers + encoded) 144 sock3 = mocket.Mocket(headers + encoded) 145 mocket.socket.call_count = 0 # Reset call count 146 mocket.socket.side_effect = [sock, sock2, sock3] 147 sock2.connect.side_effect = RuntimeError("error connecting") 148 149 adafruit_requests.set_socket(mocket, mocket.interface) 150 r = adafruit_requests.get("https://" + host + "/testwifi/index.html") 151 152 sock.send.assert_has_calls( 153 [mock.call(b"testwifi/index.html"),] 154 ) 155 156 sock.send.assert_has_calls( 157 [mock.call(b"Host: "), mock.call(host.encode("utf-8")), mock.call(b"\r\n"),] 158 ) 159 assert r.text == str(encoded, "utf-8") 160 161 host2 = "test.adafruit.com" 162 r = adafruit_requests.get("https://" + host2 + "/get2") 163 164 sock.connect.assert_called_once_with((host, 443), mocket.interface.TLS_MODE) 165 sock2.connect.assert_called_once_with((host2, 443), mocket.interface.TLS_MODE) 166 sock3.connect.assert_called_once_with((host2, 443), mocket.interface.TLS_MODE) 167 # Make sure that the socket is closed after send fails. 168 sock.close.assert_called_once() 169 sock2.close.assert_called_once() 170 assert sock3.close.call_count == 0 171 assert mocket.socket.call_count == 3