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()
python类NOT_DONE_YET的实例源码
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 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 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 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):
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"
