2016-03-21 66 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'))