Using multiple test_clients in parallel to simulate multi-user workflows

[heeplr]

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="test@me.com"):
            app.security.datastore.create_user(email="test@me.com", 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="ca@me.com",
        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="cb@me.com",
            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.

[Mrudu30]

Any answer on this or what did you do??