Reworked test_service.py.

service.parse will need to be rewritten. It's useless right now.

Author: Frederick Ross

splunklib/client.py

@@ -107,6 +107,29 @@
 
 MATCH_ENTRY_CONTENT = "%s/%s/*" % (XNAME_ENTRY, XNAME_CONTENT)
 
+capabilities = record({k:k for k in [
+            "admin_all_objects", "change_authentication", 
+            "change_own_password", "delete_by_keyword",
+            "edit_deployment_client", "edit_deployment_server",
+            "edit_dist_peer", "edit_forwarders", "edit_httpauths",
+            "edit_input_defaults", "edit_monitor", "edit_roles",
+            "edit_scripted", "edit_search_server", "edit_server",
+            "edit_splunktcp", "edit_splunktcp_ssl", "edit_tcp",
+            "edit_udp", "edit_user", "edit_web_settings", "get_metadata",
+            "get_typeahead", "indexes_edit", "license_edit", "license_tab",
+            "list_deployment_client", "list_forwarders", "list_httpauths",
+            "list_inputs", "request_remote_tok", "rest_apps_management",
+            "rest_apps_view", "rest_properties_get", "rest_properties_set",
+            "restart_splunkd", "rtsearch", "schedule_search", "search",
+            "use_file_operator"]})
+
+
+class NoSuchUserException(Exception):
+    pass
+
+class NoSuchApplicationException(Exception):
+    pass
+
 class IllegalOperationException(Exception):
     pass
 

tests/test_service.py

@@ -15,159 +15,111 @@
 # under the License.
 
 import testlib
+import logging
 
+import unittest
+
+import splunklib.data as data
 
 import splunklib.client as client
 from splunklib.binding import HTTPError
 
 class TestCase(testlib.TestCase):
     def test_capabilities(self):
-        service = client.connect(**self.opts.kwargs)
-
-        expected = [
-            "admin_all_objects", "change_authentication", 
-            "change_own_password", "delete_by_keyword",
-            "edit_deployment_client", "edit_deployment_server",
-            "edit_dist_peer", "edit_forwarders", "edit_httpauths",
-            "edit_input_defaults", "edit_monitor", "edit_roles",
-            "edit_scripted", "edit_search_server", "edit_server",
-            "edit_splunktcp", "edit_splunktcp_ssl", "edit_tcp",
-            "edit_udp", "edit_user", "edit_web_settings", "get_metadata",
-            "get_typeahead", "indexes_edit", "license_edit", "license_tab",
-            "list_deployment_client", "list_forwarders", "list_httpauths",
-            "list_inputs", "request_remote_tok", "rest_apps_management",
-            "rest_apps_view", "rest_properties_get", "rest_properties_set",
-            "restart_splunkd", "rtsearch", "schedule_search", "search",
-            "use_file_operator" ]
-
-        capabilities = service.capabilities
-        for item in expected: self.assertTrue(item in capabilities)
+        capabilities = self.service.capabilities
+        for item in client.capabilities:
+            self.assertTrue(item in capabilities)
 
     def test_info(self):
-        service = client.connect(**self.opts.kwargs)
-
-        info = service.info
-        keys = [
-            "build", "cpu_arch", "guid", "isFree", "isTrial", "licenseKeys",
+        info = self.service.info
+        keys = ["build", "cpu_arch", "guid", "isFree", "isTrial", "licenseKeys",
             "licenseSignature", "licenseState", "master_guid", "mode", 
-            "os_build", "os_name", "os_version", "serverName", "version" ]
-        for key in keys: self.assertTrue(key in info.keys())
-
-    def test_namespaces(self):
-        kwargs = self.opts.kwargs.copy()
+            "os_build", "os_name", "os_version", "serverName", "version"]
+        for key in keys: 
+            self.assertTrue(key in info.keys())
 
-        # Verify connect with no namespace
-        service = client.connect(**kwargs)
+    def test_without_namespace(self):
+        service = client.connect(**self.opts.kwargs)
         service.apps.list()
 
-        # Verify namespace permutations using standard app & owner args 
-        kwargs.update({ 'app': "search", 'owner': None })
+    def test_app_namespace(self):
+        kwargs = self.opts.kwargs.copy()
+        kwargs.update({'app': "search", 'owner': None})
         service_ns = client.connect(**kwargs)
         service_ns.apps.list()
 
+    def test_owner_wildcard(self):
+        kwargs = self.opts.kwargs.copy()
         kwargs.update({ 'app': "search", 'owner': "-" })
         service_ns = client.connect(**kwargs)
         service_ns.apps.list()
 
+    def test_default_app(self):
+        kwargs = self.opts.kwargs.copy()
         kwargs.update({ 'app': None, 'owner': "admin" })
         service_ns = client.connect(**kwargs)
         service_ns.apps.list()
 
+    def test_app_wildcard(self):
+        kwargs = self.opts.kwargs.copy()
         kwargs.update({ 'app': "-", 'owner': "admin" })
         service_ns = client.connect(**kwargs)
         service_ns.apps.list()
 
+    def test_user_namespace(self):
+        kwargs = self.opts.kwargs.copy()
         kwargs.update({ 'app': "search", 'owner': "admin" })
         service_ns = client.connect(**kwargs)
         service_ns.apps.list()
 
-        appname = "sdk-test-app"
-        username = "sdk-test-user"
-        testlib.delete_app(service, appname)
-        if username in service.users: service.users.delete(username)
-        self.assertFalse(service.apps.contains(appname))
-        self.assertFalse(service.users.contains(username))
-
-        # App & owner dont exist, verify that the following errors
-        kwargs.update({ 'app': appname, 'owner': username })
-        with self.assertRaises(HTTPError):
+    def test_nonexistant_user(self):
+        username = testlib.tmpname()
+        self.assertTrue(username not in self.service.users)
+        with self.assertRaises(client.NoSuchUserException):
+            kwargs = self.opts.kwargs.copy()
+            kwargs.update({'owner': username})
             service_ns = client.connect(**kwargs)
             service_ns.apps.list()
 
-        # Validate namespace permutations with new app & user
-        service.apps.create(appname)
-        service.users.create(username, password="changeme", roles="power")
-
-        kwargs.update({ 'app': appname, 'owner': None })
-        service_ns = client.connect(**kwargs)
-        service_ns.apps.list()
-
-        kwargs.update({ 'app': appname, 'owner': "-" })
-        service_ns = client.connect(**kwargs)
-        service_ns.apps.list()
-
-        kwargs.update({ 'app': None, 'owner': username })
-        service_ns = client.connect(**kwargs)
-        service_ns.apps.list()
-
-        kwargs.update({ 'app': "-", 'owner': username })
-        service_ns = client.connect(**kwargs)
-        service_ns.apps.list()
-
-        kwargs.update({ 'app': appname, 'owner': username })
-        service_ns = client.connect(**kwargs)
-        service_ns.apps.list()
-
-        # Cleanup
-        testlib.delete_app(service, appname)
-        service.users.delete(username)
-
-        self.assertFalse(service.apps.contains(appname))
-        self.assertFalse(service.users.contains(username))
-
-    def test_trailing(self):
-        self.assertRaises(ValueError, client.trailing, 'this is a test', 'boris')
-        self.assertRaises(ValueError, client.trailing, 'this is a test', 's is', 'boris')
-        template = '/servicesNS/boris/search/another/path/segment/that runs on'
-        self.assertEqual(template, client.trailing(template))
-        self.assertEqual('boris/search/another/path/segment/that runs on', client.trailing(template, 'ervicesNS/'))
-        self.assertEqual('another/path/segment/that runs on', client.trailing(template, 'servicesNS/', '/', '/'))
-
+    def test_nonexistant_app(self):
+        app_name = testlib.tmpname()
+        self.assertTrue(app_name not in self.service.apps)
+        with self.assertRaises(client.NoSuchApplicationException):
+            kwargs = self.opts.kwargs.copy()
+            kwargs.update({'app': app_name})
+            service_ns = client.connect(**kwargs)
+            service_ns.apps.list()
 
     def test_parse(self):
-        service = client.connect(**self.opts.kwargs)
-
-        response = service.parse("search *")
+        # Awaiting new parse method.
+        response = self.service.parse('search * abc="def" | dedup abc')
         self.assertEqual(response.status, 200)
-
-        response = service.parse("search index=twitter status_count=* | stats count(status_source) as count by status_source | sort -count | head 20")
-        self.assertEqual(response.status, 200)
-
-        try:
-            service.parse("xyzzy")
-            self.fail()
-        except HTTPError, e:
-            self.assertEqual(e.status, 400)
-        except:
-            self.fail()
+        # try:
+        #     service.parse("xyzzy")
+        #     self.fail()
+        # except HTTPError, e:
+        #     self.assertEqual(e.status, 400)
+        # except:
+        #     self.fail()
 
     def test_restart(self):
         service = client.connect(**self.opts.kwargs)
         testlib.restart(service)
         service.login() # Make sure we are awake
 
-    def test_settings(self):
-        service = client.connect(**self.opts.kwargs)
-        settings = service.settings
-
+class TestSettings(testlib.TestCase):
+    def test_read_settings(self):
+        settings = self.service.settings
         # Verify that settings contains the keys we expect
         keys = [
             "SPLUNK_DB", "SPLUNK_HOME", "enableSplunkWebSSL", "host",
             "httpport", "mgmtHostPort", "minFreeSpace", "pass4SymmKey",
             "serverName", "sessionTimeout", "startwebserver", "trustedIP"
         ]
-        for key in keys: self.assertTrue(key in settings.content)
+        for key in keys:
+            self.assertTrue(key in settings)
 
+    def test_update_settings(self):
         # Verify that we can update the settings
         original = settings['sessionTimeout']
         self.assertTrue(original != "42h")
@@ -182,5 +134,24 @@ def test_settings(self):
         updated = settings['sessionTimeout']
         self.assertEqual(updated, original)
 
+class TestTrailing(unittest.TestCase):
+    template = '/servicesNS/boris/search/another/path/segment/that runs on'
+
+    def test_raises_when_not_found_first(self):
+        self.assertRaises(ValueError, client.trailing, 'this is a test', 'boris')
+
+    def test_raises_when_not_found_second(self):
+        self.assertRaises(ValueError, client.trailing, 'this is a test', 's is', 'boris')
+
+    def test_no_args_is_identity(self):
+        self.assertEqual(self.template, client.trailing(self.template))
+
+    def test_trailing_with_one_arg_works(self):
+        self.assertEqual('boris/search/another/path/segment/that runs on', client.trailing(self.template, 'ervicesNS/'))
+
+    def test_trailing_with_n_args_works(self):
+        self.assertEqual('another/path/segment/that runs on', client.trailing(self.template, 'servicesNS/', '/', '/'))
+
+
 if __name__ == "__main__":
     testlib.main()