def testPostMethodDeCompressesDeflatedBody_deflate(self):
        self.requestData = None
        def handler(**kwargs):
            self.requestData = kwargs

        reactor = Reactor()
        server = HttpServer(reactor, self.port, handler, timeout=0.01)
        server.listen()
        sok = socket()
        sok.connect(('localhost', self.port))
        bodyData = 'bodydatabodydata'
        bodyDataCompressed = compress(bodyData)
        contentLengthCompressed = len(bodyDataCompressed)
        sok.send(('POST / HTTP/1.0\r\nContent-Type: application/x-www-form-urlencoded\r\nContent-Length: %d\r\nContent-Encoding: deflate\r\n\r\n' % contentLengthCompressed) + bodyDataCompressed)

        while not self.requestData:
            reactor.step()
        self.assertEquals(dict, type(self.requestData))
        self.assertTrue('Headers' in self.requestData)
        headers = self.requestData['Headers']
        self.assertEquals('POST', self.requestData['Method'])
        self.assertEquals('application/x-www-form-urlencoded', headers['Content-Type'])
        self.assertEquals(contentLengthCompressed, int(headers['Content-Length']))  # TS: is this correct?, maybe decompressed length?

        self.assertTrue('Body' in self.requestData)
        self.assertEquals('bodydatabodydata', self.requestData['Body'])

 def testInvalidPOSTRequestStartsOnlyOneTimer(self):
     # problem in found in OAS, timers not removed properly when whole body hasnt been read yet
     _httpserver.RECVSIZE = 1
     reactor = Reactor()
     timers = []
     orgAddTimer = reactor.addTimer
     def addTimerInterceptor(*timer):
         timers.append(timer)
         return orgAddTimer(*timer)
     reactor.addTimer = addTimerInterceptor
     server = HttpServer(reactor, self.port, lambda **kwargs: (x for x in []), timeout=0.01)
     server.listen()
     sok = socket()
     sok.connect(('localhost', self.port))
     sok.send('POST / HTTP/1.0\r\nContent-Length: 10\r\n\r\n')
     reactor.step()
     sok.send(".")
     sleep(0.1)
     reactor.step()
     sok.send(".")
     reactor.step()
     sleep(0.1)
     while select([sok],[], [], 0) != ([sok], [], []):
         reactor.step()
     self.assertEquals(2, len(timers))

    def sendRequestAndReceiveResponse(self, request, response='The Response', recvSize=4096, compressResponse=False, extraStepAfterCompress=True):
        self.responseCalled = False
        @compose
        def responseGenFunc(**kwargs):
            yield response
            yield ''
            self.responseCalled = True
        server = HttpServer(self.reactor, self.port, responseGenFunc, recvSize=recvSize, compressResponse=compressResponse)
        server.listen()
        sok = socket()
        sok.connect(('localhost', self.port))
        sok.send(request)

        mockStdout = None
        with self.stdout_replaced() as mockStdout:
            while not self.responseCalled:
                self.reactor.step()
            if compressResponse and extraStepAfterCompress: #not everythingSent???: 
                self.reactor.step()
        stdoutValue = mockStdout.getvalue()
        if stdoutValue:
            print stdoutValue

        server.shutdown()
        r = sok.recv(4096)
        sok.close()
        return r

    def testPostMethodDeCompressesDeflatedBody_gzip(self):
        self.requestData = None
        def handler(**kwargs):
            self.requestData = kwargs

        reactor = Reactor()
        server = HttpServer(reactor, self.port, handler, timeout=0.01)
        server.listen()
        sok = socket()
        sok.connect(('localhost', self.port))
        bodyData = 'bodydatabodydata'
        _sio = StringIO()
        _gzFileObj = GzipFile(filename=None, mode='wb', compresslevel=6, fileobj=_sio)
        _gzFileObj.write(bodyData); _gzFileObj.close()
        compressedBodyData = _sio.getvalue()
        bodyDataCompressed = compress(bodyData)
        contentLengthCompressed = len(bodyDataCompressed)
        sok.send(('POST / HTTP/1.0\r\nContent-Type: application/x-www-form-urlencoded\r\nContent-Length: %d\r\nContent-Encoding: gzip\r\n\r\n' % contentLengthCompressed) + bodyDataCompressed)

        while not self.requestData:
            reactor.step()
        self.assertEquals(dict, type(self.requestData))
        self.assertTrue('Headers' in self.requestData)
        headers = self.requestData['Headers']
        self.assertEquals('POST', self.requestData['Method'])
        self.assertEquals('application/x-www-form-urlencoded', headers['Content-Type'])
        self.assertEquals(contentLengthCompressed, int(headers['Content-Length']))

        self.assertTrue('Body' in self.requestData)
        self.assertEquals('bodydatabodydata', self.requestData['Body'])

 def testSuspendProtocolWithThrow(self):
     with Reactor() as reactor:
         suspend = Suspend()
         def handler(**httpvars):
             yield 'before suspend'
             yield suspend
             try:
                 suspend.getResult()
                 self.fail()
             except ValueError, e:
                 tbstring = format_exc()
                 yield "result = %s" % tbstring
             yield 'after suspend'
         listener = MyMockSocket()
         port = PortNumberGenerator.next()
         httpserver = HttpServer(reactor, port, handler, sok=listener)
         httpserver.listen()
         reactor.removeReader(listener) # avoid new connections
         httpserver._acceptor._accept()
         reactor.step()
         reactor.step()
         reactor.step()
         self.assertEquals(reactor, suspend._reactor)
         self.assertEquals(0, len(reactor._fds))
         def raiser():
             raise ValueError("BAD VALUE")
         try:
             raiser()
         except ValueError, e:
             exc_type, exc_value, exc_traceback = exc_info()
             suspend.throw(exc_type, exc_value, exc_traceback)

    def testSuspendProtocol(self):
        with Reactor() as reactor:
            suspend = Suspend()
            def handler(**httpvars):
                yield 'before suspend'
                yield suspend
                yield "result = %s" % suspend.getResult()
                yield 'after suspend'
            listener = MyMockSocket()
            port = PortNumberGenerator.next()
            httpserver = HttpServer(reactor, port, handler, sok=listener)
            httpserver.listen()
            reactor.removeReader(listener) # avoid new connections
            httpserver._acceptor._accept()
            self.assertEquals(1, len(reactor._fds))
            self.assertEquals([READ_INTENT], [v.intent for v in reactor._fds.values()])
            reactor.step()
            self.assertEquals(1, len(reactor._fds))
            self.assertEquals([WRITE_INTENT], [v.intent for v in reactor._fds.values()])
            reactor.step()
            reactor.step()
            self.assertEquals(reactor, suspend._reactor)
            self.assertEquals(0, len(reactor._fds))
            suspend.resume('RESPONSE')
            self.assertEquals(1, len(reactor._fds))
            self.assertEquals([WRITE_INTENT], [v.intent for v in reactor._fds.values()])
            reactor.step()
            reactor.step()
            reactor.step()
            self.assertEquals(['before suspend', 'result = RESPONSE', 'after suspend'], listener.data)
            self.assertEquals(0, len(reactor._fds))

            # cleanup (most) fd's
            listener.close()

 def testSuspendProtocol(self):
     reactor = Reactor(select_func=mockselect)
     suspend = Suspend()
     def handler(**httpvars):
         yield 'before suspend'
         yield suspend
         yield "result = %s" % suspend.getResult()
         yield 'after suspend'
     listener = MyMockSocket()
     port = 9
     httpserver = HttpServer(reactor, port, handler, sok=listener)
     httpserver.listen()
     reactor.removeReader(listener) # avoid new connections
     httpserver._acceptor._accept()
     reactor.step()
     reactor.step()
     self.assertEquals(1, len(reactor._writers))
     reactor.step()
     self.assertEquals(reactor, suspend._reactor)
     self.assertEquals(0, len(reactor._writers))
     suspend.resume('RESPONSE')
     reactor.step()
     reactor.step()
     reactor.step()
     self.assertEquals(['before suspend', 'result = RESPONSE', 'after suspend'], listener.data)

class ObservableHttpServer(Observable):
    def __init__(self, reactor, port, timeout=1, prio=None, sok=None, maxConnections=None, compressResponse=True, bindAddress=None):
        Observable.__init__(self)
        self._port = port
        self._reactor = reactor
        self._timeout = timeout
        self._started = False
        self._prio = prio
        self._sok = sok
        self._maxConnections = maxConnections
        self._compressResponse = compressResponse
        self._bindAddress = bindAddress

    def startServer(self):
        """Starts server,

        When running a http server on port 80, this method should be called by the
        root user. In other cases it will be started when initializing all observers,
        see observer_init()
        """
        self._httpserver = HttpServer(
                self._reactor,
                self._port,
                self._connect,
                timeout=self._timeout,
                prio=self._prio,
                sok=self._sok,
                maxConnections=self._maxConnections,
                errorHandler=self._error,
                compressResponse=self._compressResponse,
                bindAddress=self._bindAddress
            )
        self._httpserver.listen()
        self._started = True

    def observer_init(self):
        if not self._started:
            self.startServer()

    def _connect(self, **kwargs):
        return compose(self.handleRequest(port=self._port, **kwargs))

    def _error(self, **kwargs):
        yield serverUnavailableHtml +\
        '<html><head></head><body><h1>Service Unavailable</h1></body></html>'
        self.do.logHttpError(**kwargs)

    def handleRequest(self, RequestURI=None, **kwargs):
        scheme, netloc, path, query, fragments = urlsplit(RequestURI)
        arguments = parse_qs(query, keep_blank_values=True)
        requestArguments = {
            'scheme': scheme, 'netloc': netloc, 'path': path, 'query': query, 'fragments': fragments,
            'arguments': arguments,
            'RequestURI': RequestURI}
        requestArguments.update(kwargs)
        yield self.all.handleRequest(**requestArguments)

    def setMaxConnections(self, m):
        self._httpserver.setMaxConnections(m)

class AsyncReaderTest(WeightlessTestCase):

    def dispatch(self, *args, **kwargs):
        return compose(self.handler(*args, **kwargs))

    def setUp(self):
        WeightlessTestCase.setUp(self)
        self.reactor = Reactor()
        self.port = randint(2**10, 2**16)
        self.httpserver = HttpServer(self.reactor, self.port, self.dispatch)
        self.httpserver.listen()

    def tearDown(self):
        self.httpserver.shutdown()
        self.reactor.shutdown()
        WeightlessTestCase.tearDown(self)

    def testHttpRequest(self):
        self.assertEquals('GET / HTTP/1.0\r\n', _httpRequest('GET', '/'))
        self.assertEquals('POST / HTTP/1.0\r\n', _httpRequest('POST', '/'))

    def testPassRequestThruToBackOfficeServer(self):
        done = [False]
        backofficeport = self.port + 1
        def passthruhandler(*args, **kwargs):
            request = kwargs['RequestURI']
            response = yield httpget('localhost', backofficeport, request)
            yield response
            done[0] = True
        self.handler = passthruhandler
        expectedrequest = "GET /depot?arg=1&arg=2 HTTP/1.0\r\n\r\n"
        responses = (i for i in ['hel', 'lo!'])
        backofficeserver = testserver(backofficeport, responses, expectedrequest)
        client = clientget('localhost', self.port, '/depot?arg=1&arg=2')
        while not done[0]:
            self.reactor.step()
        response = client.recv(99)
        self.assertEquals('hello!', response)

    def testConnectFails(self):
        exceptions = []
        def failingserver(*args, **kwarg):
            try:
                response = yield httpget(*target)
            except Exception, e:
                exceptions.append(exc_info())
        self.handler = failingserver

        clientget('localhost', self.port, '/')
        target = ('localhost', 'port', '/') # non-numeric port
        try:
            with self.stderr_replaced():
                with self.loopingReactor():
                    while not exceptions:
                        pass
        except Exception, e:
            pass

 def testValidRequestResetsTimer(self):
     reactor = Reactor()
     server = HttpServer(reactor, self.port, lambda **kwargs: ('a' for a in range(3)), timeout=0.01, recvSize=3)
     server.listen()
     sok = socket()
     sok.connect(('localhost', self.port))
     sok.send('GET / HTTP/1.0\r\n\r\n')
     sleep(0.02)
     for i in range(11):
         reactor.step()
     response = sok.recv(4096)
     self.assertEquals('aaa', response)

    def testSuspendThrowBackwardsCompatibleWithInstanceOnlyThrow_YouWillMissTracebackHistory(self):
        with Reactor() as reactor:
            suspend = Suspend()
            def handler(**httpvars):
                yield 'before suspend'
                yield suspend
                try:
                    suspend.getResult()
                    self.fail()
                except ValueError, e:
                    tbstring = format_exc()
                    yield "result = %s" % tbstring
                yield 'after suspend'
            listener = MyMockSocket()
            port = PortNumberGenerator.next()
            httpserver = HttpServer(reactor, port, handler, sok=listener)
            httpserver.listen()
            reactor.removeReader(listener) # avoid new connections
            httpserver._acceptor._accept()
            self.assertEquals(1, len(reactor._fds))
            reactor.step()
            reactor.step()
            reactor.step()
            self.assertEquals(reactor, suspend._reactor)
            self.assertEquals(0, len(reactor._fds))
            def raiser():
                raise ValueError("BAD VALUE")
            try:
                raiser()
            except:
                exc_value = exc_info()[1]
                suspend.throw(exc_value)
            self.assertEquals(1, len(reactor._fds))
            reactor.step()
            reactor.step()
            reactor.step()
            expectedTraceback = ignoreLineNumbers("""Traceback (most recent call last):
      File "%(__file__)s", line 201, in handler
        suspend.getResult()
      File "%(suspend.py)s", line 62, in getResult
        raise self._exception[0], self._exception[1], self._exception[2]
    ValueError: BAD VALUE
            """ % fileDict)
            self.assertEquals(3, len(listener.data))
            self.assertEquals('before suspend', listener.data[0])
            self.assertEqualsWS("result = %s" % expectedTraceback, ignoreLineNumbers(listener.data[1]))
            self.assertEquals('after suspend', listener.data[2])
            self.assertEquals(0, len(reactor._fds))

            # cleanup (most) fd's
            listener.close()

 def testSendHeader(self):
     self.kwargs = None
     def response(**kwargs):
         self.kwargs = kwargs
         yield 'nosense'
     reactor = Reactor()
     server = HttpServer(reactor, self.port, response)
     server.listen()
     sok = socket()
     sok.connect(('localhost', self.port))
     sok.send('GET /path/here HTTP/1.0\r\nConnection: close\r\nApe-Nut: Mies\r\n\r\n')
     while not self.kwargs:
         reactor.step()
     self.assertEquals({'Body': '', 'RequestURI': '/path/here', 'HTTPVersion': '1.0', 'Method': 'GET', 'Headers': {'Connection': 'close', 'Ape-Nut': 'Mies'}, 'Client': ('127.0.0.1', MATCHALL)}, self.kwargs)

 def testConnect(self):
     self.req = False
     def onRequest(**kwargs):
         self.req = True
         yield 'nosens'
     reactor = Reactor()
     server = HttpServer(reactor, self.port, onRequest)
     server.listen()
     sok = socket()
     sok.connect(('localhost', self.port))
     sok.send('GET / HTTP/1.0\r\n\r\n')
     reactor.step() # connect/accept
     reactor.step() # read GET request
     reactor.step() # call onRequest for response data
     self.assertEquals(True, self.req)

    def testReadChunkedPost(self):
        self.requestData = {}
        def handler(**kwargs):
            self.requestData = kwargs

        reactor = Reactor()
        server = HttpServer(reactor, self.port, handler, timeout=0.01, recvSize=3)
        server.listen()
        sok = socket()
        sok.connect(('localhost', self.port))
        sok.send('POST / HTTP/1.0\r\nContent-Type: application/x-www-form-urlencoded\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nabcde\r\n5\r\nfghij\r\n0\r\n')

        reactor.addTimer(0.2, lambda: self.fail("Test Stuck"))
        while self.requestData.get('Body', None) != 'abcdefghij':
            reactor.step()

 def testInvalidRequestWithHalfHeader(self):
     reactor = Reactor()
     server = HttpServer(reactor, self.port, None, timeout=0.1)
     server.listen()
     sok = socket()
     sok.connect(('localhost', self.port))
     sok.send('POST / HTTP/1.0\r\n') 
     sok.send('Expect: something\r\n')
     sok.send('Content-Length: 5\r\n')
     sok.send('\r\n1234')
     sok.close()
     with self.stderr_replaced() as s:
         for i in range(4):
             reactor.step()
         self.assertEquals(1, len(reactor._readers))

    def testDefaultErrorHandler(self):
        def handler(**kwargs):
            yield "OK"

        server = HttpServer(self.reactor, self.port, handler, maxConnections=5)
        server.listen()

        self.reactor.getOpenConnections = lambda: 10

        sock = socket()
        sock.connect(('localhost', self.port))
        self.reactor.step()
        sock.send("GET / HTTP/1.0\r\n\r\n")
        self.reactor.step().step().step()

        self.assertEquals('HTTP/1.0 503 Service Unavailable\r\n\r\n<html><head></head><body><h1>Service Unavailable</h1></body></html>', sock.recv(1024))
        server.shutdown()