autopep8: avoid running on non-tracked files.

tests/gold_tests/autest-site/cli_tools.test.ext

@@ -31,19 +31,19 @@ Tools to help with TestRun commands
 #
 # Note that by default, the Default Process is created in this function.
 
-def spawn_commands(self, cmdstr, count, retcode = 0, use_default=True):
-    ret=[]
+def spawn_commands(self, cmdstr, count, retcode=0, use_default=True):
+    ret = []
 
     if use_default:
         count = int(count) - 1
-    for cnt in range(0,count):
+    for cnt in range(0, count):
         ret.append(
             self.Processes.Process(
                 name="cmdline-{num}".format(num=cnt),
-                cmdstr = cmdstr,
-                returncode = retcode
-                )
+                cmdstr=cmdstr,
+                returncode=retcode
             )
+        )
     if use_default:
         self.Processes.Default.Command = cmdstr
         self.Processes.Default.ReturnCode = retcode
@@ -52,4 +52,5 @@ def spawn_commands(self, cmdstr, count, retcode = 0, use_default=True):
         )
     return ret
 
+
 ExtendTestRun(spawn_commands, name="SpawnCommands")

tests/gold_tests/autest-site/conditions.test.ext

@@ -20,9 +20,10 @@ import subprocess
 import json
 import re
 
+
 def HasOpenSSLVersion(self, version):
     output = subprocess.check_output(
-        os.path.join(self.Variables.BINDIR, "traffic_layout") + " info --versions --json", shell = True
+        os.path.join(self.Variables.BINDIR, "traffic_layout") + " info --versions --json", shell=True
     )
     json_data = output.decode('utf-8')
     openssl_str = json.loads(json_data)['openssl_str']
@@ -34,8 +35,10 @@ def HasOpenSSLVersion(self, version):
         "openssl library version is " + exe_ver + ", must be at least " + version
     )
 
+
 def HasCurlVersion(self, version):
-    return self.EnsureVersion(["curl","--version"],min_version=version)
+    return self.EnsureVersion(["curl", "--version"], min_version=version)
+
 
 def HasCurlFeature(self, feature):
 
@@ -58,6 +61,8 @@ def HasCurlFeature(self, feature):
         default,
         "Curl needs to support feature: {feature}".format(feature=feature)
     )
+
+
 def HasCurlOption(self, option):
     def default(output):
         tag = option.lower()
@@ -75,20 +80,23 @@ def HasCurlOption(self, option):
         "Curl needs to support option: {option}".format(option=option)
     )
 
+
 def HasATSFeature(self, feature):
 
     val = self.Variables.get(feature, None)
 
     return self.Condition(
-        lambda: val == True,
+        lambda: val,
         "ATS feature not enabled: {feature}".format(feature=feature)
     )
 
-#test if a plugin exists in the libexec folder
+# test if a plugin exists in the libexec folder
+
+
 def PluginExists(self, pluginname):
 
     path = os.path.join(self.Variables.PLUGINDIR, pluginname)
-    return self.Condition(lambda: os.path.isfile(path) == True, path + " not found." )
+    return self.Condition(lambda: os.path.isfile(path), path + " not found.")
 
 
 ExtendCondition(HasOpenSSLVersion)
@@ -97,4 +105,3 @@ ExtendCondition(HasCurlVersion)
 ExtendCondition(HasCurlFeature)
 ExtendCondition(HasCurlOption)
 ExtendCondition(PluginExists)
-

tests/gold_tests/autest-site/copy_config.test.ext

@@ -34,7 +34,7 @@ class CopyATSConfig(SetupItem):
         process = self.process if self.process else self
         try:
             ts_dir = process.Env['TS_ROOT']
-        except:
+        except BaseException:
             if self.process:
                 raise SetupError(
                     'TS_ROOT is not defined. Cannot copy ats config file without location to copy to.'

tests/gold_tests/autest-site/curl_header.test.ext

@@ -22,6 +22,7 @@ import hosts.output as host
 
 import re
 
+
 class CurlHeader(Tester):
     def __init__(self,
                  value,
@@ -38,13 +39,13 @@ class CurlHeader(Tester):
                 stack=host.getCurrentStack(1)
             )
 
-        ops = ("equal","equal_re")
+        ops = ("equal", "equal_re")
         gold_dict = value
 
         headers = tuple(gold_dict.keys())
         if description is None:
-            description = 'Checking that all {} headers exist and have matching values: {}'.format(len(headers), \
-                ', '.join(headers) if len(headers) <= 10 else ', '.join(headers[0:10]) + ', etc.')
+            description = 'Checking that all {} headers exist and have matching values: {}'.format(
+                len(headers), ', '.join(headers) if len(headers) <= 10 else ', '.join(headers[0:10]) + ', etc.')
 
         # sanity check for input dictionary format
         for header, target in gold_dict.items():
@@ -54,34 +55,30 @@ class CurlHeader(Tester):
                     stack=self._stack
                 )
 
-            if target == None or isinstance(target, str):
+            if target is None or isinstance(target, str):
                 continue
             elif isinstance(target, dict):
                 for op, pos_val in target.items():
                     if op not in ops:
                         host.WriteError(
-                            'CurlHeader Input: Unsupported operation \'{}\' for value at header \'{}\'. The available operations are: {}.'.format(op, header, ', '.join(ops)),
-                            stack=self._stack
-                        )
-                    elif pos_val == None or isinstance(pos_val, str):
+                            'CurlHeader Input: Unsupported operation \'{}\' for value at header \'{}\'. The available operations are: {}.'.format(
+                                op, header, ', '.join(ops)), stack=self._stack)
+                    elif pos_val is None or isinstance(pos_val, str):
                         continue
                     elif isinstance(pos_val, list):
                         for str_ in pos_val:
-                            if not isinstance(str_, str) and str_ != None:
+                            if not isinstance(str_, str) and str_ is not None:
                                 host.WriteError(
-                                    'CurlHeader Input: Value {} has unsupported type \'{}\' for header \'{}\'. Need to provide a string or None.'.format(str_, str_.__class__.__name__, header),
-                                    stack=self._stack
-                                )
+                                    'CurlHeader Input: Value {} has unsupported type \'{}\' for header \'{}\'. Need to provide a string or None.'.format(
+                                        str_, str_.__class__.__name__, header), stack=self._stack)
                     else:
                         host.WriteError(
-                            'CurlHeader Input: Value {} has unsupported type \'{}\' for header \'{}\'. Need to provide a string, a list or None for possible curl values.'.format(pos_val, pos_val.__class__.__name__, header),
-                            stack=self._stack
-                        )
+                            'CurlHeader Input: Value {} has unsupported type \'{}\' for header \'{}\'. Need to provide a string, a list or None for possible curl values.'.format(
+                                pos_val, pos_val.__class__.__name__, header), stack=self._stack)
             else:
                 host.WriteError(
-                    'CurlHeader Input: Value {} has unsupported type \'{}\' for header \'{}\'. Need to provide either a string, a dictionary or None.'.format(target, target.__class__.__name__, header),
-                    stack=self._stack
-                )
+                    'CurlHeader Input: Value {} has unsupported type \'{}\' for header \'{}\'. Need to provide either a string, a dictionary or None.'.format(
+                        target, target.__class__.__name__, header), stack=self._stack)
 
         super(CurlHeader, self).__init__(
             value=value,
@@ -120,16 +117,16 @@ class CurlHeader(Tester):
                 val_dict[vals[0].lower()] = ': '.join(vals[1:])
 
         # generate a gold dictionary with lowercase header
-        gold_dict = {k.lower() : v for k, v in gold_dict.items()}
+        gold_dict = {k.lower(): v for k, v in gold_dict.items()}
 
         p_flag = 1
         reason = ''
 
         for header in gold_dict.keys():
             v = val_dict.get(header)
-            if v != None:
+            if v is not None:
                 res = self.match(gold_dict, header, v)
-                if res == None:
+                if res is None:
                     continue
                 else:
                     reason += 'In field: {} \n'.format(header) + res
@@ -165,7 +162,7 @@ class CurlHeader(Tester):
     def match(self, gold_dictionary, header, val_value):
         target = gold_dictionary[header]
 
-        if target == None:
+        if target is None:
             return None
         elif isinstance(target, str):
             # if given a string, check for an exact match
@@ -175,21 +172,21 @@ class CurlHeader(Tester):
         elif isinstance(target, dict):
             # if given a dict, check for valid operations indicated by the keys
             for op, pos_val in target.items():
-                if pos_val == None:
+                if pos_val is None:
                     return None
                 elif isinstance(pos_val, list):
                     # print('Need to provide a list of possible values.')
                     # continue
                     if op == 'equal':
                         for str_ in pos_val:
-                            if val_value == str_ or str_ == None:
+                            if val_value == str_ or str_ is None:
                                 return None
                         # return 'No matching strings in the list.'
                     elif op == 'equal_re':
                         for regex in pos_val:
-                            if regex == None:
+                            if regex is None:
                                 return None
-                            elif re.fullmatch(regex, val_value) != None:
+                            elif re.fullmatch(regex, val_value) is not None:
                                 return None
 
                 elif isinstance(pos_val, str):
@@ -199,17 +196,18 @@ class CurlHeader(Tester):
                             # else 'Not an exact match.\nExpected : {}\nActual   : {}'.format(pos_val, val_value)
 
                     elif op == 'equal_re':
-                        if re.fullmatch(pos_val, val_value) != None:
+                        if re.fullmatch(pos_val, val_value) is not None:
                             return None
 
             ret = ''
-            ops = {'equal' : 'Any of the following strings: ',
-                   'equal_re' : 'Any of the following regular expression: '}
+            ops = {'equal': 'Any of the following strings: ',
+                   'equal_re': 'Any of the following regular expression: '}
 
             for op, pos_val in target.items():
                 ret += '    {}: \'{}\'\n'.format(ops[op], '\', \''.join(pos_val)) if isinstance(pos_val, list) \
                     else '    {}: \'{}\'\n'.format(ops[op], pos_val)
 
             return 'Value \'{}\' matches none of: \n{}'.format(val_value, ret.strip('\n'))
 
+
 AddTester(CurlHeader)

tests/gold_tests/autest-site/init.cli.ext

@@ -26,7 +26,7 @@ autest_version = "1.8.1"
 if AuTestVersion() < autest_version:
     host.WriteError(
         "Tests need AuTest version {needed_version} or better, found version {found_version}\n"
-            "Please update AuTest:\n  pip install --upgrade autest\n".format(
+        "Please update AuTest:\n  pip install --upgrade autest\n".format(
             needed_version=autest_version,
             found_version=AuTestVersion()),
         show_stack=False)

tests/gold_tests/autest-site/ip.test.ext

@@ -27,11 +27,14 @@ from ports import get_port
 # For each argument, a port will be reserved, and its number will be assigned to the new variable for the
 # argument.
 #
+
+
 def get_tcp_port(obj, *newVariables):
     for v in newVariables:
         if not isinstance(v, str):
             raise TypeError("all function arguments must be strings")
         get_port(obj, v)
 
+
 #AddTestEntityMember(get_tcp_port, name="GetTcpPort")
 ExtendTest(get_tcp_port, name="GetTcpPort")

tests/gold_tests/autest-site/microserver.test.ext

@@ -59,8 +59,14 @@ def getHeaderFieldVal(request_header, field):
 
 # addResponse adds customized response with respect to request_header. request_header and response_header are both dictionaries
 def addResponse(self, filename, request_header, response_header):
-    client_request = Request.fromRequestLine(request_header["headers"], request_header["body"], None if "options" not in request_header else request_header["options"])
-    server_response = Response.fromRequestLine(response_header["headers"], response_header["body"], None if "options" not in response_header else response_header["options"])
+    client_request = Request.fromRequestLine(
+        request_header["headers"],
+        request_header["body"],
+        None if "options" not in request_header else request_header["options"])
+    server_response = Response.fromRequestLine(
+        response_header["headers"],
+        response_header["body"],
+        None if "options" not in response_header else response_header["options"])
 
     # timestamp field is left None because that needs to be revised for better implementation
     txn = Transaction(client_request, None, server_response, None, None, None)
@@ -81,7 +87,7 @@ def addTransactionToSession(txn, JFile):
 
     # hard coding only 1 session per file
     # since for the purpose of testing, we don't need multiple sessions in a file
-    if jsondata == None:
+    if jsondata is None:
         jsondata = {}
         jsondata["sessions"] = []
 
@@ -165,7 +171,19 @@ def uServerUpAndRunning(serverHost, port, isSsl, isIPv6, request, clientcert='',
 AddWhenFunction(uServerUpAndRunning)
 
 
-def MakeOriginServer(obj, name, port=None, s_port=None, ip='INADDR_LOOPBACK', delay=None, ssl=False, lookup_key=DEFAULT_LOOKUP_KEY, clientcert='', clientkey='', both=False, options={}):
+def MakeOriginServer(
+        obj,
+        name,
+        port=None,
+        s_port=None,
+        ip='INADDR_LOOPBACK',
+        delay=None,
+        ssl=False,
+        lookup_key=DEFAULT_LOOKUP_KEY,
+        clientcert='',
+        clientkey='',
+        both=False,
+        options={}):
     data_dir = os.path.join(obj.RunDirectory, name)
     p = obj.Processes.Process(name)
 
@@ -229,7 +247,14 @@ def MakeOriginServer(obj, name, port=None, s_port=None, ip='INADDR_LOOPBACK', de
         "options": {"skipHooks": None}
     })
 
-    p.Ready = When.uServerUpAndRunning(ipaddr, s_port if ssl else port, ssl, IPConstants.isIPv6(ip), healthcheck_request["headers"], clientcert=clientcert, clientkey=clientkey)
+    p.Ready = When.uServerUpAndRunning(
+        ipaddr,
+        s_port if ssl else port,
+        ssl,
+        IPConstants.isIPv6(ip),
+        healthcheck_request["headers"],
+        clientcert=clientcert,
+        clientkey=clientkey)
     p.ReturnCode = Any(None, 0)
 
     return p

tests/gold_tests/autest-site/setup.cli.ext

@@ -36,7 +36,7 @@ if ENV['ATS_BIN'] is not None:
         hint = ''
         if os.path.isfile(os.path.join(ENV['ATS_BIN'], 'bin', 'traffic_layout')):
             hint = "\nDid you mean '--ats-bin {}'?".\
-                format(os.path.join(ENV['ATS_BIN'],'bin'))
+                format(os.path.join(ENV['ATS_BIN'], 'bin'))
         host.WriteError("traffic_layout is not found. Aborting tests - Bad build or install.{}".format(hint), show_stack=False)
     try:
         out = subprocess.check_output([traffic_layout, "--json"])

tests/gold_tests/autest-site/traffic_replay.test.ext

@@ -17,9 +17,11 @@
 #  limitations under the License.
 
 # default 'mixed' for connection type since it doesn't hurt
+
+
 def Replay(obj, name, replay_dir, key=None, cert=None, conn_type='mixed', options={}):
     # ATS setup - one line because we leave records and remap config to user
-    ts = obj.MakeATSProcess("ts", select_ports=False) # select ports can be disabled once we add ssl port selection in extension
+    ts = obj.MakeATSProcess("ts", select_ports=False)  # select ports can be disabled once we add ssl port selection in extension
 
     # TEMP
     ts.Variables.ssl_port = 4443
@@ -62,7 +64,8 @@ def Replay(obj, name, replay_dir, key=None, cert=None, conn_type='mixed', option
     if not cert:
         cert = os.path.join(obj.Variables["AtsTestToolsDir"], "microserver", "ssl", "server.crt")
 
-    command = 'traffic-replay --log_dir {0} --type {1} --verify --host {2} --port {3} --s_port {4} '.format(data_dir, conn_type, hostIP, ts.Variables.port, ts.Variables.ssl_port)
+    command = 'traffic-replay --log_dir {0} --type {1} --verify --host {2} --port {3} --s_port {4} '.format(
+        data_dir, conn_type, hostIP, ts.Variables.port, ts.Variables.ssl_port)
 
     if key:
         command += "-k {0} ".format(key)
@@ -88,4 +91,5 @@ def Replay(obj, name, replay_dir, key=None, cert=None, conn_type='mixed', option
     # return all the stuff in case user wants to do extra optimization
     return (ts, server, dns, tr)
 
+
 AddTestRunSet(Replay)

tests/gold_tests/autest-site/trafficserver.test.ext

@@ -63,7 +63,7 @@ def MakeATSProcess(obj, name, command='traffic_server', select_ports=True, enabl
     p = obj.Processes.Process(name, command)
     #p_debug = obj.Processes.Process("port-debug", "ss --listen --tcp --process")
     #p_debug.Env['PATH'] = "/usr/sbin" + os.pathsep + p.ComposeEnv()['PATH']
-    #p.StartBefore(p_debug)
+    # p.StartBefore(p_debug)
     # we want to have a few directories more fixed
     # this helps with debugging as location are common
     # we do this by overiding locations from the "layout"
@@ -289,7 +289,8 @@ def MakeATSProcess(obj, name, command='traffic_server', select_ports=True, enabl
         port_str = "{port} {v6_port}:ipv6 ".format(port=p.Variables.port, v6_port=p.Variables.portv6)
 
         if enable_tls:
-            port_str += "{ssl_port}:ssl {ssl_portv6}:ssl:ipv6".format(ssl_port=p.Variables.ssl_port,ssl_portv6=p.Variables.ssl_portv6)
+            port_str += "{ssl_port}:ssl {ssl_portv6}:ssl:ipv6".format(
+                ssl_port=p.Variables.ssl_port, ssl_portv6=p.Variables.ssl_portv6)
 
         p.Env['PROXY_CONFIG_HTTP_SERVER_PORTS'] = port_str