Skip to content

test(ec2): add SG probe/readback rst and refresh SG lifecycle docs #39

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
audreyll2 wants to merge 6 commits into
base: main
Choose a base branch
from
Open

test(ec2): add SG probe/readback rst and refresh SG lifecycle docs #39

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
c43c824
test(ec2): add lifecycle test cases for subnet, internet gateway, and…
audreyll2 Mar 16, 2026
525c0cc
test: add EC2 networking dependency lifecycle rst examples
audreyll2 Mar 20, 2026
9401a96
test: add security group probe script and output log
audreyll2 Mar 27, 2026
0c7c142
Added SG probe/readback rst
audreyll2 Mar 29, 2026
266cf41
Merge branch 'project-vera:main' into tests
audreyll2 Apr 3, 2026
cef7f4c
Fix ENI groupSet serialization and keep local SG patches
audreyll2 Apr 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
144 changes: 125 additions & 19 deletions emulators/aws-ec2/emulator_core/services/elasticnetworkinterface.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -368,8 +368,8 @@ def CreateNetworkInterface(self, params: Dict[str, Any]):
if not sg:
return create_error_response("InvalidGroup.NotFound", f"The ID '{group_id}' does not exist")
group_set.append({
"GroupId": group_id,
"GroupName": getattr(sg, "group_name", ""),
"groupId": group_id,
"groupName": getattr(sg, "group_name", ""),
})

tag_set: List[Dict[str, Any]] = []
Expand Down Expand Up @@ -996,8 +996,8 @@ def ModifyNetworkInterfaceAttribute(self, params: Dict[str, Any]):
if not sg:
return create_error_response("InvalidGroup.NotFound", f"The ID '{group_id}' does not exist")
group_set.append({
"GroupId": group_id,
"GroupName": getattr(sg, "group_name", ""),
"groupId": group_id,
"groupName": getattr(sg, "group_name", ""),
})
network_interface.group_set = group_set

Expand Down Expand Up @@ -1378,6 +1378,27 @@ def _serialize_nested_fields(d: Dict[str, Any], indent_level: int) -> List[str]:
xml_parts.append(f'{indent}<{key}>{esc(str(value))}</{key}>')
return xml_parts


@staticmethod
def _serialize_group_set(group_set: List[Dict[str, Any]], indent_level: int) -> List[str]:
xml_parts = []
indent = ' ' * indent_level
if group_set:
xml_parts.append(f'{indent}<groupSet>')
for item in group_set:
xml_parts.append(f'{indent} <item>')
group_id = item.get("groupId") or item.get("GroupId") or ""
group_name = item.get("groupName") or item.get("GroupName") or ""
if group_id != "":
xml_parts.append(f'{indent} <groupId>{esc(str(group_id))}</groupId>')
if group_name != "":
xml_parts.append(f'{indent} <groupName>{esc(str(group_name))}</groupName>')
xml_parts.append(f'{indent} </item>')
xml_parts.append(f'{indent}</groupSet>')
else:
xml_parts.append(f'{indent}<groupSet/>')
return xml_parts

@staticmethod
def serialize_assign_ipv6_addresses_response(data: Dict[str, Any], request_id: str) -> str:
xml_parts = []
Expand Down Expand Up @@ -1516,6 +1537,26 @@ def serialize_attach_network_interface_response(data: Dict[str, Any], request_id
xml_parts.append(f'</AttachNetworkInterfaceResponse>')
return "\n".join(xml_parts)

@staticmethod
def _serialize_group_set(group_set: List[Dict[str, Any]], indent_level: int) -> List[str]:
xml_parts = []
indent = ' ' * indent_level
if group_set:
xml_parts.append(f'{indent}<groupSet>')
for item in group_set:
xml_parts.append(f'{indent} <item>')
group_id = item.get("groupId") or item.get("GroupId") or ""
group_name = item.get("groupName") or item.get("GroupName") or ""
if group_id != "":
xml_parts.append(f'{indent} <groupId>{esc(str(group_id))}</groupId>')
if group_name != "":
xml_parts.append(f'{indent} <groupName>{esc(str(group_name))}</groupName>')
xml_parts.append(f'{indent} </item>')
xml_parts.append(f'{indent}</groupSet>')
else:
xml_parts.append(f'{indent}<groupSet/>')
return xml_parts

@staticmethod
def serialize_create_network_interface_response(data: Dict[str, Any], request_id: str) -> str:
xml_parts = []
Expand All @@ -1541,10 +1582,44 @@ def serialize_create_network_interface_response(data: Dict[str, Any], request_id
param_data = data[_networkInterface_key]
indent_str = " " * 1
xml_parts.append(f'{indent_str}<networkInterface>')
xml_parts.extend(elasticnetworkinterface_ResponseSerializer._serialize_nested_fields(param_data, 2))
for key, value in param_data.items():
if key == "groupSet":
xml_parts.extend(
elasticnetworkinterface_ResponseSerializer._serialize_group_set(
value, 2
)
)
elif value is None:
continue
elif isinstance(value, dict):
xml_parts.append(f' <{key}>')
xml_parts.extend(
elasticnetworkinterface_ResponseSerializer._serialize_nested_fields(
value, 3
)
)
xml_parts.append(f' </{key}>')
elif isinstance(value, list):
xml_parts.append(f' <{key}>')
for item in value:
if isinstance(item, dict):
xml_parts.append(f' <item>')
xml_parts.extend(
elasticnetworkinterface_ResponseSerializer._serialize_nested_fields(
item, 4
)
)
xml_parts.append(f' </item>')
else:
xml_parts.append(f' <item>{esc(str(item))}</item>')
xml_parts.append(f' </{key}>')
elif isinstance(value, bool):
xml_parts.append(f' <{key}>{str(value).lower()}</{key}>')
else:
xml_parts.append(f' <{key}>{esc(str(value))}</{key}>')
xml_parts.append(f'{indent_str}</networkInterface>')
xml_parts.append(f'</CreateNetworkInterfaceResponse>')
return "\n".join(xml_parts)
return "".join(xml_parts)

@staticmethod
def serialize_create_network_interface_permission_response(data: Dict[str, Any], request_id: str) -> str:
Expand Down Expand Up @@ -1657,16 +1732,11 @@ def serialize_describe_network_interface_attribute_response(data: Dict[str, Any]
_groupSet_key = "Groups"
if _groupSet_key:
param_data = data[_groupSet_key]
indent_str = " " * 1
if param_data:
xml_parts.append(f'{indent_str}<groupSet>')
for item in param_data:
xml_parts.append(f'{indent_str} <item>')
xml_parts.extend(elasticnetworkinterface_ResponseSerializer._serialize_nested_fields(item, 2))
xml_parts.append(f'{indent_str} </item>')
xml_parts.append(f'{indent_str}</groupSet>')
else:
xml_parts.append(f'{indent_str}<groupSet/>')
xml_parts.extend(
elasticnetworkinterface_ResponseSerializer._serialize_group_set(
param_data, 1
)
)
# Serialize networkInterfaceId
_networkInterfaceId_key = None
if "networkInterfaceId" in data:
Expand All @@ -1690,7 +1760,7 @@ def serialize_describe_network_interface_attribute_response(data: Dict[str, Any]
xml_parts.extend(elasticnetworkinterface_ResponseSerializer._serialize_nested_fields(param_data, 2))
xml_parts.append(f'{indent_str}</sourceDestCheck>')
xml_parts.append(f'</DescribeNetworkInterfaceAttributeResponse>')
return "\n".join(xml_parts)
return "".join(xml_parts)

@staticmethod
def serialize_describe_network_interface_permissions_response(data: Dict[str, Any], request_id: str) -> str:
Expand Down Expand Up @@ -1748,7 +1818,43 @@ def serialize_describe_network_interfaces_response(data: Dict[str, Any], request
xml_parts.append(f'{indent_str}<networkInterfaceSet>')
for item in param_data:
xml_parts.append(f'{indent_str} <item>')
xml_parts.extend(elasticnetworkinterface_ResponseSerializer._serialize_nested_fields(item, 2))
for key, value in item.items():
if key == "groupSet":
xml_parts.extend(
elasticnetworkinterface_ResponseSerializer._serialize_group_set(
value, 3
)
)
elif value is None:
continue
elif isinstance(value, dict):
xml_parts.append(f' <{key}>')
xml_parts.extend(
elasticnetworkinterface_ResponseSerializer._serialize_nested_fields(
value, 4
)
)
xml_parts.append(f' </{key}>')
elif isinstance(value, list):
xml_parts.append(f' <{key}>')
for sub_item in value:
if isinstance(sub_item, dict):
xml_parts.append(f' <item>')
xml_parts.extend(
elasticnetworkinterface_ResponseSerializer._serialize_nested_fields(
sub_item, 5
)
)
xml_parts.append(f' </item>')
else:
xml_parts.append(
f' <item>{esc(str(sub_item))}</item>'
)
xml_parts.append(f' </{key}>')
elif isinstance(value, bool):
xml_parts.append(f' <{key}>{str(value).lower()}</{key}>')
else:
xml_parts.append(f' <{key}>{esc(str(value))}</{key}>')
xml_parts.append(f'{indent_str} </item>')
xml_parts.append(f'{indent_str}</networkInterfaceSet>')
else:
Expand All @@ -1764,7 +1870,7 @@ def serialize_describe_network_interfaces_response(data: Dict[str, Any], request
indent_str = " " * 1
xml_parts.append(f'{indent_str}<nextToken>{esc(str(param_data))}</nextToken>')
xml_parts.append(f'</DescribeNetworkInterfacesResponse>')
return "\n".join(xml_parts)
return "".join(xml_parts)

@staticmethod
def serialize_detach_network_interface_response(data: Dict[str, Any], request_id: str) -> str:
Expand Down
Loading