title: Synchronous Testing with Celery in Flask App
body: <p>I have a Flask-based web API, that uses the Asynchronous Task Manager <a href="https://github.com/celery/celery" rel="nofollow noreferrer">celery</a>. I created an API endpoint that simply runs my desired asynchronous task, and then added another endpoint that exposes the status &amp; results of this task. </p>
  
  <p>Integration Testing through this API endpoint works fine &amp; passes, however I would also like to test these functions synchronously using the <code>task_always_eager</code> setting in celery. But no matter how I try to set this config setting, the application still runs the tasks asynchronously (I know this by observing the celery logs while the tests are running; celery is still booting up workers and assigning them to this task.</p>
  
  <h3>I've Tried modifying the config variables mid-test file</h3>
  
  <pre><code># TEST STATISTICS API
  current_app.config.update(task_always_eager=True)
  </code></pre>
  
  <h3>As well as in the config.py &amp; <strong>init</strong>.py files</h3>
  
  <pre><code>&lt;config.py, next to the rest of the working config settings&gt;
  task_always_eager = True
  
  &lt;__init__.py, I use the "application factory" pattern&gt;
  celery = Celery(__name__, broker=Config.CELERY_BROKER_URL)
  def create_app(config_name):
      app = Flask(__name__)
      celery.conf.update(app.config)
  </code></pre>
  
  <p>Yet still, celery continues to run workers on this task. This is a problem for unit testing (can't effectively test the function on it's own bc of the bind=True param), and because tested celery tasks don't show up on our test coverage suite with the coverage.py module. Any help would be greatly appreciated!</p>
  
  <h3>The API</h3>
  
  <p>Exposed API Endpoint:</p>
  
  <pre><code>@api.route('/stats/&lt;start_time&gt;/&lt;end_time&gt;', methods=['GET'])
  def stats_endpoint(start_time, end_time):
      task = async_stats.delay(start_time, end_time)
      return jsonify({}), 202, {'Location': url_for('api.stats_status', task_id=task.id)}
  </code></pre>
  
  <p>Backend Async API Task:</p>
  
  <pre><code>@celery.task(bind=True)
  def async_stats(self, start_time, end_time):
      &lt;do stuff, update task state, do more stuff&gt;
      return
  </code></pre>
  
  <p>Async Task Status/Result Endpoint:</p>
  
  <pre><code>@api.route('/stats/status/&lt;task_id&gt;')
  def stats_status(task_id):
      task = async_stats.AsyncResult(task_id)
      result = task.result['result']
      status = task.result['status']
      if task.state in ('PENDING', 'STARTED', 'PROGRESS', 'SUCCESS'):
          # task was created, in various stages of completion
          response = jsonify(result)
          response.status_code = status
      &lt;more response handling...&gt;
      return response
  </code></pre>
  
  <p>UPDATE: setting the <code>CELERY_ALWAYS_EAGER</code> variable to True during config successfully runs my test scripts synchronously. However trying to do so mid-script with <code>current_app.config.update</code> doesn't work, I suspect bc the workers that are running are already configured a certain way by the time the test script runs. Would love some insight on this, if anyone has experience in this field. Having both Unit &amp; Integration tests side by side would require on-demand config changing, but would be ideal</p>