如何在Pytest测试类的每次迭代前清除会话数据？

问题描述 #

测试Flask账户管理应用的用户更新接口时，使用Pytest参数化测试类，第二次迭代出现会话残留，导致登录断言失败。错误信息如下：

E       AssertionError: assert {'comment': '...nvalid usage'} == {'response': ...successfully'}
E         Left contains 2 more items:
E         {'comment': 'User session is active', 'error': 'Invalid usage'}
E         Right contains 1 more item:
E         {'response': 'Log in procedures successfully'}
E         Use -v to get more diff

尝试添加类级teardown清除会话但无效。

相关代码 #

1. user.py中的用户更新接口 #

@bp.route('/update', methods = ('PUT',))
@login_required
def update_user_info():
    json_req = request.json
    try:
        id = json_req["id"]
    except:
        return jsonify({"error":"ID must be provided"})
    
    db_session = get_session()
    if g.user_id != id:
        error = 'Provide your account ID!'
        return jsonify({"error":error})    

    json_req = request.json
    request_keys = list(json_req.keys())

    if len(request_keys) == 0:
        error = 'Can not update empty values!'
        return jsonify({"error":error})

    #Validate request input if contain forbidden parameters
    if len(set(request_keys) - {'name','password','mail','gender','id'}) != 0:
        error = 'Invalid key value used in request'
        return jsonify({"error":error})

    #Validate password if exists in request
    if 'password' in request_keys:
        if not validate_password(json_req['password']):
            error = '''Password have to contain minimum eight characters,
                    at least one uppercase letter, one lowercase letter,
                    one number and one special character'''
            return jsonify({"error":error.replace('\n','').replace('  ','')})
        else:
            json_req['password'] = generate_password_hash(json_req['password'])

    #Validate mail if exists in request
    if 'mail' in request_keys:
        if not validate_email(json_req['mail']):
            error = 'Invalid email'
            return jsonify({"error":error})

    #Validate gender value if exists in request
    if 'gender' in request_keys:
        if not validate_gender(json_req['gender']):
            error = 'Invalid gender value'
            return jsonify({"error":error})


    try:
        timestamp = datetime.now()
        json_req['updated'] = timestamp
        db_session.execute(
            update(User)
            .where(User.id == id)
            .values(json_req)
        )
        db_session.commit()
        return jsonify({"response":f'User info updated successfully'})
    except exc.SQLAlchemyError as er:
        error = 'Can not connect with database, try later or report bug'
        db_session.rollback()
        current_app.logger.warning(er)
        return jsonify({"error": error})

2. conftest.py中的测试客户端Fixture #

@pytest.fixture(scope='class')
def test_client_class(app,request):

    # Create the test client
    test_client = app.test_client()

    # Share the test client across class methods
    request.cls.test_client = test_client
    yield test_client

3. test_update.py中的测试类（含无效的类级teardown） #

import pytest
import json
from api.database import get_session
from api.models import User
from sqlalchemy import select
from sqlalchemy import and_
from datetime import datetime

@pytest.mark.usefixtures('test_client_class')
@pytest.mark.parametrize("login_request, login_response, update_request, update_response", 
                         [
                             ( 
                                '{"login": "mail@gmail.com","password": "asdefvn#W1D2d"}',
                                '{"response": "Log in procedures successfully"}',
                                '{"id": 1,"name": "newname2"}',
                                '{"response": "User info updated successfully"}'
                             ),
                             ( 
                                '{"login": "mail2@gmail.com","password": "asd3efvn#W11D2d"}',
                                '{"response": "Log in procedures successfully"}',
                                '{"id": 2,"name": "newname2", "mail": "newmail3@gmail.com"}',
                                '{"response": "User info updated successfully"}'
                             )
                        ]
                        )
class TestUpdateUserAccount:
    user_id = None
    rendered_request_login = None
    rendered_request_update = None
    update_timestamp = None


    def test_login(self, app, init_db, test_client_class, populate_database, login_request, login_response, update_request, update_response):

        print(f'{login_request=}')
        TestUpdateUserAccount.rendered_request_login = json.loads(login_request)
        response = test_client_class.post("/user/login",json=TestUpdateUserAccount.rendered_request_login)
        print(f'{TestUpdateUserAccount.rendered_request_login=}')
        assert response.json == json.loads(login_response)



        mail_value = TestUpdateUserAccount.rendered_request_login["login"]
        # get user_id from database
        with app.app_context():
            session = get_session()
            sql_stmt = select(
                User.id).where(
                    and_(
                        User.mail == mail_value,
                        User.active == True
                    )
                )
            TestUpdateUserAccount.user_id = session.execute(sql_stmt).scalar_one()

        assert type(TestUpdateUserAccount.user_id) == type(3)

    def test_update_user_info(self, test_client_class, login_request, login_response, update_request, update_response):
        
        TestUpdateUserAccount.rendered_request_update = json.loads(update_request)
        response = test_client_class.put("/user/update",json=TestUpdateUserAccount.rendered_request_update)
        TestUpdateUserAccount.update_timestamp = datetime.now()
        print(f'{TestUpdateUserAccount.rendered_request_update=}')
        assert response.json == json.loads(update_response)


    def test_verify_database(self, app, login_request, login_response, update_request, update_response):

        with app.app_context():
            session = get_session()
            sql_stmt = select(User.name, User.updated).where(User.id == TestUpdateUserAccount.user_id)

            (new_user_name,updated_timestamp) = session.execute(sql_stmt).one()
            print(f'{updated_timestamp=}')
            assert new_user_name == TestUpdateUserAccount.rendered_request_update["name"]
            assert (TestUpdateUserAccount.update_timestamp - updated_timestamp).total_seconds() < 1

    def test_logout(self, test_client_class, login_request, login_response, update_request, update_response):

        logout_respone = test_client_class.delete("/user/logout",json={})
        assert logout_respone.json["response"] == "session data cleared successfully"

    @classmethod
    def teardown_class(cls):
        # Perform teardown actions after each iteration of the class
        # Clear the session data
        with cls.test_client.session_transaction() as session:
            session.clear()

问题原因 #

1. 测试客户端共享：test_client_class fixture的scope是class，整个测试类仅创建一个测试客户端，参数化的多次迭代会共享同一个客户端的会话。
2. teardown时机错误：teardown_class是在整个类所有测试完成后执行，而非每次参数化迭代后，因此无法清除每次迭代后的会话。
3. 类属性共享：测试类的类属性（如user_id、rendered_request_login）会在参数化迭代间共享，可能导致数据污染。

解决方法 #

方法1：修改Fixture作用域为每次测试 #

将test_client_class的scope改为function，让每次参数化迭代都生成全新的测试客户端：

# conftest.py
@pytest.fixture(scope='function')  # 修改scope为function
def test_client(app):
    return app.test_client()

同时更新测试类的usefixtures和参数：

# test_update.py
@pytest.mark.usefixtures('test_client')
# ... 测试方法中使用test_client参数而非test_client_class

方法2：每次迭代后清除会话 #

在测试类中添加teardown_method，每次测试方法执行后清除会话（或注销登录）：

# test_update.py
class TestUpdateUserAccount:
    # ... 原有代码

    def teardown_method(self):
        # 方式1：调用注销接口
        self.test_client.delete("/user/logout", json={})
        # 方式2：直接清除客户端会话
        with self.test_client.session_transaction() as session:
            session.clear()

方法3：重置类属性为实例属性 #

将类属性改为实例属性，避免参数化迭代间的数据共享：

# test_update.py
class TestUpdateUserAccount:
    def __init__(self):
        self.user_id = None
        self.rendered_request_login = None
        self.rendered_request_update = None
        self.update_timestamp = None

    # 测试方法中使用self.xxx替代TestUpdateUserAccount.xxx
    def test_login(self, app, init_db, test_client_class, populate_database, login_request, login_response, update_request, update_response):
        self.rendered_request_login = json.loads(login_request)
        response = test_client_class.post("/user/login", json=self.rendered_request_login)
        assert response.json == json.loads(login_response)

        mail_value = self.rendered_request_login["login"]
        with app.app_context():
            session = get_session()
            sql_stmt = select(User.id).where(and_(User.mail == mail_value, User.active == True))
            self.user_id = session.execute(sql_stmt).scalar_one()

        assert type(self.user_id) == int

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