def n(self, proto, handler, buf):
     if buf == b'6':
         x, y = handler.reportCursorPosition()
         proto.transport.write(b'\x1b['
                               + intToBytes(x+1)
                               + b';'
                               + intToBytes(y+1)
                               + b'R')
     else:
         handler.unhandledControlSequence(b'\x1b[' + buf + b'n')

 def setScrollRegion(self, first=None, last=None):
     if first is not None:
         first = intToBytes(first)
     else:
         first = b''
     if last is not None:
         last = intToBytes(last)
     else:
         last = b''
     self.write(b'\x1b[' + first + b';' + last + b'r')

 def test_selectGraphicRendition(self):
     """
     L{ServerProtocol.selectGraphicRendition} writes a control
     sequence containing the requested attributes and ending with
     L{CSFinalByte.SGR}
     """
     self.protocol.selectGraphicRendition(str(BLINK), str(UNDERLINE))
     self.assertEqual(self.transport.value(),
                      self.CSI +
                      intToBytes(BLINK) + b';' + intToBytes(UNDERLINE) +
                      CSFinalByte.SGR.value)

        def cbStarted(ignored):
            connectionRefused = client.startedDeferred = defer.Deferred()
            client.transport.connect("127.0.0.1", 80)

            for i in range(10):
                client.transport.write(intToBytes(i))
                server.transport.write(intToBytes(i), ("127.0.0.1", 80))

            return self.assertFailure(
                connectionRefused,
                error.ConnectionRefusedError)

    def __bytes__(self):
        """
        Return a byte string representation of the channel
        """
        name = self.name
        if not name:
            name = b'None'

        return (b'<SSHChannel ' + name +
                b' (lw ' + intToBytes(self.localWindowLeft) +
                b' rw ' + intToBytes(self.remoteWindowLeft) +
                b')>')

    def testSimpleCardinals(self):
        self.parser.dataReceived(
            b''.join(
                    [b''.join([b'\x1b[' + n + ch
                             for n in (b'', intToBytes(2), intToBytes(20), intToBytes(200))]
                           ) for ch in iterbytes(b'BACD')
                    ]))
        occs = occurrences(self.proto)

        for meth in ("Down", "Up", "Forward", "Backward"):
            for count in (1, 2, 20, 200):
                result = self.assertCall(occs.pop(0), "cursor" + meth, (count,))
                self.assertFalse(occurrences(result))
        self.assertFalse(occs)

    def __init__(self, code, message=None, response=None):
        """
        Initializes a basic exception.

        @type code: L{bytes} or L{int}
        @param code: Refers to an HTTP status code (for example, 200) either as
            an integer or a bytestring representing such. If no C{message} is
            given, C{code} is mapped to a descriptive bytestring that is used
            instead.

        @type message: L{bytes}
        @param message: A short error message, for example "NOT FOUND".

        @type response: L{bytes}
        @param response: A complete HTML document for an error page.
        """
        message = message or _codeToMessage(code)

        Exception.__init__(self, code, message, response)

        if isinstance(code, int):
            # If we're given an int, convert it to a bytestring
            # downloadPage gives a bytes, Agent gives an int, and it worked by
            # accident previously, so just make it keep working.
            code = intToBytes(code)

        self.status = code
        self.message = message
        self.response = response

    def test_blksize(self):
        self.s = MockSession()
        opts = self.proto.processOptions(OrderedDict({b'blksize':b'8'}))
        self.proto.applyOptions(self.s, opts)
        self.assertEqual(self.s.block_size, 8)
        self.assertEqual(opts, OrderedDict({b'blksize':b'8'}))

        self.s = MockSession()
        opts = self.proto.processOptions(OrderedDict({b'blksize':b'foo'}))
        self.proto.applyOptions(self.s, opts)
        self.assertEqual(self.s.block_size, 512)
        self.assertEqual(opts, OrderedDict())

        self.s = MockSession()
        opts = self.proto.processOptions(OrderedDict({b'blksize':b'65464'}))
        self.proto.applyOptions(self.s, opts)
        self.assertEqual(self.s.block_size, MAX_BLOCK_SIZE)
        self.assertEqual(opts, OrderedDict({b'blksize':intToBytes(MAX_BLOCK_SIZE)}))

        self.s = MockSession()
        opts = self.proto.processOptions(OrderedDict({b'blksize':b'65465'}))
        self.proto.applyOptions(self.s, opts)
        self.assertEqual(self.s.block_size, 512)
        self.assertEqual(opts, OrderedDict())

        self.s = MockSession()
        opts = self.proto.processOptions(OrderedDict({b'blksize':b'7'}))
        self.proto.applyOptions(self.s, opts)
        self.assertEqual(self.s.block_size, 512)
        self.assertEqual(opts, OrderedDict())

 def render_GET(self, request):
     data = json.dumps(request.args)
     request.setHeader(b"content-type", networkString("application/json"))
     request.setHeader(b"content-length", intToBytes(len(data)))
     if request.method == b"HEAD":
         return b""
     return data

def _spawn(script, outputFD):
    """
    Start a script that is a peer of this test as a subprocess.

    @param script: the module name of the script in this directory (no
        package prefix, no '.py')
    @type script: C{str}

    @rtype: L{StartStopProcessProtocol}
    """
    pyExe = FilePath(sys.executable).asBytesMode().path
    env = bytesEnviron()
    env[b"PYTHONPATH"] = FilePath(
        pathsep.join(sys.path)).asBytesMode().path
    sspp = StartStopProcessProtocol()
    reactor.spawnProcess(
        sspp, pyExe, [
            pyExe,
            FilePath(__file__).sibling(script + ".py").asBytesMode().path,
            intToBytes(outputFD),
        ],
        env=env,
        childFDs={0: "w", 1: "r", 2: "r", outputFD: outputFD}
    )
    return sspp

 def render_GET(self, request):
     data = json.dumps(request.args)
     request.setHeader(b'content-type', networkString('application/json'))
     request.setHeader(b'content-length', intToBytes(len(data)))
     if request.method == b'HEAD':
         return b''
     return data

    def render_GET(self, request):
        def decode(x):
            if isinstance(x, bytes):
                return bytes2unicode(x)
            elif isinstance(x, (list, tuple)):
                return [bytes2unicode(y) for y in x]
            elif isinstance(x, dict):
                newArgs = {}
                for a, b in x.items():
                    newArgs[decode(a)] = decode(b)
                return newArgs
            return x

        args = decode(request.args)
        content_type = request.getHeader(b'content-type')
        if content_type == b"application/json":
            jsonBytes = request.content.read()
            jsonStr = bytes2unicode(jsonBytes)
            args['json_received'] = json.loads(jsonStr)

        data = json.dumps(args)
        data = unicode2bytes(data)
        request.setHeader(b'content-type', b'application/json')
        request.setHeader(b'content-length', intToBytes(len(data)))
        if request.method == b'HEAD':
            return b''
        return data

    def _cbRender(self, result, request, responseFailed=None):
        if responseFailed:
            return

        if isinstance(result, Handler):
            result = result.result
        if not isinstance(result, Fault):
            result = (result,)
        try:
            try:
                content = xmlrpclib.dumps(
                    result, methodresponse=True,
                    allow_none=self.allowNone)
            except Exception as e:
                f = Fault(self.FAILURE, "Can't serialize output: %s" % (e,))
                content = xmlrpclib.dumps(f, methodresponse=True,
                                          allow_none=self.allowNone)

            if isinstance(content, unicode):
                content = content.encode('utf8')
            request.setHeader(
                b"content-length", intToBytes(len(content)))
            request.write(content)
        except:
            log.err()
        request.finish()

 def spin():
     for value in count:
         if value == howMany:
             connection.loseConnection()
             return
         connection.write(intToBytes(value))
         break
     reactor.callLater(0, spin)

    def contentFinish(self, content):
        if self.obj.client_supports_gzip:
            self.setHeader(b'content-encoding', b'gzip')
            content = self.zip(content, True)

        self.setHeader(b'content-length', intToBytes(len(content)))
        self.write(content)
        self.finish()

def _extractURLparts(request):
    """
    Extracts and decodes URI parts from C{request}.

    All strings must be UTF8-decodable.

    @param request: A Twisted Web request.
    @type request: L{twisted.web.iweb.IRequest}

    @raise URLDecodeError: If one of the parts could not be decoded as UTF-8.

    @return: L{tuple} of the URL scheme, the server name, the server port, the
        path info and the script name.
    @rtype: L{tuple} of L{unicode}, L{unicode}, L{int}, L{unicode}, L{unicode}
    """
    server_name = request.getRequestHostname()
    if hasattr(request.getHost(), 'port'):
        server_port = request.getHost().port
    else:
        server_port = 0
    if (bool(request.isSecure()), server_port) not in [
            (True, 443), (False, 80), (False, 0), (True, 0)]:
        server_name = server_name + b":" + intToBytes(server_port)

    script_name = b''
    if request.prepath:
        script_name = b'/'.join(request.prepath)

        if not script_name.startswith(b'/'):
            script_name = b'/' + script_name

    path_info = b''
    if request.postpath:
        path_info = b'/'.join(request.postpath)

        if not path_info.startswith(b'/'):
            path_info = b'/' + path_info

    url_scheme = u'https' if request.isSecure() else u'http'

    utf8Failures = []
    try:
        server_name = server_name.decode("utf-8")
    except UnicodeDecodeError:
        utf8Failures.append(("SERVER_NAME", failure.Failure()))
    try:
        path_info = path_info.decode("utf-8")
    except UnicodeDecodeError:
        utf8Failures.append(("PATH_INFO", failure.Failure()))
    try:
        script_name = script_name.decode("utf-8")
    except UnicodeDecodeError:
        utf8Failures.append(("SCRIPT_NAME", failure.Failure()))

    if utf8Failures:
        raise _URLDecodeError(utf8Failures)

    return url_scheme, server_name, server_port, path_info, script_name

    def _render(self, request):
        origin = request.getHeader('Origin')
        if origin:
            request.setHeader('Access-Control-Allow-Origin', origin)
            request.setHeader('Access-Control-Allow-Credentials', 'true')
        else:
            request.setHeader('Access-Control-Allow-Origin', '*')
        request.setHeader('Access-Control-Allow-Methods', 'GET, PUT, POST, DELETE, OPTIONS, HEAD')
        request.setHeader('Access-Control-Allow-Headers',
                          'Origin, Content-Type, Cache-Control, X-Requested-With, Authorization')

        ret = None
        try:
            ret = yield self.handle_request(request)
            # allow views to take full control of output streaming
            if ret is not NOT_DONE_YET and ret is not EmptyResponse:
                request.setHeader('Content-Type', 'application/json')
                json_data = json.dumps(ret, indent=2, cls=JsonSetEncoder)
                request.setHeader('Content-Length', intToBytes(len(json_data)))
                request.write(json_data)
        except HttpStatus as exc:
            request.setResponseCode(exc.status_code, exc.status_description)
            for name, value in exc.headers.items():
                request.responseHeaders.addRawHeader(name, value)
            # emit a header with location of the trusted keystone instance
            if get_config().getboolean('auth', 'use_keystone', False):
                keystone_uri = get_config().getstring('keystone', 'keystone_uri')
                request.setHeader('WWW-Authenticate', 'Keystone uri=%s' % keystone_uri)
            if exc.body:
                request.write(json.dumps(exc.body))
            else:
                request.write("%s %s\n" % (exc.status_code, exc.status_description))
            if exc.message:
                request.write("%s\n" % exc.message)
        except Exception:
            request.setHeader('Content-Type', 'text/plain')
            request.setResponseCode(500, "Server Error")
            error_message = "%s %s\n\n" % (500, "Server Error")
            request.setHeader('Content-Length', intToBytes(len(error_message)))
            request.write(error_message)
            log.err(system='httprest')
            failure.Failure().printTraceback(request)
        finally:
            if ret is not NOT_DONE_YET:
                request.finish()

 def lineReceived(self, line):
     self.i += 1
     if self.i < 4:
         # by this point we should have connection closed,
         # but just in case we didn't we won't ever send 'Hello 4'
         self.sendLine(b"Hello " + intToBytes(self.i))
     SimpleProtocol.lineReceived(self, line)
     if self.lines[-1] == b"Hello 3":
         self.transport.loseConnection()

 def testHostHeader(self):
     # if we pass Host header explicitly, it should be used, otherwise
     # it should extract from url
     return defer.gatherResults([
         client.getPage(self.getURL("host")).addCallback(
                 self.assertEqual, b"127.0.0.1:" + intToBytes(self.portno)),
         client.getPage(self.getURL("host"),
                        headers={b"Host": b"www.example.com"}).addCallback(
                 self.assertEqual, b"www.example.com")])

    def forwardData(self, data, end=False):
        if not self.startedWriting:
            if self.obj.client_supports_gzip:
                self.setHeader(b'content-encoding', b'gzip')

            if data != '' and end:
                self.setHeader(b'content-length', intToBytes(len(data)))

        if data != '':
            self.write(data)