From fef7ce13271e41ee19645e6b5e594a9bd0faea4b Mon Sep 17 00:00:00 2001 From: Andrew Teixeira Date: Wed, 14 Sep 2022 14:34:45 -0400 Subject: [PATCH] Fix for Python 3.10 Allow newer versions of requests-toolbelt Fix all uses of collections --- pytos/common/rest_requests.py | 7 +++++-- pytos/securechange/xml_objects/rest.py | 7 +++++-- pytos/securetrack/helpers.py | 23 +++++++++++++---------- setup.py | 2 +- 4 files changed, 24 insertions(+), 15 deletions(-) diff --git a/pytos/common/rest_requests.py b/pytos/common/rest_requests.py index 04d0d47..832f699 100644 --- a/pytos/common/rest_requests.py +++ b/pytos/common/rest_requests.py @@ -1,5 +1,8 @@ -import collections +try: + from collections import Iterable +except: + from collections.abc import Iterable import datetime import hashlib import http.cookiejar @@ -222,7 +225,7 @@ def _ensure_response_status(self): status_code_ok = False if status_code_ok: - if isinstance(self.expected_status_codes, collections.Iterable): + if isinstance(self.expected_status_codes, Iterable): if self.response.status_code not in self.expected_status_codes: status_code_ok = False elif isinstance(self.expected_status_codes, int): diff --git a/pytos/securechange/xml_objects/rest.py b/pytos/securechange/xml_objects/rest.py index ea3e1e6..3f7f8f7 100644 --- a/pytos/securechange/xml_objects/rest.py +++ b/pytos/securechange/xml_objects/rest.py @@ -1,7 +1,11 @@ import datetime import time import enum -from collections import OrderedDict +try: + from collections import OrderedDict +except: + from collections.abc import OrderedDict + # For backward compatibility import FileLock from pytos.securechange.xml_objects.restapi.step.rule_decommission.rule_decommission import Step_Field_Rule_Decommission @@ -2132,4 +2136,3 @@ def from_xml_node(cls, xml_node): class RejectComment(Comment): def __init__(self, comment): super().__init__(comment, Elements.REJECT_COMMENT) - diff --git a/pytos/securetrack/helpers.py b/pytos/securetrack/helpers.py index 5d66482..664ed49 100644 --- a/pytos/securetrack/helpers.py +++ b/pytos/securetrack/helpers.py @@ -1,5 +1,8 @@ # coding=utf-8 -import collections +try: + from collections import Iterable +except: + from collections.abc import Iterable import csv import io import itertools @@ -1320,7 +1323,7 @@ def get_rule_documentation_by_device_id_and_rule_id(self, device_id, rule_id): def get_rule_base(self, get_documentation=False, device_ids=None): """Get the rules for each of the devices configured in SecureTrack. - :type device_ids: collections.Iterable[int] + :type device_ids: Iterable[int] :param device_ids: If specified, get the rule base only for the specified devices. :param get_documentation: Whether or not to get the rule documentation together with the rule base. :type get_documentation: bool @@ -2140,13 +2143,13 @@ def get_services_by_device_and_object_ids(self, device_id, service_ids): :param device_id: The device ID for which we want to get network objects. :type device_id: int :param service_ids: The ID of the service - :type service_ids: int|collections.Iterable[int] + :type service_ids: int|Iterable[int] :return: The service for the specified device with the specified ID. :rtype: Single_Service|Group_Service :raise ValueError: If a device with the specified ID does not exist. :raise IOError: If there was a communication problem trying to get the network objects. """ - if isinstance(service_ids, collections.Iterable): + if isinstance(service_ids, Iterable): service_ids = ",".join([str(service_id) for service_id in service_ids]) logger.info("Getting service with ID %s for device %s.", service_ids, device_id) try: @@ -2169,13 +2172,13 @@ def get_services_by_revision_and_object_ids(self, revision_id, service_ids=""): :param revision_id: The revision ID for which we want to get network objects. :type revision_id: int :param service_ids: The ID of the service - :type service_ids: int|collections.Iterable[int] + :type service_ids: int|Iterable[int] :return: The service for the specified revision with the specified ID. :rtype: Services_List :raise ValueError: If a revision with the specified ID does not exist. :raise IOError: If there was a communication problem trying to get the services. """ - if isinstance(service_ids, collections.Iterable): + if isinstance(service_ids, Iterable): service_ids = ",".join([str(service_id) for service_id in service_ids]) logger.info("Getting service with ID %s for revision %s.", service_ids, revision_id) try: @@ -2198,14 +2201,14 @@ def get_network_objects_by_revision_and_object_ids(self, revision_id, network_ob :param revision_id: The revision ID for which we want to get network objects. :type revision_id: int :param network_object_ids: The ID of the network object to get - :type network_object_ids: int|collections.Iterable[int] + :type network_object_ids: int|Iterable[int] :return: The network objects for the specified revision. :rtype: Network_Objects_List :raise ValueError: If a revision with the specified ID does not exist. :raise IOError: Ifp there was a communication problem trying to get the network objects. """ logger.info("Getting network object with ID %s for revision %s.", network_object_ids, revision_id) - if isinstance(network_object_ids, collections.Iterable): + if isinstance(network_object_ids, Iterable): network_object_ids = ",".join([str(network_object_id) for network_object_id in network_object_ids]) try: response_string = self.get_uri( @@ -2227,14 +2230,14 @@ def get_network_objects_by_device_and_object_ids(self, device_id, network_object :param device_id: The device ID for which we want to get network objects. :type device_id: int :param network_object_ids: The ID of the network object to get - :type network_object_ids: int|collections.Iterable[int] + :type network_object_ids: int|Iterable[int] :return: The network objects for the specified device. :rtype: Network_Objects_List :raise ValueError: If a device with the specified ID does not exist. :raise IOError: Ifp there was a communication problem trying to get the network objects. """ logger.info("Getting network object with ID %s for device %s.", network_object_ids, device_id) - if isinstance(network_object_ids, collections.Iterable): + if isinstance(network_object_ids, Iterable): network_object_ids = ",".join([str(network_object_id) for network_object_id in network_object_ids]) try: response_string = self.get_uri( diff --git a/setup.py b/setup.py index a3fa5b0..d0cee0f 100644 --- a/setup.py +++ b/setup.py @@ -40,7 +40,7 @@ def get_version(): 'netaddr>=0.7.14', 'paramiko>=1.15.2', 'requests>=2.6.0', - 'requests_toolbelt==0.7.1', + 'requests_toolbelt>=0.7.1', 'netifaces==0.10.9', 'dnspython3==1.15.0', 'Mako',