Dan Tenenbaum
Dan Tenenbaum

Reputation: 1889

python 3: mock a method of the boto3 S3 client

I want to unit test some code that calls a method of the boto3 s3 client. I can't use moto because this particular method (put_bucket_lifecycle_configuration) is not yet implemented in moto. I want to mock the S3 client and assure that this method was called with specific parameters.

The code I want to test looks something like this:

# sut.py
import boto3

class S3Bucket(object):
    def __init__(self, name, lifecycle_config):
        self.name = name
        self.lifecycle_config = lifecycle_config

    def create(self):
        client = boto3.client("s3") 
        client.create_bucket(Bucket=self.name)
        rules = # some code that computes rules from self.lifecycle_config
        # I want to test that `rules` is correct in the following call:
        client.put_bucket_lifecycle_configuration(Bucket=self.name, \
          LifecycleConfiguration={"Rules": rules})

def create_a_bucket(name):
    lifecycle_policy = # a dict with a bunch of key/value pairs
    bucket = S3Bucket(name, lifecycle_policy)
    bucket.create()
    return bucket

In my test, I'd like to call create_a_bucket() (though instantiating an S3Bucket directly is also an option) and make sure that the call to put_bucket_lifecycle_configuration was made with the correct parameters.

I have messed around with unittest.mock and botocore.stub.Stubber but have not managed to crack this. Unless otherwise urged, I am not posting my attempts since they have not been successful so far.

I am open to suggestions on restructuring the code I'm trying to test in order to make it easier to test.

Upvotes: 3

Views: 10450

Answers (1)

Dan Tenenbaum
Dan Tenenbaum

Reputation: 1889

Got the test to work with the following, where ... is the remainder of the arguments that are expected to be passed to s3.put_bucket_lifecycle_configuration().

# test.py
from unittest.mock import patch
import unittest

import sut

class MyTestCase(unittest.TestCase):
    @patch("sut.boto3")
    def test_lifecycle_config(self, cli):
        s3 = cli.client.return_value
        sut.create_a_bucket("foo")
        s3.put_bucket_lifecycle_configuration.assert_called_once_with(Bucket="foo", ...)


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

Upvotes: 4

Related Questions