code changes

Author: ashah-splunk

tests/__init__.py

@@ -20,8 +20,8 @@
 import sys
 import logging
 
-import pytest
 from unittest import main, TestCase
+import pytest
 from splunklib.six.moves import cStringIO as StringIO
 
 

tests/modularinput/__init__.py

@@ -20,21 +20,17 @@
 import sys
 
 from io import TextIOWrapper
+import pytest
 
 from splunklib.searchcommands import Configuration, Option, environment, validators
 from splunklib.searchcommands.decorators import ConfigurationSetting
 from splunklib.searchcommands.internals import json_encode_string
 from splunklib.searchcommands.search_command import SearchCommand
 
-try:
-    from tests.searchcommands import rebase_environment
-except ImportError:
-    # Skip on Python 2.6
-    pass
+from tests.searchcommands import rebase_environment
 
 from splunklib import six
 
-import pytest
 
 
 @Configuration()
@@ -302,7 +298,7 @@ def fix_up(cls, command_class):
                 except Exception as error:
                     self.assertIsInstance(error, ValueError, 'Expected ValueError, not {}({}) for {}={}'.format(type(error).__name__, error, name, repr(value)))
                 else:
-                    self.fail('Expected ValueError, not success for {}={}'.format(name, repr(value)))
+                    self.fail(f'Expected ValueError, not success for {name}={repr(value)}')
 
                 settings_class = new_configuration_settings_class()
                 settings_instance = settings_class(command=None)
@@ -451,7 +447,7 @@ def test_option(self):
             elif type(x.validator).__name__ == 'RegularExpression':
                 self.assertEqual(expected[x.name], x.value.pattern)
             elif isinstance(x.value, TextIOWrapper):
-                self.assertEqual(expected[x.name], "'%s'" % x.value.name)
+                self.assertEqual(expected[x.name], f"'{x.value.name}'" )
             elif not isinstance(x.value, (bool,) + (float,) + (six.text_type,) + (six.binary_type,) + tuplewrap(six.integer_types)):
                 self.assertEqual(expected[x.name], repr(x.value))
             else:

tests/searchcommands/test_builtin_options.py

@@ -23,7 +23,7 @@ def generate(self):
     ds = chunky.ChunkedDataStream(out_stream)
     is_first_chunk = True
     finished_seen = False
-    expected = set([str(i) for i in range(1, 10)])
+    expected = set(str(i) for i in range(1, 10))
     seen = set()
     for chunk in ds:
         if is_first_chunk:

tests/searchcommands/test_decorators.py

@@ -14,26 +14,23 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+from contextlib import closing
+from unittest import main, TestCase
+import os
+import pytest
+from functools import reduce
 
 from splunklib.searchcommands.internals import CommandLineParser, InputHeader, RecordWriterV1
 from splunklib.searchcommands.decorators import Configuration, Option
 from splunklib.searchcommands.validators import Boolean
 
 from splunklib.searchcommands.search_command import SearchCommand
 
-from contextlib import closing
 from splunklib.six import StringIO, BytesIO
 
-from splunklib.six.moves import zip as izip
-
-from unittest import main, TestCase
-
-import os
 from splunklib import six
 from splunklib.six.moves import range
-from functools import reduce
 
-import pytest
 
 @pytest.mark.smoke
 class TestInternals(TestCase):
@@ -93,7 +90,8 @@ def fix_up(cls, command_class): pass
         CommandLineParser.parse(command, ['required_option=true'] + fieldnames)
 
         for option in six.itervalues(command.options):
-            if option.name in ['unnecessary_option', 'logging_configuration', 'logging_level', 'record', 'show_configuration']:
+            if option.name in ['unnecessary_option', 'logging_configuration', 'logging_level', 'record',
+                               'show_configuration']:
                 self.assertFalse(option.is_set)
                 continue
             self.assertTrue(option.is_set)
@@ -111,7 +109,8 @@ def fix_up(cls, command_class): pass
 
         # Command line with unrecognized options
 
-        self.assertRaises(ValueError, CommandLineParser.parse, command, ['unrecognized_option_1=foo', 'unrecognized_option_2=bar'])
+        self.assertRaises(ValueError, CommandLineParser.parse, command,
+                          ['unrecognized_option_1=foo', 'unrecognized_option_2=bar'])
 
         # Command line with a variety of quoted/escaped text options
 
@@ -175,24 +174,23 @@ def fix_up(cls, command_class): pass
             argv = [string]
             self.assertRaises(SyntaxError, CommandLineParser.parse, command, argv)
 
-
     def test_command_line_parser_unquote(self):
         parser = CommandLineParser
 
         options = [
-            r'foo',                 # unquoted string with no escaped characters
-            r'fo\o\ b\"a\\r',       # unquoted string with some escaped characters
-            r'"foo"',               # quoted string with no special characters
-            r'"""foobar1"""',       # quoted string with quotes escaped like this: ""
-            r'"\"foobar2\""',       # quoted string with quotes escaped like this: \"
-            r'"foo ""x"" bar"',     # quoted string with quotes escaped like this: ""
-            r'"foo \"x\" bar"',     # quoted string with quotes escaped like this: \"
-            r'"\\foobar"',          # quoted string with an escaped backslash
-            r'"foo \\ bar"',        # quoted string with an escaped backslash
-            r'"foobar\\"',          # quoted string with an escaped backslash
-            r'foo\\\bar',           # quoted string with an escaped backslash and an escaped 'b'
-            r'""',                  # pair of quotes
-            r'']                    # empty string
+            r'foo',  # unquoted string with no escaped characters
+            r'fo\o\ b\"a\\r',  # unquoted string with some escaped characters
+            r'"foo"',  # quoted string with no special characters
+            r'"""foobar1"""',  # quoted string with quotes escaped like this: ""
+            r'"\"foobar2\""',  # quoted string with quotes escaped like this: \"
+            r'"foo ""x"" bar"',  # quoted string with quotes escaped like this: ""
+            r'"foo \"x\" bar"',  # quoted string with quotes escaped like this: \"
+            r'"\\foobar"',  # quoted string with an escaped backslash
+            r'"foo \\ bar"',  # quoted string with an escaped backslash
+            r'"foobar\\"',  # quoted string with an escaped backslash
+            r'foo\\\bar',  # quoted string with an escaped backslash and an escaped 'b'
+            r'""',  # pair of quotes
+            r'']  # empty string
 
         expected = [
             r'foo',
@@ -286,7 +284,7 @@ def test_input_header(self):
             'sentence': 'hello world!'}
 
         input_header = InputHeader()
-        text = reduce(lambda value, item: value + '{}:{}\n'.format(item[0], item[1]), six.iteritems(collection), '') + '\n'
+        text = reduce(lambda value, item: value + f'{item[0]}:{item[1]}\n', six.iteritems(collection), '') + '\n'
 
         with closing(StringIO(text)) as input_file:
             input_header.read(input_file)
@@ -310,7 +308,6 @@ def test_messages_header(self):
 
         @Configuration()
         class TestMessagesHeaderCommand(SearchCommand):
-
             class ConfigurationSettings(SearchCommand.ConfigurationSettings):
 
                 @classmethod

tests/searchcommands/test_generator_command.py

@@ -15,38 +15,31 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import gzip
+import io
+import json
+import os
+import random
 
-from splunklib.searchcommands.internals import MetadataDecoder, MetadataEncoder, Recorder, RecordWriterV2
-from splunklib.searchcommands import SearchMetric
-from splunklib import six
-from splunklib.six.moves import range
-from collections import OrderedDict
-from collections import namedtuple, deque
-from splunklib.six import BytesIO as BytesIO
+import pytest
 from functools import wraps
-from glob import iglob
 from itertools import chain
-from splunklib.six.moves import filter as ifilter
-from splunklib.six.moves import map as imap
-from splunklib.six.moves import zip as izip
 from sys import float_info
 from tempfile import mktemp
 from time import time
 from types import MethodType
-from sys import version_info as python_version
-try:
-    from unittest2 import main, TestCase
-except ImportError:
-    from unittest import main, TestCase
+from unittest import main, TestCase
 
+from collections import OrderedDict
+from collections import namedtuple, deque
+
+from splunklib.searchcommands.internals import MetadataDecoder, MetadataEncoder, Recorder, RecordWriterV2
+from splunklib.searchcommands import SearchMetric
+from splunklib import six
+from splunklib.six.moves import range
+from splunklib.six import BytesIO as BytesIO
 import splunklib.six.moves.cPickle as pickle
-import gzip
-import io
-import json
-import os
-import random
 
-import pytest
 
 # region Functions for producing random apps
 
@@ -228,8 +221,8 @@ def test_record_writer_with_random_data(self, save_recording=False):
         self.assertListEqual(writer._inspector['messages'], messages)
 
         self.assertDictEqual(
-            dict([k_v for k_v in six.iteritems(writer._inspector) if k_v[0].startswith('metric.')]),
-            dict([('metric.' + k_v1[0], k_v1[1]) for k_v1 in six.iteritems(metrics)]))
+            dict(k_v for k_v in six.iteritems(writer._inspector) if k_v[0].startswith('metric.')),
+            dict(('metric.' + k_v1[0], k_v1[1]) for k_v1 in six.iteritems(metrics)))
 
         writer.flush(finished=True)
 

tests/searchcommands/test_internals_v1.py

@@ -4,7 +4,6 @@
 
 from os import path
 
-from splunklib import six
 from splunklib.searchcommands import StreamingCommand, Configuration
 
 

tests/searchcommands/test_internals_v2.py

@@ -15,15 +15,6 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
-
-from splunklib import six
-from splunklib.searchcommands import Configuration, StreamingCommand
-from splunklib.searchcommands.decorators import ConfigurationSetting, Option
-from splunklib.searchcommands.search_command import SearchCommand
-from splunklib.client import Service
-
-from splunklib.six import StringIO, BytesIO
-from splunklib.six.moves import zip as izip
 from json.encoder import encode_basestring as encode_string
 from unittest import main, TestCase
 
@@ -36,11 +27,18 @@
 
 import pytest
 
+from splunklib import six
+from splunklib.searchcommands import Configuration, StreamingCommand
+from splunklib.searchcommands.decorators import ConfigurationSetting, Option
+from splunklib.searchcommands.search_command import SearchCommand
+from splunklib.client import Service
+
+from splunklib.six import StringIO, BytesIO
+
 
 def build_command_input(getinfo_metadata, execute_metadata, execute_body):
-    input = ('chunked 1.0,{},0\n{}'.format(len(six.ensure_binary(getinfo_metadata)), getinfo_metadata) +
-             'chunked 1.0,{},{}\n{}{}'.format(len(six.ensure_binary(execute_metadata)),
-                                              len(six.ensure_binary(execute_body)), execute_metadata, execute_body))
+    input = (f'chunked 1.0,{len(six.ensure_binary(getinfo_metadata))},0\n{getinfo_metadata}' +
+             f'chunked 1.0,{len(six.ensure_binary(execute_metadata))},{len(six.ensure_binary(execute_body))}\n{execute_metadata}{execute_body}')
 
     ifile = BytesIO(six.ensure_binary(input))
 
@@ -311,7 +309,7 @@ def test_process_scpv1(self):
             # noinspection PyTypeChecker
             command.process(argv, ifile, ofile=result)
         except BaseException as error:
-            self.fail('Expected no exception, but caught {}: {}'.format(type(error).__name__, error))
+            self.fail(f'Expected no exception, but caught {type(error).__name__}: {error}')
         else:
             six.assertRegex(
                 self,

tests/searchcommands/test_multibyte_processing.py

@@ -25,15 +25,9 @@
 
 
 from collections import namedtuple
-from splunklib.six.moves import cStringIO as StringIO
 from datetime import datetime
 
-from splunklib.six.moves import filter as ifilter
-from splunklib.six.moves import map as imap
-from splunklib.six.moves import zip as izip
-
 from subprocess import PIPE, Popen
-from splunklib import six
 
 from unittest import main, skipUnless, TestCase
 
@@ -43,14 +37,12 @@
 import io
 import os
 import sys
+import pytest
 
-try:
-    from tests.searchcommands import project_root
-except ImportError:
-    # Python 2.6
-    pass
+from splunklib.six.moves import cStringIO as StringIO
+from splunklib import six
 
-import pytest
+from tests.searchcommands import project_root
 
 
 def pypy():
@@ -287,8 +279,7 @@ def _compare_csv_files_time_insensitive(self, expected, output):
                 output_row['_time'] = expected_row['_time']
 
             self.assertDictEqual(
-                expected_row, output_row, 'Error on line {0}: expected {1}, not {2}'.format(
-                    line_number, expected_row, output_row))
+                expected_row, output_row, f'Error on line {line_number}: expected {expected_row}, not {output_row}')
 
             line_number += 1
 
@@ -310,8 +301,7 @@ def _compare_csv_files_time_sensitive(self, expected, output):
         for expected_row in expected:
             output_row = next(output)
             self.assertDictEqual(
-                expected_row, output_row, 'Error on line {0}: expected {1}, not {2}'.format(
-                    line_number, expected_row, output_row))
+                expected_row, output_row, f'Error on line {line_number}: expected {expected_row}, not {output_row}')
             line_number += 1
 
     def _get_search_command_path(self, name):