diff --git a/CHANGELOG.md b/CHANGELOG.md index 7d1babbf..22d90af5 100644 --- 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 index 468ed1de..4fc7fccc 100644 --- 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 `__. If the issue or question you have is not already covered, please -`file a new issue `__ and the +`file a new issue `__ or +start a `discussion `__ 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 index 0f583bb0..f3125f7d 100644 --- 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 index d437b1d0..e07a33fa 100644 --- 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 index 92d6a99c..2ee45d41 100644 --- 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 index 71a81275..722f5127 100644 --- 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 index eec81df7..367088d0 100644 --- 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")