如何在Python中测试S3

2021-06-20 02:46:08

在某些时候,每个工程师都必须决定是否为某些东西写测试或只是运送功能并继续前进。

在一段时间嘎吱嘎吱的情况下,我将经常为简单的东西写测试(例如纯函数)或写下为他们的降压提供最大的爆炸的测试(例如,服务的端到端集成测试)。

测试代码与外部系统相互作用,如数据库或S3,需要更多的努力。但是,重要的业务逻辑经常发生在此代码中,最近我对测试它更感兴趣。

来自Dataclasses Import DataClass Import Json Import Boto3 S3_Bucket ="食谱" def get_s3():返回boto3。客户端(" s3")@ dataclass类配方:名称:str说明:str @ classmethod def get_by_name(cls,name:str):"""抬起食谱按名称args:name(str):食谱名称返回食谱对象""" response = get_s3()。 get_object(bucket = s3_bucket,key = name)响应= json。加载(响应["身体"]。read())返回cls(响应["名称"],响应["指令"])@ classMethod def update_instructions( CLS,名称:str,new_instructions:str):"""更新食谱args的说明:name(str):要更新new_instructions(str)的配方的名称:新的指令和#34;""食谱= CLS。 get_by_name(name)配方。说明= new_instructions返回配方@ classmethod def delete(cls,name:str):"""删除食谱args:name(str):删除配方的名称和#34;& #34;" get_s3()。 delete_object(bucket = s3_bucket,key = name)def to_json(self):"""序列化json returns:str:json表示配方"" "返回json。转储({"名称" self。名字,"指示" self。说明})def save(self):"""""持续存在FEREAPE到S3""" serialized_recipe = self。 to_json()。编码(" UTF-8")get_s3()。 put_object(bucket = s3_bucket,key = self。名称,body = serialized_recipe)

Moto是一个Python库,可以轻松地模拟AWS在测试中的服务。让我们使用它来测试我们的应用程序。

首先,创建一个创建我们S3存储桶的PyTest。 Mock_s3上下文管理器中的所有S3交互将在Moto的Virtual AWS帐户中定向。

从Moto Import Mock_s3导入Boto3从食谱导入配方,S3_Bucket @ PyTest进口Pytest。夹具def s3():"""在假moto aws帐户中创建食谱桶的Pytest夹具产生假的Boto3 S3客户""" Mock_s3():s3 = boto3。客户(" s3")s3。 CREATE_BUCKET(BUCKET = S3_BUCKET)产量S3

def test_create_and_get(s3):配方(name ="玉米虫",指示="在芯片上融合奶酪")。保存()配方=配方。 get_by_name(" Nachos")断言食谱。 name =="玉米片"断言食谱。说明=="在芯片上融化奶酪"

如果我们尝试获取不存在的配方,则应提出异常。此测试涵盖了这种情况。

我们还可以更新配方。此测试确认在调用Save()后更新数据。

def test_update(s3):old_instructions ="芯片上的熔体奶酪" new_instructions ="微波一盘装满玉米片和奶酪"食谱(名称=" Nachos",指令= Old_Instructions)。保存()new_recipe =配方。 update_instructions(name =" nachos",new_instructions = new_instructions)#在调用save()配方=配方之前没有更改。 get_by_name(" Nachos")断言食谱。指令== old_instructions new_recipe。保存()#保存配方=配方后的配方更新。 get_by_name(" Nachos")断言食谱。指令== new_instructions.

def test_delete(s3):食谱(名称="玉米虫",指示="在芯片上融合奶酪")。保存()响应= S3。 list_objects_v2(桶= s3_bucket)断言Len(响应["内容"])== 1断言响应["内容" ] [0] [" key" ] ==" Nachos"食谱 。删除(" Nachos")#删除配方响应= S3后,S3中的数据消失了。 list_objects_v2(桶= s3_bucket)断言"内容"不是回应。键()

总的来说,Moto在实施S3 API时做得很好。安装易于安装,感觉就像真实的S3一样,并且不需要任何代码更改。

以下是创建S3存根的PyTest夹具。由于其他S3客户端不会使用此存根,我们还需要修补get_s3并用存根替换其返回值 - 从而强制在配方类中的所有S3客户端来使用我们的存根。

导入DateTime Import JSON来自Dateutil.tz导入Tzutc从Io Import Bytesio从unittest.stub导入补丁导入boto3从botocore.stub导入stubber,任何来自botocore.response导入流媒体导入pytest从配方导入配方,s3_bucket @ pytest。夹具def s3_stub():"""用s3客户端存根模拟get_s3函数的pytest灯具会产生s3客户端&#34的脱茬;"" s3 = boto3。客户端(" s3")stubber = stuber(s3)与补丁(" recipe.get_s3",return_value = s3):产量脱茬

然后,我们可以将put_object和get_object s3 apis停留响应。使用那些存根,我们可以运行创建并随后获取配方的测试。

def test_create_and_get(s3_stub):#stup out put_object响应#注意:这些存根是不完整的 - 我省略了诸如#brevity pured_object_response = {" respectemetadata&#34等事情。 :{" quequentId" :" 5994D680BF127CE3" ," httpstatuscode" :200," retryattempts" :1,}," Etag" :'" 6299528715bad0e3510d1e4c4952ee7e"" ,} put_object_expected_pa​​rams = {"铲斗" :任何," key" :任何,"身体" :任何} s3_stub。 add_response(" put_object" put_object_response,put_object_expecty_params)#创建Get_Object Encoded_Message = JSON返回的流体。转储({"姓名":"玉米虫","指示":"碎片上的熔体奶酪"})。编码(" utf-8")raw_stream = streamingbody(bytesio(encoded_message),len(encoded_message))#stup out get_object响应get_object_response = {" respectemetadata" :{" quequentId" :" 6BFC00970E62BC8F" ," httpstatuscode" :200," retryattempts" :1,}," LastModified" : 约会时间 。 DateTime(2020,4,6,5,39,29,Tzinfo = Tzutc())," contentLength" :58," Etag" :'" 6299528715bad0e3510d1e4c4952ee7e"" ," contenttype" :"二元/八元门流" ,"元数据" :{},"身体" :raw_stream,} get_object_expected_pa​​rams = {"铲斗" :任何," key" :任何} s3_stub。 add_response(" get_object_response,get_object_eppect_params)#用s3_stub激活stuber:crecipe = crecipe(name ="玉米虫",指示="碎片搅拌奶酪" ) 食谱 。保存()配方=配方。 get_by_name(" Nachos")断言食谱。 name =="玉米片"断言食谱。说明=="在芯片上融化奶酪"

虽然Botocore Stubs是功能的,但我不喜欢用它们工作有几个原因:

他们需要更多的准备。创建存根是耗时的。即使您交互方式运行真实代码并复制响应,也需要替换一些事情 - 例如上面的流体。

他们是脆弱而假的。首先返回响应,首先出局 - 因此,如果您以不同的顺序调用S3 API,则它将抛出错误。如果您在如何调用API中有一个错误,可能不会被捕获。

要使存根看起来有点逼真,你必须嘲笑你的代码不关心的很多领域,并将您的测试与假响应融为一体。

它们从正在测试的模块中泄漏实现详细信息。例如,如果模块从使用s3.list_objects切换到s3.list_objects_v2,则测试将失败,因为它取决于被调用的特定API。这会创建对模块的私有API的不必要依赖,而不是测试公共API。

第三个选项是LocalStack,它允许您在本地提出整个AWS云堆栈。

版本:" 3.7"服务:测试:图片:S3_TESTING:最新网络: - 应用程序输入点: - /pp/wait-for-it.sh--t - " 30" - localstack:4572 - - - - pytest - 测试/环境: - aws_access_key_id = fake - aws_default_region = fake - aws_secret_access_key = fake localstack:图片:localstack / localstack端口: - " 4566-4599:4566-4599"网络: - 应用环境: - Services = S3网络:应用程序:驱动程序:桥接器:桥

接下来,我们再次模拟get_s3,此时间用连接到localstack的s3客户端替换它。

来自unittest.mock import补丁导入boto3从食谱导入配方,s3_bucket @ pytest导入pytest。夹具DEF S3_LOCALSTACK():S3 = BOTO3。客户(" s3",endpoint_url =" http:// localstack:4572")s3。 CREATE_BUCKET(BUCKET = S3_BUCKET)带补丁(" recipe.get_s3",return_value = s3):产量s3

通过这种模拟,我们可以使用Moto运行的相同测试。

def test_create_and_get(s3_localstack):食谱(name =" Nachos",指示="在芯片上融合奶酪")。保存()配方=配方。 get_by_name(" Nachos")断言食谱。 name =="玉米片"断言食谱。说明=="在芯片上融化奶酪" def test_get_does_not_exist(s3_localstack):使用pytest。提升(S3_Localstack。例外。Nosuchkey):食谱=配方。 get_by_name("三明治")

LopalStack非常易于使用,但在我的机器上旋转需要近30秒。 Moto和LocalStack都非常强大,易于使用。 两个解决方案都做好实现S3 API,并且它们还支持其他AWS服务,包括EC2,RDS,Lambda等。 除了Python之外,它们还可以用于测试代码以其他语言。 lopalstack可能是实际连接到aws的最接近的事情,但对于上面提出的简单用例,我无法证明旋转堆栈所需的额外开销和时间。 因此,我推荐Moto,因为它是正确实现S3 API的最轻量级的解决方案。 对于测试S3性能的更复杂的项目,LocalStack可能是一个不错的选择。 Botocore Stubs不会削减。 谢谢阅读! 如果您有任何反馈,我很乐意收到您的消息 - 在Twitter上关注我或在LinkedIn上留言。