Adding urlhandling test script. Removed unecessesary UrlEncoded check. Formatting

Author: stelles

splunklib/binding.py

@@ -122,7 +122,7 @@ def __new__(self, val='', skip_encode=False, encode_slash=False):
         elif skip_encode:
             return str.__new__(self, val)
         elif encode_slash:
-            return str.__new__(self,urllib.quote(val, ''))
+            return str.__new__(self, urllib.quote(val, ''))
         else:
             # When subclassing str, just call str's __new__ method
             # with your class and the value you want to have in the

splunklib/client.py

@@ -676,7 +676,7 @@ def get(self, path_segment="", owner=None, app=None, sharing=None, **query):
         # self.path to the Endpoint is relative in the SDK, so passing
         # owner, app, sharing, etc. along will produce the correct
         # namespace in the final request.
-        if not isinstance(path_segment, UrlEncoded): path_segment=UrlEncoded(path_segment, encode_slash=True)
+        path_segment = UrlEncoded(path_segment, encode_slash=True)
         if path_segment.startswith('/'):
             path = path_segment
         else:
@@ -1161,7 +1161,6 @@ def __getitem__(self, key):
                 response = self.get(key, owner=ns.owner, app=ns.app)
             else:
                 response = self.get(key)
-
             entries = self._load_list(response)
             if len(entries) > 1:
                 raise AmbiguousReferenceException("Found multiple entities named '%s'; please specify a namespace." % key)
@@ -1520,7 +1519,8 @@ def delete(self, name, **params):
             saved_searches.delete('my_saved_search')
             assert 'my_saved_search' not in saved_searches
         """
-        if not isinstance(name, UrlEncoded): name = UrlEncoded(name, encode_slash=True)
+
+        name = UrlEncoded(name, encode_slash=True)
         if 'namespace' in params:
             namespace = params.pop('namespace')
             params['owner'] = namespace.owner
@@ -1964,7 +1964,7 @@ def __getitem__(self, key):
         if isinstance(key, tuple) and len(key) == 2:
             # Fetch a single kind
             key, kind = key
-            if not isinstance(key, UrlEncoded): key = UrlEncoded(key, encode_slash=True)
+            key = UrlEncoded(key, encode_slash=True)
             try:
                 response = self.get(self.kindpath(kind) + "/" + key)
                 entries = self._load_list(response)
@@ -1981,7 +1981,7 @@ def __getitem__(self, key):
                     raise
         else:
             # Iterate over all the kinds looking for matches.
-            if not isinstance(key, UrlEncoded): key = UrlEncoded(key, encode_slash=True)
+            key = UrlEncoded(key, encode_slash=True)
             kind = None
             candidate = None
             for kind in self.kinds:
@@ -2002,7 +2002,7 @@ def __getitem__(self, key):
                     else:
                         raise
             if candidate is None:
-                raise KeyError(key) # Never found a match. Refer to name matching <http://docs.splunk.com/Documentation/Splunk/6.1.1/RESTAPI/RESTinput#data.2Finputs.2Fmonitor.2F.7Bname.7D>
+                raise KeyError(key) # Never found a match.
             else:
                 return candidate
 
@@ -2015,6 +2015,7 @@ def __contains__(self, key):
             except KeyError:
                 return False
         else:
+            key = UrlEncoded(key, encode_slash=True)
             # Without a kind, we want to minimize the number of round trips to the server, so we
             # reimplement some of the behavior of __getitem__ in order to be able to stop searching
             # on the first hit.
@@ -2077,8 +2078,7 @@ def create(self, name, kind, **kwargs):
         # and we have to adjust accordingly.
 
         # Must check if the name passed in is UrlEncoded and change it accordingly.
-        if not isinstance(name, UrlEncoded): name = UrlEncoded(name, encode_slash=True)
-        
+        name = UrlEncoded(name, encode_slash=True)
         path = _path(
             self.path + kindpath,
             '%s:%s' % (kwargs['restrictToHost'], name) \

tests/test_urlhandling.py

@@ -0,0 +1,126 @@
+import os
+from subprocess import PIPE, Popen
+from os.path import expanduser
+import time
+import sys
+import urllib
+import testlib 
+
+import splunklib.client as client
+from splunklib.binding import UrlEncoded
+
+try:
+    from utils import *
+except ImportError:
+    raise Exception("Add the SDK repository to your PYTHONPATH to run the examples "
+                    "(e.g., export PYTHONPATH=~/splunk-sdk-python.")
+
+# Creates a file in the default Home directory
+# Probably won't work with windows
+def make_file(filename):
+    Popen(("touch '%s/%s'"%(expanduser('~'),filename)), shell=True)
+
+
+def delete_file(filename):
+    Popen(("rm '%s/%s'"%(expanduser('~'),filename)), shell=True)
+
+
+class ExamplesTestCase(testlib.SDKTestCase):
+    def setUp(self):
+        if os.name == "nt":
+            sys.exit("Windows not supported")
+
+        opts = parse(sys.argv[1:], config=".splunkrc")
+        kwargs = dslice(opts.kwargs, FLAGS_SPLUNK)
+        self.service = client.connect(**kwargs)
+        super(ExamplesTestCase, self).setUp()
+
+    def test_creating_monitor_inputs(self):
+        input_names = [
+            "myfile#",
+            "myfile!",
+            "myfile$",
+            "myfile(abc).txt"
+            ]
+        for name in input_names:
+            make_file(name)
+            pathname = '/'.join([expanduser('~'), name])
+            if self.service.inputs.__contains__(pathname):
+                self.service.inputs.delete(name=pathname, kind="monitor")
+                time.sleep(1) # Need to wait for delete to be processed or else there is a failure.
+                inp = self.service.inputs.create(name=pathname, kind="monitor")
+            else:
+                inp = self.service.inputs.create(name=pathname, kind="monitor")
+            test = self.service.inputs[pathname]
+            test.refresh()
+            inp.refresh()
+            self.assertTrue(inp.name == pathname)
+            self.assertTrue(inp.access == test.access)
+            self.assertTrue(inp.content == test.content)
+            self.assertTrue(inp.fields == test.fields)
+            self.assertTrue(inp.links == test.links)
+            self.assertTrue(inp.state == test.state)
+            self.service.inputs.delete(name=pathname)
+            delete_file(name)
+
+    def test_saved_searches(self):
+        search_name = [
+                    "mysearch#",
+                    "my/search@",
+                    "my/search/test(abc)"
+                    ]
+        search_param = "search *"
+        for search in search_name:
+            if self.service.saved_searches.__contains__(search):
+                service.saved_searches.delete(search)
+                sear = self.service.saved_searches.create(name=search, search=search_param)
+            else:
+                sear = self.service.saved_searches.create(name=search, search=search_param)
+            test = self.service.saved_searches[search]
+            test.refresh()
+            sear.refresh()
+            self.assertTrue(sear.name == search)
+            self.assertTrue(sear.name == test.name)
+            self.assertTrue(sear.access == test.access)
+            self.assertTrue(sear.content == test.content)
+            self.service.saved_searches.delete(search)
+
+    def test_urlhandling_inputs(self):
+        for inp in self.service.inputs:
+            raw = self.service.inputs.get(name=inp.name)
+            url_name = UrlEncoded(inp.name, encode_slash=True)
+            encoded = self.service.inputs.get(name=url_name)
+            self.assertTrue(raw.status == encoded.status)
+            self.assertTrue(raw.headers == encoded.headers)
+            self.assertTrue(raw.reason == encoded.reason)
+            self.assertTrue(str(raw.body)== str(encoded.body))
+
+    def test_urlhandling_saved_searchs(self):
+        for ss in self.service.saved_searches:
+            raw = self.service.saved_searches.get(ss.name)
+            encoded = self.service.saved_searches.get(UrlEncoded(ss.name, encode_slash=True))
+            self.assertTrue(raw.status == encoded.status)
+            self.assertTrue(raw.headers == encoded.headers)
+            self.assertTrue(raw.reason == encoded.reason)
+            self.assertTrue(str(raw.body)== str(encoded.body))
+
+    def test_urlhandling(self):
+        for inp in self.service.inputs:
+            name = inp.name
+            self.assertTrue(urllib.quote(name) == UrlEncoded(name))
+            self.assertTrue(urllib.quote(name, '') == UrlEncoded(name, encode_slash=True))
+            self.assertTrue(name == UrlEncoded(name, skip_encode=True))
+
+
+if __name__=="__main__":
+    os.chdir("../examples")
+    try:
+        import unittest2 as unittest2
+    except ImportError:
+        import unittest
+    unittest.main()
+
+
+
+
+