Add new options for REST API devices (#96)

* feat(rest_api_device): Added the options to bypass SSL verification and allow URL redirects for RESTAPI devices. Co-authored-by: Collin Charvat <[email protected]> * docs: Updated the Maintainers section of the Readme. Co-authored-by: Shashank P <[email protected]> --------- Co-authored-by: Collin Charvat <[email protected]> Co-authored-by: Shashank P <[email protected]>
tektronix · Nov 8, 2023 · 78e0ed7 · 78e0ed7
1 parent a64fe56
commit 78e0ed7
Show file tree

Hide file tree

Showing 7 changed files with 90 additions and 17 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -23,6 +23,8 @@ Things to be included in the next release go here.
 - New examples added to the basic usage guide showing how to use the commands for some scope drivers
 - Added an example showing how to change the VISA backend that is used for connecting to devices
 - Added a new support table in the Readme showing the API support for Software Solutions
+- Added an option to bypass SSL certificate verification for RESTful API devices
+- Added an option to allow URL redirects for RESTful API devices
 
 ### Changed
 

diff --git a/README.rst b/README.rst
@@ -285,11 +285,12 @@ Maintainers
 Before reaching out to any maintainers directly, please first check if your issue or question is already
 covered by any `open issues <https://github.com/tektronix/tm_devices/issues>`__. If the issue or
 question you have is not already covered, please
-`file a new issue <https://github.com/tektronix/tm_devices/issues/new/choose>`__ and the
+`file a new issue <https://github.com/tektronix/tm_devices/issues/new/choose>`__ or
+start a `discussion <https://github.com/tektronix/tm_devices/discussions>`__ and the
 maintainers will review and respond there.
 
--  Tektronix opensource@tektronix.com
--  Nicholas Felt nicholas.felt@tektronix.com
+-  tmdevicessupport@tektronix.com - For technical support and questions.
+-  opensource@tektronix.com - For open-source policy and license questions.
 
 Contributing
 ------------

diff --git a/src/tm_devices/drivers/api/rest_api/rest_api_device.py b/src/tm_devices/drivers/api/rest_api/rest_api_device.py
@@ -50,6 +50,7 @@ def base_url(self) -> str:
     ################################################################################################
     # Public Methods
     ################################################################################################
+    # pylint: disable=too-many-arguments
     def delete(  # noqa: PLR0913
         self,
         url: str,
@@ -58,6 +59,8 @@ def delete(  # noqa: PLR0913
         timeout: Optional[float] = None,
         return_bytes: bool = False,
         allow_errors: bool = False,
+        verify_ssl: bool = True,
+        allow_redirects: bool = False,
         verbose: bool = True,
     ) -> Tuple[bool, Union[Dict[str, Any], bytes], int, Optional[requests.RequestException]]:
         """Perform a DELETE request with the given url and headers.
@@ -69,6 +72,8 @@ def delete(  # noqa: PLR0913
             timeout: How many seconds to wait for the server to send data before giving up.
             return_bytes: A boolean indicating if the response content should be returned
                 instead of the response json.
+            verify_ssl: A bool that indicates if the SSL certificate should be verified.
+            allow_redirects: A bool that indicates if URL redirects should be allowed.
             allow_errors: A boolean indicating if errors are allowed.
             verbose: Set this to False in order to disable printouts.
 
@@ -83,9 +88,12 @@ def delete(  # noqa: PLR0913
             timeout=timeout,
             return_bytes=return_bytes,
             allow_errors=allow_errors,
+            verify_ssl=verify_ssl,
+            allow_redirects=allow_redirects,
             verbose=verbose,
         )
 
+    # pylint: disable=too-many-arguments
     def get(  # noqa: PLR0913
         self,
         url: str,
@@ -94,6 +102,8 @@ def get(  # noqa: PLR0913
         timeout: Optional[float] = None,
         return_bytes: bool = False,
         allow_errors: bool = False,
+        verify_ssl: bool = True,
+        allow_redirects: bool = False,
         verbose: bool = True,
     ) -> Tuple[bool, Union[Dict[str, Any], bytes], int, Optional[requests.RequestException]]:
         """Perform a GET request with the given url and headers.
@@ -105,6 +115,8 @@ def get(  # noqa: PLR0913
             timeout: How many seconds to wait for the server to send data before giving up.
             return_bytes: A boolean indicating if the response content should be returned
                 instead of the response json.
+            verify_ssl: A bool that indicates if the SSL certificate should be verified.
+            allow_redirects: A bool that indicates if URL redirects should be allowed.
             allow_errors: A boolean indicating if errors are allowed.
             verbose: Set this to False in order to disable printouts.
 
@@ -119,9 +131,12 @@ def get(  # noqa: PLR0913
             timeout=timeout,
             return_bytes=return_bytes,
             allow_errors=allow_errors,
+            verify_ssl=verify_ssl,
+            allow_redirects=allow_redirects,
             verbose=verbose,
         )
 
+    # pylint: disable=too-many-arguments
     def patch(  # noqa: PLR0913
         self,
         url: str,
@@ -131,6 +146,8 @@ def patch(  # noqa: PLR0913
         timeout: Optional[float] = None,
         return_bytes: bool = False,
         allow_errors: bool = False,
+        verify_ssl: bool = True,
+        allow_redirects: bool = False,
         verbose: bool = True,
     ) -> Tuple[bool, Union[Dict[str, Any], bytes], int, Optional[requests.RequestException]]:
         """Perform a PATCH request with the given url and headers.
@@ -143,6 +160,8 @@ def patch(  # noqa: PLR0913
             timeout: How many seconds to wait for the server to send data before giving up.
             return_bytes: A boolean indicating if the response content should be returned
                 instead of the response json.
+            verify_ssl: A bool that indicates if the SSL certificate should be verified.
+            allow_redirects: A bool that indicates if URL redirects should be allowed.
             allow_errors: A boolean indicating if errors are allowed.
             verbose: Set this to False in order to disable printouts.
 
@@ -158,9 +177,12 @@ def patch(  # noqa: PLR0913
             timeout=timeout,
             return_bytes=return_bytes,
             allow_errors=allow_errors,
+            verify_ssl=verify_ssl,
+            allow_redirects=allow_redirects,
             verbose=verbose,
         )
 
+    # pylint: disable=too-many-arguments
     def post(  # noqa: PLR0913
         self,
         url: str,
@@ -170,6 +192,8 @@ def post(  # noqa: PLR0913
         timeout: Optional[float] = None,
         return_bytes: bool = False,
         allow_errors: bool = False,
+        verify_ssl: bool = True,
+        allow_redirects: bool = False,
         verbose: bool = True,
     ) -> Tuple[bool, Union[Dict[str, Any], bytes], int, Optional[requests.RequestException]]:
         """Perform a POST request with the given url and headers.
@@ -182,6 +206,8 @@ def post(  # noqa: PLR0913
             timeout: How many seconds to wait for the server to send data before giving up.
             return_bytes: A boolean indicating if the response content should be returned
                 instead of the response json.
+            verify_ssl: A bool that indicates if the SSL certificate should be verified.
+            allow_redirects: A bool that indicates if URL redirects should be allowed.
             allow_errors: A boolean indicating if errors are allowed.
             verbose: Set this to False in order to disable printouts.
 
@@ -197,9 +223,12 @@ def post(  # noqa: PLR0913
             timeout=timeout,
             return_bytes=return_bytes,
             allow_errors=allow_errors,
+            verify_ssl=verify_ssl,
+            allow_redirects=allow_redirects,
             verbose=verbose,
         )
 
+    # pylint: disable=too-many-arguments
     def put(  # noqa: PLR0913
         self,
         url: str,
@@ -209,6 +238,8 @@ def put(  # noqa: PLR0913
         timeout: Optional[float] = None,
         return_bytes: bool = False,
         allow_errors: bool = False,
+        verify_ssl: bool = True,
+        allow_redirects: bool = False,
         verbose: bool = True,
     ) -> Tuple[bool, Union[Dict[str, Any], bytes], int, Optional[requests.RequestException]]:
         """Perform a PUT request with the given url and headers.
@@ -221,6 +252,8 @@ def put(  # noqa: PLR0913
             timeout: How many seconds to wait for the server to send data before giving up.
             return_bytes: A boolean indicating if the response content should be returned
                 instead of the response json.
+            verify_ssl: A bool that indicates if the SSL certificate should be verified.
+            allow_redirects: A bool that indicates if URL redirects should be allowed.
             allow_errors: A boolean indicating if errors are allowed.
             verbose: Set this to False in order to disable printouts.
 
@@ -236,6 +269,8 @@ def put(  # noqa: PLR0913
             timeout=timeout,
             return_bytes=return_bytes,
             allow_errors=allow_errors,
+            verify_ssl=verify_ssl,
+            allow_redirects=allow_redirects,
             verbose=verbose,
         )
 
@@ -281,7 +316,7 @@ def wait_for_api_connection(
         start_time = time.perf_counter()
         while (time.perf_counter() - start_time) <= wait_time:
             if api_connection := self._check_api_connection():
-                # pylint: disable=compare-to-zero
+                # pylint: disable=use-implicit-booleaness-not-comparison-to-zero
                 if attempt_num != 0 or accept_immediate_connection:
                     break
                 msg = (
@@ -321,7 +356,7 @@ def _check_api_connection(self) -> bool:
         """
         raise NotImplementedError
 
-    # pylint: disable=too-many-branches
+    # pylint: disable=too-many-branches,too-many-arguments,too-many-locals
     def _send_request(  # noqa: PLR0913,PLR0912,C901
         self,
         request_type: SupportedRequestTypes,
@@ -332,6 +367,8 @@ def _send_request(  # noqa: PLR0913,PLR0912,C901
         timeout: Optional[float] = None,
         return_bytes: bool = False,
         allow_errors: bool = False,
+        verify_ssl: bool = True,
+        allow_redirects: bool = False,
         verbose: bool = True,
     ) -> Tuple[bool, Union[Dict[str, Any], bytes], int, Optional[requests.RequestException]]:
         """Perform a request with the given url and headers.
@@ -345,6 +382,8 @@ def _send_request(  # noqa: PLR0913,PLR0912,C901
             timeout: How many seconds to wait for the server to send data before giving up.
             return_bytes: A boolean indicating if the response content should be returned
                 instead of the response json.
+            verify_ssl: A bool that indicates if the SSL certificate should be verified.
+            allow_redirects: A bool that indicates if URL redirects should be allowed.
             allow_errors: A boolean indicating if errors are allowed.
             verbose: Set this to False in order to disable printouts.
 
@@ -362,7 +401,6 @@ def _send_request(  # noqa: PLR0913,PLR0912,C901
                 url = self._base_url + url
             else:
                 url = self._api_url + url
-
         response = cast(requests.Response, None)
         retval: Union[Dict[str, Any], bytes] = {}
         if self._verbose and verbose:
@@ -372,23 +410,54 @@ def _send_request(  # noqa: PLR0913,PLR0912,C901
             if json_body:
                 print(f", {json_body=}", end="")
             print("")
-
         try:
             if request_type == SupportedRequestTypes.DELETE:
-                response = requests.delete(url, headers=headers, auth=auth, timeout=timeout)
+                response = requests.delete(
+                    url,
+                    headers=headers,
+                    auth=auth,
+                    timeout=timeout,
+                    verify=verify_ssl,
+                    allow_redirects=allow_redirects,
+                )
             elif request_type == SupportedRequestTypes.GET:
-                response = requests.get(url, headers=headers, auth=auth, timeout=timeout)
+                response = requests.get(
+                    url,
+                    headers=headers,
+                    auth=auth,
+                    timeout=timeout,
+                    verify=verify_ssl,
+                    allow_redirects=allow_redirects,
+                )
             elif request_type == SupportedRequestTypes.PATCH:
                 response = requests.patch(
-                    url, headers=headers, auth=auth, json=json_body, timeout=timeout
+                    url,
+                    headers=headers,
+                    auth=auth,
+                    json=json_body,
+                    timeout=timeout,
+                    verify=verify_ssl,
+                    allow_redirects=allow_redirects,
                 )
             elif request_type == SupportedRequestTypes.POST:
                 response = requests.post(
-                    url, headers=headers, auth=auth, json=json_body, timeout=timeout
+                    url,
+                    headers=headers,
+                    auth=auth,
+                    json=json_body,
+                    timeout=timeout,
+                    verify=verify_ssl,
+                    allow_redirects=allow_redirects,
                 )
             elif request_type == SupportedRequestTypes.PUT:
                 response = requests.put(
-                    url, headers=headers, auth=auth, json=json_body, timeout=timeout
+                    url,
+                    headers=headers,
+                    auth=auth,
+                    json=json_body,
+                    timeout=timeout,
+                    verify=verify_ssl,
+                    allow_redirects=allow_redirects,
                 )
             else:
                 msg = f"{request_type} is an unsupported request type."

diff --git a/src/tm_devices/drivers/pi/scopes/tekscope/tekscope.py b/src/tm_devices/drivers/pi/scopes/tekscope/tekscope.py
@@ -168,7 +168,7 @@ def channel(self) -> "MappingProxyType[str, TekScopeChannel]":
 
             # Set scope PI verbosity back to previous value
             self.set_and_check(":VERBose", old_pi_verbosity)
-        return MappingProxyType(channel_map)  # pyright: ignore[reportUnknownVariableType]
+        return MappingProxyType(channel_map)
 
     @property
     def commands(

diff --git a/src/tm_devices/helpers/constants_and_dataclasses.py b/src/tm_devices/helpers/constants_and_dataclasses.py
@@ -110,7 +110,8 @@ class SerialConfig(AsDictionaryUseEnumNameUseCustEnumStrValueMixin, _ConfigEntry
     flow_control: Optional[FlowControl] = None
     """Optional[FlowControl]: Control for pausing/resuming data stream between slower devices.
 
-    One of ``SerialConfig.FlowControl.[none|xon_xoff|dtr_dsr|rts_cts]``."""
+    One of ``SerialConfig.FlowControl.[none|xon_xoff|dtr_dsr|rts_cts]``.
+    """
     parity: Optional[Parity] = None
     """Optional[Parity]: Parity adds a checksum bit to each data character.
      Checksum bit enables the target device to determine whether the data was received correctly.

diff --git a/tests/test_margin_testers.py b/tests/test_margin_testers.py
@@ -69,8 +69,8 @@ def test_margin_tester(tmt4: MarginTester, device_manager: DeviceManager) -> Non
         del tmt4.fpga_version
     # should be same as mocked version
     assert tmt4.sw_version == Version("1.0.0.0")
-    assert tmt4.fw_version == Version("1.0.0.1")
-    assert tmt4.fpga_version == Version("1")
+    assert tmt4.fw_version == Version("1.0.0.1")  # pyright: ignore[reportUnknownMemberType]
+    assert tmt4.fpga_version == Version("1")  # pyright: ignore[reportUnknownMemberType]
     assert tmt4.manufacturer == "UNIT_TEST manufacturer"
     assert tmt4.model == "UNIT_TEST model"
     assert tmt4.serial == "UNIT_TEST serialNumber"

diff --git a/tests/test_scopes.py b/tests/test_scopes.py
@@ -44,7 +44,7 @@ def test_tekscope(device_manager: DeviceManager) -> None:  # noqa: PLR0915
     del scope.hostname
 
     # Assert 5 series device was added and aliased properly (USB)
-    assert scope.hostname == "MSO56"
+    assert scope.hostname == "MSO56"  # pyright: ignore[reportUnknownMemberType]
     assert id(device_manager.get_scope(number_or_alias="mso56")) == id(scope)
     assert id(device_manager.get_scope(number_or_alias=scope.device_number)) == id(scope)
     assert scope.all_channel_names_list == ("CH1", "CH2", "CH3", "CH4", "CH5", "CH6")