所以我有这个flask应用程序，我试图用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
我试着测试一些端点，比如创建一个商店，我在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
还有创建商店的测试
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
Store模式
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
我得到的错误是这个
============================= 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
这就像我没有发送任何数据在cliet.post()方法，事实上，如果我删除'data={'name': 'test'}'我得到相同的结果。我试图将数据字典放在方法之外，并将其转换为json json.dumps(数据)仍然得到结果
flask-smorest期望您将json数据作为默认值发布，因此自动将其发送为json或手动调整您的头部以反映您正在发送的数据。
修改你的代码为:
response = client.post("/store", json={"name": "test_store"}, headers=headers)