我们从Python开源项目中，提取了以下45个代码示例，用于说明如何使用twisted.web.server.NOT_DONE_YET。
def render(self, resource): """ Render the given resource as a response to this request. This implementation only handles a few of the most common behaviors of resources. It can handle a render method that returns a string or C{NOT_DONE_YET}. It doesn't know anything about the semantics of request methods (eg HEAD) nor how to set any particular headers. Basically, it's largely broken, but sufficient for some tests at least. It should B{not} be expanded to do all the same stuff L{Request} does. Instead, L{DummyRequest} should be phased out and L{Request} (or some other real code factored in a different way) used. """ result = resource.render(self) if result is NOT_DONE_YET: return self.write(result) self.finish()
def getJson(self, request): deferred = self.getParent().getProcessList() def cb(processes): self.logger.debug("getJson() cb(%r)", processes) request.write(json.dumps({"processes": processes})) request.finish() def eb(failure): self.logger.debug("getJson() eb(%r)", failure) if isinstance(failure.type, Exception): util.logTwistedFailure(self.logger, failure, "Exception thrown while getting process list") self.serverError(request) request.write(json.dumps({"error": "500 Internal Server Error"})) request.finish() deferred.addCallbacks(cb, eb) return server.NOT_DONE_YET # # Create your application class, inheriting from the Guernsey RootResource class #
def testPostWithCorrectPasswordReturnsCorrectKeys(self): """ A C{POST} to C{/users/user/verify} with the correct password returns a JSON object with all the expected keys, including valid = True. """ with login(None, None, self.transact) as session: resource = VerifyUserPasswordResource(None, session, 'user') payload = dumps({'password': 'pass'}) headers = {'Content-Length': [str(len(payload))], 'Content-Type': ['application/json'], 'X-Forwarded-Protocol': ['https']} request = FakeRequest(method='POST', headers=Headers(headers), body=payload) self.assertEqual(NOT_DONE_YET, resource.render(request)) yield resource.deferred self.assertEqual(request.code, http.OK) result = loads(request.response) self.assertEqual( ['accessToken', 'fullname', 'renewalToken', 'role', 'valid'], sorted(result.keys())) self.assertTrue(result['valid'])
def testPostWithCorrectPasswordDoesNotCauseALogWarning(self): """ A C{POST} to C{/users/user/verify} with the correct password should not cause a complaint about unknown return payload fields in the logging system. """ with login(None, None, self.transact) as session: resource = VerifyUserPasswordResource(None, session, 'user') payload = dumps({'password': 'pass'}) headers = {'Content-Length': [str(len(payload))], 'Content-Type': ['application/json'], 'X-Forwarded-Protocol': ['https']} request = FakeRequest(method='POST', headers=Headers(headers), body=payload) self.assertEqual(NOT_DONE_YET, resource.render(request)) yield resource.deferred logOutput = self.log.getvalue() self.assertNotIn("unknown response payload field 'renewalToken'", logOutput) self.assertNotIn("unknown response payload field 'accessToken'", logOutput)
def testPostWithCorrectPasswordReturnsCorrectRole(self): """ A C{POST} to C{/users/user/verify} with the correct password returns a JSON object with the correct user role. """ with login(None, None, self.transact) as session: resource = VerifyUserPasswordResource(None, session, 'user') payload = dumps({'password': 'pass'}) headers = {'Content-Length': [str(len(payload))], 'Content-Type': ['application/json'], 'X-Forwarded-Protocol': ['https']} request = FakeRequest(method='POST', headers=Headers(headers), body=payload) self.assertEqual(NOT_DONE_YET, resource.render(request)) yield resource.deferred self.assertEqual(request.code, http.OK) result = loads(request.response) self.assertEqual('USER', result['role'])
def testPostWithCorrectPasswordReturnsCorrectFullname(self): """ A C{POST} to C{/users/user/verify} with the correct password returns a JSON object with the user's correct full name. """ with login(None, None, self.transact) as session: resource = VerifyUserPasswordResource(None, session, 'user') payload = dumps({'password': 'pass'}) headers = {'Content-Length': [str(len(payload))], 'Content-Type': ['application/json'], 'X-Forwarded-Protocol': ['https']} request = FakeRequest(method='POST', headers=Headers(headers), body=payload) self.assertEqual(NOT_DONE_YET, resource.render(request)) yield resource.deferred self.assertEqual(request.code, http.OK) result = loads(request.response) self.assertEqual(u'Peter Parker', result['fullname'])
def testPostWithIncorrectPasswordReturnsFalse(self): """ A C{POST} to C{/users/user/verify} with the incorrect password returns a C{{'valid': False}} response. """ with login(None, None, self.transact) as session: resource = VerifyUserPasswordResource(None, session, 'user') payload = dumps({'password': 'wrong'}) headers = {'Content-Length': [str(len(payload))], 'Content-Type': ['application/json'], 'X-Forwarded-Protocol': ['https']} request = FakeRequest(method='POST', headers=Headers(headers), body=payload) self.assertEqual(NOT_DONE_YET, resource.render(request)) yield resource.deferred self.assertEqual(request.code, http.OK) self.assertEqual(loads(request.response), {'valid': False})
def testPostWithoutPasswordReturnsBadRequest(self): """ A C{POST} to C{/users/user/verify} without a password returns a 400 Bad Request. """ with login(None, None, self.transact) as session: resource = VerifyUserPasswordResource(None, session, 'user') payload = '' headers = {'Content-Length': [str(len(payload))], 'Content-Type': ['application/json'], 'X-Forwarded-Protocol': ['https']} request = FakeRequest(method='POST', headers=Headers(headers), body=payload) self.assertEqual(NOT_DONE_YET, resource.render(request)) yield resource.deferred self.assertEqual(request.code, http.BAD_REQUEST)
def testPostWithExtraCrapInPayloadReturnsBadRequest(self): """ A C{POST} to C{/users/user/verify} with unexpected data in the payload returns a 400 Bad Request. """ with login(None, None, self.transact) as session: resource = VerifyUserPasswordResource(None, session, 'user') payload = dumps({'password': 'pass', 'foo': 'bar'}) headers = {'Content-Length': [str(len(payload))], 'Content-Type': ['application/json'], 'X-Forwarded-Protocol': ['https']} request = FakeRequest(method='POST', headers=Headers(headers), body=payload) self.assertEqual(NOT_DONE_YET, resource.render(request)) yield resource.deferred self.assertEqual(request.code, http.BAD_REQUEST)
def testInsecurePostIsRejected(self): """A C{POST} via HTTP is rejected if not in development mode.""" self.config.set('service', 'development', 'false') with login(None, None, self.transact) as session: resource = VerifyUserPasswordResource(None, session, 'user') payload = '' headers = {'Content-Length': [str(len(payload))], 'Content-Type': ['application/json']} request = FakeRequest(method='POST', headers=Headers(headers), body=payload) self.assertEqual(NOT_DONE_YET, resource.render(request)) yield resource.deferred self.assertEqual(request.code, http.BAD_REQUEST) self.assertEqual( request.getResponseHeader('X-FluidDB-Message'), '/users/<username>/verify requests must use HTTPS')
def test_simpleRender(self): """ L{renderElement} returns NOT_DONE_YET and eventually writes the rendered L{Element} to the request before finishing the request. """ element = TestElement() d = self.request.notifyFinish() def check(_): self.assertEqual( b"".join(self.request.written), b"<!DOCTYPE html>\n" b"<p>Hello, world.</p>") self.assertTrue(self.request.finished) d.addCallback(check) self.assertIdentical(NOT_DONE_YET, renderElement(self.request, element)) return d
def render(self, request): """Render this request, from my server. This will always be asynchronous, and therefore return NOT_DONE_YET. It spins off a request to the pb client, and either adds it to the list of pending issues or requests it immediately, depending on if the client is already connected. """ if not self.publisher: self.pending.append(request) if not self.waiting: self.waiting = 1 bf = pb.PBClientFactory() timeout = 10 if self.host == "unix": reactor.connectUNIX(self.port, bf, timeout) else: reactor.connectTCP(self.host, self.port, bf, timeout) d = bf.getRootObject() d.addCallbacks(self.connected, self.notConnected) else: i = Issue(request) self.publisher.callRemote('request', request).addCallbacks(i.finished, i.failed) return server.NOT_DONE_YET
def test_postSuccess(self): '''POSTing to an XHRResource results in a 200 and NOT_DONE_YET.''' request = DummyRequestResponseHeaders([b'serverID', b'sessionID', self.resourceName]) request.method = b'POST' result = self.xhr.render(request) self.assertEqual(len(self.sessionRecorder.requestsMaybeAttached), 1) [(actualFactory, actualRequest)] = self.sessionRecorder.requestsMaybeAttached self.assertIsInstance(actualFactory, self.factoryType) self.assertIs(actualRequest, request) contentType = (b'Content-Type', [b'application/javascript; charset=UTF-8']) self.assertIn(contentType, request.sortedResponseHeaders) self.assertIs(result, server.NOT_DONE_YET)
def test__render_POST_queue_messages(self): status_worker = Mock() status_worker.queueMessage = Mock() status_worker.queueMessage.return_value = succeed(None) resource = StatusHandlerResource(status_worker) message = { 'event_type': ( factory.make_name('type') + '/' + factory.make_name('sub_type')), 'origin': factory.make_name('origin'), 'name': factory.make_name('name'), 'description': factory.make_name('description'), } token = factory.make_name('token') request = self.make_request( content=json.dumps(message).encode('ascii'), token=token) output = resource.render_POST(request) self.assertEquals(NOT_DONE_YET, output) self.assertEquals(204, request.responseCode) self.assertThat( status_worker.queueMessage, MockCalledOnceWith(token, message))
def test_processMessages_calls_loseConnection_if_missing_type_field(self): protocol, factory = self.make_protocol() protocol.user = maas_factory.make_User() mock_loseConnection = self.patch_autospec(protocol, "loseConnection") self.patch_autospec( protocol, "handleRequest").return_value = NOT_DONE_YET messages = [ {"request_id": 1}, {"type": MSG_TYPE.REQUEST, "request_id": 2}, ] protocol.messages = deque(messages) self.expectThat([messages[0]], Equals(protocol.processMessages())) self.expectThat( mock_loseConnection, MockCalledOnceWith( STATUSES.PROTOCOL_ERROR, "Missing type field in the received message."))
def _render(resource, request): # type: (Resource, Union[DummyRequest, Request]) -> Any """ Simulate rendering of a Twisted resource. :param resource: Twisted Resource with render() method. :param request: Request (or mock object). :return: Deferred """ result = resource.render(request) if isinstance(result, bytes): request.write(result) request.finish() return succeed(None) elif result == NOT_DONE_YET: if request.finished: return succeed(None) else: return request.notifyFinish() else: raise ValueError("Unexpected return value: %r" % (result,))
def render(self, request): """First, check to see if this request is attempting to hook up the output conduit. If so, do it. Otherwise, unlink the current session's View from the MVC notification infrastructure, then render the page normally. """ # Check to see if we're hooking up an output conduit sess = request.getSession(interfaces.IWovenLivePage) #print "REQUEST.ARGS", request.args if request.args.has_key('woven_hookupOutputConduitToThisFrame'): sess.hookupOutputConduit(request) return server.NOT_DONE_YET if request.args.has_key('woven_clientSideEventName'): try: request.d = microdom.parseString('<xml/>', caseInsensitive=0, preserveCase=0) eventName = request.args['woven_clientSideEventName'][0] eventTarget = request.args['woven_clientSideEventTarget'][0] eventArgs = request.args.get('woven_clientSideEventArguments', []) #print "EVENT", eventName, eventTarget, eventArgs return self.clientToServerEvent(request, eventName, eventTarget, eventArgs) except: fail = failure.Failure() self.view.renderFailure(fail, request) return server.NOT_DONE_YET # Unlink the current page in this user's session from MVC notifications page = sess.getCurrentPage() #request.currentId = getattr(sess, 'currentId', 0) if page is not None: page.view.unlinkViews() sess.setCurrentPage(None) #print "PAGE SESSION IS NONE" self.pageSession = None return Controller.render(self, request)
def render(self, request): """Handle a SOAP command.""" data = request.content.read() p, header, body, attrs = SOAPpy.parseSOAPRPC(data, 1, 1, 1) methodName, args, kwargs, ns = p._name, p._aslist, p._asdict, p._ns # deal with changes in SOAPpy 0.11 if callable(args): args = args() if callable(kwargs): kwargs = kwargs() function = self.lookupFunction(methodName) if not function: self._methodNotFound(request, methodName) return server.NOT_DONE_YET else: if hasattr(function, "useKeywords"): keywords = {} for k, v in kwargs.items(): keywords[str(k)] = v d = defer.maybeDeferred(function, **keywords) else: d = defer.maybeDeferred(function, *args) d.addCallback(self._gotResult, request, methodName) d.addErrback(self._gotError, request, methodName) return server.NOT_DONE_YET
def render(self, request): self.d.addCallback(self._cbChild, request).addErrback( self._ebChild,request) from twisted.web.server import NOT_DONE_YET return NOT_DONE_YET
def _cbChild(self, child, request): result = resource.getChildForRequest(child, request).render(request) from twisted.web.server import NOT_DONE_YET if result == NOT_DONE_YET: return else: request.write(result) request.finish()
def getJson(self, request): deferred = self.prepareProcessList() def cb(processes): self.logger.debug("getJson() cb(%r)", processes) request.write(json.dumps({"processes": processes})) request.finish() def eb(failure): self.logger.debug("getJson() eb(%r)", failure) if isinstance(failure.type, Exception): util.logTwistedFailure(self.logger, failure, "Exception thrown while getting process list") self.serverError(request) request.write(json.dumps({"error": "500 Internal Server Error"})) request.finish() deferred.addCallbacks(cb, eb) return server.NOT_DONE_YET # # Create a resource class for processes owned by the user root. It # will produce HTML and JSON. This class inherits everything from the # ProcessesResource created above, but substitutes the # LineReceiverProtocol for a FilteredLineReceiverProtocol # instead. This resource just illustrates the use of that class. # # Note that we also reuse the template file from the parent class. If # we did not specify it here, the framework would try to use our class # name to construct a template filename, and no such template exists. #
def __getHtml(self, request): res = self.getHtml(request) if res is server.NOT_DONE_YET: return res if type(res) == str: return res return self.fillTemplate(res, request=request)
def __getJson(self, request): res = self.getJson(request) if res is server.NOT_DONE_YET: return res if type(res) == str: return res if res == None: return "" return json.dumps(res) + "\n"
def getHtml(self, request): deferred = self.prepareProcessList() def cb(processes): self.logger.debug("getHtml() cb(%r)", processes) request.write(self.fillTemplate(model = { "processes": sorted(processes, key=lambda x: int(x.pid)) })) request.finish() def eb(failure): self.logger.debug("getHtml() eb(%r)", failure) if isinstance(failure.type, Exception): util.logTwistedFailure(self.logger, failure, "Exception thrown while getting process list") self.serverError(request) request.write("Internal Server Error") request.finish() deferred.addCallbacks(cb, eb) return server.NOT_DONE_YET # # This method works the same way as the getHtml() method above, # but as expected, the callback produces JSON instead of HTML, and # does not call the template engine. # # Note that we use our own JSON encoder. This is actually just an # extension to the JSON encoder provided by the standard library, # but it has a few nice features that the standard JSON encoder # doesn't have. For example, it allows each class to define its # own JSON encoder method, called __json__(), which will be called # if available to convert an object to JSON. #
def getHtml(self, request): deferred = self.getProcessList() def cb(processes): self.logger.debug("getHtml() cb(%r)", processes) request.write(self.fillTemplate(model = { "processes": sorted(processes, key=lambda x: int(x.pid)) })) request.finish() def eb(failure): self.logger.debug("getHtml() eb(%r)", failure) if isinstance(failure.type, Exception): util.logTwistedFailure(self.logger, failure, "Exception thrown while getting process list") self.serverError(request) request.write("Internal Server Error") request.finish() deferred.addCallbacks(cb, eb) return server.NOT_DONE_YET # # The getJson() method is copied almost verbatim from example # 4. The only difference is the call to getProcessList(), since # this class does not contain any prepareProcessList() method. #
def getJson(self, request): deferred = self.getProcessList() def cb(processes): self.logger.debug("getJson() cb(%r)", processes) request.write(json.dumps({"processes": processes})) request.finish() def eb(failure): self.logger.debug("getJson() eb(%r)", failure) if isinstance(failure.type, Exception): util.logTwistedFailure(self.logger, failure, "Exception thrown while getting process list") self.serverError(request) request.write(json.dumps({"error": "500 Internal Server Error"})) request.finish() deferred.addCallbacks(cb, eb) return server.NOT_DONE_YET # # Create a resource class for processes owned by the user root. Since # this class is now a child of ProcessesResource, we can call the # getProcessList() method of that class by using getParent(). # # The data returned is processed in the callback by filtering the # process list on the ProcessModel 'user' attribute, so that the # filtered list only contains processes owned by root. The filtered # list is then passed to the template system just like in example 4. #
def render_GET(self, request): request = self.add_headers(request) request.write(b"") if self.is_valid_path(request.postpath): self.handle_request(request) request.notifyFinish().addBoth(self.responseFailed, request, request.postpath) return server.NOT_DONE_YET else: return self.invalid_request(request=request, code=403, reason='Invalid Path')
def render_GET(self, request): session = request.getSession() session.touch() log.debug("HTTP Request from %s:%s (%s) to %s", request.client.host, request.client.port, session.uid, request.uri) if not self.sessions.has_key(session.uid): log.info("New Client Session: %s" % session.uid) session._expireCall.cancel() session.sessionTimeout = HTTP_SESSION_TIMEOUT session.startCheckingExpiration() session.notifyOnExpire(self._expireSession) session.updates = [] session.isAuthenticated = not self.monast.authRequired session.username = None self.sessions[session.uid] = session if not session.isAuthenticated and request.path != "/doAuthentication": return "ERROR :: Authentication Required" handler = self.handlers.get(request.path) if handler: d = task.deferLater(reactor, 0.1, lambda: request) d.addCallback(handler) d.addErrback(self._onRequestFailure, request) return TWebServer.NOT_DONE_YET return "ERROR :: Request Not Found"
def testInsecurePostIsNotRejectedInDevelopmentMode(self): """A C{POST} via HTTP is not rejected when in development mode.""" self.config.set('service', 'development', 'true') with login(None, None, self.transact) as session: resource = VerifyUserPasswordResource(None, session, 'user') payload = dumps({'password': 'pass'}) headers = {'Content-Length': [str(len(payload))], 'Content-Type': ['application/json']} request = FakeRequest(method='POST', headers=Headers(headers), body=payload) self.assertEqual(NOT_DONE_YET, resource.render(request)) yield resource.deferred self.assertEqual(request.code, http.OK)
def render(self, request): """Handle a SOAP command.""" data = request.content.read() p, header, body, attrs = SOAPpy.parseSOAPRPC(data, 1, 1, 1) methodName, args, kwargs = p._name, p._aslist, p._asdict # deal with changes in SOAPpy 0.11 if callable(args): args = args() if callable(kwargs): kwargs = kwargs() function = self.lookupFunction(methodName) if not function: self._methodNotFound(request, methodName) return server.NOT_DONE_YET else: if hasattr(function, "useKeywords"): keywords = {} for k, v in kwargs.items(): keywords[str(k)] = v d = defer.maybeDeferred(function, **keywords) else: d = defer.maybeDeferred(function, *args) d.addCallback(self._gotResult, request, methodName) d.addErrback(self._gotError, request, methodName) return server.NOT_DONE_YET
def _render(resource, request): result = resource.render(request) if isinstance(result, bytes): request.write(result) request.finish() return succeed(None) elif result is server.NOT_DONE_YET: if request.finished: return succeed(None) else: return request.notifyFinish() else: raise ValueError("Unexpected return value: %r" % (result,))
def test_largeWrite(self): """ If a string longer than the Banana size limit is passed to the L{distrib.Request} passed to the remote resource, it is broken into smaller strings to be transported over the PB connection. """ class LargeWrite(resource.Resource): def render(self, request): request.write('x' * SIZE_LIMIT + 'y') request.finish() return server.NOT_DONE_YET request = self._requestTest(LargeWrite()) request.addCallback(self.assertEqual, 'x' * SIZE_LIMIT + 'y') return request
def render(self, request): """ Turn the request into the appropriate C{environ} C{dict} suitable to be passed to the WSGI application object and then pass it on. The WSGI application object is given almost complete control of the rendering process. C{NOT_DONE_YET} will always be returned in order and response completion will be dictated by the application object, as will the status, headers, and the response body. """ response = _WSGIResponse( self._reactor, self._threadpool, self._application, request) response.start() return NOT_DONE_YET
def finished(self, result): if result != server.NOT_DONE_YET: assert isinstance(result, str), "return value not a string" self.request.write(result) self.request.finish()
def remote_request(self, request): """ Look up the resource for the given request and render it. """ res = self.site.getResourceFor(request) log.msg( request ) result = res.render(request) if result is not server.NOT_DONE_YET: request.write(result) request.finish() return server.NOT_DONE_YET
def _render(resource, request, notifyFinish=True): result = resource.render(request) if isinstance(result, bytes): request.write(result) request.finish() return succeed(None) elif result is server.NOT_DONE_YET: if request.finished or not notifyFinish: return succeed(None) else: return request.notifyFinish() else: raise ValueError("Unexpected return value: %r" % (result,))
def render_POST(self, request): self.applyPolicies(request) if (request.postpath[-1] == self.resourceName and self.sessions.attachToSession(self.factory, request)): return server.NOT_DONE_YET request.setResponseCode(http.NOT_FOUND) return encoding.EMPTY
def test_dataReceived_adds_message_to_queue(self): protocol, factory = self.make_protocol() self.patch_autospec(protocol, "processMessages") message = {"type": MSG_TYPE.REQUEST} self.expectThat( protocol.dataReceived( json.dumps(message).encode("ascii")), Is(NOT_DONE_YET)) self.expectThat(protocol.messages, Equals(deque([message])))
def test_dataReceived_calls_processMessages(self): protocol, factory = self.make_protocol() mock_processMessages = self.patch_autospec(protocol, "processMessages") message = {"type": MSG_TYPE.REQUEST} self.expectThat( protocol.dataReceived( json.dumps(message).encode("ascii")), Is(NOT_DONE_YET)) self.expectThat(mock_processMessages, MockCalledOnceWith())
