Repository: getmoto/moto
Pull request #8879 (2025-05-06 11:13:59)

feat(ram): implement `GetResourceShareAssociations` API

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/8879?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) Report
Attention: Patch coverage is `90.00000%` with `3 lines` in your changes missing coverage. Please review.
> Project coverage is 92.85%. Comparing base [(`570069d`)](https://app.codecov.io/gh/getmoto/moto/commit/570069d50a0d3a5b1de649c0b2410d3a0397f6fb?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`56778c9`)](https://app.codecov.io/gh/getmoto/moto/commit/56778c9004595baf7640c2cd3f932a171ed9ee41?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).

| [Files with missing lines](https://app.codecov.io/gh/getmoto/moto/pull/8879?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Patch % | Lines |
|---|---|---|
| [moto/ram/models.py](https://app.codecov.io/gh/getmoto/moto/pull/8879?src=pr&el=tree&filepath=moto%2Fram%2Fmodels.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#diff-bW90by9yYW0vbW9kZWxzLnB5) | 89.28% | [3 Missing :warning: ](https://app.codecov.io/gh/getmoto/moto/pull/8879?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) |

<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##           master    #8879   +/-   ##
=======================================
  Coverage   92.85%   92.85%           
=======================================
  Files        1282     1282           
  Lines      111792   111822   +30     
=======================================
+ Hits       103804   103832   +28     
- Misses       7988     7990    +2     
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/8879/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Coverage Δ | |
|---|---|---|
| [servertests](https://app.codecov.io/gh/getmoto/moto/pull/8879/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `27.97% <0.00%> (-0.01%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/8879/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `92.82% <90.00%> (+<0.01%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#carryforward-flags-in-the-pull-request-comment) to find out more.

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/getmoto/moto/pull/8879?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).

<details><summary> :rocket: New features to boost your workflow: </summary>

- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>

Patch:
diff --git a/moto/ram/models.py b/moto/ram/models.py
index 87915b088..69fb55375 100644
--- a/moto/ram/models.py
+++ b/moto/ram/models.py
@@ -225,5 +225,82 @@ class ResourceAccessManagerBackend(BaseBackend):
 
         return dict(returnValue=True)
 
+    def get_resource_share_associations(self, **kwargs: Any) -> Dict[str, Any]:
+        association_type = kwargs["associationType"]
+        if association_type not in ["PRINCIPAL", "RESOURCE"]:
+            raise InvalidParameterException(
+                f"{association_type} is not a valid association type. "
+                "Specify either PRINCIPAL or RESOURCE and try again."
+            )
+
+        association_status = kwargs.get("associationStatus")
+        if association_status and association_status not in [
+            "ASSOCIATING",
+            "ASSOCIATED",
+            "FAILED",
+            "DISASSOCIATING",
+            "DISASSOCIATED",
+        ]:
+            raise InvalidParameterException(
+                f"{association_status} is not a valid association status."
+            )
+
+        resource_share_arns = kwargs.get("resourceShareArns", [])
+        resource_arn = kwargs.get("resourceArn")
+        principal = kwargs.get("principal")
+
+        if association_type == "PRINCIPAL" and resource_arn:
+            raise InvalidParameterException(
+                "You cannot specify a resource ARN when the association type is PRINCIPAL."
+            )
+        if association_type == "RESOURCE" and principal:
+            raise InvalidParameterException(
+                "You cannot specify a principal when the association type is RESOURCE."
+            )
+
+        associations = []
+        for resource_share in self.resource_shares:
+            if resource_share_arns and resource_share.arn not in resource_share_arns:
+                continue
+
+            if association_type == "PRINCIPAL":
+                for principal_id in resource_share.principals:
+                    if principal and principal != principal_id:
+                        continue
+                    associations.append(
+                        {
+                            "resourceShareArn": resource_share.arn,
+                            "resourceShareName": resource_share.name,
+                            "associatedEntity": principal_id,
+                            "associationType": "PRINCIPAL",
+                            "status": association_status or "ASSOCIATED",
+                            "creationTime": unix_time(resource_share.creation_time),
+                            "lastUpdatedTime": unix_time(
+                                resource_share.last_updated_time
+                            ),
+                            "external": False,
+                        }
+                    )
+            else:  # RESOURCE
+                for resource_id in resource_share.resource_arns:
+                    if resource_arn and resource_arn != resource_id:
+                        continue
+                    associations.append(
+                        {
+                            "resourceShareArn": resource_share.arn,
+                            "resourceShareName": resource_share.name,
+                            "associatedEntity": resource_id,
+                            "associationType": "RESOURCE",
+                            "status": association_status or "ASSOCIATED",
+                            "creationTime": unix_time(resource_share.creation_time),
+                            "lastUpdatedTime": unix_time(
+                                resource_share.last_updated_time
+                            ),
+                            "external": False,
+                        }
+                    )
+
+        return dict(resourceShareAssociations=associations)
+
 
 ram_backends = BackendDict(ResourceAccessManagerBackend, "ram")
diff --git a/moto/ram/responses.py b/moto/ram/responses.py
index 6f20efff7..93ea41153 100644
--- a/moto/ram/responses.py
+++ b/moto/ram/responses.py
@@ -36,3 +36,8 @@ class ResourceAccessManagerResponse(BaseResponse):
 
     def enable_sharing_with_aws_organization(self) -> str:
         return json.dumps(self.ram_backend.enable_sharing_with_aws_organization())
+
+    def get_resource_share_associations(self) -> str:
+        return json.dumps(
+            self.ram_backend.get_resource_share_associations(**self.request_params)
+        )
diff --git a/moto/ram/urls.py b/moto/ram/urls.py
index 09d782e6b..1b5775bb0 100644
--- a/moto/ram/urls.py
+++ b/moto/ram/urls.py
@@ -7,5 +7,6 @@ url_paths = {
     "{0}/deleteresourceshare/?$": ResourceAccessManagerResponse.dispatch,
     "{0}/enablesharingwithawsorganization$": ResourceAccessManagerResponse.dispatch,
     "{0}/getresourceshares$": ResourceAccessManagerResponse.dispatch,
+    "{0}/getresourceshareassociations$": ResourceAccessManagerResponse.dispatch,
     "{0}/updateresourceshare$": ResourceAccessManagerResponse.dispatch,
 }


Test patch:
diff --git a/tests/test_ram/test_ram.py b/tests/test_ram/test_ram.py
index 6b4ed32f8..da782ad8d 100644
--- a/tests/test_ram/test_ram.py
+++ b/tests/test_ram/test_ram.py
@@ -377,3 +377,156 @@ def test_enable_sharing_with_aws_organization_errors():
         "following error message: "
         "You don't have permissions to access this resource."
     )
+
+
+@mock_aws
+def test_get_resource_share_associations_with_principals():
+    # given
+    client = boto3.client("ram", region_name="us-east-1")
+    response = client.create_resource_share(
+        name="test",
+        principals=["123456789012"],
+        resourceArns=[
+            f"arn:aws:ec2:us-east-1:{ACCOUNT_ID}:transit-gateway/tgw-123456789"
+        ],
+    )
+    resource_share_arn = response["resourceShare"]["resourceShareArn"]
+
+    # when
+    response = client.get_resource_share_associations(
+        associationType="PRINCIPAL", resourceShareArns=[resource_share_arn]
+    )
+
+    # then
+    assert len(response["resourceShareAssociations"]) == 1
+    association = response["resourceShareAssociations"][0]
+    assert association["resourceShareArn"] == resource_share_arn
+    assert association["resourceShareName"] == "test"
+    assert association["associatedEntity"] == "123456789012"
+    assert association["associationType"] == "PRINCIPAL"
+    assert association["status"] == "ASSOCIATED"
+    assert isinstance(association["creationTime"], datetime)
+    assert isinstance(association["lastUpdatedTime"], datetime)
+    assert association["external"] is False
+
+
+@mock_aws
+def test_get_resource_share_associations_with_resources():
+    # given
+    client = boto3.client("ram", region_name="us-east-1")
+    resource_arn = f"arn:aws:ec2:us-east-1:{ACCOUNT_ID}:transit-gateway/tgw-123456789"
+    response = client.create_resource_share(
+        name="test",
+        principals=["123456789012"],
+        resourceArns=[resource_arn],
+    )
+    resource_share_arn = response["resourceShare"]["resourceShareArn"]
+
+    # when
+    response = client.get_resource_share_associations(
+        associationType="RESOURCE", resourceShareArns=[resource_share_arn]
+    )
+
+    # then
+    assert len(response["resourceShareAssociations"]) == 1
+    association = response["resourceShareAssociations"][0]
+    assert association["resourceShareArn"] == resource_share_arn
+    assert association["resourceShareName"] == "test"
+    assert association["associatedEntity"] == resource_arn
+    assert association["associationType"] == "RESOURCE"
+    assert association["status"] == "ASSOCIATED"
+    assert isinstance(association["creationTime"], datetime)
+    assert isinstance(association["lastUpdatedTime"], datetime)
+    assert association["external"] is False
+
+
+@mock_aws
+def test_get_resource_share_associations_with_filters():
+    # given
+    client = boto3.client("ram", region_name="us-east-1")
+    resource_arn = f"arn:aws:ec2:us-east-1:{ACCOUNT_ID}:transit-gateway/tgw-123456789"
+    response = client.create_resource_share(
+        name="test",
+        principals=["123456789012"],
+        resourceArns=[resource_arn],
+    )
+
+    # when filtering by principal
+    response = client.get_resource_share_associations(
+        associationType="PRINCIPAL", principal="123456789012"
+    )
+
+    # then
+    assert len(response["resourceShareAssociations"]) == 1
+    assert (
+        response["resourceShareAssociations"][0]["associatedEntity"] == "123456789012"
+    )
+
+    # when filtering by resource
+    response = client.get_resource_share_associations(
+        associationType="RESOURCE", resourceArn=resource_arn
+    )
+
+    # then
+    assert len(response["resourceShareAssociations"]) == 1
+    assert response["resourceShareAssociations"][0]["associatedEntity"] == resource_arn
+
+
+@mock_aws
+def test_get_resource_share_associations_errors():
+    # given
+    client = boto3.client("ram", region_name="us-east-1")
+    resource_arn = f"arn:aws:ec2:us-east-1:{ACCOUNT_ID}:transit-gateway/tgw-123456789"
+    client.create_resource_share(
+        name="test",
+        principals=["123456789012"],
+        resourceArns=[resource_arn],
+    )
+
+    # when invalid association type
+    with pytest.raises(ClientError) as e:
+        client.get_resource_share_associations(associationType="INVALID")
+    ex = e.value
+    assert ex.operation_name == "GetResourceShareAssociations"
+    assert ex.response["ResponseMetadata"]["HTTPStatusCode"] == 400
+    assert "InvalidParameterException" in ex.response["Error"]["Code"]
+    assert "is not a valid association type" in ex.response["Error"]["Message"]
+
+    # when invalid association status
+    with pytest.raises(ClientError) as e:
+        client.get_resource_share_associations(
+            associationType="PRINCIPAL", associationStatus="INVALID"
+        )
+    ex = e.value
+    assert ex.operation_name == "GetResourceShareAssociations"
+    assert ex.response["ResponseMetadata"]["HTTPStatusCode"] == 400
+    assert "InvalidParameterException" in ex.response["Error"]["Code"]
+    assert "is not a valid association status" in ex.response["Error"]["Message"]
+
+    # when resource ARN with PRINCIPAL type
+    with pytest.raises(ClientError) as e:
+        client.get_resource_share_associations(
+            associationType="PRINCIPAL", resourceArn=resource_arn
+        )
+    ex = e.value
+    assert ex.operation_name == "GetResourceShareAssociations"
+    assert ex.response["ResponseMetadata"]["HTTPStatusCode"] == 400
+    assert "InvalidParameterException" in ex.response["Error"]["Code"]
+    assert (
+        "You cannot specify a resource ARN when the association type is PRINCIPAL"
+        in ex.response["Error"]["Message"]
+    )
+
+    # when principal with RESOURCE type
+    with pytest.raises(ClientError) as e:
+        client.get_resource_share_associations(
+            associationType="RESOURCE", principal="123456789012"
+        )
+    ex = e.value
+    assert ex.operation_name == "GetResourceShareAssociations"
+    assert ex.response["ResponseMetadata"]["HTTPStatusCode"] == 400
+    assert "InvalidParameterException" in ex.response["Error"]["Code"]
+    assert (
+        "You cannot specify a principal when the association type is RESOURCE"
+        in ex.response["Error"]["Message"]
+    )