一个使用Flask-Login登录后的Pytest测试用例的坑

Posted 2022-02-28 Python之美

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了一个使用Flask-Login登录后的Pytest测试用例的坑相关的知识，希望对你有一定的参考价值。

前言

最近写一个项目，用到了Flask-Login实现用户登录状态。项目最后需要编写测试用例，但是测试代码卡在确认用户登录状态的部分就写不下去了，研究了好久才找到原因，分享一下pytest测试Flask应用上的一个坑儿.

一个最基础的例子

为了演示这个问题，我写了一个小型应用，全部代码可以从dongweiming/mp获取。

这里只列出部分核心代码。首先是测试部分代码:

app dbtest_user_info   response client   responsetest_login   db   db   response clientjson          responsetest_after_login   response clientjson          response client   responseos
pytest
app app _appdb
app   _app   _app       db   _app
   osclient   appfrom flaskclient response client      json  response clientresponseclient      apppytest items
app   _app   _app       db   _app
   osapp      _app   _app       db   _app
   ospytest items
test_after_login   db   db   response clientjson          response client   response1

也就是说，默认情况下(scope=\'function\')，数据在一个测试用例中才共用，所以得扩大fixture的范围。

使用Flask-Login注册登录

10330

使用`Flask-Login`注册登录

类似的登录页面,在前面也写过不少,也需要考虑很多的问题,比如session,账户的验证等等.

这是每一个项目,只要涉及到账户登录都会遇到的问题,因此有人为此做了一个轮子,就是Flask-Login.

1.安装

https://flask-login.readthedocs.io/en/latest/#flask-login
$ pip install flask-login

2.使用

1.初始化`Flask-Login`

和之前学习到的插件一样,都需要绑定到app上,才能真正的使用
from flask import Flask
from flask_login import LoginManager

app = Flask(__name__)
login_manager = LoginManager()
login_manager.init_app(app)
需要注意的是,默认情况下Flask-Login使用的是flask自带的session进行身份验证的,这就需要在config中设置SECRET_KEY.
# config.py
import os

DEBUG = True
TEMPLATES_AUTO_RELOAD = True
SECRET_KEY = os.urandom(24)
这样就建立其Flask app和Flask-Login插件之间的联系.然后让我们安装登录的常规思路一步一步如何实现.

Flask-Login确定登录的 URL

账户登录逻辑

验证账户密码

确认登录的账户是谁

登录可见

2.确认登录`URL`

为Flask-Login指定登录的URL.
from flask import Flask
from flask_login import LoginManager
import config

app = Flask(__name__)
app.config.from_object(config)
login_manager = LoginManager()

login_manager.init_app(app)
# 指定登录的URL
login_manager.login_view = 'login'
Flask-Login是没有默认登录URL的,如果你使用的是蓝图(Blueprint),那么就应该指定其他的URL
login_manager.login_view = 'bpmodel.users.login'
指定好URL后,就可以为这个URL编写逻辑代码
@app.route('/login/', methods=['GET', 'POST'])
def login():
    pass

3.账户登陆逻辑

按照流程顺序来说,接下来就应该验证表单,在验证数据库,如果数据库中有数据,则对比后,选择是否添加session并运行登录.

# 类似如下
# 登录页
class LoginView(views.MethodView):
    def get(self):
        return render_template('login.html')
    def post(self):
        # 验证表单
        form = LoginForm()
        if form.validate_on_submit():
            # 表单验证成功,就跳转到数据库验证
            print('表单验证成功')
            email = form.email.data 
            password = form.password.data
            lifetime = form.lifetime.data
            # 数据库判断
            print(email, password, lifetime)
            exist_user = db.session.query(BankUsers).filter(BankUsers.email==email).filter(BankUsers.password==password).first()
            print(exist_user)
            if exist_user != None:
                if lifetime:
                    session.clear()
                    session['username'] = exist_user.username
                    session.permanent = True
                    return redirect(url_for('personal'))
                else:
                    session.clear()
                    session['username'] = exist_user.username
                    return redirect(url_for('personal'))
            else:
                return render_template('login.html', info=form.errors)
        else:
            return render_template('login.html', info="用户名密码不正确,请注册")

同样的,在Flask-Login中遵循同样的流程.

1.编写用户类(`ORM`)

要想使用Flask-Login扩展,程序的User模型(ORM)必须实现几个方法

方法说明

is_authenticated 如果用户已经登录,返回True,否则返回False

is_active 如果允许用户登录,返回True,否则返回Falser,如果禁用用户,返回False

is_anonymous 匿名用户(未登录用户)返回False

get_id() 必须返回用户的唯一标识符作为ID,该ID必须为Unicode编码

这4个方法可以在模型类中作为方法直接实现,不过每次为User模型编写,会比较麻烦,Flask-Login提供了一个UserMixin类,其中包含这些方法的默认实现,并且可以实现大多数的需求,所以可以直接继承.
# sqlModel.py
# 类似如下

from flask_login import UserMixin
from exts import db 

class User(db.Model, UserMixin):
     __tablename__ = 'users'
        pass
Flask_login还要求程序必须实现一个回调函数,这个回调函数用于通过session中存储的用户ID重新加载用户对象.它应该接受用户的Unicode ID,并返回相应的用户对象.

使用指定的标识符加载用户:
# sqlModel.py
from app import login_manager 

@login_manger.user_loader
def load_user(user_id):
    if query_user(username) is not None:
        curr_user = User()
        curr_user.id = username
     return curr_user
如果ID无效,函数应该返回None(不引发异常),ID将从session中手动删除并且程序可以继续执行.

为什么需要这个回调函数?

is_authenticated就是用来判断用户是否登录是否有权来操作的,但是并不能知道当前登录的账户具体是谁,所以这里就需要一个回调函数用来判断是谁.

为了简便,这里不使用MySQL,简单的设计一个ORM
# 创建ORM映射
# 用户记录表
users = [
    {'username': 'Tom', 'password': '111111'},
    {'username': 'Michael', 'password': '123456'}
]


# 通过用户名,获取用户记录,如果不存在,返回None
def query_user(username, password):
    for user in users:
        if user['username'] == username:
            return user
        else:
            return None

class User(UserMixin):
    pass

# 如果用户名存在,就构造一个用户类对象,并使用用户名作为ID,如果不存在就返回None
# 回调函数
@login_manager.user_loader
def load_user(username):
    if query_user(username) is not None:
        curr_user = User()
        curr_user.id = username
        return curr_user
    else:
        return None
注意实现回调函数.

方法	说明
`is_authenticated`	如果用户已经登录,返回`True`,否则返回`False`
`is_active`	如果允许用户登录,返回`True`,否则返回`Falser`,如果禁用用户,返回`False`
`is_anonymous`	匿名用户(未登录用户)返回`False`
`get_id()`	必须返回用户的唯一标识符作为`ID`,该`ID`必须为`Unicode`编码

4.登录实现

实现如下

from flask import Flask, request, abort, redirect, url_for, render_template
from flask_login import LoginManager, login_user, UserMixin, login_required, logout_user
import config
from urllib.parse import urlparse, urljoin

app = Flask(__name__)
app.config.from_object(config)
login_manager = LoginManager()

login_manager.init_app(app)
# 指定登录的URL
login_manager.login_view = 'login'

# 创建ORM映射
# 用户记录表
users = [
    {'username': 'Tom', 'password': '111111'},
    {'username': 'Michael', 'password': '123456'}
]


# 通过用户名,获取用户记录,如果不存在,返回None
def query_user(username):
    for user in users:
        if user['username'] == username:
            return user
        else:
            return None

class User(UserMixin):
    pass

# 如果用户名存在,就构造一个用户类对象,并使用用户名作为ID,如果不存在就返回None
# 回调函数
@login_manager.user_loader
def load_user(username):
    if query_user(username) is not None:
        curr_user = User()
        curr_user.id = username
        return curr_user
    else:
        return None

def is_safe_url(target):
    ref_url = urlparse(request.host_url)
    test_url = urlparse(urljoin(request.host_url, target))
    return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc

@app.route('/login/', methods=['GET', 'POST'])
def login():
    # 假设通过表单验证
    # 假设通过数据库验证
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')
        # 验证表单,数据库
        user = query_user(username)
        if user and password == user['password']:
            # curr_user 是 User类的一个实例
            curr_user = User()
            curr_user.id = username
            # 通过 Flask-login的login_user来登录账户
            login_user(curr_user)
            nextD = request.args.get('next')
            print(nextD)
            # is_safe_url 用来检查url是否可以安全的重定向
            # 避免重定向攻击
            # if not is_safe_url(nextD):
            #     return abort(404)
            # return redirect(next or url_for('index'))
            return redirect(url_for('index'))
    else:
        return  render_template('login.html')

@app.route('/')
def index():
    return 'Hello World!'

# 登录可见的页面
@app.route('/personal/', methods=['GET', 'POST'])
@login_required
def personal():
    pass

# 登出
@app.route('/logout/')
@login_required
def logout():
    logout_user()
    return 'logout'

if __name__ == '__main__':
    app.run()

这里使用了一个方法login_user,通过上面的对比,我们也知道在这一步,应该实现的是session的相关信息,为什么只使用这一个方法就可以实现,我们可以查看一下源码.

from flask import session

def login_user(user, remember=False, duration=None, force=False, fresh=True):
    if not force and not user.is_active:
        return False

    user_id = getattr(user, current_app.login_manager.id_attribute)()
    session['_user_id'] = user_id
    session['_fresh'] = fresh
    session['_id'] = current_app.login_manager._session_identifier_generator()

    if remember:
        session['_remember'] = 'set'
        if duration is not None:
            try:
                # equal to timedelta.total_seconds() but works with Python 2.6
                session['_remember_seconds'] = (duration.microseconds +
                                                (duration.seconds +
                                                 duration.days * 24 * 3600) *
                                                10**6) / 10.0**6
            except AttributeError:
                raise Exception('duration must be a datetime.timedelta, '
                                'instead got: {0}'.format(duration))

    current_app.login_manager._update_request_context_with_user(user)
    user_logged_in.send(current_app._get_current_object(), user=_get_user())
    return True

以上包含的参数要内容是将用户信息放入

参数	说明
`user`	要登录的用户对象
`remember(bool)`	`session`过期后时候记住用户
`duration`	记住用户的过期时长
`force(bool)`	如果用户处于不活跃状态,设置这个参数(`True`)将强制登录用户
`fresh(bool)`	登录用户时,将用户`session`标记为`fresh`

可以看到login_user主要就是将用户信息加入到session中.确定是哪个user则需要使用回调函数,并且必须返回user本身

        def user_loader(self, callback):
        '''
        This sets the callback for reloading a user from the session. The
        function you set should take a user ID (a ``unicode``) and return a
        user object, or ``None`` if the user does not exist.

        :param callback: The callback for retrieving a user object.
        :type callback: callable
        '''
        self._user_callback = callback
        return callback

    
    
    def _load_user(self):
        '''Loads user from session or remember_me cookie as applicable'''

        if self._user_callback is None and self._request_callback is None:
            raise Exception(
                "Missing user_loader or request_loader. Refer to "
                "http://flask-login.readthedocs.io/#how-it-works "
                "for more info.")

        user_accessed.send(current_app._get_current_object())

        # Check SESSION_PROTECTION
        if self._session_protection_failed():
            return self._update_request_context_with_user()

        user = None

        # Load user from Flask Session
        user_id = session.get('_user_id')
        if user_id is not None and self._user_callback is not None:
            user = self._user_callback(user_id)