def test_create_with_colons(self):
        start_count = len(self.storage_passwords)
        username = testlib.tmpname()
        realm = testlib.tmpname()

        p = self.storage_passwords.create("changeme", username + ":end",
                                          ":start" + realm)
        self.assertEqual(start_count + 1, len(self.storage_passwords))
        self.assertEqual(p.realm, ":start" + realm)
        self.assertEqual(p.username, username + ":end")
        self.assertEqual(p.clear_password, "changeme")
        self.assertEqual(p.name,
                         "\\:start" + realm + ":" + username + "\\:end:")

        p.delete()
        self.assertEqual(start_count, len(self.storage_passwords))

        prefix = testlib.tmpname()
        realm = prefix + ":r:e:a:l:m:"
        user = ":u:s:e:r:"
        p = self.storage_passwords.create("changeme", user, realm)
        self.assertEqual(start_count + 1, len(self.storage_passwords))
        self.assertEqual(p.realm, realm)
        self.assertEqual(p.username, user)
        self.assertEqual(p.clear_password, "changeme")
        self.assertEqual(p.name,
                         prefix + "\\:r\\:e\\:a\\:l\\:m\\::\\:u\\:s\\:e\\:r\\::")

        p.delete()
        self.assertEqual(start_count, len(self.storage_passwords))

    def test_create(self):
        start_count = len(self.storage_passwords)
        realm = testlib.tmpname()
        username = testlib.tmpname()

        p = self.storage_passwords.create("changeme", username, realm)
        self.assertEqual(start_count + 1, len(self.storage_passwords))
        self.assertEqual(p.realm, realm)
        self.assertEqual(p.username, username)
        self.assertEqual(p.clear_password, "changeme")
        self.assertEqual(p.name, realm + ":" + username + ":")

        p.delete()
        self.assertEqual(start_count, len(self.storage_passwords))

    def test_alerts_on_events(self):
        self.assertEqual(self.saved_search.alert_count, 0)
        self.assertEqual(len(self.saved_search.fired_alerts), 0)

        self.index.enable()
        self.assertEventuallyTrue(lambda: self.index.refresh() and self.index['disabled'] == '0', timeout=25)

        eventCount = int(self.index['totalEventCount'])
        self.assertEqual(self.index['sync'], '0')
        self.assertEqual(self.index['disabled'], '0')
        self.index.refresh()
        self.index.submit('This is a test ' + testlib.tmpname(),
                          sourcetype='sdk_use', host='boris')
        def f():
            self.index.refresh()
            return int(self.index['totalEventCount']) == eventCount+1
        self.assertEventuallyTrue(f, timeout=50)

        def g():
            self.saved_search.refresh()
            return self.saved_search.alert_count == 1
        self.assertEventuallyTrue(g, timeout=200)

        alerts = self.saved_search.fired_alerts
        self.assertEqual(len(alerts), 1)

 def setUp(self):
     super(UserTestCase, self).setUp()
     self.username = testlib.tmpname()
     self.user = self.service.users.create(
         self.username,
         password='changeme',
         roles=['power', 'user'])

    def test_name_collision(self):
        opts = self.opts.kwargs.copy()
        opts['owner'] = '-'
        opts['app'] = '-'
        opts['sharing'] = 'user'
        service = client.connect(**opts)
        logging.debug("Namespace for collision testing: %s", service.namespace)
        saved_searches = service.saved_searches
        name = testlib.tmpname()
        
        query1 = '* earliest=-1m | head 1'
        query2 = '* earliest=-2m | head 2'
        namespace1 = client.namespace(app='search', sharing='app')
        namespace2 = client.namespace(owner='admin', app='search', sharing='user')
        saved_search2 = saved_searches.create(
            name, query2,
            namespace=namespace1)
        saved_search1 = saved_searches.create(
            name, query1,
            namespace=namespace2)

        self.assertRaises(client.AmbiguousReferenceException,
                          saved_searches.__getitem__, name)
        search1 = saved_searches[name, namespace1]
        self.check_saved_search(search1)
        search1.update(**{'action.email.from': '[email protected]'})
        search1.refresh()
        self.assertEqual(search1['action.email.from'], '[email protected]')
        search2 = saved_searches[name, namespace2]
        search2.update(**{'action.email.from': '[email protected]'})
        search2.refresh()
        self.assertEqual(search2['action.email.from'], '[email protected]')
        self.check_saved_search(search2)

 def test_read_invalid_input(self):
     name = testlib.tmpname()
     try:
         self.service.inputs.get(name)
         self.fail("Expected a 404 HTTPError")
     except HTTPError as he:
         self.assertTrue("HTTP 404 Not Found" in str(he))

    def test_delete(self):
        start_count = len(self.storage_passwords)
        username = testlib.tmpname()

        p = self.storage_passwords.create("changeme", username, "myrealm")
        self.assertEqual(start_count + 1, len(self.storage_passwords))
        self.assertEqual(p.realm, "myrealm")
        self.assertEqual(p.username, username)
        self.assertEqual(p.clear_password, "changeme")
        self.assertEqual(p.name, "myrealm:" + username + ":")

        self.storage_passwords.delete(username, "myrealm")
        self.assertEqual(start_count, len(self.storage_passwords))

        self.storage_passwords.create("changeme", username, "myrealm")
        self.assertEqual(start_count + 1, len(self.storage_passwords))

        self.storage_passwords.delete("myrealm:" + username + ":")
        self.assertEqual(start_count, len(self.storage_passwords))

        # Test named parameters
        self.storage_passwords.create(password="changeme", username=username,
                                      realm="myrealm")
        self.assertEqual(start_count + 1, len(self.storage_passwords))

        self.storage_passwords.delete(username, "myrealm")
        self.assertEqual(start_count, len(self.storage_passwords))

        self.storage_passwords.create(password="changeme", username=username + "/foo",
                                      realm="/myrealm")
        self.assertEqual(start_count + 1, len(self.storage_passwords))

        self.storage_passwords.delete(username + "/foo", "/myrealm")
        self.assertEqual(start_count, len(self.storage_passwords))

 def test_delete(self):
     name = testlib.tmpname()
     app = self.service.apps.create(name)
     self.assertTrue(name in self.service.apps)
     self.service.apps.delete(name)
     self.assertFalse(name in self.service.apps)
     self.clear_restart_message() # We don't actually have to restart here.

 def setUp(self):
     super(TestSavedSearch, self).setUp()
     saved_searches = self.service.saved_searches
     logging.debug("Saved searches namespace: %s", saved_searches.service.namespace)
     self.saved_search_name = testlib.tmpname()
     query = "search index=_internal * earliest=-1m | head 3"
     self.saved_search = saved_searches.create(self.saved_search_name, query)

    def test_create_crazy(self):
        start_count = len(self.storage_passwords)
        username = testlib.tmpname()
        realm = testlib.tmpname()

        p = self.storage_passwords.create("changeme",
                                          username + ":[email protected]#$%^&*()_+{}:|<>?",
                                          ":start::[email protected]#$%^&*()_+{}:|<>?" + realm)
        self.assertEqual(start_count + 1, len(self.storage_passwords))
        self.assertEqual(p.realm, ":start::[email protected]#$%^&*()_+{}:|<>?" + realm)
        self.assertEqual(p.username, username + ":[email protected]#$%^&*()_+{}:|<>?")
        self.assertEqual(p.clear_password, "changeme")
        self.assertEqual(p.name,
                         "\\:start\\:\\:[email protected]#$%^&*()_+{}\\:|<>?" + realm + ":" + username + "\\:[email protected]#$%^&*()_+{}\\:|<>?:")

        p.delete()
        self.assertEqual(start_count, len(self.storage_passwords))

    def test_create_with_slashes(self):
        start_count = len(self.storage_passwords)
        realm = "/" + testlib.tmpname()
        username = "/" + testlib.tmpname()

        # Prepends one escaped slash
        p = self.storage_passwords.create("changeme", username, realm)
        self.assertEqual(start_count + 1, len(self.storage_passwords))
        self.assertEqual(p.realm, realm)
        # Prepends one escaped slash
        self.assertEqual(p.username, username)
        self.assertEqual(p.clear_password, "changeme")
        # Checks for 2 escaped slashes (Splunk encodes the single slash)
        self.assertEqual(p.name, realm + ":" + username + ":")

        p.delete()
        self.assertEqual(start_count, len(self.storage_passwords))

 def test_create_delete(self):
     message_name = testlib.tmpname()
     message_value = 'Test message'
     message = self.service.messages.create(
         message_name, value=message_value)
     self.assertTrue(message_name in self.service.messages)
     self.assertEqual(message.value, message_value)
     self.check_entity(message)
     self.service.messages.delete(message_name)
     self.assertFalse(message_name in self.service.messages)

 def setUp(self):
     super(TestConfs, self).setUp()
     self.app_name = testlib.tmpname()
     self.app = self.service.apps.create(self.app_name)
     # Connect using the test app context
     kwargs = self.opts.kwargs.copy()
     kwargs['app'] = self.app_name
     kwargs['owner'] = "nobody"
     kwargs['sharing'] = "app"
     self.app_service = client.connect(**kwargs)

 def test_saved_search(self):
     temp_name = testlib.tmpname()
     self.check_commands(
         "saved_search/saved_search.py",
         ["saved_search/saved_search.py", "--help"],
         ["saved_search/saved_search.py", "list-all"],
         ["saved_search/saved_search.py", "--operation", "create", "--name", temp_name, "--search", "search * | head 5"],
         ["saved_search/saved_search.py", "list", "--name", temp_name],
         ["saved_search/saved_search.py", "list", "--operation", "delete", "--name", temp_name],
         ["saved_search/saved_search.py", "list", "--name",  "Top five sourcetypes"]
     )

 def setUp(self):
     super(FiredAlertTestCase, self).setUp()
     self.index_name = testlib.tmpname()
     self.assertFalse(self.index_name in self.service.indexes)
     self.index = self.service.indexes.create(self.index_name)
     saved_searches = self.service.saved_searches
     self.saved_search_name = testlib.tmpname()
     self.assertFalse(self.saved_search_name in saved_searches)
     query = "search index=%s" % self.index_name
     kwargs = {'alert_type': 'always',
               'alert.severity': "3",
               'alert.suppress': "0",
               'alert.track': "1",
               'dispatch.earliest_time': "-1h",
               'dispatch.latest_time': "now",
               'is_scheduled': "1",
               'cron_schedule': "* * * * *"}
     self.saved_search = saved_searches.create(
         self.saved_search_name,
         query, **kwargs)

    def setUp(self):
        BindingTestCase.setUp(self)
        self.username = testlib.tmpname()
        self.password = "changeme"
        self.roles = "power"

        # Delete user if it exists already
        try:
            response = self.context.delete(PATH_USERS + self.username)
            self.assertEqual(response.status, 200)
        except HTTPError, e:
            self.assertEqual(e.status, 400)

    def test_read_outputs_with_type(self):
        name = testlib.tmpname()
        service = client.connect(**self.opts.kwargs)
        service.post('data/outputs/tcp/syslog', name=name, type='tcp')
        entity = client.Entity(service, 'data/outputs/tcp/syslog/' + name)
        self.assertTrue('tcp', entity.content.type)

        if service.restart_required:
            self.restartSplunk()
        service = client.connect(**self.opts.kwargs)
        client.Entity(service, 'data/outputs/tcp/syslog/' + name).delete()
        if service.restart_required:
            self.restartSplunk()

    def test_confs(self):
        confs = self.app_service.confs
        conf_name = testlib.tmpname()
        self.assertRaises(KeyError, confs.__getitem__, conf_name)
        self.assertFalse(conf_name in confs)

        conf = confs.create(conf_name)
        self.assertTrue(conf_name in confs)
        self.assertEqual(conf.name, conf_name)

        # New conf should be empty
        stanzas = conf.list()
        self.assertEqual(len(stanzas), 0)

        # Creating a stanza works
        count = len(conf)
        stanza_name = testlib.tmpname()
        stanza = conf.create(stanza_name)
        self.assertEqual(len(conf), count+1)
        self.assertTrue(stanza_name in conf)

        # New stanzas are empty
        self.assertEqual(len(stanza), 0)

        # Update works
        key = testlib.tmpname()
        val = testlib.tmpname()
        stanza.update(**{key: val})
        self.assertEventuallyTrue(lambda: stanza.refresh() and len(stanza) == 1, pause_time=0.2)
        self.assertEqual(len(stanza), 1)
        self.assertTrue(key in stanza)

        values = {testlib.tmpname(): testlib.tmpname(),
                  testlib.tmpname(): testlib.tmpname()}
        stanza.submit(values)
        stanza.refresh()
        for key, value in values.iteritems():
            self.assertTrue(key in stanza)
            self.assertEqual(value, stanza[key])

        count = len(conf)
        conf.delete(stanza_name)
        self.assertFalse(stanza_name in conf)
        self.assertEqual(len(conf), count-1)

        # Can't actually delete configuration files directly, at least
        # not in current versions of Splunk.
        self.assertRaises(client.IllegalOperationException, confs.delete, conf_name)
        self.assertTrue(conf_name in confs)

    def test_create(self):
        self.event_type_name = testlib.tmpname()
        event_types = self.service.event_types
        self.assertFalse(self.event_type_name in event_types)

        kwargs = {}
        kwargs['search'] = "index=_internal *"
        kwargs['description'] = "An internal event"
        kwargs['disabled'] = 1
        kwargs['priority'] = 2

        event_type = event_types.create(self.event_type_name, **kwargs)
        self.assertTrue(self.event_type_name in event_types)
        self.assertEqual(self.event_type_name, event_type.name)

    def test_read(self):
        start_count = len(self.storage_passwords)
        username = testlib.tmpname()

        p = self.storage_passwords.create("changeme", username)
        self.assertEqual(start_count + 1, len(self.storage_passwords))

        for sp in self.storage_passwords:
            self.assertTrue(p.name in self.storage_passwords)
            # Name works with or without a trailing colon
            self.assertTrue((":" + username + ":") in self.storage_passwords)
            self.assertTrue((":" + username) in self.storage_passwords)

        p.delete()
        self.assertEqual(start_count, len(self.storage_passwords))