Repository: getmoto/moto
Pull request #7026 (2023-11-15 11:41:38)

LakeFormation: Add Parameters for list_permissions()
This implements the key behavior of filtering the results of lakeformation's list_permissions operation based on the passed parameters.

Implemented filtering by:
```
CatalogId, Principal, ResourceType, Resource
```
where:
```
ResourceType = [DATABASE, TABLE, CATALOG, DATA_LOCATION]
```
and 
```
Resource = [
    {
        "Database": {
            "CatalogId": "string", 
            "Name": "string"
        }
    },
    {
        "Table": {
            "CatalogId": "string", 
            "DatabaseName": "string"
            "Name": "string",
            "TableWildcard": {}
        }
    },
    {
        "Catalog": {}
    },
]
```

Used reference documents for behaviour:
- https://boto3.amazonaws.com/v1/documentation/api/1.26.83/reference/services/lakeformation/client/list_permissions.html
- https://docs.aws.amazon.com/cli/latest/reference/lakeformation/list-permissions.html

Note: 
I could not get all preexisting tests running on my system even though I followed the guidelines. (https://docs.getmoto.org/en/latest/docs/contributing/installation.html)
Newly implemented tests ran successfully though.



Patch:
diff --git a/moto/core/responses.py b/moto/core/responses.py
index 71dec337f..5d27e60bd 100644
--- a/moto/core/responses.py
+++ b/moto/core/responses.py
@@ -18,7 +18,7 @@ from moto.core.utils import (
     method_names_from_class,
     params_sort_function,
 )
-from moto.utilities.utils import load_resource
+from moto.utilities.utils import load_resource, load_resource_as_bytes
 from jinja2 import Environment, DictLoader, Template
 from typing import (
     Dict,
@@ -961,7 +961,12 @@ class AWSServiceSpec(object):
     """
 
     def __init__(self, path: str):
-        spec = load_resource("botocore", path)
+        try:
+            spec = load_resource("botocore", path)
+        except FileNotFoundError:
+            # botocore >= 1.32.1 sends compressed files
+            compressed = load_resource_as_bytes("botocore", f"{path}.gz")
+            spec = json.loads(gzip_decompress(compressed).decode("utf-8"))
 
         self.metadata = spec["metadata"]
         self.operations = spec["operations"]
diff --git a/moto/ecs/models.py b/moto/ecs/models.py
index cc10d2746..41f122be7 100644
--- a/moto/ecs/models.py
+++ b/moto/ecs/models.py
@@ -437,9 +437,10 @@ class Task(BaseObject, ManagedState):
             return f"arn:aws:ecs:{self.region_name}:{self._account_id}:task/{self.cluster_name}/{self.id}"
         return f"arn:aws:ecs:{self.region_name}:{self._account_id}:task/{self.id}"
 
-    @property
-    def response_object(self) -> Dict[str, Any]:  # type: ignore[misc]
+    def response_object(self, include_tags: bool = True) -> Dict[str, Any]:  # type: ignore
         response_object = self.gen_response_object()
+        if not include_tags:
+            response_object.pop("tags", None)
         response_object["taskArn"] = self.task_arn
         response_object["lastStatus"] = self.last_status
         return response_object
@@ -1471,12 +1472,7 @@ class EC2ContainerServiceBackend(BaseBackend):
             self.tasks[cluster.name][task.task_arn] = task
         return tasks
 
-    def describe_tasks(
-        self,
-        cluster_str: str,
-        tasks: Optional[str],
-        include: Optional[List[str]] = None,
-    ) -> List[Task]:
+    def describe_tasks(self, cluster_str: str, tasks: Optional[str]) -> List[Task]:
         """
         Only include=TAGS is currently supported.
         """
@@ -1495,22 +1491,18 @@ class EC2ContainerServiceBackend(BaseBackend):
                 ):
                     task.advance()
                     response.append(task)
-        if "TAGS" in (include or []):
-            return response
 
-        for task in response:
-            task.tags = []
         return response
 
     def list_tasks(
         self,
-        cluster_str: str,
-        container_instance: Optional[str],
-        family: str,
-        started_by: str,
-        service_name: str,
-        desiredStatus: str,
-    ) -> List[str]:
+        cluster_str: Optional[str] = None,
+        container_instance: Optional[str] = None,
+        family: Optional[str] = None,
+        started_by: Optional[str] = None,
+        service_name: Optional[str] = None,
+        desiredStatus: Optional[str] = None,
+    ) -> List[Task]:
         filtered_tasks = []
         for tasks in self.tasks.values():
             for task in tasks.values():
@@ -1554,7 +1546,7 @@ class EC2ContainerServiceBackend(BaseBackend):
                 filter(lambda t: t.desired_status == desiredStatus, filtered_tasks)
             )
 
-        return [t.task_arn for t in filtered_tasks]
+        return filtered_tasks
 
     def stop_task(self, cluster_str: str, task_str: str, reason: str) -> Task:
         cluster = self._get_cluster(cluster_str)
@@ -2080,6 +2072,10 @@ class EC2ContainerServiceBackend(BaseBackend):
             return task_def
         elif parsed_arn["service"] == "capacity-provider":
             return self._get_provider(parsed_arn["id"])
+        elif parsed_arn["service"] == "task":
+            for task in self.list_tasks():
+                if task.task_arn == resource_arn:
+                    return task
         raise NotImplementedError()
 
     def list_tags_for_resource(self, resource_arn: str) -> List[Dict[str, str]]:
diff --git a/moto/ecs/responses.py b/moto/ecs/responses.py
index 402aba9d4..40fa7278f 100644
--- a/moto/ecs/responses.py
+++ b/moto/ecs/responses.py
@@ -193,16 +193,19 @@ class EC2ContainerServiceResponse(BaseResponse):
             network_configuration,
         )
         return json.dumps(
-            {"tasks": [task.response_object for task in tasks], "failures": []}
+            {"tasks": [task.response_object() for task in tasks], "failures": []}
         )
 
     def describe_tasks(self) -> str:
         cluster = self._get_param("cluster", "default")
         tasks = self._get_param("tasks")
-        include = self._get_param("include")
-        data = self.ecs_backend.describe_tasks(cluster, tasks, include)
+        include_tags = "TAGS" in self._get_param("include", [])
+        data = self.ecs_backend.describe_tasks(cluster, tasks)
         return json.dumps(
-            {"tasks": [task.response_object for task in data], "failures": []}
+            {
+                "tasks": [task.response_object(include_tags) for task in data],
+                "failures": [],
+            }
         )
 
     def start_task(self) -> str:
@@ -221,7 +224,7 @@ class EC2ContainerServiceResponse(BaseResponse):
             tags,
         )
         return json.dumps(
-            {"tasks": [task.response_object for task in tasks], "failures": []}
+            {"tasks": [task.response_object() for task in tasks], "failures": []}
         )
 
     def list_tasks(self) -> str:
@@ -231,7 +234,7 @@ class EC2ContainerServiceResponse(BaseResponse):
         started_by = self._get_param("startedBy")
         service_name = self._get_param("serviceName")
         desiredStatus = self._get_param("desiredStatus")
-        task_arns = self.ecs_backend.list_tasks(
+        tasks = self.ecs_backend.list_tasks(
             cluster_str,
             container_instance,
             family,
@@ -239,14 +242,14 @@ class EC2ContainerServiceResponse(BaseResponse):
             service_name,
             desiredStatus,
         )
-        return json.dumps({"taskArns": task_arns})
+        return json.dumps({"taskArns": [t.task_arn for t in tasks]})
 
     def stop_task(self) -> str:
         cluster_str = self._get_param("cluster", "default")
         task = self._get_param("task")
         reason = self._get_param("reason")
         task = self.ecs_backend.stop_task(cluster_str, task, reason)
-        return json.dumps({"task": task.response_object})
+        return json.dumps({"task": task.response_object()})
 
     def create_service(self) -> str:
         cluster_str = self._get_param("cluster", "default")
diff --git a/moto/lakeformation/exceptions.py b/moto/lakeformation/exceptions.py
index 4e57f24cf..a0e3b4a17 100644
--- a/moto/lakeformation/exceptions.py
+++ b/moto/lakeformation/exceptions.py
@@ -5,3 +5,8 @@ from moto.core.exceptions import JsonRESTError
 class EntityNotFound(JsonRESTError):
     def __init__(self) -> None:
         super().__init__("EntityNotFoundException", "Entity not found")
+
+
+class InvalidInput(JsonRESTError):
+    def __init__(self, message: str) -> None:
+        super().__init__("InvalidInputException", message)
diff --git a/moto/lakeformation/models.py b/moto/lakeformation/models.py
index b5141dad4..0930720e0 100644
--- a/moto/lakeformation/models.py
+++ b/moto/lakeformation/models.py
@@ -1,9 +1,18 @@
+from __future__ import annotations
+from enum import Enum
 from collections import defaultdict
-from typing import Any, Dict, List, Tuple
+from typing import Any, Dict, List, Tuple, Optional
 
 from moto.core import BaseBackend, BackendDict, BaseModel
 from moto.utilities.tagging_service import TaggingService
-from .exceptions import EntityNotFound
+from .exceptions import EntityNotFound, InvalidInput
+
+
+class RessourceType(Enum):
+    catalog = "CATALOG"
+    database = "DATABASE"
+    table = "TABLE"
+    data_location = "DATA_LOCATION"
 
 
 class Resource(BaseModel):
@@ -18,6 +27,117 @@ class Resource(BaseModel):
         }
 
 
+class ListPermissionsResourceDatabase:
+    def __init__(self, catalog_id: Optional[str], name: str):
+        self.name = name
+        self.catalog_id = catalog_id
+
+
+class ListPermissionsResourceTable:
+    def __init__(
+        self,
+        catalog_id: Optional[str],
+        database_name: str,
+        name: Optional[str],
+        table_wildcard: Optional[
+            Dict[str, str]
+        ],  # Placeholder type, table_wildcard is an empty dict in docs
+    ):
+        if name is None and table_wildcard is None:
+            raise InvalidInput("Table name and table wildcard cannot both be empty.")
+        if name is not None and table_wildcard is not None:
+            raise InvalidInput("Table name and table wildcard cannot both be present.")
+        self.database_name = database_name
+        self.name = name
+        self.catalog_id = catalog_id
+        self.table_wildcard = table_wildcard
+
+
+class ExcludedColumnNames:
+    def __init__(self, excluded_column_names: List[str]):
+        self.excluded_column_names = excluded_column_names
+
+
+class ListPermissionsResourceTableWithColumns:
+    def __init__(
+        self,
+        catalog_id: Optional[str],
+        database_name: str,
+        name: str,
+        column_names: list[str],
+        column_wildcard: ExcludedColumnNames,
+    ):
+        self.database_name = database_name
+        self.name = name
+        self.catalog_id = catalog_id
+        self.column_names = column_names
+        self.column_wildcard = column_wildcard
+
+
+class ListPermissionsResourceDataLocation:
+    def __init__(self, catalog_id: Optional[str], resource_arn: str):
+        self.catalog_id = catalog_id
+        self.resource_arn = resource_arn
+
+
+class ListPermissionsResourceDataCellsFilter:
+    def __init__(
+        self, table_catalog_id: str, database_name: str, table_name: str, name: str
+    ):
+        self.table_catalog_id = table_catalog_id
+        self.database_name = database_name
+        self.table_name = table_name
+        self.name = name
+
+
+class ListPermissionsResourceLFTag:
+    def __init__(self, catalog_id: str, tag_key: str, tag_values: List[str]):
+        self.catalog_id = catalog_id
+        self.tag_key = tag_key
+        self.tag_values = tag_values
+
+
+class LFTag:
+    def __init__(self, tag_key: str, tag_values: List[str]):
+        self.tag_key = tag_key
+        self.tag_values = tag_values
+
+
+class ListPermissionsResourceLFTagPolicy:
+    def __init__(self, catalog_id: str, resource_type: str, expression: List[LFTag]):
+        self.catalog_id = catalog_id
+        self.resource_type = resource_type
+        self.expression = expression
+
+
+class ListPermissionsResource:
+    def __init__(
+        self,
+        catalog: Optional[
+            Dict[str, str]
+        ],  # Placeholder type, catalog is an empty dict in docs
+        database: Optional[ListPermissionsResourceDatabase],
+        table: Optional[ListPermissionsResourceTable],
+        table_with_columns: Optional[ListPermissionsResourceTableWithColumns],
+        data_location: Optional[ListPermissionsResourceDataLocation],
+        data_cells_filter: Optional[ListPermissionsResourceDataCellsFilter],
+        lf_tag: Optional[ListPermissionsResourceLFTag],
+        lf_tag_policy: Optional[ListPermissionsResourceLFTagPolicy],
+    ):
+        if catalog is None and database is None and table is None:
+            raise InvalidInput(
+                "Resource must have either the catalog, table or database field populated."
+            )
+        self.catalog = catalog
+        self.database = database
+        self.table = table
+        self.table_with_columns = table_with_columns
+        self.data_location = data_location
+        self.data_cells_filter = data_cells_filter
+        self.lf_tag = lf_tag
+        self.lf_tag_policy = lf_tag_policy
+
+
 def default_settings() -> Dict[str, Any]:
     return {
         "DataLakeAdmins": [],
@@ -114,11 +234,92 @@ class LakeFormationBackend(BaseBackend):
             grant for grant in self.grants[catalog_id] if grant["Permissions"] != []
         ]
 
-    def list_permissions(self, catalog_id: str) -> List[Dict[str, Any]]:
+    def list_permissions(
+        self,
+        catalog_id: str,
+        principal: Optional[Dict[str, str]] = None,
+        resource: Optional[ListPermissionsResource] = None,
+        resource_type: Optional[RessourceType] = None,
+    ) -> List[Dict[str, Any]]:
         """
-        No parameters have been implemented yet
+        No pagination has been implemented yet.
         """
-        return self.grants[catalog_id]
+        permissions = self.grants[catalog_id]
+
+        def filter_for_principal(permission: Dict[str, Any]) -> bool:
+            return permission["Principal"] == principal
+
+        if principal is not None:
+            permissions = list(filter(filter_for_principal, permissions))
+
+        def filter_for_resource_type(permission: Dict[str, Any]) -> bool:
+            if resource_type is None:  # Check for mypy
+                return False
+            resource = permission["Resource"]
+            if resource_type == RessourceType.catalog:
+                return "Catalog" in resource
+            elif resource_type == RessourceType.database:
+                return "Database" in resource
+            elif resource_type == RessourceType.data_location:
+                return "DataLocation" in resource
+            elif resource_type == RessourceType.table:
+                return "Table" in resource or "TableWithColumns" in resource
+            return False
+
+        if resource_type is not None:
+            permissions = list(filter(filter_for_resource_type, permissions))
+
+        def filter_for_resource(permission: Dict[str, Any]) -> bool:
+            """
+            If catalog is provided:
+                only matching permissions with resource-type "Catalog" are returned;
+            if catalog is not provided and database is provided:
+                only matching permissions with resource-type "Database" are returned;
+            if catalog and database are not provided and table is provided:
+                only matching permissions with resource-type "Table" are returned;
+            """
+            if resource is None:  # Check for linter
+                return False
+
+            permission_resource = permission["Resource"]
+            catalog = resource.catalog
+            if catalog is not None and "Catalog" in permission_resource:
+                return catalog == permission_resource["Catalog"]
+
+            database = resource.database
+            if database is not None and "Database" in permission_resource:
+                equals = database.name == permission_resource["Database"]["Name"]
+                if database.catalog_id is not None:
+                    equals = equals and (
+                        database.catalog_id
+                        == permission_resource["Database"]["CatalogId"]
+                    )
+                return equals
+
+            table = resource.table
+            if table is not None and "Table" in permission_resource:
+                equals = (
+                    table.database_name == permission_resource["Table"]["DatabaseName"]
+                )
+                if table.catalog_id is not None:
+                    equals = equals and (
+                        table.catalog_id == permission_resource["Table"]["CatalogId"]
+                    )
+                if table.name is not None and table.table_wildcard is None:
+                    equals = equals and (
+                        table.name == permission_resource["Table"]["Name"]
+                    )
+                if table.name is None and table.table_wildcard is not None:
+                    equals = equals and (
+                        table.table_wildcard
+                        == permission_resource["Table"]["TableWildcard"]
+                    )
+                return equals
+            return False
+
+        if resource is not None:
+            permissions = list(filter(filter_for_resource, permissions))
+        return permissions
 
     def create_lf_tag(self, catalog_id: str, key: str, values: List[str]) -> None:
         # There is no ARN that we can use, so just create another  unique identifier that's easy to recognize and reproduce
diff --git a/moto/lakeformation/responses.py b/moto/lakeformation/responses.py
index 628bbb5ba..c544c1e7f 100644
--- a/moto/lakeformation/responses.py
+++ b/moto/lakeformation/responses.py
@@ -3,7 +3,14 @@ import json
 from typing import Any, Dict
 
 from moto.core.responses import BaseResponse
-from .models import lakeformation_backends, LakeFormationBackend
+from .models import (
+    lakeformation_backends,
+    LakeFormationBackend,
+    ListPermissionsResource,
+    ListPermissionsResourceDatabase,
+    ListPermissionsResourceTable,
+    RessourceType,
+)
 
 
 class LakeFormationResponse(BaseResponse):
@@ -87,7 +94,55 @@ class LakeFormationResponse(BaseResponse):
 
     def list_permissions(self) -> str:
         catalog_id = self._get_param("CatalogId") or self.current_account
-        permissions = self.lakeformation_backend.list_permissions(catalog_id)
+        principal = self._get_param("Principal")
+        resource = self._get_param("Resource")
+        resource_type_param = self._get_param("ResourceType")
+        if resource_type_param is None:
+            resource_type = None
+        else:
+            resource_type = RessourceType(resource_type_param)
+
+        if resource is None:
+            list_permission_resource = None
+        else:
+            database_sub_dictionary = resource.get("Database")
+            table_sub_dictionary = resource.get("Table")
+            catalog_sub_dictionary = resource.get("Catalog")
+
+            if database_sub_dictionary is None:
+                database = None
+            else:
+                database = ListPermissionsResourceDatabase(
+                    name=database_sub_dictionary.get("Name"),
+                    catalog_id=database_sub_dictionary.get("CatalogId"),
+                )
+
+            if table_sub_dictionary is None:
+                table = None
+            else:
+                table = ListPermissionsResourceTable(
+                    database_name=table_sub_dictionary.get("DatabaseName"),
+                    name=table_sub_dictionary.get("Name"),
+                    catalog_id=table_sub_dictionary.get("CatalogId"),
+                    table_wildcard=table_sub_dictionary.get("TableWildcard"),
+                )
+
+            list_permission_resource = ListPermissionsResource(
+                catalog=catalog_sub_dictionary,
+                database=database,
+                table=table,
+                table_with_columns=None,
+                data_location=None,
+                data_cells_filter=None,
+                lf_tag=None,
+                lf_tag_policy=None,
+            )
+        permissions = self.lakeformation_backend.list_permissions(
+            catalog_id=catalog_id,
+            principal=principal,
+            resource=list_permission_resource,
+            resource_type=resource_type,
+        )
         return json.dumps({"PrincipalResourcePermissions": permissions})
 
     def create_lf_tag(self) -> str:


Test patch:
diff --git a/tests/test_ecs/test_ecs_boto3.py b/tests/test_ecs/test_ecs_boto3.py
index e88d3dff2..7b76b0a6b 100644
--- a/tests/test_ecs/test_ecs_boto3.py
+++ b/tests/test_ecs/test_ecs_boto3.py
@@ -2481,42 +2481,6 @@ def test_describe_tasks_empty_tags():
     assert len(response["tasks"]) == 1
 
 
-@mock_ec2
-@mock_ecs
-def test_describe_tasks_include_tags():
-    client = boto3.client("ecs", region_name=ECS_REGION)
-    test_cluster_name = "test_ecs_cluster"
-    setup_ecs_cluster_with_ec2_instance(client, test_cluster_name)
-
-    task_tags = [{"key": "Name", "value": "test_ecs_task"}]
-    tasks_arns = [
-        task["taskArn"]
-        for task in client.run_task(
-            cluster="test_ecs_cluster",
-            overrides={},
-            taskDefinition="test_ecs_task",
-            count=2,
-            startedBy="moto",
-            tags=task_tags,
-        )["tasks"]
-    ]
-    response = client.describe_tasks(
-        cluster="test_ecs_cluster", tasks=tasks_arns, include=["TAGS"]
-    )
-
-    assert len(response["tasks"]) == 2
-    assert set(
-        [response["tasks"][0]["taskArn"], response["tasks"][1]["taskArn"]]
-    ) == set(tasks_arns)
-    assert response["tasks"][0]["tags"] == task_tags
-
-    # Test we can pass task ids instead of ARNs
-    response = client.describe_tasks(
-        cluster="test_ecs_cluster", tasks=[tasks_arns[0].split("/")[-1]]
-    )
-    assert len(response["tasks"]) == 1
-
-
 @mock_ecs
 def test_describe_tasks_exceptions():
     client = boto3.client("ecs", region_name=ECS_REGION)
diff --git a/tests/test_ecs/test_ecs_task_tags.py b/tests/test_ecs/test_ecs_task_tags.py
new file mode 100644
index 000000000..9b8d08c65
--- /dev/null
+++ b/tests/test_ecs/test_ecs_task_tags.py
@@ -0,0 +1,71 @@
+import boto3
+
+from moto import mock_ec2, mock_ecs
+from .test_ecs_boto3 import setup_ecs_cluster_with_ec2_instance
+
+
+@mock_ec2
+@mock_ecs
+def test_describe_tasks_include_tags():
+    client = boto3.client("ecs", region_name="us-east-1")
+    test_cluster_name = "test_ecs_cluster"
+    setup_ecs_cluster_with_ec2_instance(client, test_cluster_name)
+
+    task_tags = [{"key": "Name", "value": "test_ecs_task"}]
+    tasks_arns = [
+        task["taskArn"]
+        for task in client.run_task(
+            cluster="test_ecs_cluster",
+            taskDefinition="test_ecs_task",
+            count=2,
+            tags=task_tags,
+        )["tasks"]
+    ]
+    response = client.describe_tasks(
+        cluster="test_ecs_cluster", tasks=tasks_arns, include=["TAGS"]
+    )
+
+    assert len(response["tasks"]) == 2
+    assert set(
+        [response["tasks"][0]["taskArn"], response["tasks"][1]["taskArn"]]
+    ) == set(tasks_arns)
+    assert response["tasks"][0]["tags"] == task_tags
+
+    # Test we can pass task ids instead of ARNs
+    response = client.describe_tasks(
+        cluster="test_ecs_cluster", tasks=[tasks_arns[0].split("/")[-1]]
+    )
+    assert len(response["tasks"]) == 1
+
+    tags = client.list_tags_for_resource(resourceArn=tasks_arns[0])["tags"]
+    assert tags == task_tags
+
+
+@mock_ec2
+@mock_ecs
+def test_add_tags_to_task():
+    client = boto3.client("ecs", region_name="us-east-1")
+    test_cluster_name = "test_ecs_cluster"
+    setup_ecs_cluster_with_ec2_instance(client, test_cluster_name)
+
+    task_tags = [{"key": "k1", "value": "v1"}]
+    task_arn = client.run_task(
+        cluster="test_ecs_cluster",
+        taskDefinition="test_ecs_task",
+        count=1,
+        tags=task_tags,
+    )["tasks"][0]["taskArn"]
+
+    client.tag_resource(resourceArn=task_arn, tags=[{"key": "k2", "value": "v2"}])
+
+    tags = client.describe_tasks(
+        cluster="test_ecs_cluster", tasks=[task_arn], include=["TAGS"]
+    )["tasks"][0]["tags"]
+    assert len(tags) == 2
+    assert {"key": "k1", "value": "v1"} in tags
+    assert {"key": "k2", "value": "v2"} in tags
+
+    client.untag_resource(resourceArn=task_arn, tagKeys=["k2"])
+
+    resp = client.list_tags_for_resource(resourceArn=task_arn)
+    assert resp["tags"] == [{"key": "k1", "value": "v1"}]
diff --git a/tests/test_lakeformation/test_lakeformation.py b/tests/test_lakeformation/test_lakeformation.py
index 69ad2d68e..03e6ad42f 100644
--- a/tests/test_lakeformation/test_lakeformation.py
+++ b/tests/test_lakeformation/test_lakeformation.py
@@ -1,4 +1,5 @@
 """Unit tests for lakeformation-supported APIs."""
+from typing import Dict
 import boto3
 import pytest
 
@@ -117,6 +118,230 @@ def test_list_permissions():
     ]
 
 
+def grant_table_permissions(
+    client: boto3.client, catalog_id: str, principal: str, db: str, table: str
+):
+    client.grant_permissions(
+        CatalogId=catalog_id,
+        Principal={"DataLakePrincipalIdentifier": principal},
+        Resource={
+            "Table": {
+                "DatabaseName": db,
+                "Name": table,
+                "CatalogId": catalog_id,
+                "TableWildcard": {},
+            }
+        },
+        Permissions=["ALL"],
+        PermissionsWithGrantOption=["SELECT"],
+    )
+
+
+def grant_db_permissions(
+    client: boto3.client, catalog_id: str, principal: str, db: str
+):
+    client.grant_permissions(
+        CatalogId=catalog_id,
+        Principal={"DataLakePrincipalIdentifier": principal},
+        Resource={"Database": {"Name": db, "CatalogId": catalog_id}},
+        Permissions=["ALL"],
+        PermissionsWithGrantOption=["SELECT"],
+    )
+
+
+def grant_catalog_permissions(client: boto3.client, catalog_id: str, principal: str):
+    client.grant_permissions(
+        CatalogId=catalog_id,
+        Principal={"DataLakePrincipalIdentifier": principal},
+        Resource={"Catalog": {}},
+        Permissions=["CREATE_DATABASE"],
+        PermissionsWithGrantOption=[],
+    )
+
+
+def db_response(principal: str, catalog_id: str, db: str) -> Dict:
+    return {
+        "Principal": {"DataLakePrincipalIdentifier": principal},
+        "Resource": {
+            "Database": {
+                "CatalogId": catalog_id,
+                "Name": db,
+            }
+        },
+        "Permissions": ["ALL"],
+        "PermissionsWithGrantOption": ["SELECT"],
+    }
+
+
+def table_response(principal: str, catalog_id: str, db: str, table: str) -> Dict:
+    return {
+        "Principal": {"DataLakePrincipalIdentifier": principal},
+        "Resource": {
+            "Table": {
+                "CatalogId": catalog_id,
+                "DatabaseName": db,
+                "Name": table,
+                "TableWildcard": {},
+            }
+        },
+        "Permissions": ["ALL"],
+        "PermissionsWithGrantOption": ["SELECT"],
+    }
+
+
+def catalog_response(principal: str) -> Dict:
+    return {
+        "Principal": {"DataLakePrincipalIdentifier": principal},
+        "Resource": {"Catalog": {}},
+        "Permissions": ["CREATE_DATABASE"],
+        "PermissionsWithGrantOption": [],
+    }
+
+
+@mock_lakeformation
+def test_list_permissions_filtered_for_catalog_id():
+    client = boto3.client("lakeformation", region_name="eu-west-2")
+    catalog_id_1 = "000000000000"
+    catalog_id_2 = "000000000001"
+    principal_1 = "principal_1"
+    principal_2 = "principal_2"
+
+    grant_catalog_permissions(
+        client=client, catalog_id=catalog_id_1, principal=principal_1
+    )
+    grant_catalog_permissions(
+        client=client, catalog_id=catalog_id_2, principal=principal_2
+    )
+
+    permissions = client.list_permissions(CatalogId=catalog_id_1)[
+        "PrincipalResourcePermissions"
+    ]
+    assert permissions == [catalog_response(principal=principal_1)]
+
+    permissions = client.list_permissions(CatalogId=catalog_id_2)[
+        "PrincipalResourcePermissions"
+    ]
+    assert permissions == [catalog_response(principal=principal_2)]
+
+
+@mock_lakeformation
+def test_list_permissions_filtered_for_resource_type():
+    client = boto3.client("lakeformation", region_name="eu-west-2")
+    catalog_id = "000000000000"
+    principal_1 = "principal_1"
+    db_1 = "db_1"
+    table_1 = "table_1"
+
+    grant_db_permissions(
+        client=client, catalog_id=catalog_id, principal=principal_1, db=db_1
+    )
+    grant_table_permissions(
+        client=client,
+        catalog_id=catalog_id,
+        principal=principal_1,
+        db=db_1,
+        table=table_1,
+    )
+    grant_catalog_permissions(
+        client=client, catalog_id=catalog_id, principal=principal_1
+    )
+
+    res = client.list_permissions(CatalogId=catalog_id, ResourceType="DATABASE")
+    assert res["PrincipalResourcePermissions"] == [
+        db_response(principal=principal_1, catalog_id=catalog_id, db=db_1),
+    ]
+
+    res = client.list_permissions(CatalogId=catalog_id, ResourceType="TABLE")
+    assert res["PrincipalResourcePermissions"] == [
+        table_response(
+            principal=principal_1, catalog_id=catalog_id, db=db_1, table=table_1
+        ),
+    ]
+
+    res = client.list_permissions(CatalogId=catalog_id, ResourceType="CATALOG")
+    assert res["PrincipalResourcePermissions"] == [
+        catalog_response(principal=principal_1),
+    ]
+
+
+@mock_lakeformation
+def test_list_permissions_filtered_for_resource_db():
+    client = boto3.client("lakeformation", region_name="eu-west-2")
+    catalog_id = "000000000000"
+    principal_1 = "principal_1"
+    db_1 = "db_1"
+    db_2 = "db_2"
+
+    grant_db_permissions(
+        client=client, catalog_id=catalog_id, principal=principal_1, db=db_1
+    )
+    grant_db_permissions(
+        client=client, catalog_id=catalog_id, principal=principal_1, db=db_2
+    )
+
+    res = client.list_permissions(
+        CatalogId=catalog_id, Resource={"Database": {"Name": db_1}}
+    )
+
+    assert res["PrincipalResourcePermissions"] == [
+        db_response(principal=principal_1, catalog_id=catalog_id, db=db_1),
+    ]
+
+    res = client.list_permissions(
+        CatalogId=catalog_id, Resource={"Database": {"Name": db_2}}
+    )
+    assert res["PrincipalResourcePermissions"] == [
+        db_response(principal=principal_1, catalog_id=catalog_id, db=db_2),
+    ]
+
+
+@mock_lakeformation
+def test_list_permissions_filtered_for_resource_table():
+    client = boto3.client("lakeformation", region_name="eu-west-2")
+    catalog_id = "000000000000"
+    principal_1 = "principal_1"
+    db_1 = "db_1"
+    table_1 = "table_1"
+    table_2 = "table_2"
+
+    grant_table_permissions(
+        client=client,
+        catalog_id=catalog_id,
+        principal=principal_1,
+        db=db_1,
+        table=table_1,
+    )
+    grant_table_permissions(
+        client=client,
+        catalog_id=catalog_id,
+        principal=principal_1,
+        db=db_1,
+        table=table_2,
+    )
+
+    res = client.list_permissions(
+        CatalogId=catalog_id,
+        Resource={"Table": {"DatabaseName": db_1, "Name": table_1}},
+    )
+
+    assert res["PrincipalResourcePermissions"] == [
+        table_response(
+            principal=principal_1, catalog_id=catalog_id, db=db_1, table=table_1
+        ),
+    ]
+
+    res = client.list_permissions(
+        CatalogId=catalog_id,
+        Resource={"Table": {"DatabaseName": db_1, "Name": table_2}},
+    )
+
+    assert res["PrincipalResourcePermissions"] == [
+        table_response(
+            principal=principal_1, catalog_id=catalog_id, db=db_1, table=table_2
+        ),
+    ]
+
+
 @mock_lakeformation
 def test_revoke_permissions():
     client = boto3.client("lakeformation", region_name="eu-west-2")