# Path: tests/testlib.py
# class NoRestartRequiredError(Exception):
# class WaitTimedOutError(Exception):
# class SDKTestCase(unittest.TestCase):
# def to_bool(x):
# def tmpname():
# def wait(predicate, timeout=60, pause_time=0.5):
#     def assertEventuallyTrue(self, predicate, timeout=30, pause_time=0.5,
#                              timeout_message="Operation timed out."):
#     def check_content(self, entity, **kwargs):
#     def check_entity(self, entity):
#     def clear_restart_message(self):
#     def fake_splunk_version(self, version):
#     def install_app_from_collection(self, name):
#     def app_collection_installed(self):
#     def pathInApp(self, appName, pathComponents):
#     def uncheckedRestartSplunk(self, timeout=240):
#     def restartSplunk(self, timeout=240):
#     def setUpClass(cls):
#     def setUp(self):
#     def tearDown(self):
# 
# Path: splunklib/binding.py
# def _log_duration(f):
#     @wraps(f)
#     def new_f(*args, **kwargs):
#         start_time = datetime.now()
#         val = f(*args, **kwargs)
#         end_time = datetime.now()
#         logging.debug("Operation took %s", end_time-start_time)
#         return val
#     return new_f
# 
# class HTTPError(Exception):
#     """This exception is raised for HTTP responses that return an error."""
#     def __init__(self, response, _message=None):
#         status = response.status
#         reason = response.reason
#         body = response.body.read()
#         try:
#             detail = XML(body).findtext("./messages/msg")
#         except ParseError as err:
#             detail = body
#         message = "HTTP %d %s%s" % (
#             status, reason, "" if detail is None else " -- %s" % detail)
#         Exception.__init__(self, _message or message)
#         self.status = status
#         self.reason = reason
#         self.headers = response.headers
#         self.body = body
#         self._response = response


from io import BytesIO
from time import sleep
from tests import testlib
from splunklib.binding import _log_duration, HTTPError
        from datetime import datetime
                    from collections import OrderedDict
                    from splunklib.ordereddict import OrderedDict
                    from collections import OrderedDict
                    from splunklib.ordereddict import OrderedDict
import io
    import unittest2 as unittest
    import unittest
import splunklib.client as client
import splunklib.results as results
import pytest
        import splunklib.client as client
        import splunklib.results as results
        import splunklib.results as results
        import splunklib.client as client
        import splunklib.results as results
        import splunklib.client as client
        import splunklib.results as results

# TODO: Determine if we should be importing ExpatError if ParseError is not available (e.g., on Python 2.6)
# There's code below that now catches SyntaxError instead of ParseError. Should we be catching ExpathError instead?

# from xml.etree.ElementTree import ParseError


class TestUtilities(testlib.SDKTestCase):
    def test_service_search(self):
        job = self.service.search('search index=_internal earliest=-1m | head 3')
        self.assertTrue(job.sid in self.service.jobs)
        job.cancel()

    def test_oneshot_with_garbage_fails(self):
        jobs = self.service.jobs
        self.assertRaises(TypeError, jobs.create, "abcd", exec_mode="oneshot")

    def test_oneshot(self):
        jobs = self.service.jobs
        stream = jobs.oneshot("search index=_internal earliest=-1m | head 3")
        result = results.ResultsReader(stream)
        ds = list(result)
        self.assertEqual(result.is_preview, False)
        self.assertTrue(isinstance(ds[0], dict) or \
                            isinstance(ds[0], results.Message))
        nonmessages = [d for d in ds if isinstance(d, dict)]
        self.assertTrue(len(nonmessages) <= 3)

    def test_export_with_garbage_fails(self):
        jobs = self.service.jobs

# Path: tests/testlib.py
# class NoRestartRequiredError(Exception):
# class WaitTimedOutError(Exception):
# class SDKTestCase(unittest.TestCase):
# def to_bool(x):
# def tmpname():
# def wait(predicate, timeout=60, pause_time=0.5):
#     def assertEventuallyTrue(self, predicate, timeout=30, pause_time=0.5,
#                              timeout_message="Operation timed out."):
#     def check_content(self, entity, **kwargs):
#     def check_entity(self, entity):
#     def clear_restart_message(self):
#     def fake_splunk_version(self, version):
#     def install_app_from_collection(self, name):
#     def app_collection_installed(self):
#     def pathInApp(self, appName, pathComponents):
#     def uncheckedRestartSplunk(self, timeout=240):
#     def restartSplunk(self, timeout=240):
#     def setUpClass(cls):
#     def setUp(self):
#     def tearDown(self):
# 
# Path: splunklib/binding.py
# def _log_duration(f):
#     @wraps(f)
#     def new_f(*args, **kwargs):
#         start_time = datetime.now()
#         val = f(*args, **kwargs)
#         end_time = datetime.now()
#         logging.debug("Operation took %s", end_time-start_time)
#         return val
#     return new_f
# 
# class HTTPError(Exception):
#     """This exception is raised for HTTP responses that return an error."""
#     def __init__(self, response, _message=None):
#         status = response.status
#         reason = response.reason
#         body = response.body.read()
#         try:
#             detail = XML(body).findtext("./messages/msg")
#         except ParseError as err:
#             detail = body
#         message = "HTTP %d %s%s" % (
#             status, reason, "" if detail is None else " -- %s" % detail)
#         Exception.__init__(self, _message or message)
#         self.status = status
#         self.reason = reason
#         self.headers = response.headers
#         self.body = body
#         self._response = response


# Path: tests/test_job.py
from io import BytesIO
from time import sleep
from tests import testlib
from splunklib.binding import _log_duration, HTTPError
        from datetime import datetime
                    from collections import OrderedDict
                    from splunklib.ordereddict import OrderedDict
                    from collections import OrderedDict
                    from splunklib.ordereddict import OrderedDict
import io
    import unittest2 as unittest
    import unittest
import splunklib.client as client
import splunklib.results as results
import pytest
        import splunklib.client as client
        import splunklib.results as results
        import splunklib.results as results
        import splunklib.client as client
        import splunklib.results as results
        import splunklib.client as client
        import splunklib.results as results


# TODO: Determine if we should be importing ExpatError if ParseError is not available (e.g., on Python 2.6)
# There's code below that now catches SyntaxError instead of ParseError. Should we be catching ExpathError instead?

# from xml.etree.ElementTree import ParseError


class TestUtilities(testlib.SDKTestCase):
    def test_service_search(self):
        job = self.service.search('search index=_internal earliest=-1m | head 3')
        self.assertTrue(job.sid in self.service.jobs)
        job.cancel()

    def test_oneshot_with_garbage_fails(self):
        jobs = self.service.jobs
        self.assertRaises(TypeError, jobs.create, "abcd", exec_mode="oneshot")

    def test_oneshot(self):
        jobs = self.service.jobs
        stream = jobs.oneshot("search index=_internal earliest=-1m | head 3")
        result = results.ResultsReader(stream)
        ds = list(result)
        self.assertEqual(result.is_preview, False)
        self.assertTrue(isinstance(ds[0], dict) or \
                            isinstance(ds[0], results.Message))
        nonmessages = [d for d in ds if isinstance(d, dict)]
        self.assertTrue(len(nonmessages) <= 3)

    def test_export_with_garbage_fails(self):
        jobs = self.service.jobs

        self.assertRaises(client.HTTPError, jobs.export, "asdaf;lkj2r23=")