Update implementation.py to check for .exploded attribute

docs/changelog/2023/may.rst

@@ -0,0 +1,9 @@
+--------------------------------------------------------------------------------
+                                      Fix
+--------------------------------------------------------------------------------
+
+* All connectors (except xpresso)
+    * Change implementation of ip/host address to remove default .exploded
+      attribute. If an ipaddress object is passed, .exploded will still be
+      used, otherwise use the raw IP data. This enables dynamic testbed
+      creation using string values for IP addresses.

src/rest/connector/libs/apic/acisdk_implementation.py

@@ -1,6 +1,7 @@
 import requests
 import urllib3
 
+from ipaddress import ip_address, IPv4Address, IPv6Address
 from functools import wraps
 from importlib import import_module
 from logging import getLogger
@@ -57,7 +58,16 @@ def __init__(self, *args, **kwargs):
         if 'host' in self.connection_info:
             ip = self.connection_info['host']
         else:
-            ip = self.connection_info['ip'].exploded
+            ip = self.connection_info['ip']
+            if not isinstance(ip, (IPv4Address, IPv6Address)):
+                ip = ip_address(ip)
+
+            # Properly format IPv6 URL if a v6 address is provided
+            if isinstance(ip, IPv6Address):
+                ip = f"[{ip.exploded}]"
+            else:
+                ip = ip.exploded
+
         if 'port' in self.connection_info:
             port = self.connection_info['port']
             self.url = f'https://{ip}:{port}/'

src/rest/connector/libs/apic/implementation.py

@@ -3,6 +3,7 @@
 import logging
 import requests
 
+from ipaddress import ip_address, IPv4Address, IPv6Address
 from requests.exceptions import RequestException
 
 
@@ -102,7 +103,16 @@ def connect(self, timeout=30, retries=3, retry_wait=10):
         if 'host' in self.connection_info:
             ip = self.connection_info['host']
         else:
-            ip = self.connection_info['ip'].exploded
+            ip = self.connection_info['ip']
+            if not isinstance(ip, (IPv4Address, IPv6Address)):
+                ip = ip_address(ip)
+
+            # Properly format IPv6 URL if a v6 address is provided
+            if isinstance(ip, IPv6Address):
+                ip = f"[{ip.exploded}]"
+            else:
+                ip = ip.exploded
+
         if 'port' in self.connection_info:
             port = self.connection_info['port']
             self.url = 'https://{ip}:{port}/'.format(ip=ip, port=port)

src/rest/connector/libs/bigip/implementation.py

@@ -2,6 +2,8 @@
 import logging
 import time
 
+from ipaddress import ip_address, IPv4Address, IPv6Address
+
 # Genie, pyATS, ROBOT imports
 # from pyats.connections import BaseConnection
 from rest.connector.utils import get_username_password
@@ -169,7 +171,16 @@ def connect(
                     "Cannot add ssh tunnel. Connection %s may not have ip/host or port.\n%s"
                     % (self.via, e))
         else:
-            self.ip = self.connection_info['ip'].exploded
+            self.ip = self.connection_info['ip']
+            if not isinstance(self.ip, (IPv4Address, IPv6Address)):
+                self.ip = ip_address(self.ip)
+
+            # Properly format IPv6 URL if a v6 address is provided
+            if isinstance(self.ip, IPv6Address):
+                self.ip = f"[{self.ip.exploded}]"
+            else:
+                self.ip = self.ip.exploded
+
             self.port = self.connection_info.get('port', port)
 
         if 'protocol' in self.connection_info:

src/rest/connector/libs/dcnm/implementation.py

@@ -3,6 +3,7 @@
 import json
 import logging
 import requests
+from ipaddress import ip_address, IPv4Address, IPv6Address
 from requests.auth import HTTPBasicAuth
 from requests.exceptions import RequestException
 
@@ -128,7 +129,16 @@ def connect(self, timeout=10, port=443, protocol='https'):
             try:
                 host = self.connection_info['host']
             except KeyError:
-                host = self.connection_info['ip'].exploded
+                host = self.connection_info['ip']
+                if not isinstance(host, (IPv4Address, IPv6Address)):
+                    host = ip_address(host)
+
+                # Properly format IPv6 URL if a v6 address is provided
+                if isinstance(host, IPv6Address):
+                    host = f"[{host.exploded}]"
+                else:
+                    host = host.exploded
+
             port = self.connection_info.get('port', port)
 
         if 'protocol' in self.connection_info:

src/rest/connector/libs/dnac/implementation.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import requests
+from ipaddress import ip_address, IPv4Address, IPv6Address
 from requests.auth import HTTPBasicAuth
 from requests.exceptions import RequestException
 
@@ -78,8 +79,16 @@ def connect(self, timeout=30):
         try:
             host = self.connection_info['host']
         except KeyError:
-            host = self.connection_info['ip'].exploded
-
+            host = self.connection_info['ip']
+            if not isinstance(host, (IPv4Address, IPv6Address)):
+                host = ip_address(host)
+
+            # Properly format IPv6 URL if a v6 address is provided
+            if isinstance(host, IPv6Address):
+                host = f"[{host.exploded}]"
+            else:
+                host = host.exploded
+
         port = self.connection_info.get('port', 443)
         self.verify = self.connection_info.get('verify', True)
 

src/rest/connector/libs/elasticsearch/implementation.py

@@ -2,6 +2,7 @@
 import logging
 import requests
 
+from ipaddress import ip_address, IPv4Address, IPv6Address
 from requests.exceptions import RequestException
 
 from pyats.connections import BaseConnection
@@ -94,7 +95,16 @@ def connect(self, timeout=30, port="9200", protocol='https'):
         if 'host' in self.connection_info:
             ip = self.connection_info['host']
         else:
-            ip = self.connection_info['ip'].exploded
+            ip = self.connection_info['ip']
+            if not isinstance(ip, (IPv4Address, IPv6Address)):
+                ip = ip_address(ip)
+
+            # Properly format IPv6 URL if a v6 address is provided
+            if isinstance(ip, IPv6Address):
+                ip = f"[{ip.exploded}]"
+            else:
+                ip = ip.exploded
+
         if 'protocol' in self.connection_info:
             protocol = self.connection_info['protocol']
         if 'port' in self.connection_info:

src/rest/connector/libs/iosxe/implementation.py

@@ -3,6 +3,7 @@
 import re
 import urllib.request
 import requests
+from ipaddress import ip_address, IPv4Address, IPv6Address
 from dict2xml import dict2xml
 from requests.exceptions import RequestException
 
@@ -114,7 +115,16 @@ def connect(self,
                     "Cannot add ssh tunnel. Connection %s may not have ip/host or port.\n%s"
                     % (self.via, e))
         else:
-            ip = self.connection_info.ip.exploded
+            ip = self.connection_info['ip']
+            if not isinstance(ip, (IPv4Address, IPv6Address)):
+                ip = ip_address(ip)
+
+            # Properly format IPv6 URL if a v6 address is provided
+            if isinstance(ip, IPv6Address):
+                ip = f"[{ip.exploded}]"
+            else:
+                ip = ip.exploded
+
             port = self.connection_info.get('port', port)
 
         if 'protocol' in self.connection_info:

src/rest/connector/libs/nd/implementation.py

@@ -4,6 +4,7 @@
 import time
 import urllib.parse
 
+from ipaddress import ip_address, IPv4Address, IPv6Address
 from requests.exceptions import RequestException
 
 from pyats.connections import BaseConnection
@@ -93,7 +94,16 @@ def connect(self, timeout=30, retries=3, retry_wait=10):
         if 'host' in self.connection_info:
             ip = self.connection_info['host']
         else:
-            ip = self.connection_info['ip'].exploded
+            ip = self.connection_info['ip']
+            if not isinstance(ip, (IPv4Address, IPv6Address)):
+                ip = ip_address(ip)
+
+            # Properly format IPv6 URL if a v6 address is provided
+            if isinstance(ip, IPv6Address):
+                ip = f"[{ip.exploded}]"
+            else:
+                ip = ip.exploded
+
         if 'port' in self.connection_info:
             port = self.connection_info['port']
             self.url = 'https://{ip}:{port}/'.format(ip=ip, port=port)

src/rest/connector/libs/nexusdashboard/implementation.py

@@ -3,6 +3,8 @@
 import json
 import logging
 import requests
+
+from ipaddress import ip_address, IPv4Address, IPv6Address
 from requests.auth import HTTPBasicAuth
 from requests.exceptions import RequestException
 
@@ -108,7 +110,15 @@ def connect(self, timeout=10, protocol='https'):
         try:
             host = self.connection_info['host']
         except KeyError:
-            host = self.connection_info['ip'].exploded
+            host = self.connection_info['ip']
+            if not isinstance(host, (IPv4Address, IPv6Address)):
+                host = ip_address(host)
+
+            # Properly format IPv6 URL if a v6 address is provided
+            if isinstance(host, IPv6Address):
+                host = f"[{host.exploded}]"
+            else:
+                host = host.exploded
 
         if 'protocol' in self.connection_info:
             protocol = self.connection_info['protocol']

src/rest/connector/libs/nso/implementation.py

@@ -2,6 +2,7 @@
 import json
 import logging
 import requests
+from ipaddress import ip_address, IPv4Address, IPv6Address
 from dict2xml import dict2xml
 from requests.exceptions import RequestException
 
@@ -108,7 +109,16 @@ def connect(self,
                     "Cannot add ssh tunnel. Connection %s may not have ip/host or port.\n%s"
                     % (self.via, e))
         else:
-            ip = self.connection_info['ip'].exploded
+            ip = self.connection_info['ip']
+            if not isinstance(ip, (IPv4Address, IPv6Address)):
+                ip = ip_address(ip)
+
+            # Properly format IPv6 URL if a v6 address is provided
+            if isinstance(ip, IPv6Address):
+                ip = f"[{ip.exploded}]"
+            else:
+                ip = ip.exploded
+
             port = self.connection_info.get('port', port)
 
         if 'protocol' in self.connection_info:

src/rest/connector/libs/nxos/aci/implementation.py

@@ -2,9 +2,9 @@
 import logging
 import requests
 
+from ipaddress import ip_address, IPv4Address, IPv6Address
 from requests.exceptions import RequestException
 
-
 from pyats.connections import BaseConnection
 from rest.connector.implementation import Implementation as Imp
 from rest.connector.utils import get_username_password
@@ -105,7 +105,16 @@ def connect(self, timeout=30):
         if 'host' in self.connection_info:
             ip = self.connection_info['host']
         else:
-            ip = self.connection_info['ip'].exploded
+            ip = self.connection_info['ip']
+            if not isinstance(ip, (IPv4Address, IPv6Address)):
+                ip = ip_address(ip)
+
+            # Properly format IPv6 URL if a v6 address is provided
+            if isinstance(ip, IPv6Address):
+                ip = f"[{ip.exploded}]"
+            else:
+                ip = ip.exploded
+
         if 'port' in self.connection_info:
             port = self.connection_info['port']
             self.url = 'https://{ip}:{port}/'.format(ip=ip, port=port)

src/rest/connector/libs/nxos/implementation.py

@@ -2,6 +2,8 @@
 import time
 import logging
 import requests
+
+from ipaddress import ip_address, IPv4Address, IPv6Address
 from requests.auth import HTTPBasicAuth
 from requests.exceptions import RequestException
 
@@ -136,7 +138,16 @@ def connect(self, timeout=30, port=443, protocol='https', retries=3, retry_wait=
                     "Cannot add ssh tunnel. Connection %s may not have ip/host or port.\n%s"
                     % (self.via, e))
         else:
-            ip = self.connection_info['ip'].exploded
+            ip = self.connection_info['ip']
+            if not isinstance(ip, (IPv4Address, IPv6Address)):
+                ip = ip_address(ip)
+
+            # Properly format IPv6 URL if a v6 address is provided
+            if isinstance(ip, IPv6Address):
+                ip = f"[{ip.exploded}]"
+            else:
+                ip = ip.exploded
+
             port = self.connection_info.get('port', port)
         if 'protocol' in self.connection_info:
             protocol = self.connection_info['protocol']

src/rest/connector/libs/viptela/implementation.py

@@ -2,6 +2,8 @@
 import logging
 import requests
 
+from ipaddress import ip_address, IPv4Address, IPv6Address
+
 from pyats.connections import BaseConnection
 from rest.connector.implementation import Implementation
 from rest.connector.utils import get_username_password
@@ -101,7 +103,16 @@ def connect(self, timeout=30, port="8443", protocol='https'):
                     "Cannot add ssh tunnel. Connection %s may not have ip/host or port.\n%s"
                     % (self.via, e))
         else:
-            ip = self.connection_info['ip'].exploded
+            ip = self.connection_info['ip']
+            if not isinstance(ip, (IPv4Address, IPv6Address)):
+                ip = ip_address(ip)
+
+            # Properly format IPv6 URL if a v6 address is provided
+            if isinstance(ip, IPv6Address):
+                ip = f"[{ip.exploded}]"
+            else:
+                ip = ip.exploded
+
             port = self.connection_info.get('port', port)
 
         if 'protocol' in self.connection_info:

src/rest/connector/libs/virl/implementation.py

@@ -1,9 +1,9 @@
 import logging
 import requests
 
+from ipaddress import ip_address, IPv4Address, IPv6Address
 from requests.exceptions import RequestException
 
-
 from pyats.connections import BaseConnection
 from rest.connector.implementation import Implementation
 from rest.connector.utils import get_username_password
@@ -117,7 +117,16 @@ def connect(self, timeout=30, port="19399", protocol='http'):
                     "Cannot add ssh tunnel. Connection %s may not have ip/host or port.\n%s"
                     % (self.via, e))
         else:
-            ip = self.connection_info['ip'].exploded
+            ip = self.connection_info['ip']
+            if not isinstance(ip, (IPv4Address, IPv6Address)):
+                ip = ip_address(ip)
+
+            # Properly format IPv6 URL if a v6 address is provided
+            if isinstance(ip, IPv6Address):
+                ip = f"[{ip.exploded}]"
+            else:
+                ip = ip.exploded
+
             port = self.connection_info.get('port', '19399')
 
         if 'protocol' in self.connection_info: