У меня есть приложение и маршруты в файле app.py
, из которого я импортирую созданное приложение для проверки маршрутов. Моя проблема в том, что я не могу удалить сеансы или БД обычным способом после запуска теста, и я не знаю почему.
Я нашел хак здесь с использованием db.reflect()
перед вызовом db.drop_all()
, который заставляет удалить базу данных, но я не могу заставить работать функции db.session.remove()
, поэтому сеанс сохраняется во всех тестах вызывая ошибку. Моя проблема похожа на this , но это не решило мою проблему.
app.py
import usermodel
class App:
def create_app(self):
app = Flask(__name__)
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.testing = True
app.secret_key = b'12345678910-not-my-real-key'
app.config['SQLALCHEMY_DATABASE_URI'] = "sqlite:///:memory:"
db = SQLAlchemy(app)
app.db = db
return {
'app': app,
'db': db
}
@app.route('/register')
def register():
try:
parsedData = request.get_json()
exists_on_class = user_model.User.exists_on_class(parsedData['id'])
if not exists_on_class:
user = user_model.User.new(parsedData['id'], parsedData)
user.insert()
auth_token = user.encode_auth_token(user.id)
responseObject = {
'status': 'success',
'message': 'registered',
'auth_token': auth_token.decode()
}
return make_response(jsonify(responseObject), 201)
else:
responseObject = {
'status': 'fail',
'message': 'User already exists. Please Log in.',
}
return make_response(jsonify(responseObject), 202)
except Exception as e:
print('exception in user_controller.register', e)
raise e
test.py
import app
import unittest
from dotenv import load_dotenv, find_dotenv
#loads ENV vars
def setup_testing_environment():
load_dotenv(find_dotenv(".env", raise_error_if_not_found=True))
class TestRoutes(unittest.TestCase):
def test_register_route_pass(self):
#create app
test_app = app.app.test_client()
credentials = {
"id":"1234567891012983",
"username": "username_321",
"password": "password123"
}
credentials = json.dumps(credentials)
setup_testing_environment()
with test_app as c:
db = app.db
r = c.post('/register', data=credentials, content_type='application/json')
#doesn't work but no error raised
db.session.remove()
#using reflect to drop DB hack
db.reflect()
db.drop_all()
usermodel.py
class User(UserMixin, db.Model):
id = db.Column(db.BigInteger, primary_key=True, unique=True)
username = db.Column(db.String(255), unique=True, nullable=False)
password = db.Column(db.LargeBinary(), nullable=False)
@classmethod
def new(cls, sender_id, data={}):
try:
if not sender_id:
raise ValueError('id cannot be None')
elif not data.get('username'):
raise ValueError('username cannot be none')
elif not data.get('password'):
raise ValueError('password cannot be none')
db.create_all()
d = cls()
d.id = sender_id
d.username = data.get('username')
password = utils.hash_password(data.get('password'))
d.password = password
return d
except Exception as e:
print('user new error', e)
raise e
def encode_auth_token(self, user_id):
try:
payload = {
'exp': datetime.datetime.utcnow() + datetime.timedelta(days=0, seconds=5),
'iat': datetime.datetime.utcnow(),
'sub': user_id
}
return jwt.encode(
payload,
os.environ['SECRET_KEY'],
algorithm='HS256'
)
except Exception as e:
print("Error in encode_auth_token", e)
raise e
@staticmethod
def decode_auth_token(auth_token):
try:
payload = jwt.decode(
auth_token, os.environ['SECRET_KEY'], algorithms='HS256')
return payload['sub']
except jwt.ExpiredSignatureError:
print('Signature expired. Please log in again.')
return 'Signature expired. Please log in again.'
except jwt.InvalidTokenError:
print('Invalid token. Please log in again.')
return 'Invalid token. Please log in again.'
def insert(self):
try:
if not self.id:
raise AssertionError('insert Error: ID cannot be None')
db.session.add(self)
db.session.commit()
print('INSERT OKAY')
except Exception as e:
print('Rollback:', e)
raise e
@classmethod
def exists_on_class(cls, lookup, lookup_type='id'):
try:
# with username
if lookup_type == 'username':
if cls.query.filter_by(username=lookup).first():
return True
# default - with id
else:
if cls.query.filter_by(id=lookup).first():
return True
return False
except sqlalchemy.exc.OperationalError as e:
print('no such table: user')
return False
except Exception as e:
print('an error in exists_on_class', e)
raise e
Если я снова вызываю тот же маршрут в тесте, /register
, я получаю эту ошибку:
New instance <User at 0x10dba9c18> with identity key (<class 'models.user_model.User'>, ('1234567891012983',), None) conflicts with persistent instance <User at 0x10dba9128
Возможно, я выбрал неправильный сеанс для удаления но я не смог понять, как это решить. Причина, по которой я так делаю, а не создаю приложение внутри тестов, заключается в том, что я могу тестировать маршруты в app.py
. Если я так делаю, создаю приложение внутри теста, а не импортирую, оно работает как положено; Тогда я не смог бы проверить маршруты.