import os import sys import logging import unittest from pdb import set_trace from google.appengine.ext import testbed from flask import Flask, session app = Flask('fs') app.secret_key = '41' from werkzeug.datastructures import CallbackDict from flask.sessions import SessionInterface, SessionMixin from itsdangerous import URLSafeTimedSerializer, BadSignature class ItsdangerousSession(CallbackDict, SessionMixin): def __init__(self, initial=None): def on_update(self): self.modified = True CallbackDict.__init__(self, initial, on_update) self.modified = False class ItsdangerousSessionInterface(SessionInterface): salt = 'cookie-session' session_class = ItsdangerousSession def get_serializer(self, app): if not app.secret_key: return None return URLSafeTimedSerializer(app.secret_key, salt=self.salt) def open_session(self, app, request): s = self.get_serializer(app) if s is None: return None val = request.cookies.get(app.session_cookie_name) if not val: return self.session_class() max_age = app.permanent_session_lifetime.total_seconds() try: data = s.loads(val, max_age=max_age) return self.session_class(data) except BadSignature: return self.session_class() def save_session(self, app, session, response): domain = self.get_cookie_domain(app) if not session: if session.modified: response.delete_cookie(app.session_cookie_name, domain=domain) return expires = self.get_expiration_time(app, session) val = self.get_serializer(app).dumps(dict(session)) response.set_cookie(app.session_cookie_name, val, expires=expires, httponly=True, domain=domain) app.session_interface = ItsdangerousSessionInterface() @app.route('/', methods=['POST']) def m(): logging.error(session) return "hello" class TestSuite(unittest.TestCase): def setUp(self): from nassau import settings self.app = app self.client = self.app.test_client() self.testbed = testbed.Testbed() self.testbed.activate() tb = self.testbed tb.init_datastore_v3_stub() tb.init_memcache_stub() tb.init_taskqueue_stub() tb.init_urlfetch_stub() tb.init_user_stub() tb.init_blobstore_stub() tb.setup_env(overwrite=True) setup_env_defaults = dict( USER_EMAIL = '%s@unit-test.example.com' % os.uname()[1], USER_ID = "4", AUTH_DOMAIN = 'testbed.example.com', USER_IS_ADMIN = '1', APPLICATION_ID = 'example.testbed.test', ) environment = dict(setup_env_defaults.items()) tb.setup_env( overwrite = True, **environment ) def test_session(self): with self.app.test_request_context() as trc: session['a'] = 1 with self.app.test_client() as c: session['b'] = 1 with c.session_transaction() as sess: session['c'] = 1 sess['d'] = 1 #c.get() print "X %s" % session c.post() print "Y %s" % session if __name__ == "__main__": unittest.main()