Update tests after merging from origin.

Author: BiffoBear

adafruit_wiznet5k/adafruit_wiznet5k.py

@@ -187,7 +187,8 @@ def __init__(
 
         # attempt to initialize the module
         self._ch_base_msb = 0
-        assert self._w5xxx_init() == 1, "Failed to initialize WIZnet module."
+        if self._w5xxx_init() != 1:
+            raise RuntimeError("Failed to initialize WIZnet module.")
         # Set MAC address
         self.mac_address = mac
         self.src_port = 0
@@ -214,7 +215,8 @@ def __init__(
             ret = self.set_dhcp(hostname, dhcp_timeout)
             if ret != 0:
                 self._dhcp_client = None
-            assert ret == 0, "Failed to configure DHCP Server!"
+            if ret != 0:
+                raise RuntimeError("Failed to configure DHCP Server!")
 
     def set_dhcp(
         self, hostname: Optional[str] = None, response_timeout: float = 30
@@ -274,7 +276,8 @@ def get_host_by_name(self, hostname: str) -> bytes:
         ret = _dns_client.gethostbyname(hostname)
         if self._debug:
             print("* Resolved IP: ", ret)
-        assert ret != -1, "Failed to resolve hostname!"
+        if ret == -1:
+            raise RuntimeError("Failed to resolve hostname!")
         return ret
 
     @property
@@ -628,7 +631,8 @@ def socket_available(self, socket_num: int, sock_type: int = SNMR_TCP) -> int:
                     socket_num, sock_type
                 )
             )
-        assert socket_num <= self.max_sockets, "Provided socket exceeds max_sockets."
+        if socket_num > self.max_sockets:
+            raise ValueError("Provided socket exceeds max_sockets.")
 
         res = self._get_rx_rcv_size(socket_num)
 
@@ -681,7 +685,8 @@ def socket_connect(
         :param int conn_mode: The connection mode. Use SNMR_TCP for TCP or SNMR_UDP for UDP,
             defaults to SNMR_TCP.
         """
-        assert self.link_status, "Ethernet cable disconnected!"
+        if not self.link_status:
+            raise ConnectionError("Ethernet cable disconnected!")
         if self._debug:
             print(
                 "* w5k socket connect, protocol={}, port={}, ip={}".format(
@@ -691,7 +696,7 @@ def socket_connect(
         # initialize a socket and set the mode
         res = self.socket_open(socket_num, conn_mode=conn_mode)
         if res == 1:
-            raise RuntimeError("Failed to initialize a connection with the socket.")
+            raise ConnectionError("Failed to initialize a connection with the socket.")
 
         # set socket destination IP and port
         self._write_sndipr(socket_num, dest)
@@ -705,7 +710,7 @@ def socket_connect(
                 if self._debug:
                     print("SN_SR:", self.socket_status(socket_num)[0])
                 if self.socket_status(socket_num)[0] == SNSR_SOCK_CLOSED:
-                    raise RuntimeError("Failed to establish connection.")
+                    raise ConnectionError("Failed to establish connection.")
         elif conn_mode == SNMR_UDP:
             self.udp_datasize[socket_num] = 0
         return 1
@@ -749,7 +754,8 @@ def socket_listen(
         :param int conn_mode: Connection mode SNMR_TCP for TCP or SNMR_UDP for
             UDP, defaults to SNMR_TCP.
         """
-        assert self.link_status, "Ethernet cable disconnected!"
+        if not self.link_status:
+            raise ConnectionError("Ethernet cable disconnected!")
         if self._debug:
             print(
                 "* Listening on port={}, ip={}".format(
@@ -809,7 +815,8 @@ def socket_open(self, socket_num: int, conn_mode: int = SNMR_TCP) -> int:
             UDP, defaults to SNMR_TCP.
         :return int: 1 if the socket was opened, 0 if not.
         """
-        assert self.link_status, "Ethernet cable disconnected!"
+        if not self.link_status:
+            raise ConnectionError("Ethernet cable disconnected!")
         if self._debug:
             print("*** Opening socket %d" % socket_num)
         status = self._read_snsr(socket_num)[0]
@@ -841,10 +848,8 @@ def socket_open(self, socket_num: int, conn_mode: int = SNMR_TCP) -> int:
             # open socket
             self._write_sncr(socket_num, CMD_SOCK_OPEN)
             self._read_sncr(socket_num)
-            assert (
-                self._read_snsr((socket_num))[0] == 0x13
-                or self._read_snsr((socket_num))[0] == 0x22
-            ), "Could not open socket in TCP or UDP mode."
+            if self._read_snsr((socket_num))[0] not in [0x13, 0x22]:
+                raise RuntimeError("Could not open socket in TCP or UDP mode.")
             return 0
         return 1
 
@@ -870,7 +875,7 @@ def socket_disconnect(self, socket_num: int) -> None:
         self._write_sncr(socket_num, CMD_SOCK_DISCON)
         self._read_sncr(socket_num)
 
-    def socket_read(
+    def socket_read(  # pylint: disable=too-many-branches
         self, socket_num: int, length: int
     ) -> Tuple[int, Union[int, bytearray]]:
         """
@@ -884,8 +889,11 @@ def socket_read(
             was unsuccessful then both items equal an error code, 0 for no data waiting and -1
             for no connection to the socket.
         """
-        assert self.link_status, "Ethernet cable disconnected!"
-        assert socket_num <= self.max_sockets, "Provided socket exceeds max_sockets."
+
+        if not self.link_status:
+            raise ConnectionError("Ethernet cable disconnected!")
+        if socket_num > self.max_sockets:
+            raise ValueError("Provided socket exceeds max_sockets.")
 
         # Check if there is data available on the socket
         ret = self._get_rx_rcv_size(socket_num)
@@ -978,7 +986,8 @@ def socket_write(
 
         :return int: The number of bytes written to the buffer.
         """
-        assert self.link_status, "Ethernet cable disconnected!"
+        if not self.link_status:
+            raise ConnectionError("Ethernet cable disconnected!")
         assert socket_num <= self.max_sockets, "Provided socket exceeds max_sockets."
         if len(buffer) > SOCK_SIZE:
             ret = SOCK_SIZE

adafruit_wiznet5k/adafruit_wiznet5k_dhcp.py

@@ -65,7 +65,7 @@
 DHCP_HLENETHERNET = const(0x06)
 DHCP_HOPS = const(0x00)
 
-MAGIC_COOKIE = const(0x63825363)
+MAGIC_COOKIE = b"c\x82Sc"  # Four bytes 99.130.83.99
 MAX_DHCP_OPT = const(0x10)
 
 # Default DHCP Server port
@@ -179,7 +179,7 @@ def send_dhcp_message(
         # Transaction ID (xid)
         self._initial_xid = htonl(self._transaction_id)
         self._initial_xid = self._initial_xid.to_bytes(4, "big")
-        _BUFF[4:7] = self._initial_xid
+        _BUFF[4:8] = self._initial_xid
 
         # seconds elapsed
         _BUFF[8] = (int(time_elapsed) & 0xFF00) >> 8
@@ -195,18 +195,15 @@ def send_dhcp_message(
         # as they're already set to 0.0.0.0
         # Except when renewing, then fill in ciaddr
         if renew:
-            _BUFF[12:15] = bytes(self.local_ip)
+            _BUFF[12:16] = bytes(self.local_ip)
 
         # chaddr
         _BUFF[28:34] = self._mac_address
 
         # NOTE:  192 octets of 0's, BOOTP legacy
 
         # Magic Cookie
-        _BUFF[236] = (MAGIC_COOKIE >> 24) & 0xFF
-        _BUFF[237] = (MAGIC_COOKIE >> 16) & 0xFF
-        _BUFF[238] = (MAGIC_COOKIE >> 8) & 0xFF
-        _BUFF[239] = MAGIC_COOKIE & 0xFF
+        _BUFF[236:240] = MAGIC_COOKIE
 
         # Option - DHCP Message Type
         _BUFF[240] = 53
@@ -262,10 +259,10 @@ def send_dhcp_message(
     # pylint: disable=too-many-branches, too-many-statements
     def parse_dhcp_response(
         self,
-    ) -> Union[Tuple[int, bytes], Tuple[int, int]]:
+    ) -> Tuple[int, bytearray]:
         """Parse DHCP response from DHCP server.
 
-        :return Union[Tuple[int, bytes], Tuple[int, int]]: DHCP packet type.
+        :return Tuple[int, bytearray]: DHCP packet type and ID.
         """
         global _BUFF  # pylint: disable=global-statement
         # store packet in buffer
@@ -275,22 +272,23 @@ def parse_dhcp_response(
 
         # -- Parse Packet, FIXED -- #
         # Validate OP
-        assert (
-            _BUFF[0] == DHCP_BOOT_REPLY
-        ), "Malformed Packet - \
+        if _BUFF[0] != DHCP_BOOT_REPLY:
+            raise RuntimeError(
+                "Malformed Packet - \
             DHCP message OP is not expected BOOT Reply."
+            )
 
         xid = _BUFF[4:8]
-        if bytes(xid) < self._initial_xid:
-            print("f")
-            return 0, 0
+        if bytes(xid) != self._initial_xid:
+            raise ValueError("DHCP response ID mismatch.")
 
         self.local_ip = tuple(_BUFF[16:20])
-        if _BUFF[28:34] == 0:
-            return 0, 0
+        # Check that there is a server ID.
+        if _BUFF[28:34] == b"\x00\x00\x00\x00\x00\x00":
+            raise ValueError("No DHCP server ID in the response.")
 
-        if int.from_bytes(_BUFF[235:240], "big") != MAGIC_COOKIE:
-            return 0, 0
+        if _BUFF[236:240] != MAGIC_COOKIE:
+            raise ValueError("No DHCP Magic Cookie in the response.")
 
         # -- Parse Packet, VARIABLE -- #
         ptr = 240
@@ -323,8 +321,8 @@ def parse_dhcp_response(
                 ptr += 1
                 opt_len = _BUFF[ptr]
                 ptr += 1
-                self.gateway_ip = tuple(_BUFF[ptr : ptr + opt_len])
-                ptr += opt_len
+                self.gateway_ip = tuple(_BUFF[ptr : ptr + 4])
+                ptr += opt_len  # still increment even though we only read 1 addr.
             elif _BUFF[ptr] == DNS_SERVERS:
                 ptr += 1
                 opt_len = _BUFF[ptr]
@@ -429,65 +427,79 @@ def _dhcp_state_machine(self) -> None:
             if self._sock._available():  # pylint: disable=protected-access
                 if self._debug:
                     print("* DHCP: Parsing OFFER")
-                msg_type, xid = self.parse_dhcp_response()
-                if msg_type == DHCP_OFFER:
-                    # Check if transaction ID matches, otherwise it may be an offer
-                    # for another device
-                    if htonl(self._transaction_id) == int.from_bytes(xid, "big"):
-                        if self._debug:
-                            print(
-                                "* DHCP: Send request to {}".format(self.dhcp_server_ip)
+                try:
+                    msg_type, xid = self.parse_dhcp_response()
+                except ValueError as error:
+                    if self._debug:
+                        print(error)
+                else:
+                    if msg_type == DHCP_OFFER:
+                        # Check if transaction ID matches, otherwise it may be an offer
+                        # for another device
+                        if htonl(self._transaction_id) == int.from_bytes(xid, "big"):
+                            if self._debug:
+                                print(
+                                    "* DHCP: Send request to {}".format(
+                                        self.dhcp_server_ip
+                                    )
+                                )
+                            self._transaction_id = (
+                                self._transaction_id + 1
+                            ) & 0x7FFFFFFF
+                            self.send_dhcp_message(
+                                DHCP_REQUEST, (time.monotonic() - self._start_time)
                             )
-                        self._transaction_id = (self._transaction_id + 1) & 0x7FFFFFFF
-                        self.send_dhcp_message(
-                            DHCP_REQUEST, (time.monotonic() - self._start_time)
-                        )
-                        self._dhcp_state = STATE_DHCP_REQUEST
+                            self._dhcp_state = STATE_DHCP_REQUEST
+                        else:
+                            if self._debug:
+                                print("* DHCP: Received OFFER with non-matching xid")
                     else:
                         if self._debug:
-                            print("* DHCP: Received OFFER with non-matching xid")
-                else:
-                    if self._debug:
-                        print("* DHCP: Received DHCP Message is not OFFER")
+                            print("* DHCP: Received DHCP Message is not OFFER")
 
         elif self._dhcp_state == STATE_DHCP_REQUEST:
             if self._sock._available():  # pylint: disable=protected-access
                 if self._debug:
                     print("* DHCP: Parsing ACK")
-                msg_type, xid = self.parse_dhcp_response()
-                # Check if transaction ID matches, otherwise it may be
-                # for another device
-                if htonl(self._transaction_id) == int.from_bytes(xid, "big"):
-                    if msg_type == DHCP_ACK:
-                        if self._debug:
-                            print("* DHCP: Successful lease")
-                        self._sock.close()
-                        self._sock = None
-                        self._dhcp_state = STATE_DHCP_LEASED
-                        self._last_lease_time = self._start_time
-                        if self._lease_time == 0:
-                            self._lease_time = DEFAULT_LEASE_TIME
-                        if self._t1 == 0:
-                            # T1 is 50% of _lease_time
-                            self._t1 = self._lease_time >> 1
-                        if self._t2 == 0:
-                            # T2 is 87.5% of _lease_time
-                            self._t2 = self._lease_time - (self._lease_time >> 3)
-                        self._renew_in_sec = self._t1
-                        self._rebind_in_sec = self._t2
-                        self._eth.ifconfig = (
-                            self.local_ip,
-                            self.subnet_mask,
-                            self.gateway_ip,
-                            self.dns_server_ip,
-                        )
-                        gc.collect()
+                try:
+                    msg_type, xid = self.parse_dhcp_response()
+                except ValueError as error:
+                    if self._debug:
+                        print(error)
+                else:
+                    # Check if transaction ID matches, otherwise it may be
+                    # for another device
+                    if htonl(self._transaction_id) == int.from_bytes(xid, "big"):
+                        if msg_type == DHCP_ACK:
+                            if self._debug:
+                                print("* DHCP: Successful lease")
+                            self._sock.close()
+                            self._sock = None
+                            self._dhcp_state = STATE_DHCP_LEASED
+                            self._last_lease_time = self._start_time
+                            if self._lease_time == 0:
+                                self._lease_time = DEFAULT_LEASE_TIME
+                            if self._t1 == 0:
+                                # T1 is 50% of _lease_time
+                                self._t1 = self._lease_time >> 1
+                            if self._t2 == 0:
+                                # T2 is 87.5% of _lease_time
+                                self._t2 = self._lease_time - (self._lease_time >> 3)
+                            self._renew_in_sec = self._t1
+                            self._rebind_in_sec = self._t2
+                            self._eth.ifconfig = (
+                                self.local_ip,
+                                self.subnet_mask,
+                                self.gateway_ip,
+                                self.dns_server_ip,
+                            )
+                            gc.collect()
+                        else:
+                            if self._debug:
+                                print("* DHCP: Received DHCP Message is not ACK")
                     else:
                         if self._debug:
-                            print("* DHCP: Received DHCP Message is not ACK")
-                else:
-                    if self._debug:
-                        print("* DHCP: Received non-matching xid")
+                            print("* DHCP: Received non-matching xid")
 
         elif self._dhcp_state == STATE_DHCP_WAIT:
             if time.monotonic() > (self._start_time + DHCP_WAIT_TIME):

adafruit_wiznet5k/adafruit_wiznet5k_socket.py

@@ -146,8 +146,6 @@ def inet_ntoa(ip_address: Union[bytes, bytearray]) -> str:
 
 
 # pylint: disable=too-many-arguments, unused-argument
-
-
 def getaddrinfo(
     host: str,
     port: int,
@@ -334,7 +332,8 @@ def listen(self, backlog: int = 0) -> None:
 
         :param int backlog: Included for compatibility but ignored.
         """
-        assert self._listen_port is not None, "Use bind to set the port before listen!"
+        if self._listen_port is None:
+            raise RuntimeError("Use bind to set the port before listen!")
         _the_interface.socket_listen(self._socknum, self._listen_port)
         self._buffer = b""
 
@@ -576,7 +575,8 @@ def _readline(self) -> bytes:
 
     def _disconnect(self) -> None:
         """Disconnect a TCP socket."""
-        assert self._sock_type == SOCK_STREAM, "Socket must be a TCP socket."
+        if self._sock_type != SOCK_STREAM:
+            raise RuntimeError("Socket must be a TCP socket.")
         _the_interface.socket_disconnect(self._socknum)
 
     def close(self) -> None:

optional_requirements.txt

@@ -1,3 +1,4 @@
 # SPDX-FileCopyrightText: 2022 Alec Delaney, for Adafruit Industries
 #
 # SPDX-License-Identifier: Unlicense
+pytest