how do I test methods using boto3 with moto

Question

I am writing test cases for a quick class to find / fetch keys from s3, using boto3. I have used moto in the past to test boto (not 3) code but am trying to move to boto3 with this project, and running into an issue:

class TestS3Actor(unittest.TestCase):
    @mock_s3
    def setUp(self):
        self.bucket_name = 'test_bucket_01'
        self.key_name = 'stats_com/fake_fake/test.json'
        self.key_contents = 'This is test data.'
        s3 = boto3.session.Session().resource('s3')
        s3.create_bucket(Bucket=self.bucket_name)
        s3.Object(self.bucket_name, self.key_name).put(Body=self.key_contents)

error:

...
File "/Library/Python/2.7/site-packages/botocore/vendored/requests/packages/urllib3/connectionpool.py", line 344, in _make_request
self._raise_timeout(err=e, url=url, timeout_value=conn.timeout)
File "/Library/Python/2.7/site-packages/botocore/vendored/requests/packages/urllib3/connectionpool.py", line 314, in _raise_timeout
if 'timed out' in str(err) or 'did not complete (read)' in str(err):  # Python 2.6
TypeError: __str__ returned non-string (type WantWriteError)
botocore.hooks: DEBUG: Event needs-retry.s3.CreateBucket: calling handler <botocore.retryhandler.RetryHandler object at 0x10ce75310>

It looks like moto is not mocking out the boto3 call correctly - how do I make that work?

I didn't see code inside moto make use of boto3. https://github.com/spulec/moto — mootmoot, Mar 23 '16 at 09:25
There is definitely some boto3 support, but maybe not in python2.7? https://github.com/spulec/moto/issues?utf8=%E2%9C%93&q=is%3Aissue+is%3Aopen+boto3 shows that many folks are having trouble. This issue is closest to mine, https://github.com/spulec/moto/issues/474 , suggests that a fix to HTTPretty will resolve it, and a PR is currently in with that fix, so hoping this can be resolved soon-ish. — user3610360, Mar 24 '16 at 09:35

kardaj · Accepted Answer · 2016-06-16T13:30:07.057

What worked for me is setting up the environment with boto before running my mocked tests with boto3.

Here's a working snippet:

import unittest
import boto
from boto.s3.key import Key
from moto import mock_s3
import boto3


class TestS3Actor(unittest.TestCase):
    mock_s3 = mock_s3()

    def setUp(self):
        self.mock_s3.start()
        self.location = "eu-west-1"
        self.bucket_name = 'test_bucket_01'
        self.key_name = 'stats_com/fake_fake/test.json'
        self.key_contents = 'This is test data.'
        s3 = boto.connect_s3()
        bucket = s3.create_bucket(self.bucket_name, location=self.location)
        k = Key(bucket)
        k.key = self.key_name
        k.set_contents_from_string(self.key_contents)

    def tearDown(self):
        self.mock_s3.stop()

    def test_s3_boto3(self):
        s3 = boto3.resource('s3', region_name=self.location)
        bucket = s3.Bucket(self.bucket_name)
        assert bucket.name == self.bucket_name
        # retrieve already setup keys
        keys = list(bucket.objects.filter(Prefix=self.key_name))
        assert len(keys) == 1
        assert keys[0].key == self.key_name
        # update key
        s3.Object(self.bucket_name, self.key_name).put(Body='new')
        key = s3.Object(self.bucket_name, self.key_name).get()
        assert 'new' == key['Body'].read()

When run with py.test test.py you get the following output:

collected 1 items 

test.py .

========================================================================================= 1 passed in 2.22 seconds =========================================================================================

I'm not sure exactly what you mean by "setting up the environment with `boto`" - can you elaborate on that a bit, please? — user3610360, Jun 16 '16 at 12:55
Thanks for pointing to `mock_s3.stop()` which helps "resetting" the mocked bucket! — Dror, Dec 03 '18 at 13:16
@user3610360 in moto we have first to create a fully mocked aws environment. So in the example code above, in the `setUp` function we create a testing bucket (in memory) with testing contents. The unit test is run on this testing bucket. — alexopoulos7, Jan 15 '21 at 11:34

score 1 · Answer 2 · answered Jul 06 '17 at 17:24

According to this information, it looks like streaming upload to s3 using Boto3 S3 Put is not yet supported.

In my case, I used following to successfully upload an object to a bucket:

s3.Object(self.s3_bucket_name, self.s3_key).put(Body=open("file_to_upload", 'rb'))

where "file_to_upload" is your local file to be uploaded to s3 bucket. For your test case, you can just create a temporary file to check this functionality:

test_file = open("test_file.json", "w")
test_file.write("some test contents")
test_file.close()
s3.Object(self.s3_bucket_name, self.s3_key).put(Body=open("test_file", 'rb'))

how do I test methods using boto3 with moto

2 Answers2