Использование AWS Moto с макетом Python для написания модульных тестов ⇐ Python

[Anonymous]

Я работаю с кодовой базой, в которой нет никаких настроек тестов. Я пытаюсь добавить несколько тестов, и в настоящее время у меня есть приведенный ниже тестовый класс test_main.py

Код: Выделить всё

from unittest import TestCase
from moto import mock_aws
import boto3

from main import collect_emr

class Test(TestCase):

def setUp(self):
pass

@mock_aws
def test_collect_emr(self):
emr_client = boto3.client('emr')

emr_client.run_job_flow(
Instances={
"InstanceCount": 3,
"KeepJobFlowAliveWhenNoSteps": True,
"MasterInstanceType": "c3.medium",
"Placement": {"AvailabilityZone": "us-east-1a"},
"SlaveInstanceType": "c3.xlarge",
},
JobFlowRole="EMR_EC2_DefaultRole",
LogUri="s3://mybucket/log",
Name="cluster",
ServiceRole="EMR_DefaultRole",
VisibleToAllUsers=True,
Tags=[
{
'Key': 'string',
'Value': 'string'
},
]
)

expected_outcome = [{'arn': 'arn:aws:elasticmapreduce:ap-southeast-2:123456789012:cluster/j-S78L4S34G4Q7U', 'tags': [{'key': 'string', 'value': 'string'}]}]
actual_outcome = list(collect_emr('us-east-1', 'arn:aws:iam::123456789012:role/role_name'))

self.assertEqual(expected_outcome, actual_outcome)

затем метод, который я тестирую из main.py:

Код: Выделить всё

def collect_emr(region_name: str, role_arn: str) -> Generator[ResourceInfo, Any, None]:
creds = assume_role(role_arn)
client = boto3.client(
"emr",
aws_access_key_id=creds["AccessKeyId"],
aws_secret_access_key=creds["SecretAccessKey"],
aws_session_token=creds["SessionToken"],
region_name=region_name,
config=config,
)

cluster_ids = []
for page in client.get_paginator("list_clusters").paginate():
for cluster in page['Clusters']:
if 'Id' not in cluster:
continue
else:
cluster_id = cluster['Id']
cluster_ids.append(cluster_id)

if cluster_ids:
for cluster_id in cluster_ids:
describe_cluster = client.describe_cluster(ClusterId = cluster_id)
if 'Cluster' not in describe_cluster:
continue
yield {
"arn": describe_cluster['Cluster']['ClusterArn'],
"tags": list(
[{"key": t["Key"], "value": t["Value"]} for t in describe_cluster['Cluster']["Tags"]]
),
}

и результаты моего теста при их запуске:

Код: Выделить всё

Ran 1 test in 0.852s

FAILED (failures=1)

[] != [{'arn': 'arn:aws:elasticmapreduce:ap-southeast-2:123456789012:cluster/j-S78L4S34G4Q7U',
'tags': [{'key': 'string', 'value': 'string'}]}]

Я понимаю, что клиент = внутри метода Collect_emr(... в main.py создает новый клиент boto3, который не Я не знаю о ресурсах, созданных emr_client, созданных в моем тестовом методе test_collect_emr(self):, но я не уверен, как решить эту проблему? Раньше я писал методы, которые всегда принимали бы клиента как параметр, например:

Код: Выделить всё

def collect_emr(boto_client):
...

а затем передайте клиент boto, который я создал в своем тестовом примере. Я не могу сделать это здесь, поскольку в базе кода есть что-то повсюду, и рефакторинг метода для работы таким образом может сломать много вещей. Можно ли объединить макет Python и aws moto в одном методе, или мне нужно будет использовать либо или? Буду очень признателен за любую помощь, так как я немного застрял в том, как лучше всего подойти к этому

Подробнее здесь: https://stackoverflow.com/questions/786 ... unit-tests