2016-03-21 43 views
3

我正在编写一个快速类的测试用例,使用boto3从s3中查找/获取密钥。我用摩托在过去的测试伯特(不是3)的代码,但我试图移动这个项目boto3,并运行到一个问题:我该如何测试使用boto3和moto的方法

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) 

错误:

... 
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> 

它看起来像moto没有正确地模拟出boto3的调用 - 我该如何做这件事?

+0

我没有看到moto中的代码使用boto3。 https://github.com/spulec/moto – mootmoot

+0

确实有一些boto3的支持,但也许不在python2.7? https://github.com/spulec/moto/issues?utf8=%E2%9C%93&q=is%3Aissue+is%3Aopen+boto3显示很多人都遇到了麻烦。这个问题离我的最近,https://github.com/spulec/moto/issues/474,暗示对HTTPretty的修复将解决它,并且目前正在使用该修补程序的PR,因此希望这可以很快解决-ish。 – user3610360

回答

3

对我来说有效的方法是用boto设置环境,然后用boto3运行我的模拟测试。

这里有一个工作片断:

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() 

py.test test.py你会得到下面的输出运行:

collected 1 items 

test.py . 

========================================================================================= 1 passed in 2.22 seconds ========================================================================================= 
+0

我不确定你的意思是“用boto设置环境” - 你能详细说明一下吗? – user3610360

+0

我添加了一个工作片段,希望它有帮助! – kardaj

1

this信息,它看起来像流媒体上传使用Boto3 S3的说就是到S3尚未支持。

在我的情况,我用下面的方法成功上传物件到值区:

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

其中“file_to_upload”被上传到S3存储本地文件。对于你的测试用例,你可以创建一个临时文件来检查这个功能:

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')) 
相关问题