def _get_session(self):
session_cookie = self._driver.get_cookie(settings.SESSION_COOKIE_NAME)
if session_cookie is None:
# Create new
session = get_session_store()
cookie_data = {'name': settings.SESSION_COOKIE_NAME,
'value': session.session_key,
'path': '/',
'secure': False,
}
if self._driver.name == 'phantomjs':
# Not sure why this is needed, but it seems to do the trick
cookie_data['domain'] = '.localhost'
self._add_cookie(cookie_data)
else:
session = get_session_store(session_key=session_cookie['value'])
return session
python类SESSION_COOKIE_NAME的实例源码
def test_non_js_post_workflow(self):
""" Post to a view ensuring that the response is like the normal 'get' response, but with
the thumber success message, and that a populated Feedback model is created.
"""
view_name = 'thumber_tests:example_form'
path = reverse(view_name)
http_referer = 'http://example.com{0}'.format(path)
# Get the form view, and 'follow' so that the session cookie gets set on the client
response = self.client.get(path, follow=True)
self.assertIn(settings.SESSION_COOKIE_NAME, response.cookies)
# Post the thumber form, and get the same page but with teh success message
data = {'satisfied': 'True', 'comment': 'test comment', 'thumber_token': 'sync'}
response = self.client.post(path, data, HTTP_REFERER=http_referer)
self.assertContains(response, 'Thank you for your feedback', status_code=200)
# Check that a Feedback model was created with the correct details
self.assertEquals(Feedback.objects.count(), 1)
feedback = Feedback.objects.all()[0]
self.assertEquals(feedback.view_name, view_name)
self.assertEquals(feedback.url, http_referer)
self.assertEquals(feedback.satisfied, True)
self.assertEquals(feedback.comment, 'test comment')
def test_per_user(self):
response = self.client.get(per_user)
self.assertEqual(
response._headers["cache-control"], ("Cache-Control", "max-age=100, s-maxage=600")
)
self.assertEqual(
response._headers["x-accel-expires"], ("X-Accel-Expires", "600")
)
self.assertEqual(
response._headers["x-hash-cookies"],
("X-Hash-Cookies", "messages|%s" % settings.SESSION_COOKIE_NAME)
)
self.login()
response = self.client.get(per_user)
self.assertEqual(
response._headers["cache-control"], ("Cache-Control", "max-age=100, s-maxage=600")
)
self.assertEqual(
response._headers["x-accel-expires"], ("X-Accel-Expires", "600")
)
self.assertEqual(
response._headers["x-hash-cookies"],
("X-Hash-Cookies", "messages|%s" % settings.SESSION_COOKIE_NAME)
)
def test_session_language(self):
page = api.create_page("home", "nav_playground.html", "en", published=True)
api.create_title('fr', "home", page)
page.publish('fr')
page.publish('en')
response = self.client.get('/')
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/en/')
engine = import_module(settings.SESSION_ENGINE)
store = engine.SessionStore()
store.save() # we need to make load() work, or the cookie is worthless
self.client.cookies[settings.SESSION_COOKIE_NAME] = store.session_key
# ugly and long set of session
session = self.client.session
session[LANGUAGE_SESSION_KEY] = 'fr'
session.save()
response = self.client.get('/')
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/fr/')
self.client.get('/en/')
self.assertEqual(self.client.session[LANGUAGE_SESSION_KEY], 'en')
response = self.client.get('/')
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/en/')
def _fastlogin(self, **credentials):
session = import_module(settings.SESSION_ENGINE).SessionStore()
session.save()
request = AttributeObject(session=session, META={})
user = authenticate(**credentials)
login(request, user)
session.save()
# We need to "warm up" the webdriver as we can only set cookies on the
# current domain
self.driver.get(self.live_server_url)
# While we don't care about the page fully loading, Django will freak
# out if we 'abort' this request, so we wait patiently for it to finish
self.wait_page_loaded()
self.driver.add_cookie({
'name': settings.SESSION_COOKIE_NAME,
'value': session.session_key,
'path': '/',
'domain': urlparse(self.live_server_url).hostname
})
self.driver.get('{0}/?{1}'.format(
self.live_server_url,
get_cms_setting('CMS_TOOLBAR_URL__EDIT_ON')
))
self.wait_page_loaded()
def create_session_cookie(username):
"""Creates an active session for username and returns the resulting
session cookie.
This is useful to fake login sessions during testing.
"""
user = Account.objects.get(login=username)
session = SessionStore()
session[ACCOUNT_ID_VAR] = user.id
session.save()
cookie = {
'name': settings.SESSION_COOKIE_NAME,
'value': session.session_key,
'secure': False,
'path': '/',
}
return cookie
def open(self, user_id):
self.client = tornadoredis.Client()
self.client.connect()
session_key = self.get_cookie(settings.SESSION_COOKIE_NAME)
session = session_engine.SessionStore(session_key)
try:
self.user_id = session["_auth_user_id"]
if user_id != self.user_id:
raise User.DoesNotExist
self.username = User.objects.get(id=self.user_id).username
except (KeyError, User.DoesNotExist):
self.close()
return
thread_list = Thread.objects.filter(
participants__id=self.user_id
).all()
self.channel_list = ['user_{}'.format(self.user_id)]
for thread in thread_list:
self.channel_list.append("thread_{}_messages".format(thread.id))
yield tornado.gen.Task(self.client.subscribe, self.channel_list)
self.client.listen(self.resend_response)
def _get_session(self):
session_key = self.app.cookies.get(settings.SESSION_COOKIE_NAME, None)
if session_key is None:
# Create new
session = get_session_store()
self._set_cookie(settings.SESSION_COOKIE_NAME,
session.session_key)
else:
session_key = session_key.strip('"')
session = get_session_store(session_key=session_key)
return session
def setSessionValues(self, **kwargs):
settings.SESSION_ENGINE = 'django.contrib.sessions.backends.file'
engine = import_module(settings.SESSION_ENGINE)
store = engine.SessionStore()
for key in kwargs:
store[key] = kwargs[key]
self.request.session[key] = kwargs[key]
store.save()
self.session = store
self.client.cookies[settings.SESSION_COOKIE_NAME] = store.session_key
def __init__(self, session_key=None):
self.storage_path = type(self)._get_storage_path()
self.file_prefix = settings.SESSION_COOKIE_NAME
super(SessionStore, self).__init__(session_key)
def clear_expired(cls):
storage_path = cls._get_storage_path()
file_prefix = settings.SESSION_COOKIE_NAME
for session_file in os.listdir(storage_path):
if not session_file.startswith(file_prefix):
continue
session_key = session_file[len(file_prefix):]
session = cls(session_key)
# When an expired session is loaded, its file is removed, and a
# new file is immediately created. Prevent this by disabling
# the create() method.
session.create = lambda: None
session.load()
def process_request(self, request):
session_key = request.COOKIES.get(settings.SESSION_COOKIE_NAME)
request.session = self.SessionStore(session_key)
def process_response(self, request, response):
"""
If request.session was modified, or if the configuration is to save the
session every time, save the changes and set a session cookie or delete
the session cookie if the session has been emptied.
"""
try:
accessed = request.session.accessed
modified = request.session.modified
empty = request.session.is_empty()
except AttributeError:
pass
else:
# First check if we need to delete this cookie.
# The session should be deleted only if the session is entirely empty
if settings.SESSION_COOKIE_NAME in request.COOKIES and empty:
response.delete_cookie(settings.SESSION_COOKIE_NAME,
domain=settings.SESSION_COOKIE_DOMAIN)
else:
if accessed:
patch_vary_headers(response, ('Cookie',))
if (modified or settings.SESSION_SAVE_EVERY_REQUEST) and not empty:
if request.session.get_expire_at_browser_close():
max_age = None
expires = None
else:
max_age = request.session.get_expiry_age()
expires_time = time.time() + max_age
expires = cookie_date(expires_time)
# Save the session data and refresh the client cookie.
# Skip session save for 500 responses, refs #3881.
if response.status_code != 500:
request.session.save()
response.set_cookie(settings.SESSION_COOKIE_NAME,
request.session.session_key, max_age=max_age,
expires=expires, domain=settings.SESSION_COOKIE_DOMAIN,
path=settings.SESSION_COOKIE_PATH,
secure=settings.SESSION_COOKIE_SECURE or None,
httponly=settings.SESSION_COOKIE_HTTPONLY or None)
return response
def _session(self):
"""
Obtains the current session variables.
"""
if apps.is_installed('django.contrib.sessions'):
engine = import_module(settings.SESSION_ENGINE)
cookie = self.cookies.get(settings.SESSION_COOKIE_NAME)
if cookie:
return engine.SessionStore(cookie.value)
else:
s = engine.SessionStore()
s.save()
self.cookies[settings.SESSION_COOKIE_NAME] = s.session_key
return s
return {}
def __init__(self, session_key=None):
self.storage_path = type(self)._get_storage_path()
self.file_prefix = settings.SESSION_COOKIE_NAME
super(SessionStore, self).__init__(session_key)
def clear_expired(cls):
storage_path = cls._get_storage_path()
file_prefix = settings.SESSION_COOKIE_NAME
for session_file in os.listdir(storage_path):
if not session_file.startswith(file_prefix):
continue
session_key = session_file[len(file_prefix):]
session = cls(session_key)
# When an expired session is loaded, its file is removed, and a
# new file is immediately created. Prevent this by disabling
# the create() method.
session.create = lambda: None
session.load()
def process_request(self, request):
session_key = request.COOKIES.get(settings.SESSION_COOKIE_NAME)
request.session = self.SessionStore(session_key)
def test_overriden_ajax_post(self):
view_name = 'thumber_tests:example_form'
path = reverse(view_name)
http_referer = 'http://example.com{0}'.format(path)
# Get the form view, and 'follow' so that the session cookie gets set on the client
response = self.client.get(path, follow=True)
self.assertIn(settings.SESSION_COOKIE_NAME, response.cookies)
# Post with thumber_token=ajax for a JSON response
data = {'satisfied': 'False', 'thumber_token': 'ajax'}
response = self.client.post(path, data, HTTP_REFERER=http_referer)
self.assertEquals(response['Content-Type'], 'application/json')
# Check we got a success message in our json
json = response.json()
self.assertIn('success', json)
self.assertEquals(json['success'], True)
pk = json['id']
self.assertEquals(Feedback.objects.count(), 1)
feedback = Feedback.objects.all()[0]
self.assertEquals(feedback.view_name, view_name)
self.assertEquals(feedback.url, http_referer)
self.assertEquals(feedback.satisfied, False)
# Resbumit now with the ID, and set the comment
data = {'thumber_token': 'ajax', 'id': pk, 'comment': 'test comment'}
response = self.client.post(path, data, HTTP_REFERER=http_referer)
self.assertEquals(response['Content-Type'], 'application/json')
json = response.json()
self.assertIn('success', json)
self.assertEquals(json['success'], True)
# There should still only be 1 model, and it shoudl now have the test comment
self.assertEquals(Feedback.objects.count(), 1)
feedback = Feedback.objects.all()[0]
self.assertEquals(feedback.comment, 'test comment')
def post(self, request, *args, **kwargs):
if request.POST.get('thumber_token', None) is not None:
pk = request.POST.get('id', None)
if pk is None or pk == '':
# No PK, this means we need to create a new Feedback object
http_referer = self.request.META.get('HTTP_REFERER')
sessionid = self.request.COOKIES[settings.SESSION_COOKIE_NAME]
user_feedback = ThumberForm(data=request.POST).save(commit=False)
user_feedback.url = http_referer
user_feedback.view_name = self._get_view_from_url(http_referer)
user_feedback.session = sessionid
user_feedback.view_args = (request.resolver_match.args, request.resolver_match.kwargs)
else:
# PK given, so this Feedback already exists and just needs the comment adding
user_feedback = Feedback.objects.get(pk=pk)
user_feedback.comment = request.POST['comment']
user_feedback.save()
if request.POST.get('thumber_token', None) == 'sync':
# Non-AJAX post, we've now done the processing, so return super's GET response
return self.get(request)
else:
# AJAX submission, inform frontend the frontend the POST was successful, and give the id back so it
# can be updated in a separate request
return JsonResponse({"success": True, "id": user_feedback.id})
else:
try:
return super().post(request, *args, **kwargs)
except AttributeError:
methods = [m.upper() for m in self.http_method_names if hasattr(self, m) and m.upper() != 'POST']
return HttpResponseNotAllowed(methods)
def __init__(self, session_key=None):
self.storage_path = type(self)._get_storage_path()
self.file_prefix = settings.SESSION_COOKIE_NAME
super(SessionStore, self).__init__(session_key)
def clear_expired(cls):
storage_path = cls._get_storage_path()
file_prefix = settings.SESSION_COOKIE_NAME
for session_file in os.listdir(storage_path):
if not session_file.startswith(file_prefix):
continue
session_key = session_file[len(file_prefix):]
session = cls(session_key)
# When an expired session is loaded, its file is removed, and a
# new file is immediately created. Prevent this by disabling
# the create() method.
session.create = lambda: None
session.load()
def process_request(self, request):
session_key = request.COOKIES.get(settings.SESSION_COOKIE_NAME)
request.session = self.SessionStore(session_key)
def __init__(self, session_key=None):
self.storage_path = type(self)._get_storage_path()
self.file_prefix = settings.SESSION_COOKIE_NAME
super(SessionStore, self).__init__(session_key)
def clear_expired(cls):
storage_path = cls._get_storage_path()
file_prefix = settings.SESSION_COOKIE_NAME
for session_file in os.listdir(storage_path):
if not session_file.startswith(file_prefix):
continue
session_key = session_file[len(file_prefix):]
session = cls(session_key)
# When an expired session is loaded, its file is removed, and a
# new file is immediately created. Prevent this by disabling
# the create() method.
session.create = lambda: None
session.load()
def process_request(self, request):
session_key = request.COOKIES.get(settings.SESSION_COOKIE_NAME)
request.session = self.SessionStore(session_key)
def login(self, **credentials):
"""
Sets the Factory to appear as if it has successfully logged into a site.
Returns True if login is possible; False if the provided credentials
are incorrect, or the user is inactive, or if the sessions framework is
not available.
"""
user = authenticate(**credentials)
if user and user.is_active:
# Create a fake request to store login details.
request = HttpRequest()
if self.session:
request.session = self.session
else:
request.session = SessionStore('Python/2.7', '127.0.0.1')
login(request, user)
# Save the session values.
request.session.save()
# Set the cookie to represent the session.
session_cookie = settings.SESSION_COOKIE_NAME
self.cookies[session_cookie] = request.session.session_key
cookie_data = {
'max-age': None,
'path': '/',
'domain': settings.SESSION_COOKIE_DOMAIN,
'secure': settings.SESSION_COOKIE_SECURE or None,
'expires': None,
}
self.cookies[session_cookie].update(cookie_data)
return True
else:
return False
def logout(self):
"""
Removes the authenticated user's cookies and session object.
Causes the authenticated user to be logged out.
"""
session_cookie = self.cookies.get(settings.SESSION_COOKIE_NAME)
if session_cookie:
if self.session:
self.session.delete(session_cookie)
del self.cookies[settings.SESSION_COOKIE_NAME]
def _session(self):
"""
Obtains the current session variables.
"""
if 'user_sessions' in settings.INSTALLED_APPS:
cookie = self.cookies.get(settings.SESSION_COOKIE_NAME, None)
if cookie:
return SessionStore('Python/2.7', '127.0.0.1', cookie.value)
def process_request(self, request):
engine = import_module(settings.SESSION_ENGINE)
session_key = request.COOKIES.get(settings.SESSION_COOKIE_NAME, None)
request.session = engine.SessionStore(
ip=request.META.get('REMOTE_ADDR', ''),
user_agent=request.META.get('HTTP_USER_AGENT', ''),
session_key=session_key
)
def process_response(self, request, response):
"""
If request.session was modified, or if the configuration is to save the
session every time, save the changes and set a session cookie.
"""
try:
accessed = request.session.accessed
modified = request.session.modified
except AttributeError:
pass
else:
if accessed:
patch_vary_headers(response, ('Cookie',))
if modified or settings.SESSION_SAVE_EVERY_REQUEST:
if request.session.get_expire_at_browser_close():
max_age = None
expires = None
else:
max_age = request.session.get_expiry_age()
expires_time = time.time() + max_age
expires = cookie_date(expires_time)
# Save the session data and refresh the client cookie.
# Skip session save for 500 responses, refs #3881.
if response.status_code != 500:
request.session.save()
response.set_cookie(
settings.SESSION_COOKIE_NAME,
request.session.session_key,
max_age=max_age,
expires=expires,
domain=settings.SESSION_COOKIE_DOMAIN,
path=settings.SESSION_COOKIE_PATH,
secure=settings.SESSION_COOKIE_SECURE or None,
httponly=settings.SESSION_COOKIE_HTTPONLY or None)
return response
