python类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 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 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 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 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 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.
    #