Fixed quote problem and added support for value formatting

The latter was required to write unit tests that validate the SearchCommandParser

Author: David Noble

examples/searchcommands_app/bin/countmatches.py

@@ -66,8 +66,7 @@ def stream(self, records):
             count = 0.0
             for fieldname in self.fieldnames:
                 matches = self.pattern.finditer(str(record[fieldname]))
-                if matches:
-                    count += len(list(matches))
+                count += len(list(matches))
             record[self.fieldname] = count
             yield record
 

examples/searchcommands_app/default/commands.conf

@@ -4,11 +4,14 @@
 [countmatches]
 filename = countmatches.py
 supports_getinfo = true
+supports_rawargs = true
 
 [simulate]
 filename = simulate.py
 supports_getinfo = true
+supports_rawargs = true
 
 [sum]
 filename = sum.py
 supports_getinfo = true
+supports_rawargs = true

splunklib/searchcommands/decorators.py

@@ -200,8 +200,16 @@ def _reset(self, other):
     #region Types
 
     class Encoder(JSONEncoder):
+        def __init__(self, item):
+            super(Option.Encoder, self).__init__()
+            self.item = item
+
         def default(self, o):
-            return str(o)
+            # Convert the value of a type unknown to the JSONEncoder
+            validator = self.item.validator
+            if validator is None:
+                return str(o)
+            return validator.format(o)
 
     class Item(object):
         """ Presents an instance/class view over a search command `Option`
@@ -218,8 +226,8 @@ def __repr__(self):
             return str(self)
 
         def __str__(self):
-            encoder = Option.Encoder()
-            text = '='.join([self._option.name, encoder.encode(self.value)])
+            encoder = Option.Encoder(self)
+            text = '='.join([self.name, encoder.encode(self.value)])
             return text
 
         #region Properties

splunklib/searchcommands/search_command_internals.py

@@ -352,7 +352,7 @@ def replace(match):
         \s*$
         """, re.VERBOSE)
 
-    _escaped_quote_re = re.compile(r"""(\\\\|\\"|""|\\."|\\])""")
+    _escaped_quote_re = re.compile(r"""(\\\\|\\"|""|\\."|\\)""")
 
     _name_re = re.compile(r"""[_a-zA-Z][[_a-zA-Z0-9]+""")
 

tests/test_searchcommands_app.py

@@ -20,11 +20,31 @@
     import unittest
 
 from subprocess import Popen
-import json
+import base64
 import os
 import shutil
 import testlib
 
+from splunklib.searchcommands import \
+    StreamingCommand, Configuration, Option, validators
+
+@Configuration()
+class StubbedCommand(StreamingCommand):
+    fieldname = Option(
+        doc='''
+        **Syntax:** **fieldname=***<fieldname>*
+        **Description:** Name of the field that will hold the match count''',
+        require=True, validate=validators.Fieldname())
+
+    pattern = Option(
+        doc='''
+        **Syntax:** **pattern=***<regular-expression>*
+        **Description:** Regular expression pattern to match''',
+        require=True, validate=validators.RegularExpression())
+
+    def stream(self, records):
+        pass
+
 
 class TestSearchCommandsApp(testlib.SDKTestCase):
 
@@ -37,15 +57,49 @@ def setUp(self):
             os.mkdir(path)
         return
 
+    def test_command_parser(self):
+        from splunklib.searchcommands.search_command_internals import \
+            SearchCommandParser
+
+        parser = SearchCommandParser()
+        command = StubbedCommand()
+
+        parser.parse(
+            [
+                'fieldname=word_count',
+                'pattern="\\\\w+"',
+                'text_field_1',
+                'text_field_2'
+            ],
+            command)
+
+        command_line = str(command)
+        self.assertEqual('stubbed fieldname="word_count" pattern="\\\\w+" text_field_1 text_field_2', command_line)
+        return
+
+    def test_option_show_configuration(self):
+        self._run(
+            'simulate', [
+                'csv=%s' % TestSearchCommandsApp._data_file("input/population.csv"),
+                'duration=00:00:10',
+                'interval=00:00:01',
+                'rate=200',
+                'seed=%s' % TestSearchCommandsApp._seed,
+                'show_configuration=true'],
+            __GETINFO__=(
+                'input/population.csv',
+                'output/samples.csv',
+                'log/test_show_configuration.log'))
+        return
+
     def test_generating_command_in_isolation(self):
-        encoder = json.JSONEncoder(ensure_ascii=False)
         self._run(
             'simulate', [
                 'csv=%s' % TestSearchCommandsApp._data_file("input/population.csv"),
                 'duration=00:00:10',
                 'interval=00:00:01',
                 'rate=200',
-                'seed=%s' % encoder.encode(TestSearchCommandsApp._seed)],
+                'seed=%s' % TestSearchCommandsApp._seed],
             __GETINFO__=(
                 'input/population.csv',
                 'output/samples.csv',
@@ -108,6 +162,8 @@ def test_streaming_command_on_server(self):
 
     def _run(self, command, args, **kwargs):
         for operation in ['__GETINFO__', '__EXECUTE__']:
+            if operation not in kwargs:
+                continue
             files = kwargs[operation]
             process = TestSearchCommandsApp._start_process(
                 ['python', command + '.py', operation] + args,
@@ -133,9 +189,11 @@ def _start_process(cls, args, stdin, stdout, stderr):
 
     package_directory = os.path.dirname(__file__)
     data_directory = os.path.join(package_directory, 'searchcommands_data')
-    app_bin = os.path.join(os.path.dirname(package_directory), "examples/searchcommands_app/bin")
+    app_bin = os.path.join(
+        os.path.dirname(package_directory), "examples/searchcommands_app/bin")
 
-    _seed = '\xcd{\xf8\xc4\x1c8=\x88\nc\xe2\xc4\xee\xdb\xcal'
+    _seed = base64.encodestring(
+        '\xcd{\xf8\xc4\x1c8=\x88\nc\xe2\xc4\xee\xdb\xcal')
 
 if __name__ == "__main__":
     unittest.main()