Does moto only mock boto3 calls?

[thedamnedrhino]

HI All,
Thanks for this great library.
tl; dr: does moto only apply when the boto3 client is used to interact with AWS? Celery does not use boto3 internally. Details below.

I'm trying to test a celery setup with an SQS backend locally, and using moto to mock the SQS calls.
In the example below: my initial call using boto3 to create the SQS queue succeeds, but calls within Celery to poll the SQS queue fail with a 403. It's making real calls to AWS. It looks like Celery doesn't use boto3 to make aws calls so I thought that might be the reason. Here is my code:

import unittest

import boto3
from celery.contrib.testing.worker import setup_app_for_worker
from celery.worker.worker import WorkController
from moto import mock_aws

from py.django.celery import FreshCelery


@mock_aws
class TestCustomCelery(unittest.TestCase):
    def setUp(self):
        self.queue_name = "test-queue"
        self.app = FreshCelery("test_app")
        self.app.config_from_object(
            {...}
        )

        self.sqs = boto3.client("sqs", region_name="us-east-1")
        self.queue_url = self.sqs.create_queue(
            QueueName=self.queue_name,
        )["QueueUrl"]

        setup_app_for_worker(self.app, loglevel="INFO", logfile=None)

    def test_celery(self):
        @self.app.task(bind=True, max_retries=0, refresh_message=True)
        def test_task(self):
            assert False

        test_task.delay()

        worker = WorkController(app=self.app)
        worker._install_signal_handlers = lambda: None
       
        # I thought I'd need the timer to kill the worker but it doesn't even start
        # timer = threading.Timer(10, worker.stop)
        # timer.start()
        worker.start()


if __name__ == "__main__":
    unittest.main()

The test passes but I get the below stack trace in stderr due to worker.start():

[2025-02-20 00:06:45,878: CRITICAL/MainProcess] Unrecoverable error: Exception('Request HTTP Error  HTTP 403  Forbidden (b\'<?xml version="1.0"?><ErrorResponse xmlns="http://queue.amazonaws.com/doc/2012-11-05/"><Error><Type>Sender</Type><Code>InvalidClientTokenId</Code><Message>The security token included in the request is invalid.</Message><Detail/></Error><RequestId>c78173d4-3abc-5b37-93a2-13e44ee929a8</RequestId></ErrorResponse>\')')
Traceback (most recent call last):
  File "/.../celery/worker/worker.py", line 202, in start
    self.blueprint.start(self)
  File "/.../celery/bootsteps.py", line 116, in start
    step.start(parent)
  File "/.../celery/bootsteps.py", line 365, in start
    return self.obj.start()
  File "/.../celery/worker/consumer/consumer.py", line 340, in start
    blueprint.start(self)
  File "/.../celery/bootsteps.py", line 116, in start
    step.start(parent)
  File "/.../celery/worker/consumer/consumer.py", line 746, in start
    c.loop(*c.loop_args())
  File "/.../celery/worker/loops.py", line 97, in asynloop
    next(loop)
  File "/.../kombu/asynchronous/hub.py", line 373, in create_loop
    cb(*cbargs)
  File "/.../kombu/asynchronous/http/curl.py", line 122, in on_readable
    return self._on_event(fd, _pycurl.CSELECT_IN)
  File "/.../kombu/asynchronous/http/curl.py", line 139, in _on_event
    self._process_pending_requests()
  File "/.../kombu/asynchronous/http/curl.py", line 145, in _process_pending_requests
    self._process(curl)
  File "/.../kombu/asynchronous/http/curl.py", line 191, in _process
    request.on_ready(self.Response(
  File "/.../vine/promises.py", line 168, in __call__
    svpending(*ca, **ck)
  File "/.../vine/promises.py", line 161, in __call__
    return self.throw()
  File "/.../vine/promises.py", line 158, in __call__
    retval = fun(*final_args, **final_kwargs)
  File "/.../vine/funtools.py", line 98, in _transback
    return callback(ret)
  File "/.../vine/promises.py", line 161, in __call__
    return self.throw()
  File "/.../vine/promises.py", line 158, in __call__
    retval = fun(*final_args, **final_kwargs)
  File "/.../vine/funtools.py", line 96, in _transback
    callback.throw()
  File "/.../vine/funtools.py", line 94, in _transback
    ret = filter_(*args + (ret,), **kwargs)
  File "/.../kombu/asynchronous/aws/connection.py", line 246, in _on_list_ready
    raise self._for_status(response, response.read())
Exception: Request HTTP Error  HTTP 403  Forbidden (b'<?xml version="1.0"?><ErrorResponse xmlns="http://queue.amazonaws.com/doc/2012-11-05/"><Error><Type>Sender</Type><Code>InvalidClientTokenId</Code><Message>The security token included in the request is invalid.</Message><Detail/></Error><RequestId>c78173d4-3abc-5b37-93a2-13e44ee929a8</RequestId></ErrorResponse>')                                          

[bblommers]

Hi @thedamnedrhino! Yes - the decorator hooks into boto3, so only those calls are intercepted.

Iff it is possible to configure Celery to use a different endpoint URL (i.e., use localhost instead of sqs.amazonaws.com, then you could run Moto as a server, and redirect Celery to use Moto instead.

Documentation on how to run Moto as a server, either as a stand-alone or within a Python thread: https://docs.getmoto.org/en/latest/docs/server_mode.html

I'm not too familiar with Celery, so I can't comment on whether they allow the endpoint URL configuration. Based on a quick Google search, it seems to be possible with predefined queries.