Using multiple test_clients in parallel to simulate multi-user workflows

まず issue にある client_a と client_b の fixture、および test_multi_clients_min の例から始め、次に pytest-flask の既存の test-client fixture の動作を確認します。同じテスト中に、個別のクライアントがリクエスト間でログインやログアウトを行わずに異なるログイン済みユーザーを維持でき、マルチクライアントのワークフローがテストでカバーされれば、この機能は完了です。

Is it possible to test with multiple clients, each logged in as different user to integration-test workflows, that involve multiple users?

I'm currently using flask-security and pytest for testing and wanted to give pytest-flask a try.

For example:

def test_message(client_a, client_b):
    r = client_a.post("/api/msg_send", data={"recipient": "user_b", "text": "Hello!"})
    assert r.status_code == 200
  
    r = client_b.get("/api/msg_list")
    assert r.status_code == 200
    assert len(r.json['messages']) > 0
    msg_id = r.json['messages'][0]['id']

    r = client_b.get("/api/msg_get", data={"id": msg_id})
    assert r.status_code == 200
    assert r.json['text'] == "Hello!"

    r = client_a.get("/api/msg_get", data={"id": msg_id})
    assert r.status_code == 200
    assert r.json['seen'] == True
  
    ...

With pytest, I'm preparing the test_clients in a fixture, like in this minimal working example:

import os

from flask import Flask, render_template_string
import flask_mail
from flask_security import Security, current_user, auth_required, hash_password, \
     SQLAlchemySessionUserDatastore, UserMixin, RoleMixin
from flask.json.tag import TaggedJSONSerializer
from itsdangerous import URLSafeTimedSerializer

import pytest

from sqlalchemy import create_engine, Boolean, DateTime, Column, Integer, \
                    String, ForeignKey, UnicodeText
from sqlalchemy.orm import relationship, backref, scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base


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

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'))


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()
    Base.metadata.create_all(bind=engine)


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

# Generate a nice key using secrets.token_urlsafe()
app.config['SECRET_KEY'] = os.environ.get("SECRET_KEY", 'pf9Wkove4IKEAXvy-cQkeDPhv9Cb3Ag-wyJILbq_dFw')
app.config['WTF_CSRF_ENABLED'] = False
# 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')
app.config['SECURITY_REGISTERABLE'] = True
app.config['SECURITY_EMAIL_VALIDATOR_ARGS'] = {"check_deliverability": False}
app.config['SECURITY_PASSWORD_HASH'] = "plaintext"

# Setup Flask-Security
user_datastore = SQLAlchemySessionUserDatastore(db_session, User, Role)
app.security = Security(app, user_datastore)
# flask-mail extension
mail = flask_mail.Mail()
# initialize mail extension
mail.init_app(app)

# Views
@app.route("/")
@auth_required()
def home():
    return render_template_string('Hello {{email}} !', email=current_user.email)

# one time setup
with app.app_context():
    # Create a user to test with
    init_db()

if __name__ == '__main__':
    # run application (can also use flask run)
    app.run()


@pytest.fixture()
def application():
    with app.app_context():
        if not app.security.datastore.find_user(email="[email protected]"):
            app.security.datastore.create_user(email="[email protected]", password=hash_password("password"))
        db_session.commit()

    yield app

@pytest.fixture()
def client_a(application):
    client_a = application.test_client()
    response = client_a.post("/register", data=dict(
        email="[email protected]",
        password="client A password",
        password_confirm="client A password"),
        follow_redirects=False
    )
    yield client_a

@pytest.fixture()
def client_b(application):
    client_b = application.test_client()
    response_b = client_b.post(
        "/register", data=dict(
            email="[email protected]",
            password="client B password",
            password_confirm="client B password"
        ),
        follow_redirects=False
    )
    yield client_b

def get_existing_session(client):
    cookie = next(
        (cookie for cookie in client.cookie_jar if cookie.name == "session"), None
    )
    if cookie:
        serializer = URLSafeTimedSerializer("secret", serializer=TaggedJSONSerializer())
        val = serializer.loads_unsafe(cookie.value)
        return val[1]

def test_multi_clients_min(application, client_a, client_b):
    client_a_session = get_existing_session(client_a)
    client_b_session = get_existing_session(client_b)
    assert client_a_session["_user_id"] != client_b_session["_user_id"]

It would be awesome if this would be possible with pytest-flask without logging in/out after every request that comes from a differnt user, as it would probably be much cleaner than with pytest alone.