code: def test_create_sg_multinode(iam_client_stub, ec2_client_stub):
      """
      Test AWS Bootstrap logic when config being bootstrapped has the
      following properties:
  
      (1) auth config does not specify ssh key path
      (2) available_node_types is provided
      (3) security group name and ip permissions set in provider field
      (4) Available node types have SubnetIds field set and this
          field is of form SubnetIds: [subnet-xxxxx].
          Both node types specify the same subnet-xxxxx.
  
      Tests creation of a security group and key pair under these conditions.
      """
  
      # Generate a config of the desired form.
      subnet_id = DEFAULT_SUBNET["SubnetId"]
  
      # security group info to go in provider field
      provider_data = helpers.load_aws_example_config_file(
          "example-security-group.yaml"
      )["provider"]
  
      # a multi-node-type config -- will add head/worker stuff and security group
      # info to this.
      base_config = helpers.load_aws_example_config_file(
          "example-full.yaml"
      )
  
      config = copy.deepcopy(base_config)
      # Add security group data
      config["provider"] = provider_data
      # Add head and worker fields.
      head_node_config = config["available_node_types"][
          "ray.head.default"
      ]["node_config"]
      worker_node_config = config["available_node_types"][
          "ray.worker.default"
      ]["node_config"]
      head_node_config["SubnetIds"] = [subnet_id]
      worker_node_config["SubnetIds"] = [subnet_id]
  
      # Generate stubs
      stubs.configure_iam_role_default(iam_client_stub)
      stubs.configure_key_pair_default(ec2_client_stub)
  
      # Only one of these (the one specified in the available_node_types)
      # is in the correct vpc.
      # This list of subnets is generated by the ec2.subnets.all() call
      # and then ignored, since available_node_types already specify
      # subnet_ids.
      stubs.describe_a_thousand_subnets_in_different_vpcs(
          ec2_client_stub
      )
  
      # The rest of the stubbing logic is copied from
      # test_create_sg_with_custom_inbound_rules_and_name.
  
      # expect to describe the head subnet ID
      stubs.describe_subnets_echo(ec2_client_stub, [DEFAULT_SUBNET])
      # given no existing security groups within the VPC...
      stubs.describe_no_security_groups(ec2_client_stub)
      # expect to create a security group on the head node VPC
      stubs.create_sg_echo(ec2_client_stub, DEFAULT_SG_WITH_NAME)
      # expect new head security group details to be retrieved after creation
      stubs.describe_sgs_on_vpc(
          ec2_client_stub,
          [DEFAULT_SUBNET["VpcId"]],
          [DEFAULT_SG_WITH_NAME],
      )
  
      # given custom existing default head security group inbound rules...
      # expect to authorize both default and custom inbound rules
      stubs.authorize_sg_ingress(
          ec2_client_stub,
          DEFAULT_SG_WITH_NAME_AND_RULES,
      )
  
      # given the prior modification to the head security group...
      # expect the next read of a head security group property to reload it
      stubs.describe_sg_echo(
          ec2_client_stub, DEFAULT_SG_WITH_NAME_AND_RULES
      )
  
      _get_subnets_or_die.cache_clear()
  
      # given our mocks and the config as input...
      # expect the config to be validated and bootstrapped successfully
      bootstrapped_config = helpers.bootstrap_aws_config(config)
  
      # expect the bootstrapped config to have the custom security group...
      # name and in bound rules
      assert (
          bootstrapped_config["provider"]["security_group"]["GroupName"]
          == DEFAULT_SG_WITH_NAME_AND_RULES["GroupName"]
      )
      assert (
          bootstrapped_config["provider"]["security_group"][
              "IpPermissions"
          ]
          == CUSTOM_IN_BOUND_RULES
      )
  
      # Confirming correct security group got filled for head and workers
      sg_id = DEFAULT_SG["GroupId"]
      for node_type in bootstrapped_config[
          "available_node_types"
      ].values():
          node_config = node_type["node_config"]
          assert node_config["SecurityGroupIds"] == [sg_id]
  
      # Confirming boostrap config updates available node types with
      # default KeyName
      for node_type in bootstrapped_config[
          "available_node_types"
      ].values():
          node_config = node_type["node_config"]
          assert node_config["KeyName"] == DEFAULT_KEY_PAIR["KeyName"]
  
      # Confirm security group is in the right VPC.
      # (Doesn't really confirm anything except for the structure of this test
      # data.)
      bootstrapped_head_type = bootstrapped_config["head_node_type"]
      bootstrapped_types = bootstrapped_config["available_node_types"]
      bootstrapped_head_config = bootstrapped_types[
          bootstrapped_head_type
      ]["node_config"]
      assert DEFAULT_SG["VpcId"] == DEFAULT_SUBNET["VpcId"]
      assert (
          DEFAULT_SUBNET["SubnetId"]
          == bootstrapped_head_config["SubnetIds"][0]
      )
  
      # ssh private key filled in
      assert "ssh_private_key" in bootstrapped_config["auth"]
  
      # expect no pending responses left in IAM or EC2 client stub queues
      iam_client_stub.assert_no_pending_responses()
      ec2_client_stub.assert_no_pending_responses()
  
imports: import copy
  from unittest.mock import Mock, patch
  import pytest
  from click.exceptions import ClickException
  import ray.tests.aws.utils.helpers
  import ray.tests.aws.utils.stubs
  from ray.autoscaler._private.aws.config import DEFAULT_AMI, _configure_subnet, _get_subnets_or_die, bootstrap_aws, log_to_cli
  from ray.autoscaler._private.aws.node_provider import AWSNodeProvider
  from ray.autoscaler._private.providers import _get_node_provider
  from ray.tests.aws.utils.constants import AUX_SG, AUX_SUBNET, CUSTOM_IN_BOUND_RULES, DEFAULT_CLUSTER_NAME, DEFAULT_INSTANCE_PROFILE, DEFAULT_KEY_PAIR, DEFAULT_LT, DEFAULT_SG, DEFAULT_SG_AUX_SUBNET, DEFAULT_SG_DUAL_GROUP_RULES, DEFAULT_SG_WITH_NAME, DEFAULT_SG_WITH_NAME_AND_RULES, DEFAULT_SG_WITH_RULES, DEFAULT_SG_WITH_RULES_AUX_SUBNET, DEFAULT_SUBNET
label_desc: Use image node_id 7 for calling a global function with example usage: test_create_sg_multinode(iam_client_stub, ec2_client_stub) without return types