app.yaml

application: yourappid
version: testgaesessions
runtime: python
api_version: 1

handlers:
- url: /stats.*
  script: $PYTHON_LIB/google/appengine/ext/appstats/ui.py
- url: /.*
  script: main.py

main.py

import os
from google.appengine.api import memcache
from google.appengine.ext import db, webapp
from google.appengine.ext.appstats import recording
from google.appengine.ext.webapp.util import run_wsgi_app

COOKIE_KEY = os.urandom(64)
TEST_GAESESSIONS_HYBRID = False
TEST_GAESESSIONS_MC = False
TEST_GAESESSIONS_CO = True
TEST_BEAKER = False
TEST_SUAS = False
TEST_GMEMSESS = False
TEST_GAEUTILITIES_COOKIE_ONLY = False
req_handler_cls = webapp.RequestHandler
start_session = lambda s : None
TEST_GAESESSIONS = TEST_GAESESSIONS_HYBRID or TEST_GAESESSIONS_MC or TEST_GAESESSIONS_CO
if TEST_GAESESSIONS:
    from gaesessions import get_current_session, SessionMiddleware
    get_session = lambda h : get_current_session()
    save_session = lambda s : None
elif TEST_BEAKER:
    from beaker.middleware import SessionMiddleware
    get_session = lambda h : h.request.environ['beaker.session']
    save_session = lambda s : s.save()
elif TEST_SUAS:
    from suas.session import RequestHandler as SUASRequestHandler
    req_handler_cls = SUASRequestHandler
    get_session = lambda h : h.session
    save_session = lambda s : None
    start_session = lambda s : s.start(None, True)
elif TEST_GMEMSESS:
    from gmemsess import Session
    get_session = lambda h : Session(h)
    save_session = lambda s : s.save()
else:
    from appengine_utilities.sessions import Session
    if TEST_GAEUTILITIES_COOKIE_ONLY:
        get_session = lambda h : Session(
                cookie_name = 'test',
                integrate_flash = False,
                session_expire_time = 180 * 24 * 60 * 60, # 180 days
                set_cookie_expires = True,
                writer = 'cookie',
                wsgiref_headers = h.request.headers,
            )
    else:
        get_session = lambda h : Session()
    save_session = lambda s : None

class TestModel(db.Model):
    s = db.StringProperty()
    i = db.IntegerProperty()
    f = db.FloatProperty()

# note: these entities are about 900B when stored as a protobuf
def get_test_entity(i):
    """Create the entity just like it would be in the datastore (so our tests don't actually go to the datastore)."""
    return TestModel(key=db.Key.from_path('TestModel', str(i)), s="a"*500, i=i, f=i*10.0)

class EmptyPage(req_handler_cls):
    def get(self):
        self.response.headers['Content-Type'] = 'text/plain'
        self.response.out.write('this page does nothing with sessions at all - site configured to test ')
        if TEST_GAESESSIONS_HYBRID:
            self.response.out.write('gaesessions - default (hybrid)')
        if TEST_GAESESSIONS_CO:
            self.response.out.write('gaesessions - cookies only')
        elif TEST_GAESESSIONS_MC:
            self.response.out.write('gaesessions memcache only')
        elif TEST_BEAKER:
            self.response.out.write('beaker.session')
        elif TEST_GMEMSESS:
            self.response.out.write('gmemsess')
        elif TEST_SUAS:
            self.response.out.write('suas.session')
        else:
            self.response.out.write('gaeutilities.sessions')

class NoOpSession(req_handler_cls):
    def get(self):
        session = get_session(self)
        start_session(session)
        self.response.headers['Content-Type'] = 'text/plain'
        self.response.out.write('this page retrieves the session object but does nothing to it')

class ClearSession(req_handler_cls):
    def get(self):
        session = get_session(self)
        session.clear()
        memcache.flush_all()
        save_session(session)
        self.response.headers['Content-Type'] = 'text/plain'
        self.response.out.write('this page clears all data from the session object and flushes memcache')

class ReadInts(req_handler_cls):
    def get(self, n):
        session = get_session(self)
        for i in xrange(int(n)):
            x = session['i%d' % i]
        self.response.headers['Content-Type'] = 'text/plain'
        self.response.out.write('this page retrieved %s ints from the session' % n)

class WriteInts(req_handler_cls):
    def get(self, n):
        session = get_session(self)
        start_session(session)
        for i in xrange(int(n)):
            session['i%d' % i] = i
        save_session(session)
        self.response.headers['Content-Type'] = 'text/plain'
        self.response.out.write('this page wrote %s ints to the session' % n)

class ReadModels(req_handler_cls):
    def get(self, n):
        session = get_session(self)
        for i in xrange(int(n)):
            x = session['m%d' % i]
            #self.response.out.write(str(x))
            #self.response.out.write(str(type(x)))
        self.response.headers['Content-Type'] = 'text/plain'
        self.response.out.write('this page retrieved %s models from the session' % n)

class WriteModels(req_handler_cls):
    def get(self, n):
        session = get_session(self)
        start_session(session)
        for i in xrange(int(n)):
            session['m%d' % i] = get_test_entity(i)
        save_session(session)
        self.response.headers['Content-Type'] = 'text/plain'
        self.response.out.write('this page wrote %s models to the session' % n)

class WriteBoth(req_handler_cls):
    def get(self, ni, nm):
        session = get_session(self)
        start_session(session)
        for i in xrange(int(ni)):
            session['i%d' % i] = i
        for i in xrange(int(nm)):
            session['m%d' % i] = get_test_entity(i)
        save_session(session)
        self.response.headers['Content-Type'] = 'text/plain'
        self.response.out.write('this page wrote %s ints and %s models to the session' % (ni, nm))

class ErrorPage(webapp.RequestHandler):
    def get(self):
        self.response.headers['Content-Type'] = 'text/plain'
        self.response.out.write('404: page does not exist!')

app = webapp.WSGIApplication([('/',                        EmptyPage),
                              ('/no-op',                   NoOpSession),
                              ('/clear',                   ClearSession),
                              ('/int/read/(\d+)',          ReadInts),
                              ('/int/write/(\d+)',         WriteInts),
                              ('/model/read/(\d+)',        ReadModels),
                              ('/model/write/(\d+)',       WriteModels),
                              ('/both/write/(\d+)/(\d+)',  WriteBoth),
                              ('/.*',                      ErrorPage)
                             ], debug=True)
if TEST_GAESESSIONS_HYBRID:
    app = SessionMiddleware(app, COOKIE_KEY)
elif TEST_GAESESSIONS_MC:
    app = SessionMiddleware(app, COOKIE_KEY, no_datastore=True, cookie_only_threshold=0)
elif TEST_GAESESSIONS_CO:
    app = SessionMiddleware(app, COOKIE_KEY, cookie_only_threshold=14*1024)
elif TEST_BEAKER:
    session_opts = { 'session.type': 'ext:google',  'session.auto': False }
    app = SessionMiddleware(app, session_opts)

app = recording.appstats_wsgi_middleware(app)
def main(): run_wsgi_app(app)
if __name__ == '__main__': main()

tester.py

#!/usr/bin/env python

import cookielib
import logging
from optparse import OptionParser
import sys
import urllib2

def fetch(OPENER, url):
    try:
        r = OPENER.open(url)
        ret = r.read()
        logging.info(ret)
        if 'Traceback' in ret or '404' in ret:
            logging.error('got error page for ' + url)
            sys.exit(-1)
        r.close()
    except Exception, e:
        logging.error("unable to fetch %s: %s" % (url, e))
        #sys.exit(-1)
        raise

def main(argv=sys.argv[1:]):
    parser = OptionParser()
    parser.add_option("-b", "--beaker",
                      action="store_true", default=False,
                      help="test beaker session (default: test gaeutiltiies)")
    parser.add_option("-c", "--cookie",
                      action="store_true", default=False,
                      help="test gaeutilities with cookie-only session (default: test gaeutiltiies)")
    parser.add_option("-g", "--gmemsess",
                      action="store_true", default=False,
                      help="test gmemsess (default: test gaeutiltiies)")
    parser.add_option("-C", "--gaesessionsco",
                      action="store_true", default=False,
                      help="test gae-sessions with cookies only (default: test gaeutilities)")
    parser.add_option("-m", "--gaesessionsmc",
                      action="store_true", default=False,
                      help="test gae-sessions with memcache only (default: test gaeutilities)")
    parser.add_option("-s", "--gaesessions",
                      action="store_true", default=False,
                      help="test gae-sessions (default: test gaeutilities)")
    parser.add_option("-S", "--suas",
                      action="store_true", default=False,
                      help="test suas secure cookie session (default: test gaeutiltiies)")
    parser.add_option("-l", "--local",
                      action="store_true", default=False,
                      help="test on the local dev server [default: test on the production server)")

    (options, args) = parser.parse_args(argv)
    if len(args) > 0:
        parser.error("too many arguments")

    if options.gaesessions:
        VERSION = "testgaesessions"
    elif options.gaesessionsmc:
        VERSION = "testgaesessionsmc"
    elif options.gaesessionsco:
        VERSION = "testgaesessionsco"
    elif options.beaker:
        VERSION = 'testbeaker'
    elif options.suas:
        VERSION = 'testsuas'
    elif options.gmemsess:
        VERSION = 'testgmemsess'
    elif options.cookie:
        VERSION = 'testgaeutilcookies'
    else:
        VERSION = "testgaeutilities"

    logging.basicConfig(level=logging.DEBUG,
                        format='%(message)s',
                        filename='./.results/%s-tester.log' % VERSION[4:],
                        filemode='w')
    console = logging.StreamHandler()
    console.setLevel(logging.DEBUG)
    formatter = logging.Formatter('%(message)s')
    console.setFormatter(formatter)
    logging.getLogger('').addHandler(console)

    if options.local:
        logging.debug('assuming you have ' + VERSION + ' running on the local dev server ...')
        PREFIX = 'http://localhost:8080/'
    else:
        PREFIX = 'http://' + VERSION + '.latest.dound.appspot.com/'

    START_FLAG = PREFIX + '?start=' + VERSION
    EMPTY_PAGE = PREFIX
    NOOP_SESS  = PREFIX + 'no-op'
    CLEAR_ALL  = PREFIX + 'clear'
    RD_INTS    = PREFIX + 'int/read/%d'
    WR_INTS    = PREFIX + 'int/write/%d'
    RD_MODELS  = PREFIX + 'model/read/%d'
    WR_MODELS  = PREFIX + 'model/write/%d'
    WR_BOTH    = PREFIX + 'both/write/%d/%d'

    def new_session():
        return urllib2.build_opener(urllib2.HTTPCookieProcessor(cookielib.FileCookieJar("cookies")))

    def fetch_n(OPENER, url, n, txt):
        for i in xrange(n):
            logging.info('running: ' + txt + ' (test %d of %d)' % (i+1, n))
            fetch(OPENER, url)
        raw_input('done with test; press <ENTER> to continue (you might want to look at AppStats now) ...')
        fetch(OPENER, START_FLAG + '&restart')  # just in case the app went out of memory while we looked at AppStats

    logging.debug('making sure the app is loaded and cleanup anything from the past')
    s = new_session()
    fetch(s, CLEAR_ALL)

    # mark that we started (make it easier to find the start point in appstats)
    fetch(s, START_FLAG)
    fetch_n(s, EMPTY_PAGE, 10, 'test empty page')
    fetch_n(s, NOOP_SESS, 10, 'test no-op empty session')

    for sz in [1, 10, 100]:
        fetch_n(s, WR_INTS % sz, 10, 'int tests - writing %d ints' % sz)
        fetch_n(s, RD_INTS % sz, 10, 'int tests - reading %d ints' % sz)
        fetch(s, CLEAR_ALL)

        fetch_n(s, WR_MODELS % sz, 10, 'model tests - writing %d entities' % sz)
        fetch_n(s, RD_MODELS % sz, 10, 'model tests - reading %d entities' % sz)
        fetch(s, CLEAR_ALL)

    # note that session_data_sz does NOT include any overhead and is approximate
    ENTITY_SIZE = 900  # approx protobuf size in bytes
    def test_small_reads_with_large_session(session_data_sz, num_ints):
        num_entities = (session_data_sz - num_ints*4) / ENTITY_SIZE
        num_ints = max(0, num_ints)
        num_entities = max(0, num_entities)
        test_info = 'test small reads with %.1fKB of session data (%d ints and %d models)' % (session_data_sz/1024.0, num_ints, num_entities)
        try:
            for i in xrange(10):
                print test_info + ' (%d of 10)' % (i+1)
                s = new_session()
                fetch(s, WR_BOTH % (num_ints, num_entities))
                fetch(s, NOOP_SESS)
                fetch(s, RD_INTS % 1)
        except Exception,e :
            print 'died: ', e
        raw_input('done with test; press <ENTER> to continue (you might want to look at AppStats now) ...')
        fetch(s, CLEAR_ALL)

    for num_ints in (1, 10, 100, 10000):
        for sz_KB in (10, 100, 500):
            test_small_reads_with_large_session(sz_KB * 1024, num_ints)

    logging.debug('all done!')

if __name__ == '__main__': main()

I've updated the code to also support testing beaker. The results link in the above comment now includes the results of testing beaker 1.5.3 too.

Updated the code to benchmark gmemsess and suas too.