Werkzeug测试客户端互相干扰：未登录客户端可访问受保护路由？

我使用werkzeug.test.Client在不启动服务器的情况下测试Flask应用。
我希望验证已登录用户可以访问特定资源，而未登录用户无法访问。

• 通过test_client=Client(app)创建名为test_client的客户端，然后通过flask_security_too的/login路由完成登录。
• 通过mal_client=Client(app)创建另一个名为mal_client的客户端，不对其进行登录，但当我用它访问受@login_required保护的路由时，它却被允许访问，这不符合预期。
• 已确认mal_client的请求中除Host外没有其他请求头（未发送test_client的Cookie），因此不确定该请求为何能通过。

我制作了一个超简单的Flask应用来演示该问题。由于难以通过pdb调试@login_required装饰器，我创建了一个名为is_logged_in的路由，仅返回current_user.is_authenticated（@login_required本应引用该值）。结果显示mal_client未发送任何会话Cookie，但却处于已登录状态。

该演示应用大部分改编自Flask-Security-Too快速入门文档。
is_logged_in.py是我的is_logged_in路由，logging_hooks.py用于检查收发的请求头。除此之外，主要逻辑位于user_test_app.py，可通过python3 ./user_test_app.py运行。

user_test_app.py #

#!/usr/bin/python3

from flask_security import current_user, auth_required, hash_password
from flask import render_template_string
from database import db_session, init_db
from is_logged_in import is_logged_in
from werkzeug.test import Client
from logging_hooks import *
from app import app
import os

def run_tests(app):
    
    app.config['WTF_CSRF_ENABLED'] = False
    
    test_client=login_test_client()
    
    mal=assure_mal_not_logged_in()
    
def assure_mal_not_logged_in():
    mal_client=Client(app)
    
    response=mal_client.get('/is_logged_in')
    assert response.status_code == 200
    
    assert response.json==False
    
    return mal_client

    
def login_test_client():
    test_client=Client(app)
    
    result=test_client.post('/login', json={
        'next': '/', 
        'email': 'test@me.com',
        'password' : 'password',
        'remember': 'y', 
        'submit': 'Login',
    })
    assert result.status_code == 200
    
    response=test_client.get('/is_logged_in')
    assert response.status_code == 200
    
    assert response.json==True
    
    return test_client

if __name__ == '__main__':
    with app.app_context():
        # Create a user to test with
        init_db()
        if not app.security.datastore.find_user(email="test@me.com"):
            app.security.datastore.create_user(email="test@me.com", password=hash_password("password"))
        db_session.commit()
        
        run_tests(app)
        
        #app.run(host='0.0.0.0', port=9000)

is_logged_in.py #

from flask_security import current_user, login_required
from app import app
import flask

@app.route('/is_logged_in')
def is_logged_in():

    is_logged_in=current_user.is_authenticated
    
    return flask.jsonify(is_logged_in)

app.py #

#!/usr/bin/python3

import os

from flask import Flask
from flask_security import Security, SQLAlchemySessionUserDatastore
from database import db_session
from models import User, Role

# Create app
app = Flask(__name__)
app.config['DEBUG'] = True

# Generate a nice key using secrets.token_urlsafe()
app.config['SECRET_KEY'] = os.environ.get("SECRET_KEY", 'pf9Wkove4IKEAXvy-cQkeDPhv9Cb3Ag-wyJILbq_dFw')
# Bcrypt is set as default SECURITY_PASSWORD_HASH, which requires a salt
# Generate a good salt using: secrets.SystemRandom().getrandbits(128)
app.config['SECURITY_PASSWORD_SALT'] = os.environ.get("SECURITY_PASSWORD_SALT", '146585145368132386173505678016728509634')

# Setup Flask-Security
user_datastore = SQLAlchemySessionUserDatastore(db_session, User, Role)
app.security = Security(app, user_datastore)

database.py #

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:////tmp/test.db')
db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))
Base = declarative_base()
Base.query = db_session.query_property()

def init_db():
    # import all modules here that might define models so that
    # they will be registered properly on the metadata.  Otherwise
    # you will have to import them first before calling init_db()
    import models
    Base.metadata.create_all(bind=engine)

models.py #

from database import Base
from flask_security import UserMixin, RoleMixin
from sqlalchemy import create_engine
from sqlalchemy.orm import relationship, backref
from sqlalchemy import Boolean, DateTime, Column, Integer, \
                    String, ForeignKey, UnicodeText

class RolesUsers(Base):
    __tablename__ = 'roles_users'
    id = Column(Integer(), primary_key=True)
    user_id = Column('user_id', Integer(), ForeignKey('user.id'))
    role_id = Column('role_id', Integer(), ForeignKey('role.id'))

class Role(Base, RoleMixin):
    __tablename__ = 'role'
    id = Column(Integer(), primary_key=True)
    name = Column(String(80), unique=True)
    description = Column(String(255))
    permissions = Column(UnicodeText)

class User(Base, UserMixin):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    email = Column(String(255), unique=True)
    username = Column(String(255), unique=True, nullable=True)
    password = Column(String(255), nullable=False)
    last_login_at = Column(DateTime())
    current_login_at = Column(DateTime())
    last_login_ip = Column(String(100))
    current_login_ip = Column(String(100))
    login_count = Column(Integer)
    active = Column(Boolean())
    fs_uniquifier = Column(String(255), unique=True, nullable=False)
    confirmed_at = Column(DateTime())
    roles = relationship('Role', secondary='roles_users',
                         backref=backref('users', lazy='dynamic'))

logging_hooks.py #

from app import app
from flask import request

@app.before_request
def log_request_info():
    message=f'''
↓↓↓↓↓↓↓↓↓↓↓↓↓
↓↓ {request.method} request from {request.remote_addr} for {request.url}

Request Headers: 
{request.headers}Request Body: 
{request.get_data()}

======'''

    app.logger.debug(message)

@app.after_request
def log_response_info(response):
    message=f'''
======
Response Summary:
{str(response)}

Response Headers:
{str(response.headers)}'''
    #Only log details for responses that aren't pass-through (static files are pass-through, for example):
    try:
        if not response.direct_passthrough:
            message +=f'Response Body: \n{response.get_data().decode("utf-8")}'
    except: pass
    
    message+='\n↑↑\n↑↑↑↑↑↑↑↑↑↑↑↑↑↑'
    
    app.logger.debug(message)

    return response

1. 这是一个Bug吗？
2. 我是否使用方式有误？
3. 我遗漏了什么？
4. 如何实现验证@login_required功能是否正常的测试目标？

1. 这是Bug吗？ #

不是Bug，是测试时的上下文隔离和会话处理不当导致的问题。

2. 是否使用方式有误？ #

是的，存在两处核心错误：

• 直接使用werkzeug.test.Client而非Flask自带的app.test_client()，后者会自动处理请求上下文隔离和会话清理，避免不同客户端间的会话泄漏。
• is_logged_in路由没有添加@login_required装饰器，无法验证登录保护逻辑，直接返回的current_user.is_authenticated会受之前请求残留的会话影响。

3. 遗漏了什么？ #

• 请求上下文隔离：不同测试客户端需要独立的请求上下文，避免会话共享。
• SQLAlchemy会话清理：scoped_session在测试后未清理，导致用户数据残留到下一个请求。
• 登录保护验证逻辑：测试未验证未登录用户访问受保护路由时的拦截行为（比如重定向到登录页），仅检查current_user.is_authenticated无法体现@login_required的作用。

4. 如何实现测试目标？ #

修改以下关键部分即可完成正确的测试：

第一步：给路由添加登录保护

修改is_logged_in.py，添加@login_required装饰器：

from flask_security import current_user, login_required
from app import app
import flask

@app.route('/is_logged_in')
@login_required  # 让路由受登录保护
def is_logged_in():
    is_logged_in = current_user.is_authenticated
    return flask.jsonify(is_logged_in)

第二步：修改测试代码，使用Flask测试客户端并隔离上下文

修改user_test_app.py：

#!/usr/bin/python3

from flask_security import hash_password
from database import db_session, init_db
from app import app
import os

def run_tests(app):
    app.config['WTF_CSRF_ENABLED'] = False
    
    # 使用上下文管理器创建独立客户端，自动隔离会话
    with app.test_client() as test_client:
        login_test_client(test_client)
    
    # 用全新客户端测试未登录访问
    with app.test_client() as mal_client:
        assure_mal_not_logged_in(mal_client)
    
    # 清理SQLAlchemy会话，避免数据残留
    db_session.remove()

def assure_mal_not_logged_in(client):
    # 未登录用户访问受保护路由，默认会被重定向到登录页（状态码302）
    response = client.get('/is_logged_in')
    assert response.status_code == 302, "未登录用户不应访问受保护路由"

def login_test_client(client):
    result = client.post('/login', json={
        'next': '/', 
        'email': 'test@me.com',
        'password': 'password',
        'remember': 'y', 
        'submit': 'Login',
    })
    assert result.status_code == 200
    
    # 登录后访问受保护路由，应返回200
    response = client.get('/is_logged_in', follow_redirects=True)
    assert response.status_code == 200
    assert response.json == True

if __name__ == '__main__':
    with app.app_context():
        init_db()
        if not app.security.datastore.find_user(email="test@me.com"):
            app.security.datastore.create_user(
                email="test@me.com", 
                password=hash_password("password")
            )
        db_session.commit()
        
        run_tests(app)

第三步：验证测试效果

运行python3 ./user_test_app.py，如果测试通过，说明：

• 已登录用户可以正常访问受保护路由
• 未登录用户会被拦截（重定向到登录页）

内容的提问来源于stack exchange，提问作者Andy Doucette