Closed caffeinatedMike closed 3 years ago
So, after reviewing the test setup for this package I realized there was a straight-forward way to utilize fakeredis
for tests. The only thing I was missing, and what was presumably leading to the the connection error was the push_connection
function from rq
. The below setup works when the app is in testing mode.
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_wtf import CSRFProtect
from rq_scheduler import Scheduler
from redis import Redis
from rq import Queue
db = SQLAlchemy()
csrf = CSRFProtect()
def create_app(config_obj='config.DevelopmentConfig'):
app = Flask(__name__)
app.config.from_object(config_obj)
# Initialize CSRF protection
csrf.init_app(app)
# Setup database components
db.init_app(app)
with app.app_context():
from app import models
# create db models
db.create_all()
# import routes
from app import views
# Initialize redis components
if app.testing:
from fakeredis import FakeStrictRedis
from rq import push_connection
app.redis = FakeStrictRedis()
push_connection(app.redis) # <= The missing piece
app.task_queue = Queue('reporting-tasks', is_async=False, connection=app.redis)
else:
app.redis = Redis.from_url(app.config['REDIS_URL'])
app.task_queue = Queue('reporting-tasks', connection=app.redis)
# Initialize scheduler and associated crons
app.scheduler = Scheduler(queue=app.task_queue)
app.scheduler.cron('*/5 * * * *', func='app.tasks:check_failed_jobs', id='check_failed_jobs')
app.scheduler.cron('* * */1 * *', func='app.tasks:purge_old_logs', id='purge_old_logs')
return app
The current
rq
module suggests usingfakeredis.FakeStrictRedis
when running tests, but doing so with this scheduler causes anrq.connections.NoRedisConnectionException: Could not resolve a Redis connection
error. Exampleapp_factory
function below: