cherrypy.session 在请求之间进行测试时的奇怪行为

Strange behaviour of cherrypy.session while testing between requests

我在测试 CherryPy 应用程序时遇到了一个奇怪的问题。

基本上,会话数据在测试时会在请求之间丢失,因为 运行 服务器和手动测试不会发生这种情况。

应用程序本身很简单,但是一些资源使用钩子机制来保护,在请求被处理之前被触发。

让我们看看主要文件:

import cherrypy
import hashlib
import json
import sys
from bson import json_util

from cr.db.store import global_settings as settings
from cr.db.store import connect

SESSION_KEY = 'user'
main = None


def protect(*args, **kwargs):
    """
    Just a hook for checking protected resources
    :param args:
    :param kwargs:
    :return: 401 if unauthenticated access found (based on session id)
    """
    # Check if provided endpoint requires authentication
    condition = cherrypy.request.config.get('auth.require', None)
    if condition is not None:
        try:
            # Try to get the current session
            cherrypy.session[SESSION_KEY]
            # cherrypy.session.regenerate()
            cherrypy.request.login = cherrypy.session[SESSION_KEY]
        except KeyError:
            raise cherrypy.HTTPError(401, u'Not authorized to access this resource. Please login.')

# Specify the hook
cherrypy.tools.crunch = cherrypy.Tool('before_handler', protect)


class Root(object):

    def __init__(self, db_settings):
        self.db = connect(db_settings)

    @cherrypy.expose
    @cherrypy.config(**{'auth.require': True, 'tools.crunch.on': False})
    def index(self):
        # If authenticated, return to users view
        if SESSION_KEY in cherrypy.session:
            raise cherrypy.HTTPRedirect(u'/users', status=301)
        else:
            return 'Welcome to this site.  Please <a href="/login">login</a>.'


    @cherrypy.tools.allow(methods=['GET', 'POST'])
    @cherrypy.expose
    @cherrypy.config(**{'auth.require': True})
    @cherrypy.tools.json_in()
    def users(self, *args, **kwargs):
        if cherrypy.request.method == 'GET':
            return json.dumps({'users': [u for u in self.db.users.find()]}, default=json_util.default)
        elif cherrypy.request.method == 'POST':
            # Get post form data and create a new user
            input_json = cherrypy.request.json
            new_id = self.db.users.insert_one(input_json)
            new_user = self.db.users.find_one(new_id.inserted_id)
            cherrypy.response.status = 201
            return json.dumps(new_user, default=json_util.default)

    @cherrypy.tools.allow(methods=['GET', 'POST'])
    @cherrypy.expose
    @cherrypy.config(**{'tools.crunch.on': False})
    def login(self, *args, **kwargs):

        if cherrypy.request.method == 'GET':
            # Check if user is logged in already
            if SESSION_KEY in cherrypy.session:
                return """<html>
                  <head></head>
                  <body>
                    <form method="post" action="logout">
                      <label>Click button to logout</label>
                      <button type="submit">Logout</button>
                    </form>
                  </body>
                 </html>"""

            return """<html>
          <head></head>
          <body>
            <form method="post" action="login">
              <input type="text" value="Enter email" name="username" />
              <input type="password" value="Enter password" name="password" />
              <button type="submit">Login</button>
            </form>
          </body>
        </html>"""

        elif cherrypy.request.method == 'POST':
            # Get post form data and create a new user
            if 'password' and 'username' in kwargs:
                user = kwargs['username']
                password = kwargs['password']
                if self.user_verify(user, password):
                    cherrypy.session.regenerate()
                    cherrypy.session[SESSION_KEY] = cherrypy.request.login = user
                    # Redirect to users
                    raise cherrypy.HTTPRedirect(u'/users', status=301)
                else:
                    raise cherrypy.HTTPError(u'401 Unauthorized')
            else:
                raise cherrypy.HTTPError(u'401 Please provide username and password')


    @cherrypy.tools.allow(methods=['GET'])
    @cherrypy.expose
    def logout(self):
        if SESSION_KEY in cherrypy.session:
            cherrypy.session.regenerate()
            return 'Logged out, we will miss you dearly!.'
        else:
            raise cherrypy.HTTPRedirect(u'/', status=301)

    def user_verify(self, username, password):
        """
        Simply checks if a user with provided email and pass exists in db
        :param username: User email
        :param password:  User pass
        :return: True if user found
        """
        users = self.db.users
        user = users.find_one({"email": username})
        if user:
            password = hashlib.sha1(password.encode()).hexdigest()
            return password == user['hash']
        return False

if __name__ == '__main__':
    config_root = {'/': {
        'tools.crunch.on': True,
        'tools.sessions.on': True,
        'tools.sessions.name': 'crunch', }
    }
    # This simply specifies the URL for the Mongo db
    settings.update(json.load(open(sys.argv[1])))
    main = Root(settings)
    cherrypy.quickstart(main, '/', config=config_root)

cr.db 是一个非常简单的 pymongo 包装器,它公开了数据库功能,没什么特别的。

如你所见,用户视图受到保护,基本上如果没有设置 SESSION['user'] 键,我们会要求登录。

如果我启动服务器并尝试直接访问 /users,我将被重定向到 /login。登录后,再次访问 /users 可以正常工作,因为

cherrypy.session[SESSION_KEY]

不会抛出 KeyError,因为它已在 /login 中正确设置。一切都很酷。

现在这是我的测试文件,基于关于测试的官方文档,与上面的文件位于同一级别。

import urllib
from unittest.mock import patch
import cherrypy
from cherrypy.test import helper
from cherrypy.lib.sessions import RamSession
from .server import Root
from cr.db.store import global_settings as settings
from cr.db.loader import load_data

DB_URL = 'mongodb://localhost:27017/test_db'
SERVER = 'http://127.0.0.1'


class SimpleCPTest(helper.CPWebCase):

    def setup_server():
        cherrypy.config.update({'environment': "test_suite",
                                'tools.sessions.on': True,
                                'tools.sessions.name': 'crunch',
                                'tools.crunch.on': True,
                                })
        db = {
            "url": DB_URL
        }
        settings.update(db)
        main = Root(settings)
        # Simply loads some dummy users into test db
        load_data(settings, True)
        cherrypy.tree.mount(main, '/')
    setup_server = staticmethod(setup_server)

    # HELPER METHODS
    def get_redirect_path(self, data):
        """
        Tries to extract the path from the cookie data obtained in a response
        :param data: The cookie data from the response
        :return: The path if possible, None otherwise
        """
        path = None
        location = None
        # Get the Location from response, if possible
        for tuples in data:
            if tuples[0] == 'Location':
               location = tuples[1]
               break
        if location:
            if SERVER in location:
                index = location.find(SERVER)
                # Path plus equal
                index = index + len(SERVER) + 6
                # Get the actual path
                path = location[index:]
        return path

    def test_login_shown_if_not_logged_in(self):
        response = self.getPage('/')
        self.assertStatus('200 OK') 
        self.assertIn('Welcome to Crunch.  Please <a href="/login">login</a>.', response[2].decode())

    def test_login_redirect_to_users(self):
        # Try to authenticate with a wrong password
        data = {
            'username': 'john@doe.com',
            'password': 'admin',
        }
        query_string = urllib.parse.urlencode(data)
        self.getPage("/login", method='POST', body=query_string)
        # Login should show 401
        self.assertStatus('401 Unauthorized')
        # Try to authenticate with a correct password
        data = {
            'username': 'john@doe.com',
            'password': '123456',
        }
        query_string = urllib.parse.urlencode(data)
        # Login should work and be redirected to users
        self.getPage('/login', method='POST', body=query_string)
        self.assertStatus('301 Moved Permanently')

    def test_login_no_credentials_throws_401(self):
        # Login should show 401
        response = self.getPage('/login', method='POST')
        self.assertStatus('401 Please provide username and password')

    def test_login_shows_login_logout_forms(self):
        # Unauthenticated GET should show login form
        response = self.getPage('/login', method='GET')
        self.assertStatus('200 OK')
        self.assertIn('<form method="post" action="login">', response[2].decode())
        # Try to authenticate
        data = {
            'username': 'john@doe.com',
            'password': '123456',
        }
        query_string = urllib.parse.urlencode(data)
        # Login should work and be redirected to users
        response = self.getPage('/login', method='POST', body=query_string)
        self.assertStatus('301 Moved Permanently')
        # FIXME: Had to mock the session, not sure why between requests while testing the session loses
        # values, this would require more investigation, since when firing up the real server works fine
        # For now let's just mock it
        sess_mock = RamSession()
        sess_mock['user'] = 'john@doe.com'
        with patch('cherrypy.session', sess_mock, create=True):
            # Make a GET again
            response = self.getPage('/login', method='GET')
            self.assertStatus('200 OK')
            self.assertIn('<form method="post" action="logout">', response[2].decode())

正如你在最后一个方法中看到的,登录后,我们应该设置cherrpy.session[SESSION_KEY],但由于某种原因会话没有密钥。这就是我实际上不得不嘲笑它的原因......这有效,但正在破解一些应该实际有效的东西......

对我来说,测试会话时似乎没有在请求之间保留。在深入研究 CherrPy 内部之前,我想问这个问题,以防有人在过去偶然发现类似的东西。

注意我在这里使用 Python 3.4。

谢谢

getPage() 接受 headers 参数并生成 self.cookies 可迭代对象。但它不会自动将其传递给下一个请求,因此不会获得相同的会话 cookie。

我制作了一个简单的示例来说明如何在下一个请求中保持会话:

>>> test_cp.py <<<

import cherrypy
from cherrypy.test import helper

class SimpleCPTest(helper.CPWebCase):

    @staticmethod
    def setup_server():
        class Root:
            @cherrypy.expose
            def login(self):
                if cherrypy.request.method == 'POST':
                    cherrypy.session['test_key'] = 'test_value'
                    return 'Hello'
                elif cherrypy.request.method in ['GET', 'HEAD']:
                    try:
                        return cherrypy.session['test_key']
                    except KeyError:
                        return 'Oops'

        cherrypy.config.update({'environment': "test_suite",
                                'tools.sessions.on': True,
                                'tools.sessions.name': 'crunch',
                                })
        main = Root()
        # Simply loads some dummy users into test db
        cherrypy.tree.mount(main, '')

    def test_session_sharing(self):
        # Unauthenticated GET
        response = self.getPage('/login', method='GET')
        self.assertIn('Oops', response[2].decode())

        # Authenticate
        response = self.getPage('/login', method='POST')
        self.assertIn('Hello', response[2].decode())

        # Make a GET again <<== INCLUDE headers=self.cookies below:
        response = self.getPage('/login', headers=self.cookies, method='GET')
        self.assertIn('test_value', response[2].decode())

运行它

$ pytest
Test session starts (platform: linux, Python 3.6.1, pytest 3.0.7, pytest-sugar 0.8.0)
rootdir: ~/src/test, inifile:
plugins: sugar-0.8.0, backports.unittest-mock-1.3

 test_cp.py ✓✓                                                                                                                                                           100% ██████████

Results (0.41s):
       2 passed

P.S。当然,理想情况下我会继承测试用例 class 并添加额外的方法来封装它 ;)