File Manager

003 File Manager

Current Path: /usr/local/lib/python3.8/site-packages/salt/ext/tornado/test

usr / local / lib / python3.8 / site-packages / salt / ext / tornado / test /

📁 ..
📄 __init__.py(20 B)
📄 __main__.py(430 B)
📁 __pycache__
📄 asyncio_test.py(4.78 KB)
📄 auth_test.py(22.19 KB)
📄 concurrent_test.py(13.66 KB)
📄 curl_httpclient_test.py(4.83 KB)
📄 escape_test.py(10.9 KB)
📄 gen_test.py(46.36 KB)
📄 http1connection_test.py(2.02 KB)
📄 httpclient_test.py(26.79 KB)
📄 httpserver_test.py(41.43 KB)
📄 httputil_test.py(17.17 KB)
📄 import_test.py(1.73 KB)
📄 ioloop_test.py(24.55 KB)
📄 iostream_test.py(42.53 KB)
📄 locale_test.py(5.96 KB)
📄 locks_test.py(15.74 KB)
📄 log_test.py(9.51 KB)
📄 netutil_test.py(7.6 KB)
📄 options_test.py(10.83 KB)
📄 process_test.py(11.44 KB)
📄 queues_test.py(12.79 KB)
📄 resolve_test_helper.py(514 B)
📄 routing_test.py(7.48 KB)
📄 runtests.py(7.59 KB)
📄 simple_httpclient_test.py(30.33 KB)
📄 stack_context_test.py(10.9 KB)
📄 tcpclient_test.py(11.31 KB)
📄 tcpserver_test.py(2.32 KB)
📄 template_test.py(18.63 KB)
📄 testing_test.py(8.57 KB)
📄 twisted_test.py(27.33 KB)
📄 util.py(4.65 KB)
📄 util_test.py(7.29 KB)
📄 web_test.py(111.58 KB)
📄 websocket_test.py(22.37 KB)
📄 windows_test.py(877 B)
📄 wsgi_test.py(3.71 KB)

Editing: testing_test.py

#!/usr/bin/env python
# pylint: skip-file

from __future__ import absolute_import, division, print_function

from salt.ext.tornado import gen, ioloop
from salt.ext.tornado.log import app_log
from salt.ext.tornado.testing import AsyncTestCase, gen_test, ExpectLog
from salt.ext.tornado.test.util import unittest, skipBefore35, exec_test
import contextlib
import os
import traceback
import warnings

@contextlib.contextmanager
def set_environ(name, value):
    old_value = os.environ.get(name)
    os.environ[name] = value

try:
        yield
    finally:
        if old_value is None:
            del os.environ[name]
        else:
            os.environ[name] = old_value

class AsyncTestCaseTest(AsyncTestCase):
    def test_exception_in_callback(self):
        self.io_loop.add_callback(lambda: 1 / 0)
        try:
            self.wait()
            self.fail("did not get expected exception")
        except ZeroDivisionError:
            pass

def test_wait_timeout(self):
        time = self.io_loop.time

# Accept default 5-second timeout, no error
        self.io_loop.add_timeout(time() + 0.01, self.stop)
        self.wait()

# Timeout passed to wait()
        self.io_loop.add_timeout(time() + 1, self.stop)
        with self.assertRaises(self.failureException):
            self.wait(timeout=0.01)

# Timeout set with environment variable
        self.io_loop.add_timeout(time() + 1, self.stop)
        with set_environ('ASYNC_TEST_TIMEOUT', '0.01'):
            with self.assertRaises(self.failureException):
                self.wait()

def test_subsequent_wait_calls(self):
        """
        This test makes sure that a second call to wait()
        clears the first timeout.
        """
        self.io_loop.add_timeout(self.io_loop.time() + 0.00, self.stop)
        self.wait(timeout=0.02)
        self.io_loop.add_timeout(self.io_loop.time() + 0.03, self.stop)
        self.wait(timeout=0.15)

def test_multiple_errors(self):
        def fail(message):
            raise Exception(message)
        self.io_loop.add_callback(lambda: fail("error one"))
        self.io_loop.add_callback(lambda: fail("error two"))
        # The first error gets raised; the second gets logged.
        with ExpectLog(app_log, "multiple unhandled exceptions"):
            with self.assertRaises(Exception) as cm:
                self.wait()
        self.assertEqual(str(cm.exception), "error one")

class AsyncTestCaseWrapperTest(unittest.TestCase):
    def test_undecorated_generator(self):
        class Test(AsyncTestCase):
            def test_gen(self):
                yield
        test = Test('test_gen')
        result = unittest.TestResult()
        test.run(result)
        self.assertEqual(len(result.errors), 1)
        self.assertIn("should be decorated", result.errors[0][1])

@skipBefore35
    def test_undecorated_coroutine(self):
        namespace = exec_test(globals(), locals(), """
        class Test(AsyncTestCase):
            async def test_coro(self):
                pass
        """)

test_class = namespace['Test']
        test = test_class('test_coro')
        result = unittest.TestResult()

# Silence "RuntimeWarning: coroutine 'test_coro' was never awaited".
        with warnings.catch_warnings():
            warnings.simplefilter('ignore')
            test.run(result)

self.assertEqual(len(result.errors), 1)
        self.assertIn("should be decorated", result.errors[0][1])

def test_undecorated_generator_with_skip(self):
        class Test(AsyncTestCase):
            @unittest.skip("don't run this")
            def test_gen(self):
                yield
        test = Test('test_gen')
        result = unittest.TestResult()
        test.run(result)
        self.assertEqual(len(result.errors), 0)
        self.assertEqual(len(result.skipped), 1)

def test_other_return(self):
        class Test(AsyncTestCase):
            def test_other_return(self):
                return 42
        test = Test('test_other_return')
        result = unittest.TestResult()
        test.run(result)
        self.assertEqual(len(result.errors), 1)
        self.assertIn("Return value from test method ignored", result.errors[0][1])

class SetUpTearDownTest(unittest.TestCase):
    def test_set_up_tear_down(self):
        """
        This test makes sure that AsyncTestCase calls super methods for
        setUp and tearDown.

InheritBoth is a subclass of both AsyncTestCase and
        SetUpTearDown, with the ordering so that the super of
        AsyncTestCase will be SetUpTearDown.
        """
        events = []
        result = unittest.TestResult()

class SetUpTearDown(unittest.TestCase):
            def setUp(self):
                events.append('setUp')

def tearDown(self):
                events.append('tearDown')

class InheritBoth(AsyncTestCase, SetUpTearDown):
            def test(self):
                events.append('test')

InheritBoth('test').run(result)
        expected = ['setUp', 'test', 'tearDown']
        self.assertEqual(expected, events)

class GenTest(AsyncTestCase):
    def setUp(self):
        super(GenTest, self).setUp()
        self.finished = False

def tearDown(self):
        self.assertTrue(self.finished)
        super(GenTest, self).tearDown()

@gen_test
    def test_sync(self):
        self.finished = True

@gen_test
    def test_async(self):
        yield gen.Task(self.io_loop.add_callback)
        self.finished = True

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

# This can't use assertRaises because we need to inspect the
        # exc_info triple (and not just the exception object)
        try:
            test(self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            # The stack trace should blame the add_timeout line, not just
            # unrelated IOLoop/testing internals.
            self.assertIn(
                "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                traceback.format_exc())

self.finished = True

def test_no_timeout(self):
        # A test that does not exceed its timeout should succeed.
        @gen_test(timeout=1)
        def test(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 0.1)

test(self)
        self.finished = True

def test_timeout_environment_variable(self):
        @gen_test(timeout=0.5)
        def test_long_timeout(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 0.25)

# Uses provided timeout of 0.5 seconds, doesn't time out.
        with set_environ('ASYNC_TEST_TIMEOUT', '0.1'):
            test_long_timeout(self)

self.finished = True

def test_no_timeout_environment_variable(self):
        @gen_test(timeout=0.01)
        def test_short_timeout(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 1)

# Uses environment-variable timeout of 0.1, times out.
        with set_environ('ASYNC_TEST_TIMEOUT', '0.1'):
            with self.assertRaises(ioloop.TimeoutError):
                test_short_timeout(self)

self.finished = True

def test_with_method_args(self):
        @gen_test
        def test_with_args(self, *args):
            self.assertEqual(args, ('test',))
            yield gen.Task(self.io_loop.add_callback)

test_with_args(self, 'test')
        self.finished = True

def test_with_method_kwargs(self):
        @gen_test
        def test_with_kwargs(self, **kwargs):
            self.assertDictEqual(kwargs, {'test': 'test'})
            yield gen.Task(self.io_loop.add_callback)

test_with_kwargs(self, test='test')
        self.finished = True

@skipBefore35
    def test_native_coroutine(self):
        namespace = exec_test(globals(), locals(), """
        @gen_test
        async def test(self):
            self.finished = True
        """)

namespace['test'](self)

@skipBefore35
    def test_native_coroutine_timeout(self):
        # Set a short timeout and exceed it.
        namespace = exec_test(globals(), locals(), """
        @gen_test(timeout=0.1)
        async def test(self):
            await gen.sleep(1)
        """)

try:
            namespace['test'](self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            self.finished = True

if __name__ == '__main__':
    unittest.main()

003 File Manager

Editing: testing_test.py

Upload File

Create Folder