Блог

У меня есть метод python, который перечисляет определенные сервисы ECS Fargate на основе некоторых тегов и некоторых имен, используя boto3 get_paginator.

 def list_service_name(environment,
                      resource_owner_name,
                      ecs_client):
    list_of_service = list()
    cluster_name = "my cluster name "   environment
    target_string = "-somedummy"
    resource_owner_tag = resource_owner_name
    service_paginator = ecs_client.get_paginator('list_services')
    for page in service_paginator.paginate(cluster=cluster_name,
                                           launchType='FARGATE'):
       
        for service in page['serviceArns']:

           
            response = ecs_client.list_tags_for_resource(resourceArn=service)
           
            for tags in response['tags']:
                if tags['key'] == 'ResourceOwner' and 
                        tags['value'] == resource_owner_tag and 
                        service.endswith(target_string):
                    list_of_service.append(service)

    return list_of_service
 

Теперь я хотел бы проверить это с помощью moto.
Поэтому я создал conftest.py , где я определил все фиктивные подключения moto к таким сервисам, как ecs. Кроме того, я создал test_main.py файл, как показано ниже, в котором я создал фиктивные службы, подключенные к ECS Fargate. Но по какой-то причине, если я попытаюсь подтвердить результат основного метода в тестовом файле, списки служб вернутся empty . В то время как я хотел бы видеть test-service-for-successful в качестве результата. Есть ли что-то, чего мне не хватает, или разбиение на страницы все еще недоступно в moto?

 from my_code.main import *


@pytest.fixture
def env_name():
    return "local"


@pytest.fixture
def cluster_name(env_name):
    return "my dummy"   env_name   "cluster_name"


@pytest.fixture
def successful_service_name():
    return "test-service-for-successful"


@pytest.fixture
def un_successful_service_name():
    return "test-service-for-un-successful"


@pytest.fixture
def resource_owner():
    return "dummy_tag"


@pytest.fixture
def test_create_service(ecs_client,
                        cluster_name,
                        successful_service_name,
                        un_successful_service_name,
                        resource_owner):
    _ = ecs_client.create_cluster(clusterName=cluster_name)
    _ = ecs_client.register_task_definition(
        family="test_ecs_task",
        containerDefinitions=[
            {
                "name": "hello_world",
                "image": "docker/hello-world:latest",
                "cpu": 1024,
                "memory": 400,
                "essential": True,
                "environment": [
                    {"name": "environment", "value": "local"}
                ],
                "logConfiguration": {"logDriver": "json-file"},
            }
        ],
    )
    ecs_client.create_service(
        cluster=cluster_name,
        serviceName=successful_service_name,
        taskDefinition="test_ecs_task",
        desiredCount=0,
        launchType="FARGATE",
        tags=[{"key": "resource_owner", "value": resource_owner}]
    )

    ecs_client.create_service(
        cluster=cluster_name,
        serviceName=un_successful_service_name,
        taskDefinition="test_ecs_task",
        desiredCount=0,
        launchType="FARGATE",
        tags=[{"key": "resource_owner", "value": resource_owner}]
    )
    yield


def test_list_service_name(env_name,
                           resource_owner,
                           ecs_client):

    objects = list_service_name(env_name,
                                resource_owner,
                                ecs_client)
    # here object is []
    # Where as I should see successful_service_name