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: http1connection_test.py

# pylint: skip-file
from __future__ import absolute_import, division, print_function

import socket

from salt.ext.tornado.http1connection import HTTP1Connection
from salt.ext.tornado.httputil import HTTPMessageDelegate
from salt.ext.tornado.iostream import IOStream
from salt.ext.tornado.locks import Event
from salt.ext.tornado.netutil import add_accept_handler
from salt.ext.tornado.testing import AsyncTestCase, bind_unused_port, gen_test

class HTTP1ConnectionTest(AsyncTestCase):
    def setUp(self):
        super(HTTP1ConnectionTest, self).setUp()
        self.asyncSetUp()

@gen_test
    def asyncSetUp(self):
        listener, port = bind_unused_port()
        event = Event()

def accept_callback(conn, addr):
            self.server_stream = IOStream(conn)
            self.addCleanup(self.server_stream.close)
            event.set()

add_accept_handler(listener, accept_callback)
        self.client_stream = IOStream(socket.socket())
        self.addCleanup(self.client_stream.close)
        yield [self.client_stream.connect(('127.0.0.1', port)),
               event.wait()]
        self.io_loop.remove_handler(listener)
        listener.close()

@gen_test
    def test_http10_no_content_length(self):
        # Regression test for a bug in which can_keep_alive would crash
        # for an HTTP/1.0 (not 1.1) response with no content-length.
        conn = HTTP1Connection(self.client_stream, True)
        self.server_stream.write(b"HTTP/1.0 200 Not Modified\r\n\r\nhello")
        self.server_stream.close()

event = Event()
        test = self
        body = []

class Delegate(HTTPMessageDelegate):
            def headers_received(self, start_line, headers):
                test.code = start_line.code

def data_received(self, data):
                body.append(data)

def finish(self):
                event.set()

yield conn.read_response(Delegate())
        yield event.wait()
        self.assertEqual(self.code, 200)
        self.assertEqual(b''.join(body), b'hello')

003 File Manager

Editing: http1connection_test.py

Upload File

Create Folder