FM / test

/usr/lib/python3/dist-packages/twisted/web/test/

/usr/lib/python3/dist-packages/twisted/web/test

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
injectionhelpers.py	5627	0644	edit dl rm
requesthelper.py	13406	0644	edit dl rm
test_agent.py	117805	0644	edit dl rm
test_cgi.py	13761	0644	edit dl rm
test_client.py	1372	0644	edit dl rm
test_distrib.py	18288	0644	edit dl rm
test_domhelpers.py	11103	0644	edit dl rm
test_error.py	16210	0644	edit dl rm
test_flatten.py	18259	0644	edit dl rm
test_html.py	1264	0644	edit dl rm
test_http.py	136566	0644	edit dl rm
test_http2.py	109983	0644	edit dl rm
test_httpauth.py	24128	0644	edit dl rm
test_http_headers.py	20386	0644	edit dl rm
test_newclient.py	108751	0644	edit dl rm
test_proxy.py	20092	0644	edit dl rm
test_resource.py	8214	0644	edit dl rm
test_script.py	3784	0644	edit dl rm
test_stan.py	5667	0644	edit dl rm
test_static.py	67858	0644	edit dl rm
test_tap.py	10947	0644	edit dl rm
test_template.py	25977	0644	edit dl rm
test_util.py	13900	0644	edit dl rm
test_vhost.py	7917	0644	edit dl rm
test_web.py	65996	0644	edit dl rm
test_webclient.py	59615	0644	edit dl rm
test_web__responses.py	877	0644	edit dl rm
test_wsgi.py	78210	0644	edit dl rm
test_xml.py	42357	0644	edit dl rm
test_xmlrpc.py	30302	0644	edit dl rm
_util.py	2568	0644	edit dl rm
__init__.py	108	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/twisted/web/test/test_cgi.py (13761B)

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for L{twisted.web.twcgi}.
"""

import sys
import os
import json
from io import BytesIO

from twisted.trial import unittest
from twisted.internet import address, reactor, interfaces, error
from twisted.python import util, failure, log
from twisted.web.http import NOT_FOUND, INTERNAL_SERVER_ERROR
from twisted.web import client, twcgi, server, resource, http_headers
from twisted.web.test._util import _render
from twisted.web.test.test_web import DummyRequest

DUMMY_CGI = '''\
print("Header: OK")
print("")
print("cgi output")
'''

DUAL_HEADER_CGI = '''\
print("Header: spam")
print("Header: eggs")
print("")
print("cgi output")
'''

BROKEN_HEADER_CGI = '''\
print("XYZ")
print("")
print("cgi output")
'''

SPECIAL_HEADER_CGI = '''\
print("Server: monkeys")
print("Date: last year")
print("")
print("cgi output")
'''

READINPUT_CGI = '''\
# This is an example of a correctly-written CGI script which reads a body
# from stdin, which only reads env['CONTENT_LENGTH'] bytes.

import os, sys

body_length = int(os.environ.get('CONTENT_LENGTH',0))
indata = sys.stdin.read(body_length)
print("Header: OK")
print("")
print("readinput ok")
'''

READALLINPUT_CGI = '''\
# This is an example of the typical (incorrect) CGI script which expects
# the server to close stdin when the body of the request is complete.
# A correct CGI should only read env['CONTENT_LENGTH'] bytes.

import sys

indata = sys.stdin.read()
print("Header: OK")
print("")
print("readallinput ok")
'''

NO_DUPLICATE_CONTENT_TYPE_HEADER_CGI = '''\
print("content-type: text/cgi-duplicate-test")
print("")
print("cgi output")
'''

HEADER_OUTPUT_CGI = '''\
import json
import os
print("")
print("")
vals = {x:y for x,y in os.environ.items() if x.startswith("HTTP_")}
print(json.dumps(vals))
'''

class PythonScript(twcgi.FilteredScript):
    filter = sys.executable

class CGITests(unittest.TestCase):
    """
    Tests for L{twcgi.FilteredScript}.
    """

if not interfaces.IReactorProcess.providedBy(reactor):
        skip = "CGI tests require a functional reactor.spawnProcess()"

def startServer(self, cgi):
        root = resource.Resource()
        cgipath = util.sibpath(__file__, cgi)
        root.putChild(b"cgi", PythonScript(cgipath))
        site = server.Site(root)
        self.p = reactor.listenTCP(0, site)
        return self.p.getHost().port

def tearDown(self):
        if getattr(self, 'p', None):
            return self.p.stopListening()

def writeCGI(self, source):
        cgiFilename = os.path.abspath(self.mktemp())
        with open(cgiFilename, 'wt') as cgiFile:
            cgiFile.write(source)
        return cgiFilename

def test_CGI(self):
        cgiFilename = self.writeCGI(DUMMY_CGI)

portnum = self.startServer(cgiFilename)
        url = 'http://localhost:%d/cgi' % (portnum,)
        url = url.encode("ascii")
        d = client.Agent(reactor).request(b"GET", url)
        d.addCallback(client.readBody)
        d.addCallback(self._testCGI_1)
        return d

def _testCGI_1(self, res):
        self.assertEqual(res, b"cgi output" + os.linesep.encode("ascii"))

def test_protectedServerAndDate(self):
        """
        If the CGI script emits a I{Server} or I{Date} header, these are
        ignored.
        """
        cgiFilename = self.writeCGI(SPECIAL_HEADER_CGI)

portnum = self.startServer(cgiFilename)
        url = "http://localhost:%d/cgi" % (portnum,)
        url = url.encode("ascii")
        agent = client.Agent(reactor)
        d = agent.request(b"GET", url)
        d.addCallback(discardBody)
        def checkResponse(response):
            self.assertNotIn('monkeys',
                             response.headers.getRawHeaders('server'))
            self.assertNotIn('last year',
                             response.headers.getRawHeaders('date'))
        d.addCallback(checkResponse)
        return d

def test_noDuplicateContentTypeHeaders(self):
        """
        If the CGI script emits a I{content-type} header, make sure that the
        server doesn't add an additional (duplicate) one, as per ticket 4786.
        """
        cgiFilename = self.writeCGI(NO_DUPLICATE_CONTENT_TYPE_HEADER_CGI)

portnum = self.startServer(cgiFilename)
        url = "http://localhost:%d/cgi" % (portnum,)
        url = url.encode("ascii")
        agent = client.Agent(reactor)
        d = agent.request(b"GET", url)
        d.addCallback(discardBody)
        def checkResponse(response):
            self.assertEqual(
                response.headers.getRawHeaders('content-type'),
                ['text/cgi-duplicate-test'])
            return response
        d.addCallback(checkResponse)
        return d

def test_noProxyPassthrough(self):
        """
        The CGI script is never called with the Proxy header passed through.
        """
        cgiFilename = self.writeCGI(HEADER_OUTPUT_CGI)

portnum = self.startServer(cgiFilename)
        url = "http://localhost:%d/cgi" % (portnum,)
        url = url.encode("ascii")

agent = client.Agent(reactor)

headers = http_headers.Headers({b"Proxy": [b"foo"],
                                        b"X-Innocent-Header": [b"bar"]})
        d = agent.request(b"GET", url, headers=headers)

def checkResponse(response):
            headers = json.loads(response.decode("ascii"))
            self.assertEqual(
                set(headers.keys()),
                {"HTTP_HOST", "HTTP_CONNECTION", "HTTP_X_INNOCENT_HEADER"})

d.addCallback(client.readBody)
        d.addCallback(checkResponse)
        return d

def test_duplicateHeaderCGI(self):
        """
        If a CGI script emits two instances of the same header, both are sent
        in the response.
        """
        cgiFilename = self.writeCGI(DUAL_HEADER_CGI)

portnum = self.startServer(cgiFilename)
        url = "http://localhost:%d/cgi" % (portnum,)
        url = url.encode("ascii")
        agent = client.Agent(reactor)
        d = agent.request(b"GET", url)
        d.addCallback(discardBody)
        def checkResponse(response):
            self.assertEqual(
                response.headers.getRawHeaders('header'), ['spam', 'eggs'])
        d.addCallback(checkResponse)
        return d

def test_malformedHeaderCGI(self):
        """
        Check for the error message in the duplicated header
        """
        cgiFilename = self.writeCGI(BROKEN_HEADER_CGI)

def addMessage(eventDict):
            loggedMessages.append(log.textFromEventDict(eventDict))

log.addObserver(addMessage)
        self.addCleanup(log.removeObserver, addMessage)

def checkResponse(ignored):
            self.assertIn("ignoring malformed CGI header: " + repr(b'XYZ'),
                          loggedMessages)

d.addCallback(checkResponse)
        return d

def test_ReadEmptyInput(self):
        cgiFilename = os.path.abspath(self.mktemp())
        with open(cgiFilename, 'wt') as cgiFile:
            cgiFile.write(READINPUT_CGI)

portnum = self.startServer(cgiFilename)
        agent = client.Agent(reactor)
        url = "http://localhost:%d/cgi" % (portnum,)
        url = url.encode("ascii")
        d = agent.request(b"GET", url)
        d.addCallback(client.readBody)
        d.addCallback(self._test_ReadEmptyInput_1)
        return d
    test_ReadEmptyInput.timeout = 5

def _test_ReadEmptyInput_1(self, res):
        expected = "readinput ok{}".format(os.linesep)
        expected = expected.encode("ascii")
        self.assertEqual(res, expected)

def test_ReadInput(self):
        cgiFilename = os.path.abspath(self.mktemp())
        with open(cgiFilename, 'wt') as cgiFile:
            cgiFile.write(READINPUT_CGI)

portnum = self.startServer(cgiFilename)
        agent = client.Agent(reactor)
        url = "http://localhost:%d/cgi" % (portnum,)
        url = url.encode("ascii")
        d = agent.request(
            uri=url,
            method=b"POST",
            bodyProducer=client.FileBodyProducer(
                BytesIO(b"Here is your stdin")),
        )
        d.addCallback(client.readBody)
        d.addCallback(self._test_ReadInput_1)
        return d
    test_ReadInput.timeout = 5

def _test_ReadInput_1(self, res):
        expected = "readinput ok{}".format(os.linesep)
        expected = expected.encode("ascii")
        self.assertEqual(res, expected)

def test_ReadAllInput(self):
        cgiFilename = os.path.abspath(self.mktemp())
        with open(cgiFilename, 'wt') as cgiFile:
            cgiFile.write(READALLINPUT_CGI)

portnum = self.startServer(cgiFilename)
        url = "http://localhost:%d/cgi" % (portnum,)
        url = url.encode("ascii")
        d = client.Agent(reactor).request(
            uri=url,
            method=b"POST",
            bodyProducer=client.FileBodyProducer(
                BytesIO(b"Here is your stdin")),
        )
        d.addCallback(client.readBody)
        d.addCallback(self._test_ReadAllInput_1)
        return d
    test_ReadAllInput.timeout = 5

def _test_ReadAllInput_1(self, res):
        expected = "readallinput ok{}".format(os.linesep)
        expected = expected.encode("ascii")
        self.assertEqual(res, expected)

def test_useReactorArgument(self):
        """
        L{twcgi.FilteredScript.runProcess} uses the reactor passed as an
        argument to the constructor.
        """
        class FakeReactor:
            """
            A fake reactor recording whether spawnProcess is called.
            """
            called = False
            def spawnProcess(self, *args, **kwargs):
                """
                Set the C{called} flag to C{True} if C{spawnProcess} is called.

@param args: Positional arguments.
                @param kwargs: Keyword arguments.
                """
                self.called = True

fakeReactor = FakeReactor()
        request = DummyRequest(['a', 'b'])
        request.client = address.IPv4Address('TCP', '127.0.0.1', 12345)
        resource = twcgi.FilteredScript("dummy-file", reactor=fakeReactor)
        _render(resource, request)

self.assertTrue(fakeReactor.called)

class CGIScriptTests(unittest.TestCase):
    """
    Tests for L{twcgi.CGIScript}.
    """

def test_pathInfo(self):
        """
        L{twcgi.CGIScript.render} sets the process environment
        I{PATH_INFO} from the request path.
        """
        class FakeReactor:
            """
            A fake reactor recording the environment passed to spawnProcess.
            """
            def spawnProcess(self, process, filename, args, env, wdir):
                """
                Store the C{env} L{dict} to an instance attribute.

@param process: Ignored
                @param filename: Ignored
                @param args: Ignored
                @param env: The environment L{dict} which will be stored
                @param wdir: Ignored
                """
                self.process_env = env

_reactor = FakeReactor()
        resource = twcgi.CGIScript(self.mktemp(), reactor=_reactor)
        request = DummyRequest(['a', 'b'])
        request.client = address.IPv4Address('TCP', '127.0.0.1', 12345)
        _render(resource, request)

self.assertEqual(_reactor.process_env["PATH_INFO"],
                         "/a/b")

class CGIDirectoryTests(unittest.TestCase):
    """
    Tests for L{twcgi.CGIDirectory}.
    """
    def test_render(self):
        """
        L{twcgi.CGIDirectory.render} sets the HTTP response code to I{NOT
        FOUND}.
        """
        resource = twcgi.CGIDirectory(self.mktemp())
        request = DummyRequest([''])
        d = _render(resource, request)
        def cbRendered(ignored):
            self.assertEqual(request.responseCode, NOT_FOUND)
        d.addCallback(cbRendered)
        return d

def test_notFoundChild(self):
        """
        L{twcgi.CGIDirectory.getChild} returns a resource which renders an
        response with the HTTP I{NOT FOUND} status code if the indicated child
        does not exist as an entry in the directory used to initialized the
        L{twcgi.CGIDirectory}.
        """
        path = self.mktemp()
        os.makedirs(path)
        resource = twcgi.CGIDirectory(path)
        request = DummyRequest(['foo'])
        child = resource.getChild("foo", request)
        d = _render(child, request)
        def cbRendered(ignored):
            self.assertEqual(request.responseCode, NOT_FOUND)
        d.addCallback(cbRendered)
        return d

class CGIProcessProtocolTests(unittest.TestCase):
    """
    Tests for L{twcgi.CGIProcessProtocol}.
    """
    def test_prematureEndOfHeaders(self):
        """
        If the process communicating with L{CGIProcessProtocol} ends before
        finishing writing out headers, the response has I{INTERNAL SERVER
        ERROR} as its status code.
        """
        request = DummyRequest([''])
        protocol = twcgi.CGIProcessProtocol(request)
        protocol.processEnded(failure.Failure(error.ProcessTerminated()))
        self.assertEqual(request.responseCode, INTERNAL_SERVER_ERROR)

def discardBody(response):
    """
    Discard the body of a HTTP response.

@param response: The response.

@return: The response.
    """
    return client.readBody(response).addCallback(lambda _: response)

cmd: