My goal is to be able to test routes with the Flask app test_client using self.client.post(url, .... The problem that I'm running into is any time I make a request using it, I hit a DetachedInstanceError from SQLAlchemy, saying that either refresh or lazy loading operations cannot occur.
Here's my setup:
class BaseTestCase(unittest.TestCase):
def setUp(self):
self.postgresql = Postgresql()
config = {
'FLASK_ENV': 'test',
'SQLALCHEMY_DATABASE_URI': self.postgresql.url()
}
with patch.dict('os.environ', config):
self.app = create_app()
self.client = self.app.test_client()
ctx = self.app.app_context()
ctx.push()
db.create_all()
self.assertEqual(app.testing, True)
reset_fake_data()
def tearDown(self):
db.session.remove()
self.postgresql.stop()
def post(self, url, data={}, headers={}):
headers = {'Content-Type': 'application/json', **headers}
return self.client.post(url, headers=headers, data=json.dumps(data))
def post_with_auth_token(self, url, data={}, headers={}):
headers = {'Authorization': f'Bearer {self.auth_token}'}
return self.post(url, headers=headers, data=data)
Then in another file, I'm calling this with response = self.post('/api/users/register', data=data) and getting the error:
sqlalchemy.orm.exc.DetachedInstanceError: Parent instance <User at 0x10ea43f98> is not bound to a Session; lazy load operation of attribute 'courses' cannot proceed (Background on this error at: http://sqlalche.me/e/bhk3)
Below is the route. I create a user object and then attempt to add a course to it but apparently, when I try to do user.courses.append, it's unable to lazy load the courses. I've looked into disabling the expire_on_commit attribute, but I'm not sure how to do that only in the cases of testing, or if there's a better way to go about this.
I read that converting from lazy to eager loading could also fix it, but I'd rather keep things loading lazily and it didn't seem to fix it when I set the relationship to be lazy='subquery' anyways.
@users.route('/register', methods=['POST'])
def register():
data = request.get_json()
try:
user = User.from_data(**data)
except UnprocessableEntity:
return jsonify(msg='Missing parameters'), 422
except Conflict:
return jsonify(msg='Email already exists'), 409
except Unauthorized:
return jsonify(msg='Unauthorized'), 401
if data.get('invitation_token'):
invite = CourseInvite.query.filter_by(token=data['invitation_token']).first_or_404('Invite not found')
if invite.email and invite.email != user.email:
return jsonify(msg='Email mismatch'), 401
elif invite.used:
return jsonify(msg='Invite already used'), 401
user.courses.append(CourseEnrollment(course_id=invite.course.id))
invite.used = True
db.session.add(user)
db.session.commit()
return jsonify(msg='User successfully created'), 201