[sso-admin] Implement DescribeAccountAssignmentCreation/DeletionStatus (

#7605)

moto/ssoadmin/models.py

@@ -38,7 +38,9 @@ def __init__(
         self.principal_id = principal_id
         self.created_date = unix_time()
 
-    def to_json(self, include_creation_date: bool = False) -> Dict[str, Any]:
+    def to_json(
+        self, include_creation_date: bool = False, include_request_id: bool = False
+    ) -> Dict[str, Any]:
         summary: Dict[str, Any] = {
             "TargetId": self.target_id,
             "TargetType": self.target_type,
@@ -48,6 +50,8 @@ def to_json(self, include_creation_date: bool = False) -> Dict[str, Any]:
         }
         if include_creation_date:
             summary["CreatedDate"] = self.created_date
+        if include_request_id:
+            summary["RequestId"] = self.request_id
         return summary
 
 
@@ -126,6 +130,7 @@ class SSOAdminBackend(BaseBackend):
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
         self.account_assignments: List[AccountAssignment] = list()
+        self.deleted_account_assignments: List[AccountAssignment] = list()
         self.permission_sets: List[PermissionSet] = list()
         self.aws_managed_policies: Optional[Dict[str, Any]] = None
 
@@ -147,7 +152,7 @@ def create_account_assignment(
             principal_id,
         )
         self.account_assignments.append(assignment)
-        return assignment.to_json()
+        return assignment.to_json(include_creation_date=True, include_request_id=True)
 
     def delete_account_assignment(
         self,
@@ -166,8 +171,9 @@ def delete_account_assignment(
             principal_type,
             principal_id,
         )
+        self.deleted_account_assignments.append(account)
         self.account_assignments.remove(account)
-        return account.to_json(include_creation_date=True)
+        return account.to_json(include_creation_date=True, include_request_id=True)
 
     def _find_account(
         self,
@@ -513,5 +519,27 @@ def detach_customer_managed_policy_reference_from_permission_set(
             customer_managed_policy_reference=customer_managed_policy_reference,
         )
 
+    def describe_account_assignment_creation_status(
+        self, account_assignment_creation_request_id: str, instance_arn: str
+    ) -> Dict[str, Any]:
+        for account in self.account_assignments:
+            if account.request_id == account_assignment_creation_request_id:
+                return account.to_json(
+                    include_creation_date=True, include_request_id=True
+                )
+
+        raise ResourceNotFoundException
+
+    def describe_account_assignment_deletion_status(
+        self, account_assignment_deletion_request_id: str, instance_arn: str
+    ) -> Dict[str, Any]:
+        for account in self.deleted_account_assignments:
+            if account.request_id == account_assignment_deletion_request_id:
+                return account.to_json(
+                    include_creation_date=True, include_request_id=True
+                )
+
+        raise ResourceNotFoundException
+
 
 ssoadmin_backends = BackendDict(SSOAdminBackend, "sso")

moto/ssoadmin/responses.py

@@ -1,7 +1,6 @@
 import json
 
 from moto.core.responses import BaseResponse
-from moto.moto_api._internal import mock_random
 
 from .models import SSOAdminBackend, ssoadmin_backends
 
@@ -34,7 +33,6 @@ def create_account_assignment(self) -> str:
             principal_id=principal_id,
         )
         summary["Status"] = "SUCCEEDED"
-        summary["RequestId"] = str(mock_random.uuid4())
         return json.dumps({"AccountAssignmentCreationStatus": summary})
 
     def delete_account_assignment(self) -> str:
@@ -54,7 +52,6 @@ def delete_account_assignment(self) -> str:
             principal_id=principal_id,
         )
         summary["Status"] = "SUCCEEDED"
-        summary["RequestId"] = str(mock_random.uuid4())
         return json.dumps({"AccountAssignmentDeletionStatus": summary})
 
     def list_account_assignments(self) -> str:
@@ -300,3 +297,31 @@ def detach_customer_managed_policy_reference_from_permission_set(self) -> str:
             customer_managed_policy_reference=customer_managed_policy_reference,
         )
         return json.dumps({})
+
+    def describe_account_assignment_creation_status(self) -> str:
+        account_assignment_creation_request_id = self._get_param(
+            "AccountAssignmentCreationRequestId"
+        )
+        instance_arn = self._get_param("InstanceArn")
+        account_assignment_creation_status = self.ssoadmin_backend.describe_account_assignment_creation_status(
+            account_assignment_creation_request_id=account_assignment_creation_request_id,
+            instance_arn=instance_arn,
+        )
+        account_assignment_creation_status["Status"] = "SUCCEEDED"
+        return json.dumps(
+            dict(AccountAssignmentCreationStatus=account_assignment_creation_status)
+        )
+
+    def describe_account_assignment_deletion_status(self) -> str:
+        account_assignment_deletion_request_id = self._get_param(
+            "AccountAssignmentDeletionRequestId"
+        )
+        instance_arn = self._get_param("InstanceArn")
+        account_assignment_deletion_status = self.ssoadmin_backend.describe_account_assignment_deletion_status(
+            account_assignment_deletion_request_id=account_assignment_deletion_request_id,
+            instance_arn=instance_arn,
+        )
+        account_assignment_deletion_status["Status"] = "SUCCEEDED"
+        return json.dumps(
+            dict(AccountAssignmentDeletionStatus=account_assignment_deletion_status)
+        )

tests/test_ssoadmin/test_ssoadmin.py

@@ -583,3 +583,75 @@ def test_list_permission_sets_pagination():
     )
     for page in page_iterator:
         assert len(page["PermissionSets"]) <= 5
+
+
+@mock_aws
+def test_describe_account_assignment_creation_status():
+    client = boto3.client("sso-admin", region_name="eu-west-1")
+
+    # Test that we can get the account assignment info for existing ones
+    request_id = client.create_account_assignment(
+        InstanceArn="arn:aws:sso:::instance/ins-aaaabbbbccccdddd",
+        PermissionSetArn="arn:aws:sso:::permissionSet/ins-eeeeffffgggghhhh/ps-hhhhkkkkppppoooo",
+        PrincipalType="USER",
+        PrincipalId="some-id",
+        TargetType="AWS_ACCOUNT",
+        TargetId="123123123123",
+    )["AccountAssignmentCreationStatus"]["RequestId"]
+
+    resp = client.describe_account_assignment_creation_status(
+        AccountAssignmentCreationRequestId=request_id,
+        InstanceArn="arn:aws:sso:::instance/ins-aaaabbbbccccdddd",
+    )
+    assert resp["AccountAssignmentCreationStatus"]["Status"] == "SUCCEEDED"
+    assert resp["AccountAssignmentCreationStatus"]["PrincipalId"] == "some-id"
+
+    # Test that non-existent ones raise an exception
+    with pytest.raises(ClientError) as exc:
+        client.describe_account_assignment_creation_status(
+            AccountAssignmentCreationRequestId="non-existent-create-account-assignment-id",
+            InstanceArn="arn:aws:sso:::instance/ins-aaaabbbbccccdddd",
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+
+
+@mock_aws
+def test_describe_account_assignment_deletion_status():
+    client = boto3.client("sso-admin", region_name="eu-west-1")
+
+    # Create & delete an account assignment
+    client.create_account_assignment(
+        InstanceArn="arn:aws:sso:::instance/ins-aaaabbbbccccdddd",
+        PermissionSetArn="arn:aws:sso:::permissionSet/ins-eeeeffffgggghhhh/ps-hhhhkkkkppppoooo",
+        PrincipalType="USER",
+        PrincipalId="some-id",
+        TargetType="AWS_ACCOUNT",
+        TargetId="123123123123",
+    )
+
+    request_id = client.delete_account_assignment(
+        InstanceArn="arn:aws:sso:::instance/ins-aaaabbbbccccdddd",
+        PermissionSetArn="arn:aws:sso:::permissionSet/ins-eeeeffffgggghhhh/ps-hhhhkkkkppppoooo",
+        PrincipalType="USER",
+        PrincipalId="some-id",
+        TargetType="AWS_ACCOUNT",
+        TargetId="123123123123",
+    )["AccountAssignmentDeletionStatus"]["RequestId"]
+
+    # Test that we can get the account assignment info for existing ones
+    resp = client.describe_account_assignment_deletion_status(
+        AccountAssignmentDeletionRequestId=request_id,
+        InstanceArn="arn:aws:sso:::instance/ins-aaaabbbbccccdddd",
+    )
+    assert resp["AccountAssignmentDeletionStatus"]["Status"] == "SUCCEEDED"
+    assert resp["AccountAssignmentDeletionStatus"]["PrincipalId"] == "some-id"
+
+    # Test that non-existent ones raise an exception
+    with pytest.raises(ClientError) as exc:
+        client.describe_account_assignment_deletion_status(
+            AccountAssignmentDeletionRequestId="non-existent-create-account-assignment-id",
+            InstanceArn="arn:aws:sso:::instance/ins-aaaabbbbccccdddd",
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"