So I have this flask application and I'm trying to test it with pytest app.py:
import os
from flask import Flask, jsonify
from flask_smorest import Api
from flask_jwt_extended import JWTManager
from flask_migrate import Migrate
from db import db
from resources.item import blp as ItemBlueprint
from resources.store import blp as StoreBlueprint
from resources.tag import blp as TagBlueprint
from resources.user import blp as UserBlueprint
from blocklist import BLOCKLIST
def create_app(db_url=None):
app = Flask(__name__)
app.config["PROPAGATE_EXCEPTIONS"] = True
app.config["API_TITLE"] = "Stores REST API"
app.config["API_VERSION"] = "v1"
app.config["OPENAPI_VERSION"] = "3.0.3"
app.config["OPENAPI_URL_PREFIX"] = "/"
app.config["OPENAPI_SWAGGER_UI_PATH"] = "/swagger-ui"
app.config["OPENAPI_SWAGGER_UI_URL"] = "https://cdn.jsdelivr.net/npm/swagger-ui-dist/"
app.config["SQLALCHEMY_DATABASE_URI"] = db_url or os.getenv("DATABASE_URL", "sqlite:///data.db")
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = True
db.init_app(app)
api = Api(app)
migrate = Migrate(app, db)
app.config['JWT_SECRET_KEY'] = '69490938337699758397870296439802775085'
jwt = JWTManager(app)
@jwt.needs_fresh_token_loader
def token_not_fresh_callback(jwt_header, jwt_payload):
return (
jsonify(
{
"description": "The token is not fresh.",
"error": "fresh_token_required",
}
),
401,
)
@jwt.token_in_blocklist_loader
def check_if_token_in_blocklist(jwt_header, jwt_payload):
return jwt_payload['jti'] in BLOCKLIST
@jwt.revoked_token_loader
def revoked_token_callback(jwt_header, jwt_payload):
return (
jsonify(
{
"description": "The token has been revoked.",
"error": "token_revoked"
}
)
)
@jwt.additional_claims_loader
def add_claims_to_jwt(identity):
if identity == 1:
return {"is_admin": True}
return {"is_admin": False}
@jwt.expired_token_loader
def expired_token_callback(jwt_header, jwt_payload):
return (
jsonify(
{
"message": "The token has expired.",
"error": "token_expired"
}
),
401,
)
@jwt.invalid_token_loader
def invalid_token_callback(error):
return (
jsonify(
{
"message": "Signature verification failed.",
"error": "invalid_token"
}
), 401,
)
@jwt.unauthorized_loader
def missing_token_callback(error):
return (
jsonify(
{
"description": "Request does not contain an access token.",
"error": "authorization_required",
}
), 401,
)
# @app.before_first_request
# def create_tables():
# db.create_all()
api.register_blueprint(ItemBlueprint)
api.register_blueprint(StoreBlueprint)
api.register_blueprint(TagBlueprint)
api.register_blueprint(UserBlueprint)
return app
And I am trying to test some endpoints like create a store, so I'm doing this in the tests/conftest.py
from app import create_app
@pytest.fixture(scope='module')
def client():
flask_app = create_app()
flask_app.testing = True
flask_app.testing = True
with flask_app.test_client() as testing_client:
with flask_app.app_context():
yield testing_client
And there is the test for creation of a store
from flask_jwt_extended import create_access_token
def test_store_creation(client):
"""
GIVEN a Store model
WHEN a new Store is created
THEN check the name field is defined correctly
"""
access_token = create_access_token('admin')
headers = {
'Authorization': 'Bearer {}'.format(access_token)
}
response = client.post('/store', data={"name": "test_store"}, headers=headers)
print(response.get_json())
assert response.status_code == 201
The Store schema
class PlainItemSchema(Schema):
id = fields.Str(dump_only=True)
name = fields.Str(required=True)
price = fields.Float(required=True)
class PlainTagSchema(Schema):
id = fields.Int(dump_only=True)
name = fields.Str()
class PlainStoreSchema(Schema):
id = fields.Str(dump_only=True)
name = fields.Str(required=True)
class StoreSchema(PlainStoreSchema):
items = fields.List(fields.Nested(PlainItemSchema()), dump_only=True)
tags = fields.List(fields.Nested(PlainTagSchema()), dump_only=True)
My Store Blueprint
from flask.views import MethodView
from flask_smorest import abort, Blueprint
from sqlalchemy.exc import SQLAlchemyError, IntegrityError
from flask_jwt_extended import jwt_required, get_jwt
from schemas import StoreSchema
from db import db
from models import StoreModel
blp = Blueprint('stores', __name__, description='Operations on stores')
@blp.route('/store/<int:store_id>')
class Store(MethodView):
@blp.response(200, StoreSchema)
def get(self, store_id):
store = StoreModel.query.get_or_404(store_id)
return store
@jwt_required(fresh=True)
def delete(self, store_id):
"""Only admins can delete stores"""
jwt = get_jwt()
if not jwt.get('is_admin'):
abort(400, message='Admin privilege required.')
store = StoreModel.query.get_or_404(store_id)
db.session.delete(store)
db.session.commit()
return {"message": "Store deleted"}
@blp.route("/store")
class StoreList(MethodView):
@blp.response(200, StoreSchema(many=True))
def get(self):
return StoreModel.query.all()
@jwt_required()
@blp.arguments(StoreSchema)
@blp.response(201, StoreSchema)
def post(self, store_data):
store = StoreModel(**store_data)
try:
db.session.add(store)
db.session.commit()
except IntegrityError:
abort(
400,
message="A store with that name already exists.",
)
except SQLAlchemyError:
abort(500, message="An error occurred creating the store.")
return store
The error I'm getting is this
============================= test session starts ==============================
collecting ... collected 1 item
test_stores.py::test_store_creation FAILED [100%]{'code': 422, 'errors': {'json': {'name': ['Missing data for required field.']}}, 'status': 'Unprocessable Entity'}
test_stores.py:4 (test_store_creation)
422 != 201
Expected :201
Actual :422
<Click to see difference>
client = <FlaskClient <Flask 'app'>>
def test_store_creation(client):
"""
GIVEN a Store model
WHEN a new Store is created
THEN check the name field is defined correctly
"""
data = {
"name": "test"
}
access_token = create_access_token('admin')
headers = {
'Authorization': 'Bearer {}'.format(access_token)
}
response = client.post('/store', headers=headers, data=data)
print(response.get_json())
> assert response.status_code == 201
E assert 422 == 201
E where 422 = <WrapperTestResponse 109 bytes [422 UNPROCESSABLE ENTITY]>.status_code
test_stores.py:22: AssertionError
It's like I'm not sending any data in the cliet.post() method, In fact if I remove the 'data={'name': 'test'}' I get the same result. I tried to put data dict outside the method and convert it to json with json.dumps(data) still getting the result
CodePudding user response：
flask-smorest is expecting you to post json data as default, so automatically send it as json OR adjust your header manually to reflect the data your are sending.
Fix your code as:
response = client.post("/store", json={"name": "test_store"}, headers=headers)