我们从Python开源项目中，提取了以下20个代码示例，用于说明如何使用twisted.web.server.Session()。
def __init__(self, postpath, session=None): self.sitepath = [] self.written = [] self.finished = 0 self.postpath = postpath self.prepath = [] self.session = None self.protoSession = session or Session(0, self) self.args = {} self.outgoingHeaders = {} self.requestHeaders = Headers() self.responseHeaders = Headers() self.responseCode = None self.headers = {} self._finishedDeferreds = [] self._serverName = b"dummy" self.clientproto = b"HTTP/1.0"
def test_sessionUIDGeneration(self): """ L{site.getSession} generates L{Session} objects with distinct UIDs from a secure source of entropy. """ site = server.Site(resource.Resource()) # Ensure that we _would_ use the unpredictable random source if the # test didn't stub it. self.assertIdentical(site._entropy, os.urandom) def predictableEntropy(n): predictableEntropy.x += 1 return (unichr(predictableEntropy.x) * n).encode("charmap") predictableEntropy.x = 0 self.patch(site, "_entropy", predictableEntropy) a = self.getAutoExpiringSession(site) b = self.getAutoExpiringSession(site) self.assertEqual(a.uid, b"01" * 0x20) self.assertEqual(b.uid, b"02" * 0x20) # This functionality is silly (the value is no longer used in session # generation), but 'counter' was a public attribute since time # immemorial so we should make sure if anyone was using it to get site # metrics or something it keeps working. self.assertEqual(site.counter, 2)
def test_startCheckingExpiration(self): """ L{server.Session.startCheckingExpiration} causes the session to expire after L{server.Session.sessionTimeout} seconds without activity. """ self.session.startCheckingExpiration() # Advance to almost the timeout - nothing should happen. self.clock.advance(self.session.sessionTimeout - 1) self.assertIn(self.uid, self.site.sessions) # Advance to the timeout, the session should expire. self.clock.advance(1) self.assertNotIn(self.uid, self.site.sessions) # There should be no calls left over, either. self.assertFalse(self.clock.calls)
def test_sessionAttribute(self): """ On a L{Request}, the C{session} attribute retrieves the associated L{Session} only if it has been initialized. If the request is secure, it retrieves the secure session. """ site = server.Site(resource.Resource()) d = DummyChannel() d.transport = DummyChannel.SSL() request = server.Request(d, 1) request.site = site request.sitepath = [] self.assertIs(request.session, None) insecureSession = request.getSession(forceNotSecure=True) self.addCleanup(insecureSession.expire) self.assertIs(request.session, None) secureSession = request.getSession() self.addCleanup(secureSession.expire) self.assertIsNot(secureSession, None) self.assertIsNot(secureSession, insecureSession) self.assertIs(request.session, secureSession)
def getSession(self): if self.session: return self.session assert not self.written, "Session cannot be requested after data has been written." self.session = self.protoSession return self.session
def __init__(self, postpath, session=None): self.sitepath = [] self.written = [] self.finished = 0 self.postpath = postpath self.prepath = [] self.session = None self.protoSession = session or server.Session(0, self) self.args = {} self.outgoingHeaders = {}
def startSession(): global startedSession if not startedSession: print "Started session adapter" registerAdapter(User, Session, IUser) startedSession = True
def __init__(self, postpath, session=None): self.sitepath = [] self.written = [] self.finished = 0 self.postpath = postpath self.prepath = [] self.session = None self.protoSession = session or Session(0, self) self.args = {} self.requestHeaders = Headers() self.responseHeaders = Headers() self.responseCode = None self._finishedDeferreds = [] self._serverName = b"dummy" self.clientproto = b"HTTP/1.0"
def getAutoExpiringSession(self, site): """ Create a new session which auto expires at cleanup. @param site: The site on which the session is created. @type site: L{server.Site} @return: A newly created session. @rtype: L{server.Session} """ session = site.makeSession() # Clean delayed calls from session expiration. self.addCleanup(session.expire) return session
def test_makeSession(self): """ L{site.getSession} generates a new C{Session} instance with an uid of type L{bytes}. """ site = server.Site(resource.Resource()) session = self.getAutoExpiringSession(site) self.assertIsInstance(session, server.Session) self.assertIsInstance(session.uid, bytes)
def test_defaultReactor(self): """ If not value is passed to L{server.Session.__init__}, the global reactor is used. """ session = server.Session(server.Site(resource.Resource()), b'123') self.assertIdentical(session._reactor, reactor)
def test_expire(self): """ L{server.Session.expire} expires the session. """ self.session.expire() # It should be gone from the session dictionary. self.assertNotIn(self.uid, self.site.sessions) # And there should be no pending delayed calls. self.assertFalse(self.clock.calls)
def test_expireWhileChecking(self): """ L{server.Session.expire} expires the session even if the timeout call isn't due yet. """ self.session.startCheckingExpiration() self.test_expire()
def test_notifyOnExpire(self): """ A function registered with L{server.Session.notifyOnExpire} is called when the session expires. """ callbackRan = [False] def expired(): callbackRan[0] = True self.session.notifyOnExpire(expired) self.session.expire() self.assertTrue(callbackRan[0])
def test_retrieveExistingSession(self): """ L{Request.getSession} retrieves an existing session if the relevant cookie is set in the incoming request. """ site = server.Site(resource.Resource()) d = DummyChannel() request = server.Request(d, 1) request.site = site request.sitepath = [] mySession = server.Session(b"special-id", site) site.sessions[mySession.uid] = mySession request.received_cookies[b'TWISTED_SESSION'] = mySession.uid self.assertIs(request.getSession(), mySession)
def getHtml(self, request): self.logger.debug("getHtml(%r)", request) if self.checkPermission(request): self.logger.debug("User has permission to access this resource") else: self.logger.debug("User does NOT have permission to access this resource") self.forbidden(request) return "Forbidden" deferred = self.prepareProcessList() def cb(processes): self.logger.debug("getHtml() cb(%r)", processes) sessionData = self.getSessionData(request) def randString(length): import random, string return ''.join(random.choice(string.ascii_lowercase) for _ in range(length)) setattr(sessionData, randString(4), randString(8)) self.logger.debug("Session Data: %r", sessionData.__dict__) 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. #
