Please add support for the ResilienceHub API

[StevenCaswellVR]

I am using moto 5.0.2, Python mocks with Python 3.11.3, boto3 1.26.68, botocore 1.29.165.

I have the following code:

@pytest.fixture(scope="function")
def aws_credentials():
    """Mocked AWS Credentials for moto."""
    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
    os.environ["AWS_SECURITY_TOKEN"] = "testing"
    os.environ["AWS_SESSION_TOKEN"] = "testing"
    os.environ["AWS_DEFAULT_REGION"] = "us-east-1"

@pytest.fixture(scope="function")
def resiliencehub(aws_credentials):
    with mock_aws():
        yield boto3.client("resiliencehub", region_name="us-east-2")

@pytest.fixture
def create_policy(resiliencehub):
    response = boto3.client("resiliencehub").create_resiliency_policy(
        policyName="mock-resilience-hub-basic-webapp",
        policyDescription="Mocked resiliency policy",
        policy={"AZ": {"rpoInSecs": 600, "rtoInSecs": 600}, "Hardware": {"rpoInSecs": 600, "rtoInSecs": 600}, "Region": {"rpoInSecs": 3600, "rtoInSecs": 14400}, "Software": {"rpoInSecs": 600, "rtoInSecs": 600}},
        tier="Standard",
    )
    print("create_policy response")
    pprint(response)
    return response.get("policy", {}).get("policyArn")

When running this code I get the following error:

$ pytest --capture=tee-sys test_resiliencehub_assesssment_score.py::test_runner_with_appli
cation_names
========================================================================================== test session starts ===========================================================================================
platform darwin -- Python 3.11.3, pytest-6.2.5, py-1.11.0, pluggy-1.4.0
rootdir: /Users/stevencaswell/vr/Projects/resilience-hub-application-builder/project-repo/automated-resiliency-health-index/app/app/step_runner_lambda/tests
plugins: typeguard-2.13.3
collected 1 item                                                                                                                                                                                         

test_resiliencehub_assesssment_score.py E                                                                                                                                                          [100%]

================================================================================================= ERRORS =================================================================================================
__________________________________________________________________________ ERROR at setup of test_runner_with_application_names __________________________________________________________________________

resiliencehub = <botocore.client.ResilienceHub object at 0x1077b72d0>

    @pytest.fixture
    def create_policy(resiliencehub):
>       response = boto3.client("resiliencehub").create_resiliency_policy(
            policyName="mock-resilience-hub-basic-webapp",
            policyDescription="Mocked resiliency policy",
            policy={"AZ": {"rpoInSecs": 600, "rtoInSecs": 600}, "Hardware": {"rpoInSecs": 600, "rtoInSecs": 600}, "Region": {"rpoInSecs": 3600, "rtoInSecs": 14400}, "Software": {"rpoInSecs": 600, "rtoInSecs": 600}},
            tier="Standard",
        )

test_resiliencehub_assesssment_score.py:26: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/Users/stevencaswell/Tools/pyenv/versions/vr-rhi-resilience-hub-integration-3.11.3/lib/python3.11/site-packages/botocore/client.py:530: in _api_call
    return self._make_api_call(operation_name, kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <botocore.client.ResilienceHub object at 0x10b8e9b90>, operation_name = 'CreateResiliencyPolicy'
api_params = {'clientToken': '987a4b10-28aa-4b45-9dff-79047ea9075f', 'policy': {'AZ': {'rpoInSecs': 600, 'rtoInSecs': 600}, 'Hardwa...InSecs': 600}}, 'policyDescription': 'Mocked resiliency policy', 'policyName': 'mock-resilience-hub-basic-webapp', ...}

    def _make_api_call(self, operation_name, api_params):
        operation_model = self._service_model.operation_model(operation_name)
        service_name = self._service_model.service_name
        history_recorder.record(
            'API_CALL',
            {
                'service': service_name,
                'operation': operation_name,
                'params': api_params,
            },
        )
        if operation_model.deprecated:
            logger.debug(
                'Warning: %s.%s() is deprecated', service_name, operation_name
            )
        request_context = {
            'client_region': self.meta.region_name,
            'client_config': self.meta.config,
            'has_streaming_input': operation_model.has_streaming_input,
            'auth_type': operation_model.auth_type,
        }
        api_params = self._emit_api_params(
            api_params=api_params,
            operation_model=operation_model,
            context=request_context,
        )
        endpoint_url, additional_headers = self._resolve_endpoint_ruleset(
            operation_model, api_params, request_context
        )
        request_dict = self._convert_to_request_dict(
            api_params=api_params,
            operation_model=operation_model,
            endpoint_url=endpoint_url,
            context=request_context,
            headers=additional_headers,
        )
        resolve_checksum_context(request_dict, operation_model, api_params)
    
        service_id = self._service_model.service_id.hyphenize()
        handler, event_response = self.meta.events.emit_until_response(
            'before-call.{service_id}.{operation_name}'.format(
                service_id=service_id, operation_name=operation_name
            ),
            model=operation_model,
            params=request_dict,
            request_signer=self._request_signer,
            context=request_context,
        )
    
        if event_response is not None:
            http, parsed_response = event_response
        else:
            apply_request_checksum(request_dict)
            http, parsed_response = self._make_request(
                operation_model, request_dict, request_context
            )
    
        self.meta.events.emit(
            'after-call.{service_id}.{operation_name}'.format(
                service_id=service_id, operation_name=operation_name
            ),
            http_response=http,
            parsed=parsed_response,
            model=operation_model,
            context=request_context,
        )
    
        if http.status_code >= 300:
            error_code = parsed_response.get("Error", {}).get("Code")
            error_class = self.exceptions.from_code(error_code)
>           raise error_class(parsed_response, operation_name)
E           botocore.exceptions.ClientError: An error occurred (404) when calling the CreateResiliencyPolicy operation: Not yet implemented

/Users/stevencaswell/Tools/pyenv/versions/vr-rhi-resilience-hub-integration-3.11.3/lib/python3.11/site-packages/botocore/client.py:964: ClientError

[bblommers]

Hi @StevenCaswellVR, I'll mark it as an enhancement.

Is it just the one method that you need?

[StevenCaswellVR]

Actually there are several methods I need:

• create_resiliency_policy
• create_app
• list_app
• list_app_assessments

It would be nice to have all of the list_* methods but those above are the most critical.

Thanks.

[bblommers]

Moto >= 5.0.4.dev7 now contains the following methods:

• create_app()
• create_resiliency_policy()
• describe_app()
• describe_resiliency_policy()
• list_app_assessments()
• list_apps()
• list_resiliency_policies()
• list_tags_for_resource()
• tag_resource()
• untag_resource()

Note that the list_app_assessments() currently always returns an empty list, just because there are no app assessments to list.

If I'm understanding the workflow correctly, the following methods would need to be called (at minimum) to start an assessment:

• create_app_version_resource()
• publish_app_version()
• start_app_assessment()

So before we can show actual data in the list_app_assessments, we need to implement these methods first.

Let me know if that's desired, and/or if you're running into any issues so far.

[StevenCaswellVR]

I will start checking out the 5.0.4.dev7 later today.

I think more likely the workflow sequence that would work would be:

• import_resources_to_draft_app_version with emphasis on the sourceArns parameter
• create_app_version_app_component
• create_app_version_resource
• publish_app_version

I'm somewhat guessing as I've not used this particular flow, as I've been mostly focused on the import_resources_to_draft_app_version followed by resolve_app_version_resources flow. Unfortunately I think this flow might be too difficult to mock as the import reads live resources (i.e., from a terraform state file) and populates the app structure based on that content. While I'd love to be able to have that flow mocked, it feels like that could be difficult.

Thanks.

[bblommers]

Hi @StevenCaswellVR!

Moto >= 5.0.4.dev19 now also has support for:

• create_app_version_app_component()
• create_app_version_resource()
• import_resources_to_draft_app_version()
• list_app_version_app_components()
• list_app_version_resources()
• list_app_versions()
• publish_app_version()

[bblommers]

moto >= 5.0.8.dev5 now has the ability to configure the results of the list_app_assessment()-method - see the documentation here: http://docs.getmoto.org/en/latest/docs/services/resiliencehub.html

As far as I can tell, that covers everything, so I'll close this. Feel free to open new issue if you run into problems though, or would like to see support for other features.