diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 354c761..6f11a4c 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -23,7 +23,7 @@ repos: - id: pylint name: pylint (library code) types: [python] - exclude: "^(docs/|examples/|setup.py$)" + exclude: "^(docs/|examples/|tests/|setup.py$)" - repo: local hooks: - id: pylint_examples @@ -32,3 +32,11 @@ repos: entry: /usr/bin/env bash -c args: ['([[ ! -d "examples" ]] || for example in $(find . -path "./examples/*.py"); do pylint --disable=missing-docstring,invalid-name $example; done)'] language: system +- repo: local + hooks: + - id: pylint_tests + name: pylint (tests code) + description: Run pylint rules on "tests/*.py" files + entry: /usr/bin/env bash -c + args: ['([[ ! -d "tests" ]] || for example in $(find . -path "./tests/*.py"); do pylint --disable=missing-docstring,invalid-name $example; done)'] + language: system diff --git a/adafruit_requests.py b/adafruit_requests.py index f85465e..3e681ac 100644 --- a/adafruit_requests.py +++ b/adafruit_requests.py @@ -652,9 +652,12 @@ def wrap_socket(self, socket, server_hostname=None): def set_socket(sock, iface=None): """Legacy API for setting the socket and network interface. Use a `Session` instead.""" global _default_session # pylint: disable=global-statement,invalid-name - _default_session = Session(sock, _FakeSSLContext(iface)) - if iface: - sock.set_interface(iface) + if not iface: + # pylint: disable=protected-access + _default_session = Session(sock, _FakeSSLContext(sock._the_interface)) + else: + _default_session = Session(sock, _FakeSSLContext(iface)) + sock.set_interface(iface) def request(method, url, data=None, json=None, headers=None, stream=False, timeout=1): diff --git a/examples/requests_simpletest.py b/examples/requests_simpletest.py index 519f347..14cf66c 100755 --- a/examples/requests_simpletest.py +++ b/examples/requests_simpletest.py @@ -29,6 +29,11 @@ # esp32_ready = DigitalInOut(board.D10) # esp32_reset = DigitalInOut(board.D5) +# If you have an AirLift Featherwing or ItsyBitsy Airlift: +# esp32_cs = DigitalInOut(board.D13) +# esp32_ready = DigitalInOut(board.D11) +# esp32_reset = DigitalInOut(board.D12) + spi = busio.SPI(board.SCK, board.MOSI, board.MISO) esp = adafruit_esp32spi.ESP_SPIcontrol(spi, esp32_cs, esp32_ready, esp32_reset) @@ -46,8 +51,8 @@ requests.set_socket(socket, esp) TEXT_URL = "http://wifitest.adafruit.com/testwifi/index.html" -JSON_GET_URL = "http://httpbin.org/get" -JSON_POST_URL = "http://httpbin.org/post" +JSON_GET_URL = "https://httpbin.org/get" +JSON_POST_URL = "https://httpbin.org/post" print("Fetching text from %s" % TEXT_URL) response = requests.get(TEXT_URL) diff --git a/tests/chunk_test.py b/tests/chunk_test.py index 166f67c..a03eec8 100644 --- a/tests/chunk_test.py +++ b/tests/chunk_test.py @@ -2,16 +2,18 @@ # # SPDX-License-Identifier: Unlicense +""" Chunk Tests """ + from unittest import mock import mocket import adafruit_requests -ip = "1.2.3.4" -host = "wifitest.adafruit.com" -path = "/testwifi/index.html" -text = b"This is a test of Adafruit WiFi!\r\nIf you can read this, its working :)" -headers = b"HTTP/1.0 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" -headers_extra_space = b"HTTP/1.0 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" +IP = "1.2.3.4" +HOST = "wifitest.adafruit.com" +PATH = "/testwifi/index.html" +TEXT = b"This is a test of Adafruit WiFi!\r\nIf you can read this, its working :)" +HEADERS = b"HTTP/1.0 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" +HEADERS_EXTRA_SPACE = b"HTTP/1.0 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" def _chunk(response, split, extra=b""): @@ -36,18 +38,20 @@ def _chunk(response, split, extra=b""): return chunked -def do_test_get_text(extra=b""): +def do_test_get_text( + extra=b"", +): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - c = _chunk(text, 33, extra) - print(c) - sock = mocket.Mocket(headers + c) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + chunk = _chunk(TEXT, 33, extra) + print(chunk) + sock = mocket.Mocket(HEADERS + chunk) pool.socket.return_value = sock - s = adafruit_requests.Session(pool) - r = s.get("http://" + host + path) + requests_session = adafruit_requests.Session(pool) + response = requests_session.get("http://" + HOST + PATH) - sock.connect.assert_called_once_with((ip, 80)) + sock.connect.assert_called_once_with((IP, 80)) sock.send.assert_has_calls( [ @@ -63,7 +67,7 @@ def do_test_get_text(extra=b""): mock.call(b"wifitest.adafruit.com"), ] ) - assert r.text == str(text, "utf-8") + assert response.text == str(TEXT, "utf-8") def test_get_text(): @@ -74,19 +78,22 @@ def test_get_text_extra(): do_test_get_text(b";blahblah; blah") -def do_test_close_flush(extra=b""): - """Test that a chunked response can be closed even when the request contents were not accessed.""" +def do_test_close_flush( + extra=b"", +): + """Test that a chunked response can be closed even when the + request contents were not accessed.""" pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - c = _chunk(text, 33, extra) - print(c) - sock = mocket.Mocket(headers + c) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + chunk = _chunk(TEXT, 33, extra) + print(chunk) + sock = mocket.Mocket(HEADERS + chunk) pool.socket.return_value = sock - s = adafruit_requests.Session(pool) - r = s.get("http://" + host + path) + requests_session = adafruit_requests.Session(pool) + response = requests_session.get("http://" + HOST + PATH) - sock.connect.assert_called_once_with((ip, 80)) + sock.connect.assert_called_once_with((IP, 80)) sock.send.assert_has_calls( [ @@ -103,7 +110,7 @@ def do_test_close_flush(extra=b""): ] ) - r.close() + response.close() def test_close_flush(): @@ -114,18 +121,20 @@ def test_close_flush_extra(): do_test_close_flush(b";blahblah; blah") -def do_test_get_text_extra_space(extra=b""): +def do_test_get_text_extra_space( + extra=b"", +): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - c = _chunk(text, 33, extra) - print(c) - sock = mocket.Mocket(headers_extra_space + c) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + chunk = _chunk(TEXT, 33, extra) + print(chunk) + sock = mocket.Mocket(HEADERS_EXTRA_SPACE + chunk) pool.socket.return_value = sock - s = adafruit_requests.Session(pool) - r = s.get("http://" + host + path) + requests_session = adafruit_requests.Session(pool) + response = requests_session.get("http://" + HOST + PATH) - sock.connect.assert_called_once_with((ip, 80)) + sock.connect.assert_called_once_with((IP, 80)) sock.send.assert_has_calls( [ @@ -141,4 +150,4 @@ def do_test_get_text_extra_space(extra=b""): mock.call(b"wifitest.adafruit.com"), ] ) - assert r.text == str(text, "utf-8") + assert response.text == str(TEXT, "utf-8") diff --git a/tests/concurrent_test.py b/tests/concurrent_test.py index 9b2c894..42df46c 100644 --- a/tests/concurrent_test.py +++ b/tests/concurrent_test.py @@ -2,34 +2,35 @@ # # SPDX-License-Identifier: Unlicense +""" Concurrent Tests """ + +import errno from unittest import mock import mocket -import pytest -import errno import adafruit_requests -ip = "1.2.3.4" -host = "wifitest.adafruit.com" -host2 = "wifitest2.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 +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(): +def test_second_connect_fails_memoryerror(): # pylint: disable=invalid-name 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) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(RESPONSE) + sock2 = mocket.Mocket(RESPONSE) + sock3 = mocket.Mocket(RESPONSE) pool.socket.call_count = 0 # Reset call count pool.socket.side_effect = [sock, sock2, sock3] sock2.connect.side_effect = MemoryError() ssl = mocket.SSLContext() - s = adafruit_requests.Session(pool, ssl) - r = s.get("https://" + host + path) + requests_session = adafruit_requests.Session(pool, ssl) + response = requests_session.get("https://" + HOST + PATH) sock.send.assert_has_calls( [ @@ -44,14 +45,13 @@ def test_second_connect_fails_memoryerror(): mock.call(b"\r\n"), ] ) - assert r.text == str(text, "utf-8") + assert response.text == str(TEXT, "utf-8") - host2 = "test.adafruit.com" - s.get("https://" + host2 + path) + requests_session.get("https://" + HOST2 + PATH) - 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((HOST, 443)) + sock2.connect.assert_called_once_with((HOST2, 443)) + sock3.connect.assert_called_once_with((HOST2, 443)) # Make sure that the socket is closed after send fails. sock.close.assert_called_once() sock2.close.assert_called_once() @@ -59,20 +59,20 @@ def test_second_connect_fails_memoryerror(): assert pool.socket.call_count == 3 -def test_second_connect_fails_oserror(): +def test_second_connect_fails_oserror(): # pylint: disable=invalid-name 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) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(RESPONSE) + sock2 = mocket.Mocket(RESPONSE) + sock3 = mocket.Mocket(RESPONSE) 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() - s = adafruit_requests.Session(pool, ssl) - r = s.get("https://" + host + path) + requests_session = adafruit_requests.Session(pool, ssl) + response = requests_session.get("https://" + HOST + PATH) sock.send.assert_has_calls( [ @@ -87,14 +87,13 @@ def test_second_connect_fails_oserror(): mock.call(b"\r\n"), ] ) - assert r.text == str(text, "utf-8") + assert response.text == str(TEXT, "utf-8") - host2 = "test.adafruit.com" - s.get("https://" + host2 + path) + requests_session.get("https://" + HOST2 + PATH) - 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((HOST, 443)) + sock2.connect.assert_called_once_with((HOST2, 443)) + sock3.connect.assert_called_once_with((HOST2, 443)) # Make sure that the socket is closed after send fails. sock.close.assert_called_once() sock2.close.assert_called_once() diff --git a/tests/header_test.py b/tests/header_test.py index 346bbf0..05bf88f 100644 --- a/tests/header_test.py +++ b/tests/header_test.py @@ -2,34 +2,34 @@ # # SPDX-License-Identifier: Unlicense -from unittest import mock +""" Header Tests """ + import mocket -import json import adafruit_requests -ip = "1.2.3.4" -host = "httpbin.org" -response_headers = b"HTTP/1.0 200 OK\r\nContent-Length: 0\r\n\r\n" +IP = "1.2.3.4" +HOST = "httpbin.org" +RESPONSE_HEADERS = b"HTTP/1.0 200 OK\r\nContent-Length: 0\r\n\r\n" def test_json(): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(response_headers) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(RESPONSE_HEADERS) pool.socket.return_value = sock sent = [] def _send(data): - sent.append(data) + sent.append(data) # pylint: disable=no-member return len(data) sock.send.side_effect = _send - s = adafruit_requests.Session(pool) + requests_session = adafruit_requests.Session(pool) headers = {"user-agent": "blinka/1.0.0"} - r = s.get("http://" + host + "/get", headers=headers) + requests_session.get("http://" + HOST + "/get", headers=headers) - sock.connect.assert_called_once_with((ip, 80)) + sock.connect.assert_called_once_with((IP, 80)) sent = b"".join(sent).lower() assert b"user-agent: blinka/1.0.0\r\n" in sent # The current implementation sends two user agents. Fix it, and uncomment below. diff --git a/tests/legacy_mocket.py b/tests/legacy_mocket.py index 4506a6f..8e18c4f 100644 --- a/tests/legacy_mocket.py +++ b/tests/legacy_mocket.py @@ -2,6 +2,8 @@ # # SPDX-License-Identifier: Unlicense +""" Mock for Legacy Socket """ + from unittest import mock SOCK_STREAM = 0 @@ -12,7 +14,9 @@ socket = mock.Mock() -class Mocket: +class Mocket: # pylint: disable=too-few-public-methods + """ Mock Socket """ + def __init__(self, response): self.settimeout = mock.Mock() self.close = mock.Mock() @@ -24,21 +28,20 @@ def __init__(self, response): self._response = response self._position = 0 - def _send(self, data): + def _send(self, data): # pylint: disable=unused-argument if self.fail_next_send: self.fail_next_send = False raise RuntimeError("Send failed") - return None def _readline(self): i = self._response.find(b"\r\n", self._position) - r = self._response[self._position : i + 2] + response = self._response[self._position : i + 2] self._position = i + 2 - return r + return response def _recv(self, count): end = self._position + count - r = self._response[self._position : end] + response = self._response[self._position : end] self._position = end - print(r) - return r + print(response) + return response diff --git a/tests/legacy_test.py b/tests/legacy_test.py index 7743f3f..cba604d 100644 --- a/tests/legacy_test.py +++ b/tests/legacy_test.py @@ -2,69 +2,70 @@ # # SPDX-License-Identifier: Unlicense +""" Legacy Tests """ + from unittest import mock -import legacy_mocket as mocket import json -import pytest +import legacy_mocket as mocket import adafruit_requests -ip = "1.2.3.4" -host = "httpbin.org" -response = {"Date": "July 25, 2019"} -encoded = json.dumps(response).encode("utf-8") -headers = "HTTP/1.0 200 OK\r\nContent-Length: {}\r\n\r\n".format(len(encoded)).encode( +IP = "1.2.3.4" +HOST = "httpbin.org" +RESPONSE = {"Date": "July 25, 2019"} +ENCODED = json.dumps(RESPONSE).encode("utf-8") +HEADERS = "HTTP/1.0 200 OK\r\nContent-Length: {}\r\n\r\n".format(len(ENCODED)).encode( "utf-8" ) def test_get_json(): - mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(headers + encoded) + mocket.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(HEADERS + ENCODED) mocket.socket.return_value = sock adafruit_requests.set_socket(mocket, mocket.interface) - r = adafruit_requests.get("http://" + host + "/get") + response = adafruit_requests.get("http://" + HOST + "/get") - sock.connect.assert_called_once_with((ip, 80)) - assert r.json() == response - r.close() + sock.connect.assert_called_once_with((IP, 80)) + assert response.json() == RESPONSE + response.close() def test_tls_mode(): - mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(headers + encoded) + mocket.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(HEADERS + ENCODED) mocket.socket.return_value = sock adafruit_requests.set_socket(mocket, mocket.interface) - r = adafruit_requests.get("https://" + host + "/get") + response = adafruit_requests.get("https://" + HOST + "/get") - sock.connect.assert_called_once_with((host, 443), mocket.interface.TLS_MODE) - assert r.json() == response - r.close() + sock.connect.assert_called_once_with((HOST, 443), mocket.interface.TLS_MODE) + assert response.json() == RESPONSE + response.close() def test_post_string(): - mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(headers + encoded) + mocket.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(HEADERS + ENCODED) mocket.socket.return_value = sock adafruit_requests.set_socket(mocket, mocket.interface) data = "31F" - r = adafruit_requests.post("http://" + host + "/post", data=data) - sock.connect.assert_called_once_with((ip, 80)) + response = adafruit_requests.post("http://" + HOST + "/post", data=data) + sock.connect.assert_called_once_with((IP, 80)) sock.send.assert_called_with(b"31F") - r.close() + response.close() def test_second_tls_send_fails(): - mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(headers + encoded) - sock2 = mocket.Mocket(headers + encoded) + mocket.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(HEADERS + ENCODED) + sock2 = mocket.Mocket(HEADERS + ENCODED) mocket.socket.call_count = 0 # Reset call count mocket.socket.side_effect = [sock, sock2] adafruit_requests.set_socket(mocket, mocket.interface) - r = adafruit_requests.get("https://" + host + "/testwifi/index.html") + response = adafruit_requests.get("https://" + HOST + "/testwifi/index.html") sock.send.assert_has_calls( [ @@ -75,17 +76,17 @@ def test_second_tls_send_fails(): sock.send.assert_has_calls( [ mock.call(b"Host: "), - mock.call(host.encode("utf-8")), + mock.call(HOST.encode("utf-8")), mock.call(b"\r\n"), ] ) - assert r.text == str(encoded, "utf-8") + assert response.text == str(ENCODED, "utf-8") sock.fail_next_send = True - adafruit_requests.get("https://" + host + "/get2") + adafruit_requests.get("https://" + HOST + "/get2") - sock.connect.assert_called_once_with((host, 443), mocket.interface.TLS_MODE) - sock2.connect.assert_called_once_with((host, 443), mocket.interface.TLS_MODE) + sock.connect.assert_called_once_with((HOST, 443), mocket.interface.TLS_MODE) + sock2.connect.assert_called_once_with((HOST, 443), mocket.interface.TLS_MODE) # Make sure that the socket is closed after send fails. sock.close.assert_called_once() assert sock2.close.call_count == 0 @@ -93,14 +94,14 @@ def test_second_tls_send_fails(): def test_second_send_fails(): - mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(headers + encoded) - sock2 = mocket.Mocket(headers + encoded) + mocket.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(HEADERS + ENCODED) + sock2 = mocket.Mocket(HEADERS + ENCODED) mocket.socket.call_count = 0 # Reset call count mocket.socket.side_effect = [sock, sock2] adafruit_requests.set_socket(mocket, mocket.interface) - r = adafruit_requests.get("http://" + host + "/testwifi/index.html") + response = adafruit_requests.get("http://" + HOST + "/testwifi/index.html") sock.send.assert_has_calls( [ @@ -111,17 +112,17 @@ def test_second_send_fails(): sock.send.assert_has_calls( [ mock.call(b"Host: "), - mock.call(host.encode("utf-8")), + mock.call(HOST.encode("utf-8")), mock.call(b"\r\n"), ] ) - assert r.text == str(encoded, "utf-8") + assert response.text == str(ENCODED, "utf-8") sock.fail_next_send = True - adafruit_requests.get("http://" + host + "/get2") + adafruit_requests.get("http://" + HOST + "/get2") - sock.connect.assert_called_once_with((ip, 80)) - sock2.connect.assert_called_once_with((ip, 80)) + sock.connect.assert_called_once_with((IP, 80)) + sock2.connect.assert_called_once_with((IP, 80)) # Make sure that the socket is closed after send fails. sock.close.assert_called_once() assert sock2.close.call_count == 0 @@ -129,15 +130,14 @@ def test_second_send_fails(): def test_first_read_fails(): - mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) + mocket.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) sock = mocket.Mocket(b"") - sock2 = mocket.Mocket(headers + encoded) + sock2 = mocket.Mocket(HEADERS + ENCODED) mocket.socket.call_count = 0 # Reset call count mocket.socket.side_effect = [sock, sock2] adafruit_requests.set_socket(mocket, mocket.interface) - - r = adafruit_requests.get("http://" + host + "/testwifi/index.html") + adafruit_requests.get("http://" + HOST + "/testwifi/index.html") sock.send.assert_has_calls( [ @@ -148,7 +148,7 @@ def test_first_read_fails(): sock.send.assert_has_calls( [ mock.call(b"Host: "), - mock.call(host.encode("utf-8")), + mock.call(HOST.encode("utf-8")), mock.call(b"\r\n"), ] ) @@ -156,29 +156,29 @@ def test_first_read_fails(): sock2.send.assert_has_calls( [ mock.call(b"Host: "), - mock.call(host.encode("utf-8")), + mock.call(HOST.encode("utf-8")), mock.call(b"\r\n"), ] ) - sock.connect.assert_called_once_with((ip, 80)) - sock2.connect.assert_called_once_with((ip, 80)) + sock.connect.assert_called_once_with((IP, 80)) + sock2.connect.assert_called_once_with((IP, 80)) # Make sure that the socket is closed after the first receive fails. sock.close.assert_called_once() assert mocket.socket.call_count == 2 def test_second_tls_connect_fails(): - mocket.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(headers + encoded) - sock2 = mocket.Mocket(headers + encoded) - sock3 = mocket.Mocket(headers + encoded) + mocket.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(HEADERS + ENCODED) + sock2 = mocket.Mocket(HEADERS + ENCODED) + sock3 = mocket.Mocket(HEADERS + ENCODED) mocket.socket.call_count = 0 # Reset call count mocket.socket.side_effect = [sock, sock2, sock3] sock2.connect.side_effect = RuntimeError("error connecting") adafruit_requests.set_socket(mocket, mocket.interface) - r = adafruit_requests.get("https://" + host + "/testwifi/index.html") + response = adafruit_requests.get("https://" + HOST + "/testwifi/index.html") sock.send.assert_has_calls( [ @@ -189,16 +189,16 @@ def test_second_tls_connect_fails(): sock.send.assert_has_calls( [ mock.call(b"Host: "), - mock.call(host.encode("utf-8")), + mock.call(HOST.encode("utf-8")), mock.call(b"\r\n"), ] ) - assert r.text == str(encoded, "utf-8") + assert response.text == str(ENCODED, "utf-8") host2 = "test.adafruit.com" - r = adafruit_requests.get("https://" + host2 + "/get2") + response = adafruit_requests.get("https://" + host2 + "/get2") - sock.connect.assert_called_once_with((host, 443), mocket.interface.TLS_MODE) + sock.connect.assert_called_once_with((HOST, 443), mocket.interface.TLS_MODE) sock2.connect.assert_called_once_with((host2, 443), mocket.interface.TLS_MODE) sock3.connect.assert_called_once_with((host2, 443), mocket.interface.TLS_MODE) # Make sure that the socket is closed after send fails. diff --git a/tests/mocket.py b/tests/mocket.py index 2a44ed2..578a808 100644 --- a/tests/mocket.py +++ b/tests/mocket.py @@ -2,10 +2,14 @@ # # SPDX-License-Identifier: Unlicense +""" Mock Socket """ + from unittest import mock -class MocketPool: +class MocketPool: # pylint: disable=too-few-public-methods + """ Mock SocketPool """ + SOCK_STREAM = 0 def __init__(self): @@ -13,7 +17,9 @@ def __init__(self): self.socket = mock.Mock() -class Mocket: +class Mocket: # pylint: disable=too-few-public-methods + """ Mock Socket """ + def __init__(self, response): self.settimeout = mock.Mock() self.close = mock.Mock() @@ -34,15 +40,15 @@ def _send(self, data): def _readline(self): i = self._response.find(b"\r\n", self._position) - r = self._response[self._position : i + 2] + response = self._response[self._position : i + 2] self._position = i + 2 - return r + return response def _recv(self, count): end = self._position + count - r = self._response[self._position : end] + response = self._response[self._position : end] self._position = end - return r + return response def _recv_into(self, buf, nbytes=0): assert isinstance(nbytes, int) and nbytes >= 0 @@ -56,9 +62,13 @@ def _recv_into(self, buf, nbytes=0): return read -class SSLContext: +class SSLContext: # pylint: disable=too-few-public-methods + """ Mock SSL Context """ + def __init__(self): self.wrap_socket = mock.Mock(side_effect=self._wrap_socket) - def _wrap_socket(self, sock, server_hostname=None): + def _wrap_socket( + self, sock, server_hostname=None + ): # pylint: disable=no-self-use,unused-argument return sock diff --git a/tests/parse_test.py b/tests/parse_test.py index 551b264..36ccd10 100644 --- a/tests/parse_test.py +++ b/tests/parse_test.py @@ -2,31 +2,35 @@ # # SPDX-License-Identifier: Unlicense -from unittest import mock -import mocket +""" Parse Tests """ + import json +import mocket import adafruit_requests -ip = "1.2.3.4" -host = "httpbin.org" -response = {"Date": "July 25, 2019"} -encoded = json.dumps(response).encode("utf-8") +IP = "1.2.3.4" +HOST = "httpbin.org" +RESPONSE = {"Date": "July 25, 2019"} +ENCODED = json.dumps(RESPONSE).encode("utf-8") # Padding here tests the case where a header line is exactly 32 bytes buffered by # aligning the Content-Type header after it. -headers = "HTTP/1.0 200 OK\r\npadding: 000\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n".format( - len(encoded) -).encode( - "utf-8" +HEADERS = ( + ( + "HTTP/1.0 200 OK\r\npadding: 000\r\n" + "Content-Type: application/json\r\nContent-Length: {}\r\n\r\n" + ) + .format(len(ENCODED)) + .encode("utf-8") ) def test_json(): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(headers + encoded) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(HEADERS + ENCODED) pool.socket.return_value = sock - s = adafruit_requests.Session(pool) - r = s.get("http://" + host + "/get") - sock.connect.assert_called_once_with((ip, 80)) - assert r.json() == response + requests_session = adafruit_requests.Session(pool) + response = requests_session.get("http://" + HOST + "/get") + sock.connect.assert_called_once_with((IP, 80)) + assert response.json() == RESPONSE diff --git a/tests/post_test.py b/tests/post_test.py index 0438c94..1d0364f 100644 --- a/tests/post_test.py +++ b/tests/post_test.py @@ -2,29 +2,31 @@ # # SPDX-License-Identifier: Unlicense +""" Post Tests """ + from unittest import mock -import mocket import json +import mocket import adafruit_requests -ip = "1.2.3.4" -host = "httpbin.org" -response = {} -encoded = json.dumps(response).encode("utf-8") -headers = "HTTP/1.0 200 OK\r\nContent-Length: {}\r\n\r\n".format(len(encoded)).encode( +IP = "1.2.3.4" +HOST = "httpbin.org" +RESPONSE = {} +ENCODED = json.dumps(RESPONSE).encode("utf-8") +HEADERS = "HTTP/1.0 200 OK\r\nContent-Length: {}\r\n\r\n".format(len(ENCODED)).encode( "utf-8" ) def test_method(): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(headers + encoded) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(HEADERS + ENCODED) pool.socket.return_value = sock - s = adafruit_requests.Session(pool) - r = s.post("http://" + host + "/post") - sock.connect.assert_called_once_with((ip, 80)) + requests_session = adafruit_requests.Session(pool) + requests_session.post("http://" + HOST + "/post") + sock.connect.assert_called_once_with((IP, 80)) sock.send.assert_has_calls( [ @@ -44,38 +46,38 @@ def test_method(): def test_string(): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(headers + encoded) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(HEADERS + ENCODED) pool.socket.return_value = sock - s = adafruit_requests.Session(pool) + requests_session = adafruit_requests.Session(pool) data = "31F" - r = s.post("http://" + host + "/post", data=data) - sock.connect.assert_called_once_with((ip, 80)) + requests_session.post("http://" + HOST + "/post", data=data) + sock.connect.assert_called_once_with((IP, 80)) sock.send.assert_called_with(b"31F") def test_form(): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(headers + encoded) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(HEADERS + ENCODED) pool.socket.return_value = sock - s = adafruit_requests.Session(pool) + requests_session = adafruit_requests.Session(pool) data = {"Date": "July 25, 2019"} - r = s.post("http://" + host + "/post", data=data) - sock.connect.assert_called_once_with((ip, 80)) + requests_session.post("http://" + HOST + "/post", data=data) + sock.connect.assert_called_once_with((IP, 80)) sock.send.assert_called_with(b"Date=July 25, 2019") def test_json(): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(headers + encoded) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(HEADERS + ENCODED) pool.socket.return_value = sock - s = adafruit_requests.Session(pool) + requests_session = adafruit_requests.Session(pool) json_data = {"Date": "July 25, 2019"} - r = s.post("http://" + host + "/post", json=json_data) - sock.connect.assert_called_once_with((ip, 80)) + requests_session.post("http://" + HOST + "/post", json=json_data) + sock.connect.assert_called_once_with((IP, 80)) sock.send.assert_called_with(b'{"Date": "July 25, 2019"}') diff --git a/tests/protocol_test.py b/tests/protocol_test.py index 35b3a95..fb0471b 100644 --- a/tests/protocol_test.py +++ b/tests/protocol_test.py @@ -2,40 +2,42 @@ # # SPDX-License-Identifier: Unlicense +""" Protocol Tests """ + from unittest import mock import mocket import pytest import adafruit_requests -ip = "1.2.3.4" -host = "wifitest.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 +IP = "1.2.3.4" +HOST = "wifitest.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_get_https_no_ssl(): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(response) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(RESPONSE) pool.socket.return_value = sock - s = adafruit_requests.Session(pool) + requests_session = adafruit_requests.Session(pool) with pytest.raises(RuntimeError): - r = s.get("https://" + host + path) + requests_session.get("https://" + HOST + PATH) def test_get_https_text(): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(response) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(RESPONSE) pool.socket.return_value = sock ssl = mocket.SSLContext() - s = adafruit_requests.Session(pool, ssl) - r = s.get("https://" + host + path) + requests_session = adafruit_requests.Session(pool, ssl) + response = requests_session.get("https://" + HOST + PATH) - sock.connect.assert_called_once_with((host, 443)) + sock.connect.assert_called_once_with((HOST, 443)) sock.send.assert_has_calls( [ @@ -51,22 +53,22 @@ def test_get_https_text(): mock.call(b"wifitest.adafruit.com"), ] ) - assert r.text == str(text, "utf-8") + assert response.text == str(TEXT, "utf-8") # Close isn't needed but can be called to release the socket early. - r.close() + response.close() def test_get_http_text(): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(response) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(RESPONSE) pool.socket.return_value = sock - s = adafruit_requests.Session(pool) - r = s.get("http://" + host + path) + requests_session = adafruit_requests.Session(pool) + response = requests_session.get("http://" + HOST + PATH) - sock.connect.assert_called_once_with((ip, 80)) + sock.connect.assert_called_once_with((IP, 80)) sock.send.assert_has_calls( [ @@ -82,20 +84,20 @@ def test_get_http_text(): mock.call(b"wifitest.adafruit.com"), ] ) - assert r.text == str(text, "utf-8") + assert response.text == str(TEXT, "utf-8") def test_get_close(): """Test that a response can be closed without the contents being accessed.""" pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(response) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(RESPONSE) pool.socket.return_value = sock - s = adafruit_requests.Session(pool) - r = s.get("http://" + host + path) + requests_session = adafruit_requests.Session(pool) + response = requests_session.get("http://" + HOST + PATH) - sock.connect.assert_called_once_with((ip, 80)) + sock.connect.assert_called_once_with((IP, 80)) sock.send.assert_has_calls( [ @@ -111,4 +113,4 @@ def test_get_close(): mock.call(b"wifitest.adafruit.com"), ] ) - r.close() + response.close() diff --git a/tests/reuse_test.py b/tests/reuse_test.py index def1ca8..0f9cc55 100644 --- a/tests/reuse_test.py +++ b/tests/reuse_test.py @@ -2,28 +2,30 @@ # # SPDX-License-Identifier: Unlicense +""" Reuse Tests """ + from unittest import mock import mocket import pytest import adafruit_requests -ip = "1.2.3.4" -host = "wifitest.adafruit.com" -host2 = "wifitest2.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 +IP = "1.2.3.4" +HOST = "wifitest.adafruit.com" +HOST2 = "wifitest2.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_get_twice(): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(response + response) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(RESPONSE + RESPONSE) pool.socket.return_value = sock ssl = mocket.SSLContext() - s = adafruit_requests.Session(pool, ssl) - r = s.get("https://" + host + path) + requests_session = adafruit_requests.Session(pool, ssl) + response = requests_session.get("https://" + HOST + PATH) sock.send.assert_has_calls( [ @@ -39,9 +41,9 @@ def test_get_twice(): mock.call(b"wifitest.adafruit.com"), ] ) - assert r.text == str(text, "utf-8") + assert response.text == str(TEXT, "utf-8") - r = s.get("https://" + host + path + "2") + response = requests_session.get("https://" + HOST + PATH + "2") sock.send.assert_has_calls( [ @@ -58,20 +60,20 @@ def test_get_twice(): ] ) - assert r.text == str(text, "utf-8") - sock.connect.assert_called_once_with((host, 443)) + assert response.text == str(TEXT, "utf-8") + sock.connect.assert_called_once_with((HOST, 443)) pool.socket.assert_called_once() def test_get_twice_after_second(): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(response + response) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(RESPONSE + RESPONSE) pool.socket.return_value = sock ssl = mocket.SSLContext() - s = adafruit_requests.Session(pool, ssl) - r = s.get("https://" + host + path) + requests_session = adafruit_requests.Session(pool, ssl) + response = requests_session.get("https://" + HOST + PATH) sock.send.assert_has_calls( [ @@ -88,7 +90,7 @@ def test_get_twice_after_second(): ] ) - r2 = s.get("https://" + host + path + "2") + requests_session.get("https://" + HOST + PATH + "2") sock.send.assert_has_calls( [ @@ -104,25 +106,25 @@ def test_get_twice_after_second(): mock.call(b"wifitest.adafruit.com"), ] ) - sock.connect.assert_called_once_with((host, 443)) + sock.connect.assert_called_once_with((HOST, 443)) pool.socket.assert_called_once() with pytest.raises(RuntimeError): - r.text == str(text, "utf-8") + response.text == str(TEXT, "utf-8") # pylint: disable=expression-not-assigned def test_connect_out_of_memory(): 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) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(RESPONSE) + sock2 = mocket.Mocket(RESPONSE) + sock3 = mocket.Mocket(RESPONSE) pool.socket.side_effect = [sock, sock2, sock3] sock2.connect.side_effect = MemoryError() ssl = mocket.SSLContext() - s = adafruit_requests.Session(pool, ssl) - r = s.get("https://" + host + path) + requests_session = adafruit_requests.Session(pool, ssl) + response = requests_session.get("https://" + HOST + PATH) sock.send.assert_has_calls( [ @@ -138,9 +140,9 @@ def test_connect_out_of_memory(): mock.call(b"wifitest.adafruit.com"), ] ) - assert r.text == str(text, "utf-8") + assert response.text == str(TEXT, "utf-8") - r = s.get("https://" + host2 + path) + response = requests_session.get("https://" + HOST2 + PATH) sock3.send.assert_has_calls( [ mock.call(b"GET"), @@ -156,23 +158,23 @@ def test_connect_out_of_memory(): ] ) - assert r.text == str(text, "utf-8") + assert response.text == str(TEXT, "utf-8") sock.close.assert_called_once() - sock.connect.assert_called_once_with((host, 443)) - sock3.connect.assert_called_once_with((host2, 443)) + sock.connect.assert_called_once_with((HOST, 443)) + sock3.connect.assert_called_once_with((HOST2, 443)) def test_second_send_fails(): pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(response) - sock2 = mocket.Mocket(response) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(RESPONSE) + sock2 = mocket.Mocket(RESPONSE) pool.socket.side_effect = [sock, sock2] ssl = mocket.SSLContext() - s = adafruit_requests.Session(pool, ssl) - r = s.get("https://" + host + path) + requests_session = adafruit_requests.Session(pool, ssl) + response = requests_session.get("https://" + HOST + PATH) sock.send.assert_has_calls( [ @@ -187,30 +189,30 @@ def test_second_send_fails(): mock.call(b"\r\n"), ] ) - assert r.text == str(text, "utf-8") + assert response.text == str(TEXT, "utf-8") sock.fail_next_send = True - s.get("https://" + host + path + "2") + requests_session.get("https://" + HOST + PATH + "2") - sock.connect.assert_called_once_with((host, 443)) - sock2.connect.assert_called_once_with((host, 443)) + sock.connect.assert_called_once_with((HOST, 443)) + sock2.connect.assert_called_once_with((HOST, 443)) # Make sure that the socket is closed after send fails. sock.close.assert_called_once() assert sock2.close.call_count == 0 assert pool.socket.call_count == 2 -def test_second_send_lies_recv_fails(): +def test_second_send_lies_recv_fails(): # pylint: disable=invalid-name pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),) - sock = mocket.Mocket(response) - sock2 = mocket.Mocket(response) + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) + sock = mocket.Mocket(RESPONSE) + sock2 = mocket.Mocket(RESPONSE) pool.socket.side_effect = [sock, sock2] ssl = mocket.SSLContext() - s = adafruit_requests.Session(pool, ssl) - r = s.get("https://" + host + path) + requests_session = adafruit_requests.Session(pool, ssl) + response = requests_session.get("https://" + HOST + PATH) sock.send.assert_has_calls( [ @@ -225,12 +227,12 @@ def test_second_send_lies_recv_fails(): mock.call(b"\r\n"), ] ) - assert r.text == str(text, "utf-8") + assert response.text == str(TEXT, "utf-8") - s.get("https://" + host + path + "2") + requests_session.get("https://" + HOST + PATH + "2") - sock.connect.assert_called_once_with((host, 443)) - sock2.connect.assert_called_once_with((host, 443)) + sock.connect.assert_called_once_with((HOST, 443)) + sock2.connect.assert_called_once_with((HOST, 443)) # Make sure that the socket is closed after send fails. sock.close.assert_called_once() assert sock2.close.call_count == 0