EFS: Add support for describing access points using File system ID filter (#7043)

This commit is contained in:
Ryan McGinty 2023-11-18 12:09:06 -08:00 committed by GitHub
parent e1738d7363
commit 281bd79c90
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 39 additions and 2 deletions

View File

@ -698,7 +698,11 @@ class EFSBackend(BaseBackend):
self.access_points[access_point.access_point_id] = access_point
return access_point
def describe_access_points(self, access_point_id: str) -> List[AccessPoint]:
def describe_access_points(
self,
access_point_id: Optional[str],
file_system_id: Optional[str],
) -> List[AccessPoint]:
"""
Pagination is not yet implemented
"""
@ -706,6 +710,12 @@ class EFSBackend(BaseBackend):
if access_point_id not in self.access_points:
raise AccessPointNotFound(access_point_id)
return [self.access_points[access_point_id]]
elif file_system_id:
return [
access_point
for access_point in self.access_points.values()
if access_point.file_system_id == file_system_id
]
return list(self.access_points.values())
def delete_access_point(self, access_point_id: str) -> None:

View File

@ -170,7 +170,10 @@ class EFSResponse(BaseResponse):
def describe_access_points(self) -> TYPE_RESPONSE:
access_point_id = self._get_param("AccessPointId")
access_points = self.efs_backend.describe_access_points(access_point_id)
file_system_id = self._get_param("FileSystemId")
access_points = self.efs_backend.describe_access_points(
access_point_id, file_system_id
)
resp = [ap.info_json() for ap in access_points]
return json.dumps({"AccessPoints": resp}), {"Content-Type": "application/json"}

View File

@ -100,6 +100,30 @@ def test_describe_access_points__multiple(efs, file_system):
assert len(resp["AccessPoints"]) == 2
def test_describe_access_points__filters(efs):
fs_id1 = efs.create_file_system(CreationToken="foobarbaz")["FileSystemId"]
fs_id2 = efs.create_file_system(CreationToken="bazbarfoo")["FileSystemId"]
ap_id1 = efs.create_access_point(ClientToken="ct", FileSystemId=fs_id1)[
"AccessPointId"
]
ap_id2 = efs.create_access_point(ClientToken="ct", FileSystemId=fs_id2)[
"AccessPointId"
]
resp = efs.describe_access_points(FileSystemId=fs_id1)
assert len(resp["AccessPoints"]) == 1
assert resp.get("NextToken") is None
assert resp["AccessPoints"][0]["FileSystemId"] == fs_id1
assert resp["AccessPoints"][0]["AccessPointId"] == ap_id1
resp = efs.describe_access_points(AccessPointId=ap_id2)
assert len(resp["AccessPoints"]) == 1
assert resp.get("NextToken") is None
assert resp["AccessPoints"][0]["FileSystemId"] == fs_id2
assert resp["AccessPoints"][0]["AccessPointId"] == ap_id2
def test_delete_access_points(efs, file_system):
fs_id = file_system["FileSystemId"]