diff --git a/.gitignore b/.gitignore index db3d538..a06dc67 100755 --- a/.gitignore +++ b/.gitignore @@ -46,3 +46,10 @@ _build .idea .vscode *~ + +# tox-specific files +.tox +build + +# coverage-specific files +.coverage diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 70ade69..7b20fd7 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -4,9 +4,14 @@ repos: - repo: https://github.com/python/black - rev: 23.3.0 + rev: 24.2.0 hooks: - id: black + - repo: https://github.com/PyCQA/isort + rev: 5.13.2 + hooks: + - id: isort + args: ["--profile", "black", "--filter-files"] - repo: https://github.com/fsfe/reuse-tool rev: v1.1.2 hooks: @@ -32,11 +37,11 @@ repos: types: [python] files: "^examples/" args: - - --disable=missing-docstring,invalid-name,consider-using-f-string,duplicate-code + - --disable=consider-using-f-string,duplicate-code,missing-docstring,invalid-name - id: pylint name: pylint (test code) description: Run pylint rules on "tests/*.py" files types: [python] files: "^tests/" args: - - --disable=missing-docstring,consider-using-f-string,duplicate-code + - --disable=consider-using-f-string,duplicate-code,missing-docstring,invalid-name,protected-access,redefined-outer-name diff --git a/adafruit_requests.py b/adafruit_requests.py index c700a8d..771bfcc 100644 --- a/adafruit_requests.py +++ b/adafruit_requests.py @@ -31,103 +31,29 @@ * Adafruit CircuitPython firmware for the supported boards: https://github.com/adafruit/circuitpython/releases +* Adafruit's Connection Manager library: + https://github.com/adafruit/Adafruit_CircuitPython_ConnectionManager + """ __version__ = "0.0.0+auto.0" __repo__ = "https://github.com/adafruit/Adafruit_CircuitPython_Requests.git" import errno +import json as json_module import sys -import json as json_module +from adafruit_connection_manager import get_connection_manager if not sys.implementation.name == "circuitpython": - from ssl import SSLContext - from types import ModuleType, TracebackType - from typing import Any, Dict, Optional, Tuple, Type, Union - - try: - from typing import Protocol - except ImportError: - from typing_extensions import Protocol - - # Based on https://github.com/python/typeshed/blob/master/stdlib/_socket.pyi - class CommonSocketType(Protocol): - """Describes the common structure every socket type must have.""" - - def send(self, data: bytes, flags: int = ...) -> None: - """Send data to the socket. The meaning of the optional flags kwarg is - implementation-specific.""" - - def settimeout(self, value: Optional[float]) -> None: - """Set a timeout on blocking socket operations.""" - - def close(self) -> None: - """Close the socket.""" - - class CommonCircuitPythonSocketType(CommonSocketType, Protocol): - """Describes the common structure every CircuitPython socket type must have.""" - - def connect( - self, - address: Tuple[str, int], - conntype: Optional[int] = ..., - ) -> None: - """Connect to a remote socket at the provided (host, port) address. The conntype - kwarg optionally may indicate SSL or not, depending on the underlying interface. - """ - - class SupportsRecvWithFlags(Protocol): - """Describes a type that posseses a socket recv() method supporting the flags kwarg.""" - - def recv(self, bufsize: int = ..., flags: int = ...) -> bytes: - """Receive data from the socket. The return value is a bytes object representing - the data received. The maximum amount of data to be received at once is specified - by bufsize. The meaning of the optional flags kwarg is implementation-specific. - """ - - class SupportsRecvInto(Protocol): - """Describes a type that possesses a socket recv_into() method.""" - - def recv_into( - self, buffer: bytearray, nbytes: int = ..., flags: int = ... - ) -> int: - """Receive up to nbytes bytes from the socket, storing the data into the provided - buffer. If nbytes is not specified (or 0), receive up to the size available in the - given buffer. The meaning of the optional flags kwarg is implementation-specific. - Returns the number of bytes received.""" - - class CircuitPythonSocketType( - CommonCircuitPythonSocketType, - SupportsRecvInto, - SupportsRecvWithFlags, - Protocol, - ): # pylint: disable=too-many-ancestors - """Describes the structure every modern CircuitPython socket type must have.""" - - class StandardPythonSocketType( - CommonSocketType, SupportsRecvInto, SupportsRecvWithFlags, Protocol - ): - """Describes the structure every standard Python socket type must have.""" - - def connect(self, address: Union[Tuple[Any, ...], str, bytes]) -> None: - """Connect to a remote socket at the provided address.""" - - SocketType = Union[ - CircuitPythonSocketType, - StandardPythonSocketType, - ] - - SocketpoolModuleType = ModuleType - - class InterfaceType(Protocol): - """Describes the structure every interface type must have.""" + from types import TracebackType + from typing import Any, Dict, Optional, Type - @property - def TLS_MODE(self) -> int: # pylint: disable=invalid-name - """Constant representing that a socket's connection mode is TLS.""" - - SSLContextType = Union[SSLContext, "_FakeSSLContext"] + from circuitpython_typing.socket import ( + SocketpoolModuleType, + SocketType, + SSLContextType, + ) class _RawResponse: @@ -160,7 +86,7 @@ class Response: encoding = None - def __init__(self, sock: SocketType, session: Optional["Session"] = None) -> None: + def __init__(self, sock: SocketType, session: "Session") -> None: self.socket = sock self.encoding = "utf-8" self._cached = None @@ -175,10 +101,7 @@ def __init__(self, sock: SocketType, session: Optional["Session"] = None) -> Non http = self._readto(b" ") if not http: - if session: - session._close_socket(self.socket) - else: - self.socket.close() + session._connection_manager.close_socket(self.socket) raise RuntimeError("Unable to read HTTP response.") self.status_code: int = int(bytes(self._readto(b" "))) """The status code returned by the server""" @@ -320,7 +243,8 @@ def close(self) -> None: self._throw_away(chunk_size + 2) self._parse_headers() if self._session: - self._session._free_socket(self.socket) # pylint: disable=protected-access + # pylint: disable=protected-access + self._session._connection_manager.free_socket(self.socket) else: self.socket.close() self.socket = None @@ -436,99 +360,13 @@ def __init__( self, socket_pool: SocketpoolModuleType, ssl_context: Optional[SSLContextType] = None, + session_id: Optional[str] = None, ) -> None: - self._socket_pool = socket_pool + self._connection_manager = get_connection_manager(socket_pool) self._ssl_context = ssl_context - # Hang onto open sockets so that we can reuse them. - self._open_sockets = {} - self._socket_free = {} + self._session_id = session_id self._last_response = None - def _free_socket(self, socket: SocketType) -> None: - if socket not in self._open_sockets.values(): - raise RuntimeError("Socket not from session") - self._socket_free[socket] = True - - def _close_socket(self, sock: SocketType) -> None: - sock.close() - del self._socket_free[sock] - key = None - for k in self._open_sockets: # pylint: disable=consider-using-dict-items - if self._open_sockets[k] == sock: - key = k - break - if key: - del self._open_sockets[key] - - def _free_sockets(self) -> None: - free_sockets = [] - for sock, val in self._socket_free.items(): - if val: - free_sockets.append(sock) - for sock in free_sockets: - self._close_socket(sock) - - def _get_socket( - self, host: str, port: int, proto: str, *, timeout: float = 1 - ) -> CircuitPythonSocketType: - # pylint: disable=too-many-branches - key = (host, port, proto) - if key in self._open_sockets: - sock = self._open_sockets[key] - if self._socket_free[sock]: - self._socket_free[sock] = False - return sock - if proto == "https:" and not self._ssl_context: - raise RuntimeError( - "ssl_context must be set before using adafruit_requests for https" - ) - addr_info = self._socket_pool.getaddrinfo( - host, port, 0, self._socket_pool.SOCK_STREAM - )[0] - retry_count = 0 - sock = None - last_exc = None - while retry_count < 5 and sock is None: - if retry_count > 0: - if any(self._socket_free.items()): - self._free_sockets() - else: - raise RuntimeError("Sending request failed") from last_exc - retry_count += 1 - - try: - sock = self._socket_pool.socket(addr_info[0], addr_info[1]) - except OSError as exc: - last_exc = exc - continue - except RuntimeError as exc: - last_exc = exc - continue - - connect_host = addr_info[-1][0] - if proto == "https:": - sock = self._ssl_context.wrap_socket(sock, server_hostname=host) - connect_host = host - sock.settimeout(timeout) # socket read timeout - - try: - sock.connect((connect_host, port)) - except MemoryError as exc: - last_exc = exc - sock.close() - sock = None - except OSError as exc: - last_exc = exc - sock.close() - sock = None - - if sock is None: - raise RuntimeError("Repeated socket failures") from last_exc - - self._open_sockets[key] = sock - self._socket_free[sock] = False - return sock - @staticmethod def _check_headers(headers: Dict[str, str]): if not isinstance(headers, dict): @@ -688,7 +526,14 @@ def request( last_exc = None while retry_count < 2: retry_count += 1 - socket = self._get_socket(host, port, proto, timeout=timeout) + socket = self._connection_manager.get_socket( + host, + port, + proto, + session_id=self._session_id, + timeout=timeout, + ssl_context=self._ssl_context, + ) ok = True try: self._send_request(socket, host, method, path, headers, data, json) @@ -709,7 +554,7 @@ def request( if result == b"H": # Things seem to be ok so break with socket set. break - self._close_socket(socket) + self._connection_manager.close_socket(socket) socket = None if not socket: @@ -766,103 +611,3 @@ def patch(self, url: str, **kw) -> Response: def delete(self, url: str, **kw) -> Response: """Send HTTP DELETE request""" return self.request("DELETE", url, **kw) - - -# Backwards compatible API: - -_default_session = None # pylint: disable=invalid-name - - -class _FakeSSLSocket: - def __init__(self, socket: CircuitPythonSocketType, tls_mode: int) -> None: - self._socket = socket - self._mode = tls_mode - self.settimeout = socket.settimeout - self.send = socket.send - self.recv = socket.recv - self.close = socket.close - self.recv_into = socket.recv_into - - def connect(self, address: Tuple[str, int]) -> None: - """connect wrapper to add non-standard mode parameter""" - try: - return self._socket.connect(address, self._mode) - except RuntimeError as error: - raise OSError(errno.ENOMEM) from error - - -class _FakeSSLContext: - def __init__(self, iface: InterfaceType) -> None: - self._iface = iface - - def wrap_socket( - self, socket: CircuitPythonSocketType, server_hostname: Optional[str] = None - ) -> _FakeSSLSocket: - """Return the same socket""" - # pylint: disable=unused-argument - return _FakeSSLSocket(socket, self._iface.TLS_MODE) - - -def set_socket( - sock: SocketpoolModuleType, iface: Optional[InterfaceType] = None -) -> None: - """Legacy API for setting the socket and network interface. Use a `Session` instead.""" - global _default_session # pylint: disable=global-statement,invalid-name - 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: str, - url: str, - data: Optional[Any] = None, - json: Optional[Any] = None, - headers: Optional[Dict[str, str]] = None, - stream: bool = False, - timeout: float = 1, -) -> None: - """Send HTTP request""" - # pylint: disable=too-many-arguments - _default_session.request( - method, - url, - data=data, - json=json, - headers=headers, - stream=stream, - timeout=timeout, - ) - - -def head(url: str, **kw): - """Send HTTP HEAD request""" - return _default_session.request("HEAD", url, **kw) - - -def get(url: str, **kw): - """Send HTTP GET request""" - return _default_session.request("GET", url, **kw) - - -def post(url: str, **kw): - """Send HTTP POST request""" - return _default_session.request("POST", url, **kw) - - -def put(url: str, **kw): - """Send HTTP PUT request""" - return _default_session.request("PUT", url, **kw) - - -def patch(url: str, **kw): - """Send HTTP PATCH request""" - return _default_session.request("PATCH", url, **kw) - - -def delete(url: str, **kw): - """Send HTTP DELETE request""" - return _default_session.request("DELETE", url, **kw) diff --git a/docs/conf.py b/docs/conf.py index ed4d3f1..4e19482 100755 --- a/docs/conf.py +++ b/docs/conf.py @@ -4,9 +4,9 @@ # # SPDX-License-Identifier: MIT +import datetime import os import sys -import datetime sys.path.insert(0, os.path.abspath("..")) diff --git a/docs/examples.rst b/docs/examples.rst index 4c95536..3637216 100755 --- a/docs/examples.rst +++ b/docs/examples.rst @@ -1,8 +1,32 @@ -Simple test ------------- +Examples +======== -Ensure your device works with this simple test. +Below are a few examples, for use with common boards. There are more in the examples folder of the library -.. literalinclude:: ../examples/requests_simpletest.py - :caption: examples/requests_simpletest.py +On-board WiFi +-------------- + +.. literalinclude:: ../examples/wifi/requests_wifi_simpletest.py + :caption: examples/wifi/requests_wifi_simpletest.py + :linenos: + +ESP32SPI +-------------- + +.. literalinclude:: ../examples/esp32spi/requests_esp32spi_simpletest.py + :caption: examples/esp32spi/requests_esp32spi_simpletest.py + :linenos: + +WIZNET5K +-------------- + +.. literalinclude:: ../examples/wiznet5k/requests_wiznet5k_simpletest.py + :caption: examples/wiznet5k/requests_wiznet5k_simpletest.py + :linenos: + +Fona +-------------- + +.. literalinclude:: ../examples/fona/requests_fona_simpletest.py + :caption: examples/fona/requests_fona_simpletest.py :linenos: diff --git a/examples/requests_advanced_cpython.py b/examples/cpython/requests_cpython_advanced.py similarity index 71% rename from examples/requests_advanced_cpython.py rename to examples/cpython/requests_cpython_advanced.py index 6a9e109..89d715b 100644 --- a/examples/requests_advanced_cpython.py +++ b/examples/cpython/requests_cpython_advanced.py @@ -1,18 +1,21 @@ # SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries # SPDX-License-Identifier: MIT -import socket +import socket as pool +import ssl + import adafruit_requests -http = adafruit_requests.Session(socket) +# Initialize a requests session +requests = adafruit_requests.Session(pool, ssl.create_default_context()) -JSON_GET_URL = "http://httpbin.org/get" +JSON_GET_URL = "https://httpbin.org/get" # Define a custom header as a dict. headers = {"user-agent": "blinka/1.0.0"} print("Fetching JSON data from %s..." % JSON_GET_URL) -response = http.get(JSON_GET_URL, headers=headers) +response = requests.get(JSON_GET_URL, headers=headers) print("-" * 60) json_data = response.json() @@ -24,8 +27,5 @@ print("Response HTTP Status Code: ", response.status_code) print("-" * 60) -# Read Response, as raw bytes instead of pretty text -print("Raw Response: ", response.content) - # Close, delete and collect the response data response.close() diff --git a/examples/requests_https_cpython.py b/examples/cpython/requests_cpython_simpletest.py old mode 100755 new mode 100644 similarity index 59% rename from examples/requests_https_cpython.py rename to examples/cpython/requests_cpython_simpletest.py index 39bd6dc..70bdfde --- a/examples/requests_https_cpython.py +++ b/examples/cpython/requests_cpython_simpletest.py @@ -1,48 +1,52 @@ # SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries # SPDX-License-Identifier: MIT -# adafruit_requests usage with a CPython socket -import socket +import socket as pool import ssl -import adafruit_requests as requests -https = requests.Session(socket, ssl.create_default_context()) +import adafruit_requests -TEXT_URL = "https://httpbin.org/get" +# Initialize a requests session +requests = adafruit_requests.Session(pool, ssl.create_default_context()) + +TEXT_URL = "http://wifitest.adafruit.com/testwifi/index.html" 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) -# print("-" * 40) +print("Fetching text from %s" % TEXT_URL) +response = requests.get(TEXT_URL) +print("-" * 40) -# print("Text Response: ", response.text) -# print("-" * 40) -# response.close() +print("Text Response: ", response.text) +print("-" * 40) +response.close() print("Fetching JSON data from %s" % JSON_GET_URL) -response = https.get(JSON_GET_URL) +response = requests.get(JSON_GET_URL) print("-" * 40) print("JSON Response: ", response.json()) print("-" * 40) +response.close() data = "31F" print("POSTing data to {0}: {1}".format(JSON_POST_URL, data)) -response = https.post(JSON_POST_URL, data=data) +response = requests.post(JSON_POST_URL, data=data) print("-" * 40) json_resp = response.json() # Parse out the 'data' key from json_resp dict. print("Data received from server:", json_resp["data"]) print("-" * 40) +response.close() json_data = {"Date": "July 25, 2019"} print("POSTing data to {0}: {1}".format(JSON_POST_URL, json_data)) -response = https.post(JSON_POST_URL, json=json_data) +response = requests.post(JSON_POST_URL, json=json_data) print("-" * 40) json_resp = response.json() # Parse out the 'json' key from json_resp dict. print("JSON Data received from server:", json_resp["json"]) print("-" * 40) +response.close() diff --git a/examples/requests_advanced.py b/examples/esp32spi/requests_esp32spi_advanced.py similarity index 58% rename from examples/requests_advanced.py rename to examples/esp32spi/requests_esp32spi_advanced.py index 65ec126..3b0bb41 100644 --- a/examples/requests_advanced.py +++ b/examples/esp32spi/requests_esp32spi_advanced.py @@ -1,22 +1,20 @@ # SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries # SPDX-License-Identifier: MIT +import os + +import adafruit_connection_manager +import adafruit_esp32spi.adafruit_esp32spi_socket as pool import board import busio -from digitalio import DigitalInOut -import adafruit_esp32spi.adafruit_esp32spi_socket as socket from adafruit_esp32spi import adafruit_esp32spi -import adafruit_requests as requests - -# Add a secrets.py to your filesystem that has a dictionary called secrets with "ssid" and -# "password" keys with your WiFi credentials. DO NOT share that file or commit it into Git or other -# source control. -# pylint: disable=no-name-in-module,wrong-import-order -try: - from secrets import secrets -except ImportError: - print("WiFi secrets are kept in secrets.py, please add them there!") - raise +from digitalio import DigitalInOut + +import adafruit_requests + +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") # If you are using a board with pre-defined ESP32 Pins: esp32_cs = DigitalInOut(board.ESP_CS) @@ -28,23 +26,28 @@ # 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) +radio = adafruit_esp32spi.ESP_SPIcontrol(spi, esp32_cs, esp32_ready, esp32_reset) print("Connecting to AP...") -while not esp.is_connected: +while not radio.is_connected: try: - esp.connect_AP(secrets["ssid"], secrets["password"]) + radio.connect_AP(ssid, password) except RuntimeError as e: print("could not connect to AP, retrying: ", e) continue -print("Connected to", str(esp.ssid, "utf-8"), "\tRSSI:", esp.rssi) +print("Connected to", str(radio.ssid, "utf-8"), "\tRSSI:", radio.rssi) -# Initialize a requests object with a socket and esp32spi interface -socket.set_interface(esp) -requests.set_socket(socket, esp) +# Initialize a requests session +ssl_context = adafruit_connection_manager.create_fake_ssl_context(pool, radio) +requests = adafruit_requests.Session(pool, ssl_context) -JSON_GET_URL = "http://httpbin.org/get" +JSON_GET_URL = "https://httpbin.org/get" # Define a custom header as a dict. headers = {"user-agent": "blinka/1.0.0"} diff --git a/examples/requests_simpletest.py b/examples/esp32spi/requests_esp32spi_simpletest.py old mode 100755 new mode 100644 similarity index 69% rename from examples/requests_simpletest.py rename to examples/esp32spi/requests_esp32spi_simpletest.py index 14cf66c..01d8204 --- a/examples/requests_simpletest.py +++ b/examples/esp32spi/requests_esp32spi_simpletest.py @@ -1,23 +1,20 @@ # SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries # SPDX-License-Identifier: MIT -# adafruit_requests usage with an esp32spi_socket +import os + +import adafruit_connection_manager +import adafruit_esp32spi.adafruit_esp32spi_socket as pool import board import busio -from digitalio import DigitalInOut -import adafruit_esp32spi.adafruit_esp32spi_socket as socket from adafruit_esp32spi import adafruit_esp32spi -import adafruit_requests as requests - -# Add a secrets.py to your filesystem that has a dictionary called secrets with "ssid" and -# "password" keys with your WiFi credentials. DO NOT share that file or commit it into Git or other -# source control. -# pylint: disable=no-name-in-module,wrong-import-order -try: - from secrets import secrets -except ImportError: - print("WiFi secrets are kept in secrets.py, please add them there!") - raise +from digitalio import DigitalInOut + +import adafruit_requests + +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") # If you are using a board with pre-defined ESP32 Pins: esp32_cs = DigitalInOut(board.ESP_CS) @@ -35,20 +32,20 @@ # 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) +radio = adafruit_esp32spi.ESP_SPIcontrol(spi, esp32_cs, esp32_ready, esp32_reset) print("Connecting to AP...") -while not esp.is_connected: +while not radio.is_connected: try: - esp.connect_AP(secrets["ssid"], secrets["password"]) + radio.connect_AP(ssid, password) except RuntimeError as e: print("could not connect to AP, retrying: ", e) continue -print("Connected to", str(esp.ssid, "utf-8"), "\tRSSI:", esp.rssi) +print("Connected to", str(radio.ssid, "utf-8"), "\tRSSI:", radio.rssi) -# Initialize a requests object with a socket and esp32spi interface -socket.set_interface(esp) -requests.set_socket(socket, esp) +# Initialize a requests session +ssl_context = adafruit_connection_manager.create_fake_ssl_context(pool, radio) +requests = adafruit_requests.Session(pool, ssl_context) TEXT_URL = "http://wifitest.adafruit.com/testwifi/index.html" JSON_GET_URL = "https://httpbin.org/get" diff --git a/examples/requests_advanced_cellular.py b/examples/fona/requests_fona_advanced.py old mode 100755 new mode 100644 similarity index 60% rename from examples/requests_advanced_cellular.py rename to examples/fona/requests_fona_advanced.py index ea8d512..33bc6f0 --- a/examples/requests_advanced_cellular.py +++ b/examples/fona/requests_fona_advanced.py @@ -1,38 +1,37 @@ # SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries # SPDX-License-Identifier: MIT -# pylint: disable=unused-import +import os import time + +import adafruit_connection_manager +import adafruit_fona.adafruit_fona_network as network +import adafruit_fona.adafruit_fona_socket as pool import board import busio import digitalio -from adafruit_fona.adafruit_fona import FONA -from adafruit_fona.fona_3g import FONA3G -import adafruit_fona.adafruit_fona_network as network -import adafruit_fona.adafruit_fona_socket as cellular_socket -import adafruit_requests as requests +from adafruit_fona.adafruit_fona import FONA # pylint: disable=unused-import +from adafruit_fona.fona_3g import FONA3G # pylint: disable=unused-import -# Get GPRS details and more from a secrets.py file -try: - from secrets import secrets -except ImportError: - print("GPRS secrets are kept in secrets.py, please add them there!") - raise +import adafruit_requests + +# Get GPRS details, ensure these are setup in settings.toml +apn = os.getenv("APN") +apn_username = os.getenv("APN_USERNAME") +apn_password = os.getenv("APN_PASSWORD") # Create a serial connection for the FONA connection uart = busio.UART(board.TX, board.RX) -rst = digitalio.DigitalInOut(board.D9) +rst = digitalio.DigitalInOut(board.D4) # Use this for FONA800 and FONA808 -# fona = FONA(uart, rst) +radio = FONA(uart, rst) # Use this for FONA3G -fona = FONA3G(uart, rst) +# radio = FONA3G(uart, rst) # Initialize cellular data network -network = network.CELLULAR( - fona, (secrets["apn"], secrets["apn_username"], secrets["apn_password"]) -) +network = network.CELLULAR(radio, (apn, apn_username, apn_password)) while not network.is_attached: print("Attaching to network...") @@ -45,8 +44,9 @@ time.sleep(0.5) print("Network Connected!") -# Initialize a requests object with a socket and cellular interface -requests.set_socket(cellular_socket, fona) +# Initialize a requests session +ssl_context = adafruit_connection_manager.create_fake_ssl_context(pool, radio) +requests = adafruit_requests.Session(pool, ssl_context) JSON_GET_URL = "http://httpbin.org/get" @@ -66,8 +66,5 @@ print("Response HTTP Status Code: ", response.status_code) print("-" * 60) -# Read Response, as raw bytes instead of pretty text -print("Raw Response: ", response.content) - # Close, delete and collect the response data response.close() diff --git a/examples/requests_simpletest_cellular.py b/examples/fona/requests_fona_simpletest.py old mode 100755 new mode 100644 similarity index 71% rename from examples/requests_simpletest_cellular.py rename to examples/fona/requests_fona_simpletest.py index 6b4d1a7..8841d3d --- a/examples/requests_simpletest_cellular.py +++ b/examples/fona/requests_fona_simpletest.py @@ -1,38 +1,37 @@ # SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries # SPDX-License-Identifier: MIT -# pylint: disable=unused-import +import os import time + +import adafruit_connection_manager +import adafruit_fona.adafruit_fona_network as network +import adafruit_fona.adafruit_fona_socket as pool import board import busio import digitalio -from adafruit_fona.adafruit_fona import FONA -from adafruit_fona.fona_3g import FONA3G -import adafruit_fona.adafruit_fona_network as network -import adafruit_fona.adafruit_fona_socket as cellular_socket -import adafruit_requests as requests +from adafruit_fona.adafruit_fona import FONA # pylint: disable=unused-import +from adafruit_fona.fona_3g import FONA3G # pylint: disable=unused-import + +import adafruit_requests -# Get GPRS details and more from a secrets.py file -try: - from secrets import secrets -except ImportError: - print("GPRS secrets are kept in secrets.py, please add them there!") - raise +# Get GPRS details, ensure these are setup in settings.toml +apn = os.getenv("APN") +apn_username = os.getenv("APN_USERNAME") +apn_password = os.getenv("APN_PASSWORD") # Create a serial connection for the FONA connection uart = busio.UART(board.TX, board.RX) rst = digitalio.DigitalInOut(board.D4) # Use this for FONA800 and FONA808 -fona = FONA(uart, rst) +radio = FONA(uart, rst) # Use this for FONA3G -# fona = FONA3G(uart, rst) +# radio = FONA3G(uart, rst) # Initialize cellular data network -network = network.CELLULAR( - fona, (secrets["apn"], secrets["apn_username"], secrets["apn_password"]) -) +network = network.CELLULAR(radio, (apn, apn_username, apn_password)) while not network.is_attached: print("Attaching to network...") @@ -45,8 +44,9 @@ time.sleep(0.5) print("Network Connected!") -# Initialize a requests object with a socket and cellular interface -requests.set_socket(cellular_socket, fona) +# Initialize a requests session +ssl_context = adafruit_connection_manager.create_fake_ssl_context(pool, radio) +requests = adafruit_requests.Session(pool, ssl_context) TEXT_URL = "http://wifitest.adafruit.com/testwifi/index.html" JSON_GET_URL = "http://httpbin.org/get" diff --git a/examples/requests_advanced_ethernet.py b/examples/requests_advanced_ethernet.py deleted file mode 100644 index e9ecf57..0000000 --- a/examples/requests_advanced_ethernet.py +++ /dev/null @@ -1,58 +0,0 @@ -# SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries -# SPDX-License-Identifier: MIT - -import board -import busio -from digitalio import DigitalInOut -from adafruit_wiznet5k.adafruit_wiznet5k import WIZNET5K -import adafruit_wiznet5k.adafruit_wiznet5k_socket as socket -import adafruit_requests as requests - -cs = DigitalInOut(board.D10) -spi_bus = busio.SPI(board.SCK, MOSI=board.MOSI, MISO=board.MISO) - -# Initialize ethernet interface with DHCP -eth = WIZNET5K(spi_bus, cs) - -# Initialize a requests object with a socket and ethernet interface -requests.set_socket(socket, eth) - -JSON_GET_URL = "http://httpbin.org/get" - -attempts = 3 # Number of attempts to retry each request -failure_count = 0 -response = None - -# Define a custom header as a dict. -headers = {"user-agent": "blinka/1.0.0"} - -print("Fetching JSON data from %s..." % JSON_GET_URL) -while not response: - try: - response = requests.get(JSON_GET_URL, headers=headers) - failure_count = 0 - except AssertionError as error: - print("Request failed, retrying...\n", error) - failure_count += 1 - if failure_count >= attempts: - raise AssertionError( - "Failed to resolve hostname, \ - please check your router's DNS configuration." - ) from error - continue -print("-" * 60) - -json_data = response.json() -headers = json_data["headers"] -print("Response's Custom User-Agent Header: {0}".format(headers["User-Agent"])) -print("-" * 60) - -# Read Response's HTTP status code -print("Response HTTP Status Code: ", response.status_code) -print("-" * 60) - -# Read Response, as raw bytes instead of pretty text -print("Raw Response: ", response.content) - -# Close, delete and collect the response data -response.close() diff --git a/examples/requests_github_cpython.py b/examples/requests_github_cpython.py deleted file mode 100755 index 88d7135..0000000 --- a/examples/requests_github_cpython.py +++ /dev/null @@ -1,16 +0,0 @@ -# SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries -# SPDX-License-Identifier: MIT - -# adafruit_requests usage with a CPython socket -import socket -import ssl -import adafruit_requests - -http = adafruit_requests.Session(socket, ssl.create_default_context()) - -print("Getting CircuitPython star count") -headers = {"Transfer-Encoding": "chunked"} -response = http.get( - "https://api.github.com/repos/adafruit/circuitpython", headers=headers -) -print("circuitpython stars", response.json()["stargazers_count"]) diff --git a/examples/requests_https_circuitpython.py b/examples/requests_https_circuitpython.py deleted file mode 100755 index 026aa2d..0000000 --- a/examples/requests_https_circuitpython.py +++ /dev/null @@ -1,65 +0,0 @@ -# SPDX-FileCopyrightText: 2021 jfabernathy for Adafruit Industries -# SPDX-License-Identifier: MIT - -# adafruit_requests usage with a CircuitPython socket -# this has been tested with Adafruit Metro ESP32-S2 Express - -import ssl -import wifi -import socketpool - -import adafruit_requests as requests - - -# Get wifi details and more from a secrets.py file -try: - from secrets import secrets -except ImportError: - print("WiFi secrets are kept in secrets.py, please add them there!") - raise - -print("Connecting to %s" % secrets["ssid"]) -wifi.radio.connect(secrets["ssid"], secrets["password"]) -print("Connected to %s!" % secrets["ssid"]) -print("My IP address is", wifi.radio.ipv4_address) - -socket = socketpool.SocketPool(wifi.radio) -https = requests.Session(socket, ssl.create_default_context()) - -TEXT_URL = "https://httpbin.org/get" -JSON_GET_URL = "https://httpbin.org/get" -JSON_POST_URL = "https://httpbin.org/post" - -print("Fetching text from %s" % TEXT_URL) -response = https.get(TEXT_URL) -print("-" * 40) -print("Text Response: ", response.text) -print("-" * 40) -response.close() - -print("Fetching JSON data from %s" % JSON_GET_URL) -response = https.get(JSON_GET_URL) -print("-" * 40) - -print("JSON Response: ", response.json()) -print("-" * 40) - -data = "31F" -print("POSTing data to {0}: {1}".format(JSON_POST_URL, data)) -response = https.post(JSON_POST_URL, data=data) -print("-" * 40) - -json_resp = response.json() -# Parse out the 'data' key from json_resp dict. -print("Data received from server:", json_resp["data"]) -print("-" * 40) - -json_data = {"Date": "July 25, 2019"} -print("POSTing data to {0}: {1}".format(JSON_POST_URL, json_data)) -response = https.post(JSON_POST_URL, json=json_data) -print("-" * 40) - -json_resp = response.json() -# Parse out the 'json' key from json_resp dict. -print("JSON Data received from server:", json_resp["json"]) -print("-" * 40) diff --git a/examples/requests_simpletest_ethernet.py b/examples/requests_simpletest_ethernet.py deleted file mode 100644 index a432bea..0000000 --- a/examples/requests_simpletest_ethernet.py +++ /dev/null @@ -1,115 +0,0 @@ -# SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries -# SPDX-License-Identifier: MIT - -import board -import busio -from digitalio import DigitalInOut -from adafruit_wiznet5k.adafruit_wiznet5k import WIZNET5K -import adafruit_wiznet5k.adafruit_wiznet5k_socket as socket -import adafruit_requests as requests - -cs = DigitalInOut(board.D10) -spi_bus = busio.SPI(board.SCK, MOSI=board.MOSI, MISO=board.MISO) - -# Initialize ethernet interface with DHCP -eth = WIZNET5K(spi_bus, cs) - -# Initialize a requests object with a socket and ethernet interface -requests.set_socket(socket, eth) - -TEXT_URL = "http://wifitest.adafruit.com/testwifi/index.html" -JSON_GET_URL = "http://httpbin.org/get" -JSON_POST_URL = "http://httpbin.org/post" - -attempts = 3 # Number of attempts to retry each request -failure_count = 0 -response = None - -print("Fetching text from %s" % TEXT_URL) -while not response: - try: - response = requests.get(TEXT_URL) - failure_count = 0 - except AssertionError as error: - print("Request failed, retrying...\n", error) - failure_count += 1 - if failure_count >= attempts: - raise AssertionError( - "Failed to resolve hostname, \ - please check your router's DNS configuration." - ) from error - continue -print("-" * 40) - -print("Text Response: ", response.text) -print("-" * 40) -response.close() -response = None - -print("Fetching JSON data from %s" % JSON_GET_URL) -while not response: - try: - response = requests.get(JSON_GET_URL) - failure_count = 0 - except AssertionError as error: - print("Request failed, retrying...\n", error) - failure_count += 1 - if failure_count >= attempts: - raise AssertionError( - "Failed to resolve hostname, \ - please check your router's DNS configuration." - ) from error - continue -print("-" * 40) - -print("JSON Response: ", response.json()) -print("-" * 40) -response.close() -response = None - -data = "31F" -print("POSTing data to {0}: {1}".format(JSON_POST_URL, data)) -while not response: - try: - response = requests.post(JSON_POST_URL, data=data) - failure_count = 0 - except AssertionError as error: - print("Request failed, retrying...\n", error) - failure_count += 1 - if failure_count >= attempts: - raise AssertionError( - "Failed to resolve hostname, \ - please check your router's DNS configuration." - ) from error - continue -print("-" * 40) - -json_resp = response.json() -# Parse out the 'data' key from json_resp dict. -print("Data received from server:", json_resp["data"]) -print("-" * 40) -response.close() -response = None - -json_data = {"Date": "July 25, 2019"} -print("POSTing data to {0}: {1}".format(JSON_POST_URL, json_data)) -while not response: - try: - response = requests.post(JSON_POST_URL, json=json_data) - failure_count = 0 - except AssertionError as error: - print("Request failed, retrying...\n", error) - failure_count += 1 - if failure_count >= attempts: - raise AssertionError( - "Failed to resolve hostname, \ - please check your router's DNS configuration." - ) from error - continue -print("-" * 40) - -json_resp = response.json() -# Parse out the 'json' key from json_resp dict. -print("JSON Data received from server:", json_resp["json"]) -print("-" * 40) -response.close() diff --git a/examples/requests_adafruit_discord_active_online.py b/examples/wifi/expanded/requests_wifi_adafruit_discord_active_online.py similarity index 94% rename from examples/requests_adafruit_discord_active_online.py rename to examples/wifi/expanded/requests_wifi_adafruit_discord_active_online.py index 48a201c..d43f73f 100644 --- a/examples/requests_adafruit_discord_active_online.py +++ b/examples/wifi/expanded/requests_wifi_adafruit_discord_active_online.py @@ -5,12 +5,14 @@ requests_adafruit_discord_active_online """ import gc +import json import os -import time import ssl -import json -import wifi +import time + import socketpool +import wifi + import adafruit_requests # Public API. No user or token required @@ -24,9 +26,9 @@ # 600 = 10 mins, 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour sleep_time = 900 -# this example uses settings.toml for credentials -ssid = os.getenv("WIFI_SSID") -appw = os.getenv("WIFI_PASSWORD") +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") # Converts seconds to minutes/hours/days @@ -56,7 +58,7 @@ def time_calc(input_time): requests = adafruit_requests.Session(pool, ssl.create_default_context()) while not wifi.radio.ipv4_address: try: - wifi.radio.connect(ssid, appw) + wifi.radio.connect(ssid, password) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") diff --git a/examples/requests_api_discord.py b/examples/wifi/expanded/requests_wifi_api_discord.py similarity index 95% rename from examples/requests_api_discord.py rename to examples/wifi/expanded/requests_wifi_api_discord.py index b6e0348..91b2b6c 100644 --- a/examples/requests_api_discord.py +++ b/examples/wifi/expanded/requests_wifi_api_discord.py @@ -2,12 +2,14 @@ # SPDX-License-Identifier: MIT # Coded for Circuit Python 8.2 # DJDevon3 Adafruit Feather ESP32-S3 Discord API Example +import json import os -import time import ssl -import json -import wifi +import time + import socketpool +import wifi + import adafruit_requests # Active Logged in User Account Required, no tokens required @@ -17,9 +19,9 @@ # Ensure this is in settings.toml # "Discord_Authorization": "Request Header Auth here" -# Uses settings.toml for credentials +# Get WiFi details, ensure these are setup in settings.toml ssid = os.getenv("CIRCUITPY_WIFI_SSID") -appw = os.getenv("CIRCUITPY_WIFI_PASSWORD") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") Discord_Auth = os.getenv("Discord_Authorization") # Initialize WiFi Pool (There can be only 1 pool & top of script) @@ -60,7 +62,7 @@ def time_calc(input_time): # input_time in seconds requests = adafruit_requests.Session(pool, ssl.create_default_context()) while not wifi.radio.ipv4_address: try: - wifi.radio.connect(ssid, appw) + wifi.radio.connect(ssid, password) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") diff --git a/examples/requests_api_fitbit.py b/examples/wifi/expanded/requests_wifi_api_fitbit.py similarity index 98% rename from examples/requests_api_fitbit.py rename to examples/wifi/expanded/requests_wifi_api_fitbit.py index 6494237..022ee22 100644 --- a/examples/requests_api_fitbit.py +++ b/examples/wifi/expanded/requests_wifi_api_fitbit.py @@ -3,11 +3,13 @@ # Coded for Circuit Python 8.2 import os -import time import ssl -import wifi -import socketpool +import time + import microcontroller +import socketpool +import wifi + import adafruit_requests # Initialize WiFi Pool (There can be only 1 pool & top of script) @@ -34,6 +36,10 @@ # Fitbit_First_Refresh_Token = "64 character string" # Fitbit_UserID = "UserID authorizing the ClientID" +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") + Fitbit_ClientID = os.getenv("Fitbit_ClientID") Fitbit_Token = os.getenv("Fitbit_Token") Fitbit_First_Refresh_Token = os.getenv( @@ -41,9 +47,6 @@ ) # overides nvm first run only Fitbit_UserID = os.getenv("Fitbit_UserID") -wifi_ssid = os.getenv("CIRCUITPY_WIFI_SSID") -wifi_pw = os.getenv("CIRCUITPY_WIFI_PASSWORD") - # Time between API refreshes # 300 = 5 mins, 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour sleep_time = 900 @@ -76,7 +79,7 @@ def time_calc(input_time): requests = adafruit_requests.Session(pool, ssl.create_default_context()) while not wifi.radio.ipv4_address: try: - wifi.radio.connect(wifi_ssid, wifi_pw) + wifi.radio.connect(ssid, password) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") diff --git a/examples/requests_api_github.py b/examples/wifi/expanded/requests_wifi_api_github.py similarity index 86% rename from examples/requests_api_github.py rename to examples/wifi/expanded/requests_wifi_api_github.py index a220df7..f1f86e3 100644 --- a/examples/requests_api_github.py +++ b/examples/wifi/expanded/requests_wifi_api_github.py @@ -3,17 +3,15 @@ # Coded for Circuit Python 8.0 """DJDevon3 Adafruit Feather ESP32-S2 Github_API_Example""" import gc -import time -import ssl import json -import wifi +import os +import ssl +import time + import socketpool -import adafruit_requests +import wifi -# Github developer token required. -# Ensure these are uncommented and in secrets.py or .env -# "Github_username": "Your Github Username", -# "Github_token": "Your long API token", +import adafruit_requests # Initialize WiFi Pool (There can be only 1 pool & top of script) pool = socketpool.SocketPool(wifi.radio) @@ -22,11 +20,12 @@ # 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour sleep_time = 900 -try: - from secrets import secrets -except ImportError: - print("Secrets File Import Error") - raise +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") +# Github developer token required. +github_username = os.getenv("Github_username") +github_token = os.getenv("Github_token") if sleep_time < 60: sleep_time_conversion = "seconds" @@ -41,8 +40,8 @@ sleep_int = sleep_time / 60 / 60 / 24 sleep_time_conversion = "days" -github_header = {"Authorization": " token " + secrets["Github_token"]} -GH_SOURCE = "https://api.github.com/users/" + secrets["Github_username"] +github_header = {"Authorization": " token " + github_token} +GH_SOURCE = "https://api.github.com/users/" + github_username # Connect to Wi-Fi print("\n===============================") @@ -50,7 +49,7 @@ requests = adafruit_requests.Session(pool, ssl.create_default_context()) while not wifi.radio.ipv4_address: try: - wifi.radio.connect(secrets["ssid"], secrets["password"]) + wifi.radio.connect(ssid, password) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") diff --git a/examples/requests_api_mastodon.py b/examples/wifi/expanded/requests_wifi_api_mastodon.py similarity index 95% rename from examples/requests_api_mastodon.py rename to examples/wifi/expanded/requests_wifi_api_mastodon.py index 893746c..56f3e8d 100644 --- a/examples/requests_api_mastodon.py +++ b/examples/wifi/expanded/requests_wifi_api_mastodon.py @@ -3,10 +3,13 @@ # Coded for Circuit Python 8.0 """DJDevon3 Adafruit Feather ESP32-S2 Mastodon_API_Example""" import gc -import time +import os import ssl -import wifi +import time + import socketpool +import wifi + import adafruit_requests # Mastodon V1 API - Public access (no dev creds or app required) @@ -27,11 +30,9 @@ # 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour sleep_time = 900 -try: - from secrets import secrets -except ImportError: - print("Secrets File Import Error") - raise +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") # Converts seconds in minutes/hours/days @@ -69,7 +70,7 @@ def time_calc(input_time): requests = adafruit_requests.Session(pool, ssl.create_default_context()) while not wifi.radio.ipv4_address: try: - wifi.radio.connect(secrets["ssid"], secrets["password"]) + wifi.radio.connect(ssid, password) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") diff --git a/examples/requests_api_openskynetwork_private.py b/examples/wifi/expanded/requests_wifi_api_openskynetwork_private.py similarity index 96% rename from examples/requests_api_openskynetwork_private.py rename to examples/wifi/expanded/requests_wifi_api_openskynetwork_private.py index a69eba8..486a4de 100644 --- a/examples/requests_api_openskynetwork_private.py +++ b/examples/wifi/expanded/requests_wifi_api_openskynetwork_private.py @@ -3,13 +3,15 @@ # Coded for Circuit Python 8.1 # DJDevon3 ESP32-S3 OpenSkyNetwork_Private_API_Example +import json import os -import time import ssl -import json -import wifi -import socketpool +import time + import circuitpython_base64 as base64 +import socketpool +import wifi + import adafruit_requests # OpenSky-Network.org Login required for this API @@ -28,10 +30,9 @@ # https://openskynetwork.github.io/opensky-api/rest.html#limitations sleep_time = 1800 -# this example uses settings.toml for credentials -# timezone offset is in seconds plus or minus GMT -ssid = os.getenv("AP_SSID") -appw = os.getenv("AP_PASSWORD") +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") osnu = os.getenv("OSN_Username") osnp = os.getenv("OSN_Password") @@ -91,7 +92,7 @@ def _format_datetime(datetime): request = adafruit_requests.Session(pool, ssl.create_default_context()) while not wifi.radio.ipv4_address: try: - wifi.radio.connect(ssid, appw) + wifi.radio.connect(ssid, password) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") diff --git a/examples/requests_api_openskynetwork_private_area.py b/examples/wifi/expanded/requests_wifi_api_openskynetwork_private_area.py similarity index 97% rename from examples/requests_api_openskynetwork_private_area.py rename to examples/wifi/expanded/requests_wifi_api_openskynetwork_private_area.py index e474e48..8cd08a2 100644 --- a/examples/requests_api_openskynetwork_private_area.py +++ b/examples/wifi/expanded/requests_wifi_api_openskynetwork_private_area.py @@ -3,13 +3,15 @@ # Coded for Circuit Python 8.1 # DJDevon3 ESP32-S3 OpenSkyNetwork_Private_Area_API_Example +import json import os -import time import ssl -import json -import wifi -import socketpool +import time + import circuitpython_base64 as base64 +import socketpool +import wifi + import adafruit_requests # OpenSky-Network.org Website Login required for this API @@ -30,10 +32,10 @@ # https://openskynetwork.github.io/opensky-api/rest.html#limitations sleep_time = 1800 -# this example uses settings.toml for credentials +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") # No token required, only website login -ssid = os.getenv("AP_SSID") -appw = os.getenv("AP_PASSWORD") osnu = os.getenv("OSN_Username") osnp = os.getenv("OSN_Password") @@ -105,7 +107,7 @@ def _format_datetime(datetime): request = adafruit_requests.Session(pool, ssl.create_default_context()) while not wifi.radio.ipv4_address: try: - wifi.radio.connect(ssid, appw) + wifi.radio.connect(ssid, password) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") diff --git a/examples/requests_api_openskynetwork_public.py b/examples/wifi/expanded/requests_wifi_api_openskynetwork_public.py similarity index 96% rename from examples/requests_api_openskynetwork_public.py rename to examples/wifi/expanded/requests_wifi_api_openskynetwork_public.py index e114cb7..5bcd69b 100644 --- a/examples/requests_api_openskynetwork_public.py +++ b/examples/wifi/expanded/requests_wifi_api_openskynetwork_public.py @@ -2,12 +2,14 @@ # SPDX-License-Identifier: MIT # Coded for Circuit Python 8.1 # Adafruit Feather ESP32-S3 OpenSkyNetwork_Public_API_Example +import json import os -import time import ssl -import json -import wifi +import time + import socketpool +import wifi + import adafruit_requests # No login necessary for Public API. Drastically reduced daily limit vs Private @@ -26,9 +28,9 @@ # https://openskynetwork.github.io/opensky-api/rest.html#limitations sleep_time = 1800 -# Wifi credentials pulled from settings.toml -ssid = os.getenv("AP_SSID") -appw = os.getenv("AP_PASSWORD") +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") # Requests URL - icao24 is their endpoint required for a transponder # example https://opensky-network.org/api/states/all?icao24=a808c5 @@ -69,7 +71,7 @@ def _format_datetime(datetime): requests = adafruit_requests.Session(pool, ssl.create_default_context()) while not wifi.radio.ipv4_address: try: - wifi.radio.connect(ssid, appw) + wifi.radio.connect(ssid, password) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") diff --git a/examples/requests_api_steam.py b/examples/wifi/expanded/requests_wifi_api_steam.py similarity index 95% rename from examples/requests_api_steam.py rename to examples/wifi/expanded/requests_wifi_api_steam.py index d29406c..208b6d6 100644 --- a/examples/requests_api_steam.py +++ b/examples/wifi/expanded/requests_wifi_api_steam.py @@ -2,13 +2,15 @@ # SPDX-License-Identifier: MIT # Coded for Circuit Python 8.0 """DJDevon3 Adafruit Feather ESP32-S2 api_steam Example""" -import os import gc -import time -import ssl import json -import wifi +import os +import ssl +import time + import socketpool +import wifi + import adafruit_requests # Steam API Docs: https://steamcommunity.com/dev @@ -16,10 +18,10 @@ # Steam Usernumber: Visit https://steamcommunity.com # click on your profile icon, your usernumber will be in the browser url. -# Ensure these are setup in settings.toml +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") # Requires Steam Developer API key -ssid = os.getenv("AP_SSID") -appw = os.getenv("AP_PASSWORD") steam_usernumber = os.getenv("steam_id") steam_apikey = os.getenv("steam_api_key") @@ -62,7 +64,7 @@ requests = adafruit_requests.Session(pool, ssl.create_default_context()) while not wifi.radio.ipv4_address: try: - wifi.radio.connect(ssid, appw) + wifi.radio.connect(ssid, password) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") diff --git a/examples/requests_api_twitch.py b/examples/wifi/expanded/requests_wifi_api_twitch.py similarity index 97% rename from examples/requests_api_twitch.py rename to examples/wifi/expanded/requests_wifi_api_twitch.py index 396712f..716caa9 100644 --- a/examples/requests_api_twitch.py +++ b/examples/wifi/expanded/requests_wifi_api_twitch.py @@ -4,10 +4,12 @@ # Twitch_API_Example import os -import time import ssl -import wifi +import time + import socketpool +import wifi + import adafruit_requests # Initialize WiFi Pool (There can be only 1 pool & top of script) @@ -21,9 +23,9 @@ # "Twitch_Client_Secret": "APP ID secret here", # "Twitch_UserID": "Your Twitch UserID here", -# Use settings.toml for credentials +# Get WiFi details, ensure these are setup in settings.toml ssid = os.getenv("CIRCUITPY_WIFI_SSID") -appw = os.getenv("CIRCUITPY_WIFI_PASSWORD") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") twitch_client_id = os.getenv("Twitch_ClientID") twitch_client_secret = os.getenv("Twitch_Client_Secret") # For finding your Twitch User ID @@ -57,7 +59,7 @@ def time_calc(input_time): requests = adafruit_requests.Session(pool, ssl.create_default_context()) while not wifi.radio.connected: try: - wifi.radio.connect(ssid, appw) + wifi.radio.connect(ssid, password) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") diff --git a/examples/requests_api_twitter.py b/examples/wifi/expanded/requests_wifi_api_twitter.py similarity index 90% rename from examples/requests_api_twitter.py rename to examples/wifi/expanded/requests_wifi_api_twitter.py index 995c790..4dcdfa6 100644 --- a/examples/requests_api_twitter.py +++ b/examples/wifi/expanded/requests_wifi_api_twitter.py @@ -3,11 +3,14 @@ # Coded for Circuit Python 8.0 """DJDevon3 Adafruit Feather ESP32-S2 Twitter_API_Example""" import gc -import time -import ssl import json -import wifi +import os +import ssl +import time + import socketpool +import wifi + import adafruit_requests # Twitter developer account bearer token required. @@ -22,11 +25,11 @@ # 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour sleep_time = 900 -try: - from secrets import secrets -except ImportError: - print("Secrets File Import Error") - raise +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") +tw_userid = os.getenv("TW_userid") +tw_bearer_token = os.getenv("TW_bearer_token") if sleep_time < 60: sleep_time_conversion = "seconds" @@ -42,10 +45,10 @@ sleep_time_conversion = "days" # Used with any Twitter 0auth request. -twitter_header = {"Authorization": "Bearer " + secrets["TW_bearer_token"]} +twitter_header = {"Authorization": "Bearer " + tw_bearer_token} TW_SOURCE = ( "https://api.twitter.com/2/users/" - + secrets["TW_userid"] + + tw_userid + "?user.fields=public_metrics,created_at,pinned_tweet_id" + "&expansions=pinned_tweet_id" + "&tweet.fields=created_at,public_metrics,source,context_annotations,entities" @@ -57,7 +60,7 @@ requests = adafruit_requests.Session(pool, ssl.create_default_context()) while not wifi.radio.ipv4_address: try: - wifi.radio.connect(secrets["ssid"], secrets["password"]) + wifi.radio.connect(ssid, password) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") diff --git a/examples/requests_api_youtube.py b/examples/wifi/expanded/requests_wifi_api_youtube.py similarity index 92% rename from examples/requests_api_youtube.py rename to examples/wifi/expanded/requests_wifi_api_youtube.py index 5fdacbb..e9bc6a2 100644 --- a/examples/requests_api_youtube.py +++ b/examples/wifi/expanded/requests_wifi_api_youtube.py @@ -3,11 +3,14 @@ # Coded for Circuit Python 8.0 """DJDevon3 Adafruit Feather ESP32-S2 YouTube_API_Example""" import gc -import time -import ssl import json -import wifi +import os +import ssl +import time + import socketpool +import wifi + import adafruit_requests # Ensure these are uncommented and in secrets.py or .env @@ -21,11 +24,12 @@ # 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour sleep_time = 900 -try: - from secrets import secrets -except ImportError: - print("Secrets File Import Error") - raise +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") +yt_username = os.getenv("YT_username") +yt_token = os.getenv("YT_token") + if sleep_time < 60: sleep_time_conversion = "seconds" @@ -45,9 +49,9 @@ "https://youtube.googleapis.com/youtube/v3/channels?" + "part=statistics" + "&forUsername=" - + secrets["YT_username"] + + yt_username + "&key=" - + secrets["YT_token"] + + yt_token ) # Connect to Wi-Fi @@ -56,7 +60,7 @@ requests = adafruit_requests.Session(pool, ssl.create_default_context()) while not wifi.radio.ipv4_address: try: - wifi.radio.connect(secrets["ssid"], secrets["password"]) + wifi.radio.connect(ssid, password) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") diff --git a/examples/requests_multiple_cookies.py b/examples/wifi/expanded/requests_wifi_multiple_cookies.py similarity index 83% rename from examples/requests_multiple_cookies.py rename to examples/wifi/expanded/requests_wifi_multiple_cookies.py index 45ae81f..36e4616 100644 --- a/examples/requests_multiple_cookies.py +++ b/examples/wifi/expanded/requests_wifi_multiple_cookies.py @@ -6,23 +6,23 @@ for connecting to the internet depending on your device. """ +import os import ssl -import wifi + import socketpool +import wifi + import adafruit_requests COOKIE_TEST_URL = "https://www.adafruit.com" -# Get wifi details and more from a secrets.py file -try: - from secrets import secrets -except ImportError: - print("WiFi secrets are kept in secrets.py, please add them there!") - raise +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") # Connect to the Wi-Fi network -print("Connecting to %s" % secrets["ssid"]) -wifi.radio.connect(secrets["ssid"], secrets["password"]) +print("Connecting to %s" % ssid) +wifi.radio.connect(ssid, password) # Set up the requests library pool = socketpool.SocketPool(wifi.radio) diff --git a/examples/wifi/requests_wifi_advanced.py b/examples/wifi/requests_wifi_advanced.py new file mode 100644 index 0000000..3bb0976 --- /dev/null +++ b/examples/wifi/requests_wifi_advanced.py @@ -0,0 +1,51 @@ +# SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries +# SPDX-License-Identifier: MIT + +import os +import ssl + +import socketpool +import wifi + +import adafruit_requests + +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") + +# Initialize WiFi Pool (There can be only 1 pool & top of script) +radio = wifi.radio +pool = socketpool.SocketPool(radio) + +print("Connecting to AP...") +while not wifi.radio.ipv4_address: + try: + wifi.radio.connect(ssid, password) + except ConnectionError as e: + print("could not connect to AP, retrying: ", e) +print("Connected to", str(radio.ap_info.ssid, "utf-8"), "\tRSSI:", radio.ap_info.rssi) + +# Initialize a requests session +ssl_context = ssl.create_default_context() +requests = adafruit_requests.Session(pool, ssl_context) + +JSON_GET_URL = "https://httpbin.org/get" + +# Define a custom header as a dict. +headers = {"user-agent": "blinka/1.0.0"} + +print("Fetching JSON data from %s..." % JSON_GET_URL) +response = requests.get(JSON_GET_URL, headers=headers) +print("-" * 60) + +json_data = response.json() +headers = json_data["headers"] +print("Response's Custom User-Agent Header: {0}".format(headers["User-Agent"])) +print("-" * 60) + +# Read Response's HTTP status code +print("Response HTTP Status Code: ", response.status_code) +print("-" * 60) + +# Close, delete and collect the response data +response.close() diff --git a/examples/wifi/requests_wifi_simpletest.py b/examples/wifi/requests_wifi_simpletest.py new file mode 100644 index 0000000..35b835a --- /dev/null +++ b/examples/wifi/requests_wifi_simpletest.py @@ -0,0 +1,72 @@ +# SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries +# SPDX-License-Identifier: MIT + +import os +import ssl + +import socketpool +import wifi + +import adafruit_requests + +# Get WiFi details, ensure these are setup in settings.toml +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +password = os.getenv("CIRCUITPY_WIFI_PASSWORD") + +# Initialize WiFi Pool (There can be only 1 pool & top of script) +radio = wifi.radio +pool = socketpool.SocketPool(radio) + +print("Connecting to AP...") +while not wifi.radio.ipv4_address: + try: + wifi.radio.connect(ssid, password) + except ConnectionError as e: + print("could not connect to AP, retrying: ", e) +print("Connected to", str(radio.ap_info.ssid, "utf-8"), "\tRSSI:", radio.ap_info.rssi) + +# Initialize a requests session +ssl_context = ssl.create_default_context() +requests = adafruit_requests.Session(pool, ssl_context) + +TEXT_URL = "http://wifitest.adafruit.com/testwifi/index.html" +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) +print("-" * 40) + +print("Text Response: ", response.text) +print("-" * 40) +response.close() + +print("Fetching JSON data from %s" % JSON_GET_URL) +response = requests.get(JSON_GET_URL) +print("-" * 40) + +print("JSON Response: ", response.json()) +print("-" * 40) +response.close() + +data = "31F" +print("POSTing data to {0}: {1}".format(JSON_POST_URL, data)) +response = requests.post(JSON_POST_URL, data=data) +print("-" * 40) + +json_resp = response.json() +# Parse out the 'data' key from json_resp dict. +print("Data received from server:", json_resp["data"]) +print("-" * 40) +response.close() + +json_data = {"Date": "July 25, 2019"} +print("POSTing data to {0}: {1}".format(JSON_POST_URL, json_data)) +response = requests.post(JSON_POST_URL, json=json_data) +print("-" * 40) + +json_resp = response.json() +# Parse out the 'json' key from json_resp dict. +print("JSON Data received from server:", json_resp["json"]) +print("-" * 40) +response.close() diff --git a/examples/wiznet5k/requests_wiznet5k_advanced.py b/examples/wiznet5k/requests_wiznet5k_advanced.py new file mode 100644 index 0000000..a6d9909 --- /dev/null +++ b/examples/wiznet5k/requests_wiznet5k_advanced.py @@ -0,0 +1,42 @@ +# SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries +# SPDX-License-Identifier: MIT + +import adafruit_connection_manager +import adafruit_wiznet5k.adafruit_wiznet5k_socket as pool +import board +import busio +from adafruit_wiznet5k.adafruit_wiznet5k import WIZNET5K +from digitalio import DigitalInOut + +import adafruit_requests + +cs = DigitalInOut(board.D10) +spi_bus = busio.SPI(board.SCK, MOSI=board.MOSI, MISO=board.MISO) + +# Initialize ethernet interface with DHCP +radio = WIZNET5K(spi_bus, cs) + +# Initialize a requests session +ssl_context = adafruit_connection_manager.create_fake_ssl_context(pool, radio) +requests = adafruit_requests.Session(pool, ssl_context) + +JSON_GET_URL = "http://httpbin.org/get" + +# Define a custom header as a dict. +headers = {"user-agent": "blinka/1.0.0"} + +print("Fetching JSON data from %s..." % JSON_GET_URL) +response = requests.get(JSON_GET_URL, headers=headers) +print("-" * 60) + +json_data = response.json() +headers = json_data["headers"] +print("Response's Custom User-Agent Header: {0}".format(headers["User-Agent"])) +print("-" * 60) + +# Read Response's HTTP status code +print("Response HTTP Status Code: ", response.status_code) +print("-" * 60) + +# Close, delete and collect the response data +response.close() diff --git a/examples/requests_simpletest_cpython.py b/examples/wiznet5k/requests_wiznet5k_simpletest.py old mode 100755 new mode 100644 similarity index 50% rename from examples/requests_simpletest_cpython.py rename to examples/wiznet5k/requests_wiznet5k_simpletest.py index be51d57..646107f --- a/examples/requests_simpletest_cpython.py +++ b/examples/wiznet5k/requests_wiznet5k_simpletest.py @@ -1,28 +1,39 @@ # SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries # SPDX-License-Identifier: MIT -# adafruit_requests usage with a CPython socket -import socket +import adafruit_connection_manager +import adafruit_wiznet5k.adafruit_wiznet5k_socket as pool +import board +import busio +from adafruit_wiznet5k.adafruit_wiznet5k import WIZNET5K +from digitalio import DigitalInOut + import adafruit_requests -http = adafruit_requests.Session(socket) +cs = DigitalInOut(board.D10) +spi_bus = busio.SPI(board.SCK, MOSI=board.MOSI, MISO=board.MISO) + +# Initialize ethernet interface with DHCP +radio = WIZNET5K(spi_bus, cs) + +# Initialize a requests session +ssl_context = adafruit_connection_manager.create_fake_ssl_context(pool, radio) +requests = adafruit_requests.Session(pool, ssl_context) TEXT_URL = "http://wifitest.adafruit.com/testwifi/index.html" JSON_GET_URL = "http://httpbin.org/get" JSON_POST_URL = "http://httpbin.org/post" -REDIRECT_URL = "http://httpbingo.org/redirect/1" -RELATIVE_REDIRECT_URL = "http://httpbingo.org/relative-redirect/1" -ABSOLUTE_REDIRECT_URL = "http://httpbingo.org/absolute-redirect/1" print("Fetching text from %s" % TEXT_URL) -response = http.get(TEXT_URL) +response = requests.get(TEXT_URL) print("-" * 40) print("Text Response: ", response.text) print("-" * 40) +response.close() print("Fetching JSON data from %s" % JSON_GET_URL) -response = http.get(JSON_GET_URL) +response = requests.get(JSON_GET_URL) print("-" * 40) print("JSON Response: ", response.json()) @@ -31,43 +42,22 @@ data = "31F" print("POSTing data to {0}: {1}".format(JSON_POST_URL, data)) -response = http.post(JSON_POST_URL, data=data) +response = requests.post(JSON_POST_URL, data=data) print("-" * 40) json_resp = response.json() # Parse out the 'data' key from json_resp dict. print("Data received from server:", json_resp["data"]) print("-" * 40) +response.close() json_data = {"Date": "July 25, 2019"} print("POSTing data to {0}: {1}".format(JSON_POST_URL, json_data)) -response = http.post(JSON_POST_URL, json=json_data) +response = requests.post(JSON_POST_URL, json=json_data) print("-" * 40) json_resp = response.json() # Parse out the 'json' key from json_resp dict. print("JSON Data received from server:", json_resp["json"]) print("-" * 40) - -print("Fetching JSON data from redirect url %s" % REDIRECT_URL) -response = http.get(REDIRECT_URL) -print("-" * 40) - -print("JSON Response: ", response.json()) -print("-" * 40) - -print("Fetching JSON data from relative redirect url %s" % RELATIVE_REDIRECT_URL) -response = http.get(RELATIVE_REDIRECT_URL) -print("-" * 40) - -print("JSON Response: ", response.json()) -print("-" * 40) - -print("Fetching JSON data from aboslute redirect url %s" % ABSOLUTE_REDIRECT_URL) -response = http.get(ABSOLUTE_REDIRECT_URL) -print("-" * 40) - -print("JSON Response: ", response.json()) -print("-" * 40) - response.close() diff --git a/requirements.txt b/requirements.txt index 7a984a4..2505288 100755 --- a/requirements.txt +++ b/requirements.txt @@ -3,3 +3,4 @@ # SPDX-License-Identifier: Unlicense Adafruit-Blinka +Adafruit-Circuitpython-ConnectionManager diff --git a/tests/chunk_test.py b/tests/chunk_test.py index df18bfe..ec4606d 100644 --- a/tests/chunk_test.py +++ b/tests/chunk_test.py @@ -5,7 +5,9 @@ """ Chunk Tests """ from unittest import mock + import mocket + import adafruit_requests IP = "1.2.3.4" diff --git a/tests/chunked_redirect_test.py b/tests/chunked_redirect_test.py index bb8ce9e..69f8b27 100644 --- a/tests/chunked_redirect_test.py +++ b/tests/chunked_redirect_test.py @@ -5,15 +5,33 @@ """ Redirection Tests """ from unittest import mock + import mocket from chunk_test import _chunk + import adafruit_requests IP = "1.2.3.4" HOST = "docs.google.com" -PATH = ( +PATH_BASE = ( "/spreadsheets/d/e/2PACX-1vR1WjUKz35-ek6SiR5droDfvPp51MTds4wUs57vEZNh2uDfihSTPhTaiiRo" - "vLbNe1mkeRgurppRJ_Zy/pub?output=tsv" + "vLbNe1mkeRgurppRJ_Zy/" +) +PATH = PATH_BASE + "pub?output=tsv" + +FILE_REDIRECT = ( + b"e@2PACX-1vR1WjUKz35-ek6SiR5droDfvPp51MTds4wUs57vEZNh2uDfihSTPhTai" + b"iRovLbNe1mkeRgurppRJ_Zy?output=tsv" +) +RELATIVE_RELATIVE_REDIRECT = ( + b"370cmver1f290kjsnpar5ku2h9g/3llvt5u8njbvat22m9l19db1h4/1656191325000/109226138307867586192/*/" + + FILE_REDIRECT +) +RELATIVE_ABSOLUTE_REDIRECT = ( + b"/pub/70cmver1f290kjsnpar5ku2h9g/" + RELATIVE_RELATIVE_REDIRECT +) +ABSOLUTE_ABSOLUTE_REDIRECT = ( + b"https://doc-14-2g-sheets.googleusercontent.com" + RELATIVE_ABSOLUTE_REDIRECT ) # response headers returned from the initial request @@ -24,10 +42,7 @@ b"Pragma: no-cache\r\n" b"Expires: Mon, 01 Jan 1990 00:00:00 GMT\r\n" b"Date: Sat, 25 Jun 2022 21:08:48 GMT\r\n" - b"Location: https://doc-14-2g-sheets.googleusercontent.com/pub/70cmver1f290kjsnpar5ku2h9g/3" - b"llvt5u8njbvat22m9l19db1h4/1656191325000" - b"/109226138307867586192/*/e@2PACX-1vR1WjUKz35-ek6SiR5droDfvPp51MTds4wUs57vEZNh2uDfihSTPhTai" - b"iRovLbNe1mkeRgurppRJ_Zy?output=tsv\r\n" + b"Location: NEW_LOCATION_HERE\r\n" b'P3P: CP="This is not a P3P policy! See g.co/p3phelp for more info."\r\n' b"X-Content-Type-Options: nosniff\r\n" b"X-XSS-Protection: 1; mode=block\r\n" @@ -116,13 +131,16 @@ def _recv_into(self, buf, nbytes=0): return read -def do_test_chunked_redirect(): +def test_chunked_absolute_absolute_redirect(): pool = mocket.MocketPool() pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 443)),) chunk = _chunk(BODY_REDIRECT, len(BODY_REDIRECT)) chunk2 = _chunk(BODY, len(BODY)) - sock1 = MocketRecvInto(HEADERS_REDIRECT + chunk) + redirect = HEADERS_REDIRECT.replace( + b"NEW_LOCATION_HERE", ABSOLUTE_ABSOLUTE_REDIRECT + ) + sock1 = MocketRecvInto(redirect + chunk) sock2 = mocket.Mocket(HEADERS + chunk2) pool.socket.side_effect = (sock1, sock2) @@ -133,9 +151,109 @@ def do_test_chunked_redirect(): sock2.connect.assert_called_once_with( ("doc-14-2g-sheets.googleusercontent.com", 443) ) + sock2.send.assert_has_calls([mock.call(RELATIVE_ABSOLUTE_REDIRECT[1:])]) + + assert response.text == str(BODY, "utf-8") + + +def test_chunked_relative_absolute_redirect(): + pool = mocket.MocketPool() + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 443)),) + chunk = _chunk(BODY_REDIRECT, len(BODY_REDIRECT)) + chunk2 = _chunk(BODY, len(BODY)) + + redirect = HEADERS_REDIRECT.replace( + b"NEW_LOCATION_HERE", RELATIVE_ABSOLUTE_REDIRECT + ) + sock1 = MocketRecvInto(redirect + chunk) + sock2 = mocket.Mocket(HEADERS + chunk2) + pool.socket.side_effect = (sock1, sock2) + + requests_session = adafruit_requests.Session(pool, mocket.SSLContext()) + response = requests_session.get("https://" + HOST + PATH) + + sock1.connect.assert_called_once_with((HOST, 443)) + sock2.connect.assert_called_once_with(("docs.google.com", 443)) + sock2.send.assert_has_calls([mock.call(RELATIVE_ABSOLUTE_REDIRECT[1:])]) + + assert response.text == str(BODY, "utf-8") + + +def test_chunked_relative_relative_redirect(): + pool = mocket.MocketPool() + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 443)),) + chunk = _chunk(BODY_REDIRECT, len(BODY_REDIRECT)) + chunk2 = _chunk(BODY, len(BODY)) + + redirect = HEADERS_REDIRECT.replace( + b"NEW_LOCATION_HERE", RELATIVE_RELATIVE_REDIRECT + ) + sock1 = MocketRecvInto(redirect + chunk) + sock2 = mocket.Mocket(HEADERS + chunk2) + pool.socket.side_effect = (sock1, sock2) + + requests_session = adafruit_requests.Session(pool, mocket.SSLContext()) + response = requests_session.get("https://" + HOST + PATH) + + sock1.connect.assert_called_once_with((HOST, 443)) + sock2.connect.assert_called_once_with(("docs.google.com", 443)) + sock2.send.assert_has_calls( + [mock.call(bytes(PATH_BASE[1:], "utf-8") + RELATIVE_RELATIVE_REDIRECT)] + ) + + assert response.text == str(BODY, "utf-8") + + +def test_chunked_relative_relative_redirect_backstep(): + pool = mocket.MocketPool() + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 443)),) + chunk = _chunk(BODY_REDIRECT, len(BODY_REDIRECT)) + chunk2 = _chunk(BODY, len(BODY)) + + remove_paths = 2 + backstep = b"../" * remove_paths + path_base_parts = PATH_BASE.split("/") + # PATH_BASE starts with "/" so skip it and also remove from the count + path_base = ( + "/".join(path_base_parts[1 : len(path_base_parts) - remove_paths - 1]) + "/" + ) + + redirect = HEADERS_REDIRECT.replace( + b"NEW_LOCATION_HERE", backstep + RELATIVE_RELATIVE_REDIRECT + ) + sock1 = MocketRecvInto(redirect + chunk) + sock2 = mocket.Mocket(HEADERS + chunk2) + pool.socket.side_effect = (sock1, sock2) + + requests_session = adafruit_requests.Session(pool, mocket.SSLContext()) + response = requests_session.get("https://" + HOST + PATH) + + sock1.connect.assert_called_once_with((HOST, 443)) + sock2.connect.assert_called_once_with(("docs.google.com", 443)) + sock2.send.assert_has_calls( + [mock.call(bytes(path_base, "utf-8") + RELATIVE_RELATIVE_REDIRECT)] + ) assert response.text == str(BODY, "utf-8") -def test_chunked_redirect(): - do_test_chunked_redirect() +def test_chunked_allow_redirects_false(): + pool = mocket.MocketPool() + pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 443)),) + chunk = _chunk(BODY_REDIRECT, len(BODY_REDIRECT)) + chunk2 = _chunk(BODY, len(BODY)) + + redirect = HEADERS_REDIRECT.replace( + b"NEW_LOCATION_HERE", ABSOLUTE_ABSOLUTE_REDIRECT + ) + sock1 = MocketRecvInto(redirect + chunk) + sock2 = mocket.Mocket(HEADERS + chunk2) + pool.socket.side_effect = (sock1, sock2) + + requests_session = adafruit_requests.Session(pool, mocket.SSLContext()) + response = requests_session.get("https://" + HOST + PATH, allow_redirects=False) + + sock1.connect.assert_called_once_with((HOST, 443)) + sock2.connect.assert_not_called() + + assert response.text == str(BODY_REDIRECT, "utf-8") diff --git a/tests/concurrent_test.py b/tests/concurrent_test.py index 79a32c5..58c2ade 100644 --- a/tests/concurrent_test.py +++ b/tests/concurrent_test.py @@ -6,31 +6,19 @@ import errno from unittest import mock + import mocket -import adafruit_requests - -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(): # 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) + + +def test_second_connect_fails_memoryerror(pool, requests_ssl): + sock = mocket.Mocket() + sock2 = mocket.Mocket() + sock3 = mocket.Mocket() pool.socket.call_count = 0 # Reset call count pool.socket.side_effect = [sock, sock2, sock3] sock2.connect.side_effect = MemoryError() - ssl = mocket.SSLContext() - - requests_session = adafruit_requests.Session(pool, ssl) - response = requests_session.get("https://" + HOST + PATH) + response = requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1) sock.send.assert_has_calls( [ @@ -46,13 +34,13 @@ def test_second_connect_fails_memoryerror(): # pylint: disable=invalid-name mock.call(b"\r\n"), ] ) - assert response.text == str(TEXT, "utf-8") + assert response.text == str(mocket.MOCK_RESPONSE_TEXT, "utf-8") - requests_session.get("https://" + HOST2 + PATH) + requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_2) - 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((mocket.MOCK_HOST_1, 443)) + sock2.connect.assert_called_once_with((mocket.MOCK_HOST_2, 443)) + sock3.connect.assert_called_once_with((mocket.MOCK_HOST_2, 443)) # Make sure that the socket is closed after send fails. sock.close.assert_called_once() sock2.close.assert_called_once() @@ -60,20 +48,15 @@ def test_second_connect_fails_memoryerror(): # pylint: disable=invalid-name assert pool.socket.call_count == 3 -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) +def test_second_connect_fails_oserror(pool, requests_ssl): + sock = mocket.Mocket() + sock2 = mocket.Mocket() + sock3 = mocket.Mocket() 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() - - requests_session = adafruit_requests.Session(pool, ssl) - response = requests_session.get("https://" + HOST + PATH) + response = requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1) sock.send.assert_has_calls( [ @@ -89,13 +72,13 @@ def test_second_connect_fails_oserror(): # pylint: disable=invalid-name mock.call(b"\r\n"), ] ) - assert response.text == str(TEXT, "utf-8") + assert response.text == str(mocket.MOCK_RESPONSE_TEXT, "utf-8") - requests_session.get("https://" + HOST2 + PATH) + requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_2) - 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((mocket.MOCK_HOST_1, 443)) + sock2.connect.assert_called_once_with((mocket.MOCK_HOST_2, 443)) + sock3.connect.assert_called_once_with((mocket.MOCK_HOST_2, 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/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..94023cb --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,46 @@ +# SPDX-FileCopyrightText: 2023 Justin Myers for Adafruit Industries +# +# SPDX-License-Identifier: Unlicense + +""" PyTest Setup """ + +import adafruit_connection_manager +import mocket +import pytest + +import adafruit_requests + + +@pytest.fixture(autouse=True) +def reset_connection_manager(monkeypatch): + """Reset the ConnectionManager, since it's a singlton and will hold data""" + monkeypatch.setattr( + "adafruit_requests.get_connection_manager", + adafruit_connection_manager.ConnectionManager, + ) + + +@pytest.fixture +def sock(): + return mocket.Mocket(mocket.MOCK_RESPONSE) + + +@pytest.fixture +def pool(sock): + pool = mocket.MocketPool() + pool.getaddrinfo.return_value = ( + (None, None, None, None, (mocket.MOCK_POOL_IP, 80)), + ) + pool.socket.return_value = sock + return pool + + +@pytest.fixture +def requests(pool): + return adafruit_requests.Session(pool) + + +@pytest.fixture +def requests_ssl(pool): + ssl_context = mocket.SSLContext() + return adafruit_requests.Session(pool, ssl_context) diff --git a/tests/header_test.py b/tests/header_test.py index ee0fad4..8bcb354 100644 --- a/tests/header_test.py +++ b/tests/header_test.py @@ -5,148 +5,85 @@ """ Header Tests """ import mocket -import adafruit_requests +import pytest -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_check_headers_not_dict(requests): + with pytest.raises(AttributeError) as context: + requests._check_headers("") + assert "headers must be in dict format" in str(context) -def test_host(): - pool = mocket.MocketPool() - 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) # pylint: disable=no-member - return len(data) +def test_check_headers_not_valid(requests): + with pytest.raises(AttributeError) as context: + requests._check_headers( + {"Good1": "a", "Good2": b"b", "Good3": None, "Bad1": True} + ) + assert "Header part (True) from Bad1 must be of type str or bytes" in str(context) - sock.send.side_effect = _send - requests_session = adafruit_requests.Session(pool) - headers = {} - requests_session.get("http://" + HOST + "/get", headers=headers) - - sock.connect.assert_called_once_with((IP, 80)) - sent = b"".join(sent) - assert b"Host: httpbin.org\r\n" in sent +def test_check_headers_valid(requests): + requests._check_headers({"Good1": "a", "Good2": b"b", "Good3": None}) + assert True -def test_host_replace(): - pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) - sock = mocket.Mocket(RESPONSE_HEADERS) - pool.socket.return_value = sock - sent = [] +def test_host(sock, requests): + headers = {} + requests.get("http://" + mocket.MOCK_HOST_1 + "/get", headers=headers) - def _send(data): - sent.append(data) # pylint: disable=no-member - return len(data) + sock.connect.assert_called_once_with((mocket.MOCK_POOL_IP, 80)) + sent = b"".join(sock.sent_data) + assert b"Host: wifitest.adafruit.com\r\n" in sent - sock.send.side_effect = _send - requests_session = adafruit_requests.Session(pool) - headers = {"host": IP} - requests_session.get("http://" + HOST + "/get", headers=headers) +def test_host_replace(sock, requests): + headers = {"host": mocket.MOCK_POOL_IP} + requests.get("http://" + mocket.MOCK_HOST_1 + "/get", headers=headers) - sock.connect.assert_called_once_with((IP, 80)) - sent = b"".join(sent) - assert b"host: 1.2.3.4\r\n" in sent - assert b"Host: httpbin.org\r\n" not in sent + sock.connect.assert_called_once_with((mocket.MOCK_POOL_IP, 80)) + sent = b"".join(sock.sent_data) + assert b"host: 10.10.10.10\r\n" in sent + assert b"Host: wifitest.adafruit.com\r\n" not in sent assert sent.lower().count(b"host:") == 1 -def test_user_agent(): - pool = mocket.MocketPool() - 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) # pylint: disable=no-member - return len(data) - - sock.send.side_effect = _send - - requests_session = adafruit_requests.Session(pool) +def test_user_agent(sock, requests): headers = {} - requests_session.get("http://" + HOST + "/get", headers=headers) + requests.get("http://" + mocket.MOCK_HOST_1 + "/get", headers=headers) - sock.connect.assert_called_once_with((IP, 80)) - sent = b"".join(sent) + sock.connect.assert_called_once_with((mocket.MOCK_POOL_IP, 80)) + sent = b"".join(sock.sent_data) assert b"User-Agent: Adafruit CircuitPython\r\n" in sent -def test_user_agent_replace(): - pool = mocket.MocketPool() - 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) # pylint: disable=no-member - return len(data) - - sock.send.side_effect = _send - - requests_session = adafruit_requests.Session(pool) +def test_user_agent_replace(sock, requests): headers = {"user-agent": "blinka/1.0.0"} - requests_session.get("http://" + HOST + "/get", headers=headers) + requests.get("http://" + mocket.MOCK_HOST_1 + "/get", headers=headers) - sock.connect.assert_called_once_with((IP, 80)) - sent = b"".join(sent) + sock.connect.assert_called_once_with((mocket.MOCK_POOL_IP, 80)) + sent = b"".join(sock.sent_data) assert b"user-agent: blinka/1.0.0\r\n" in sent assert b"User-Agent: Adafruit CircuitPython\r\n" not in sent assert sent.lower().count(b"user-agent:") == 1 -def test_content_type(): - pool = mocket.MocketPool() - 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) # pylint: disable=no-member - return len(data) - - sock.send.side_effect = _send - - requests_session = adafruit_requests.Session(pool) +def test_content_type(sock, requests): headers = {} data = {"test": True} - requests_session.post("http://" + HOST + "/get", data=data, headers=headers) + requests.post("http://" + mocket.MOCK_HOST_1 + "/get", data=data, headers=headers) - sock.connect.assert_called_once_with((IP, 80)) - sent = b"".join(sent) + sock.connect.assert_called_once_with((mocket.MOCK_POOL_IP, 80)) + sent = b"".join(sock.sent_data) assert b"Content-Type: application/x-www-form-urlencoded\r\n" in sent -def test_content_type_replace(): - pool = mocket.MocketPool() - 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) # pylint: disable=no-member - return len(data) - - sock.send.side_effect = _send - - requests_session = adafruit_requests.Session(pool) +def test_content_type_replace(sock, requests): headers = {"content-type": "application/test"} data = {"test": True} - requests_session.post("http://" + HOST + "/get", data=data, headers=headers) + requests.post("http://" + mocket.MOCK_HOST_1 + "/get", data=data, headers=headers) - sock.connect.assert_called_once_with((IP, 80)) - sent = b"".join(sent) + sock.connect.assert_called_once_with((mocket.MOCK_POOL_IP, 80)) + sent = b"".join(sock.sent_data) assert b"content-type: application/test\r\n" in sent assert b"Content-Type: application/x-www-form-urlencoded\r\n" not in sent assert sent.lower().count(b"content-type:") == 1 diff --git a/tests/method_test.py b/tests/method_test.py new file mode 100644 index 0000000..dac6d07 --- /dev/null +++ b/tests/method_test.py @@ -0,0 +1,80 @@ +# SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries +# +# SPDX-License-Identifier: Unlicense + +""" Post Tests """ + +from unittest import mock + +import mocket +import pytest + + +@pytest.mark.parametrize( + "call", + ( + "DELETE", + "GET", + "HEAD", + "PATCH", + "POST", + "PUT", + ), +) +def test_methods(call, sock, requests): + method = getattr(requests, call.lower()) + method("http://" + mocket.MOCK_HOST_1 + "/" + call.lower()) + sock.connect.assert_called_once_with((mocket.MOCK_POOL_IP, 80)) + + sock.send.assert_has_calls( + [ + mock.call(bytes(call, "utf-8")), + mock.call(b" /"), + mock.call(bytes(call.lower(), "utf-8")), + mock.call(b" HTTP/1.1\r\n"), + ] + ) + sock.send.assert_has_calls( + [ + mock.call(b"Host"), + mock.call(b": "), + mock.call(b"wifitest.adafruit.com"), + ] + ) + + +def test_post_string(sock, requests): + data = "31F" + requests.post("http://" + mocket.MOCK_HOST_1 + "/post", data=data) + sock.connect.assert_called_once_with((mocket.MOCK_POOL_IP, 80)) + sock.send.assert_called_with(b"31F") + + +def test_post_form(sock, requests): + data = {"Date": "July 25, 2019", "Time": "12:00"} + requests.post("http://" + mocket.MOCK_HOST_1 + "/post", data=data) + sock.connect.assert_called_once_with((mocket.MOCK_POOL_IP, 80)) + sock.send.assert_has_calls( + [ + mock.call(b"Content-Type"), + mock.call(b": "), + mock.call(b"application/x-www-form-urlencoded"), + mock.call(b"\r\n"), + ] + ) + sock.send.assert_called_with(b"Date=July 25, 2019&Time=12:00") + + +def test_post_json(sock, requests): + json_data = {"Date": "July 25, 2019", "Time": "12:00"} + requests.post("http://" + mocket.MOCK_HOST_1 + "/post", json=json_data) + sock.connect.assert_called_once_with((mocket.MOCK_POOL_IP, 80)) + sock.send.assert_has_calls( + [ + mock.call(b"Content-Type"), + mock.call(b": "), + mock.call(b"application/json"), + mock.call(b"\r\n"), + ] + ) + sock.send.assert_called_with(b'{"Date": "July 25, 2019", "Time": "12:00"}') diff --git a/tests/mocket.py b/tests/mocket.py index 3603800..3155231 100644 --- a/tests/mocket.py +++ b/tests/mocket.py @@ -1,4 +1,5 @@ # SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries +# SPDX-FileCopyrightText: 2024 Justin Myers for Adafruit Industries # # SPDX-License-Identifier: Unlicense @@ -6,21 +7,34 @@ from unittest import mock +MOCK_POOL_IP = "10.10.10.10" +MOCK_HOST_1 = "wifitest.adafruit.com" +MOCK_HOST_2 = "wifitest2.adafruit.com" +MOCK_PATH_1 = "/testwifi/index.html" +MOCK_ENDPOINT_1 = MOCK_HOST_1 + MOCK_PATH_1 +MOCK_ENDPOINT_2 = MOCK_HOST_2 + MOCK_PATH_1 +MOCK_RESPONSE_TEXT = ( + b"This is a test of Adafruit WiFi!\r\nIf you can read this, its working :)" +) +MOCK_RESPONSE = b"HTTP/1.0 200 OK\r\nContent-Length: 70\r\n\r\n" + MOCK_RESPONSE_TEXT + class MocketPool: # pylint: disable=too-few-public-methods """Mock SocketPool""" SOCK_STREAM = 0 - def __init__(self): + # pylint: disable=unused-argument + def __init__(self, radio=None): self.getaddrinfo = mock.Mock() + self.getaddrinfo.return_value = ((None, None, None, None, (MOCK_POOL_IP, 80)),) self.socket = mock.Mock() class Mocket: # pylint: disable=too-few-public-methods """Mock Socket""" - def __init__(self, response): + def __init__(self, response=MOCK_RESPONSE): self.settimeout = mock.Mock() self.close = mock.Mock() self.connect = mock.Mock() @@ -28,14 +42,17 @@ def __init__(self, response): self.readline = mock.Mock(side_effect=self._readline) self.recv = mock.Mock(side_effect=self._recv) self.recv_into = mock.Mock(side_effect=self._recv_into) + # Test helpers self._response = response self._position = 0 self.fail_next_send = False + self.sent_data = [] def _send(self, data): if self.fail_next_send: self.fail_next_send = False return 0 + self.sent_data.append(data) return len(data) def _readline(self): @@ -71,3 +88,18 @@ def _wrap_socket( self, sock, server_hostname=None ): # pylint: disable=no-self-use,unused-argument return sock + + +# pylint: disable=too-few-public-methods +class MockRadio: + class Radio: + pass + + class ESP_SPIcontrol: + TLS_MODE = 2 + + class WIZNET5K: + pass + + class Unsupported: + pass diff --git a/tests/parse_test.py b/tests/parse_test.py index 36ccd10..d9e1a56 100644 --- a/tests/parse_test.py +++ b/tests/parse_test.py @@ -5,11 +5,11 @@ """ 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") # Padding here tests the case where a header line is exactly 32 bytes buffered by @@ -24,13 +24,11 @@ ) -def test_json(): - pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) +def test_json(pool): sock = mocket.Mocket(HEADERS + ENCODED) pool.socket.return_value = sock requests_session = adafruit_requests.Session(pool) - response = requests_session.get("http://" + HOST + "/get") - sock.connect.assert_called_once_with((IP, 80)) + response = requests_session.get("http://" + mocket.MOCK_HOST_1 + "/get") + sock.connect.assert_called_once_with((mocket.MOCK_POOL_IP, 80)) assert response.json() == RESPONSE diff --git a/tests/post_test.py b/tests/post_test.py deleted file mode 100644 index 226efd4..0000000 --- a/tests/post_test.py +++ /dev/null @@ -1,100 +0,0 @@ -# SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries -# -# SPDX-License-Identifier: Unlicense - -""" Post Tests """ - -from unittest import mock -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( - "utf-8" -) - - -def test_method(): - pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) - sock = mocket.Mocket(HEADERS + ENCODED) - pool.socket.return_value = sock - - 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( - [ - mock.call(b"POST"), - mock.call(b" /"), - mock.call(b"post"), - mock.call(b" HTTP/1.1\r\n"), - ] - ) - sock.send.assert_has_calls( - [ - mock.call(b"Host"), - mock.call(b": "), - mock.call(b"httpbin.org"), - ] - ) - - -def test_string(): - pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) - sock = mocket.Mocket(HEADERS + ENCODED) - pool.socket.return_value = sock - - requests_session = adafruit_requests.Session(pool) - data = "31F" - 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.socket.return_value = sock - - requests_session = adafruit_requests.Session(pool) - data = {"Date": "July 25, 2019", "Time": "12:00"} - requests_session.post("http://" + HOST + "/post", data=data) - sock.connect.assert_called_once_with((IP, 80)) - sock.send.assert_has_calls( - [ - mock.call(b"Content-Type"), - mock.call(b": "), - mock.call(b"application/x-www-form-urlencoded"), - mock.call(b"\r\n"), - ] - ) - sock.send.assert_called_with(b"Date=July 25, 2019&Time=12:00") - - -def test_json(): - pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) - sock = mocket.Mocket(HEADERS + ENCODED) - pool.socket.return_value = sock - - requests_session = adafruit_requests.Session(pool) - json_data = {"Date": "July 25, 2019", "Time": "12:00"} - requests_session.post("http://" + HOST + "/post", json=json_data) - sock.connect.assert_called_once_with((IP, 80)) - sock.send.assert_has_calls( - [ - mock.call(b"Content-Type"), - mock.call(b": "), - mock.call(b"application/json"), - mock.call(b"\r\n"), - ] - ) - sock.send.assert_called_with(b'{"Date": "July 25, 2019", "Time": "12:00"}') diff --git a/tests/protocol_test.py b/tests/protocol_test.py index 2a5a930..4fd7770 100644 --- a/tests/protocol_test.py +++ b/tests/protocol_test.py @@ -5,39 +5,20 @@ """ 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 - -def test_get_https_no_ssl(): - pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) - sock = mocket.Mocket(RESPONSE) - pool.socket.return_value = sock - requests_session = adafruit_requests.Session(pool) - with pytest.raises(RuntimeError): - requests_session.get("https://" + HOST + PATH) +def test_get_https_no_ssl(requests): + with pytest.raises(AttributeError): + requests.get("https://" + mocket.MOCK_ENDPOINT_1) -def test_get_https_text(): - pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) - sock = mocket.Mocket(RESPONSE) - pool.socket.return_value = sock - ssl = mocket.SSLContext() +def test_get_https_text(sock, requests_ssl): + response = requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1) - 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((mocket.MOCK_HOST_1, 443)) sock.send.assert_has_calls( [ @@ -54,22 +35,16 @@ def test_get_https_text(): mock.call(b"wifitest.adafruit.com"), ] ) - assert response.text == str(TEXT, "utf-8") + assert response.text == str(mocket.MOCK_RESPONSE_TEXT, "utf-8") # Close isn't needed but can be called to release the socket early. 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.socket.return_value = sock - - requests_session = adafruit_requests.Session(pool) - response = requests_session.get("http://" + HOST + PATH) +def test_get_http_text(sock, requests): + response = requests.get("http://" + mocket.MOCK_ENDPOINT_1) - sock.connect.assert_called_once_with((IP, 80)) + sock.connect.assert_called_once_with((mocket.MOCK_POOL_IP, 80)) sock.send.assert_has_calls( [ @@ -86,20 +61,14 @@ def test_get_http_text(): mock.call(b"wifitest.adafruit.com"), ] ) - assert response.text == str(TEXT, "utf-8") + assert response.text == str(mocket.MOCK_RESPONSE_TEXT, "utf-8") -def test_get_close(): +def test_get_close(sock, requests): """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.socket.return_value = sock - - requests_session = adafruit_requests.Session(pool) - response = requests_session.get("http://" + HOST + PATH) + response = requests.get("http://" + mocket.MOCK_ENDPOINT_1) - sock.connect.assert_called_once_with((IP, 80)) + sock.connect.assert_called_once_with((mocket.MOCK_POOL_IP, 80)) sock.send.assert_has_calls( [ diff --git a/tests/reuse_test.py b/tests/reuse_test.py index b768a58..8ef5f5d 100644 --- a/tests/reuse_test.py +++ b/tests/reuse_test.py @@ -5,27 +5,16 @@ """ 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 -def test_get_twice(): - pool = mocket.MocketPool() - pool.getaddrinfo.return_value = ((None, None, None, None, (IP, 80)),) - sock = mocket.Mocket(RESPONSE + RESPONSE) +def test_get_twice(pool, requests_ssl): + sock = mocket.Mocket(mocket.MOCK_RESPONSE + mocket.MOCK_RESPONSE) pool.socket.return_value = sock - ssl = mocket.SSLContext() - requests_session = adafruit_requests.Session(pool, ssl) - response = requests_session.get("https://" + HOST + PATH) + response = requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1) sock.send.assert_has_calls( [ @@ -42,9 +31,9 @@ def test_get_twice(): mock.call(b"wifitest.adafruit.com"), ] ) - assert response.text == str(TEXT, "utf-8") + assert response.text == str(mocket.MOCK_RESPONSE_TEXT, "utf-8") - response = requests_session.get("https://" + HOST + PATH + "2") + response = requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1 + "2") sock.send.assert_has_calls( [ @@ -62,20 +51,16 @@ def test_get_twice(): ] ) - assert response.text == str(TEXT, "utf-8") - sock.connect.assert_called_once_with((HOST, 443)) + assert response.text == str(mocket.MOCK_RESPONSE_TEXT, "utf-8") + sock.connect.assert_called_once_with((mocket.MOCK_HOST_1, 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) +def test_get_twice_after_second(pool, requests_ssl): + sock = mocket.Mocket(mocket.MOCK_RESPONSE + mocket.MOCK_RESPONSE) pool.socket.return_value = sock - ssl = mocket.SSLContext() - requests_session = adafruit_requests.Session(pool, ssl) - response = requests_session.get("https://" + HOST + PATH) + response = requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1) sock.send.assert_has_calls( [ @@ -93,7 +78,7 @@ def test_get_twice_after_second(): ] ) - requests_session.get("https://" + HOST + PATH + "2") + requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1 + "2") sock.send.assert_has_calls( [ @@ -110,25 +95,22 @@ 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((mocket.MOCK_HOST_1, 443)) pool.socket.assert_called_once() - with pytest.raises(RuntimeError): - response.text == str(TEXT, "utf-8") # pylint: disable=expression-not-assigned + with pytest.raises(RuntimeError) as context: + result = response.text # pylint: disable=unused-variable + assert "Newer Response closed this one. Use Responses immediately." in str(context) -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) +def test_connect_out_of_memory(pool, requests_ssl): + sock = mocket.Mocket() + sock2 = mocket.Mocket() + sock3 = mocket.Mocket() pool.socket.side_effect = [sock, sock2, sock3] sock2.connect.side_effect = MemoryError() - ssl = mocket.SSLContext() - requests_session = adafruit_requests.Session(pool, ssl) - response = requests_session.get("https://" + HOST + PATH) + response = requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1) sock.send.assert_has_calls( [ @@ -145,9 +127,9 @@ def test_connect_out_of_memory(): mock.call(b"wifitest.adafruit.com"), ] ) - assert response.text == str(TEXT, "utf-8") + assert response.text == str(mocket.MOCK_RESPONSE_TEXT, "utf-8") - response = requests_session.get("https://" + HOST2 + PATH) + response = requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_2) sock3.send.assert_has_calls( [ mock.call(b"GET"), @@ -164,23 +146,18 @@ def test_connect_out_of_memory(): ] ) - assert response.text == str(TEXT, "utf-8") + assert response.text == str(mocket.MOCK_RESPONSE_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((mocket.MOCK_HOST_1, 443)) + sock3.connect.assert_called_once_with((mocket.MOCK_HOST_2, 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) +def test_second_send_fails(pool, requests_ssl): + sock = mocket.Mocket() + sock2 = mocket.Mocket() pool.socket.side_effect = [sock, sock2] - ssl = mocket.SSLContext() - - requests_session = adafruit_requests.Session(pool, ssl) - response = requests_session.get("https://" + HOST + PATH) + response = requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1) sock.send.assert_has_calls( [ @@ -196,30 +173,25 @@ def test_second_send_fails(): mock.call(b"\r\n"), ] ) - assert response.text == str(TEXT, "utf-8") + assert response.text == str(mocket.MOCK_RESPONSE_TEXT, "utf-8") sock.fail_next_send = True - requests_session.get("https://" + HOST + PATH + "2") + requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1 + "2") - sock.connect.assert_called_once_with((HOST, 443)) - sock2.connect.assert_called_once_with((HOST, 443)) + sock.connect.assert_called_once_with((mocket.MOCK_HOST_1, 443)) + sock2.connect.assert_called_once_with((mocket.MOCK_HOST_1, 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(): # 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) +def test_second_send_lies_recv_fails(pool, requests_ssl): + sock = mocket.Mocket() + sock2 = mocket.Mocket() pool.socket.side_effect = [sock, sock2] - ssl = mocket.SSLContext() - - requests_session = adafruit_requests.Session(pool, ssl) - response = requests_session.get("https://" + HOST + PATH) + response = requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1) sock.send.assert_has_calls( [ @@ -235,12 +207,12 @@ def test_second_send_lies_recv_fails(): # pylint: disable=invalid-name mock.call(b"\r\n"), ] ) - assert response.text == str(TEXT, "utf-8") + assert response.text == str(mocket.MOCK_RESPONSE_TEXT, "utf-8") - requests_session.get("https://" + HOST + PATH + "2") + requests_ssl.get("https://" + mocket.MOCK_ENDPOINT_1 + "2") - sock.connect.assert_called_once_with((HOST, 443)) - sock2.connect.assert_called_once_with((HOST, 443)) + sock.connect.assert_called_once_with((mocket.MOCK_HOST_1, 443)) + sock2.connect.assert_called_once_with((mocket.MOCK_HOST_1, 443)) # Make sure that the socket is closed after send fails. sock.close.assert_called_once() assert sock2.close.call_count == 0 diff --git a/tox.ini b/tox.ini index ab2df5e..85530c9 100644 --- a/tox.ini +++ b/tox.ini @@ -1,11 +1,39 @@ # SPDX-FileCopyrightText: 2022 Kevin Conley +# SPDX-FileCopyrightText: 2024 Justin Myers for Adafruit Industries # # SPDX-License-Identifier: MIT [tox] -envlist = py38 +envlist = py311 [testenv] -changedir = {toxinidir}/tests -deps = pytest==6.2.5 +description = run tests +deps = + pytest==7.4.3 commands = pytest + +[testenv:coverage] +description = run coverage +deps = + pytest==7.4.3 + pytest-cov==4.1.0 +package = editable +commands = + coverage run --source=. --omit=tests/* --branch {posargs} -m pytest + coverage report + coverage html + +[testenv:lint] +description = run linters +deps = + pre-commit==3.6.0 +skip_install = true +commands = pre-commit run {posargs} + +[testenv:docs] +description = build docs +deps = + -r requirements.txt + -r docs/requirements.txt +skip_install = true +commands = sphinx-build -E -W -b html docs/. _build/html