Integ-tests: EFS utils

Signed-off-by: Hanwen <[email protected]>

Author: hanwen-cluster

tests/integration-tests/tests/storage/storage_common.py

@@ -208,6 +208,10 @@ def check_fsx(
             assert_fsx_correctly_shared(scheduler_commands, remote_command_executor, mount_dir)
 
 
+def get_efs_ids(cluster, region):
+    return retrieve_cfn_outputs(cluster.cfn_name, region).get("EFSIds").split(",")
+
+
 def get_fsx_ids(cluster, region):
     return retrieve_cfn_outputs(cluster.cfn_name, region).get("FSXIds").split(",")
 

tests/integration-tests/tests/storage/test_efs.py

@@ -18,12 +18,12 @@
 from remote_command_executor import RemoteCommandExecutor
 from troposphere import Ref, Template, ec2
 from troposphere.efs import MountTarget
-from utils import generate_stack_name, get_vpc_snakecase_value
+from utils import generate_stack_name, get_vpc_snakecase_value, get_compute_nodes_instance_ips
 
 from tests.storage.storage_common import (
     test_efs_correctly_mounted,
     verify_directory_correctly_shared,
-    write_file_into_efs,
+    write_file_into_efs, get_efs_ids,
 )
 
 
@@ -130,6 +130,8 @@ def test_multiple_efs(
 
     run_benchmarks(remote_command_executor, scheduler_commands)
 
+    _test_efs_utils(remote_command_executor, scheduler_commands, cluster, region, all_mount_dirs, get_efs_ids(cluster, region))
+
 
 def _add_mount_targets(subnet_ids, efs_ids, security_group, template):
     subnet_response = boto3.client("ec2").describe_subnets(SubnetIds=subnet_ids)["Subnets"]
@@ -230,3 +232,21 @@ def _assert_subnet_az_relations(region, vpc_stack, expected_in_same_az):
         assert_that(head_node_subnet_az).is_equal_to(compute_subnet_az)
     else:
         assert_that(head_node_subnet_az).is_not_equal_to(compute_subnet_az)
+
+
+def _test_efs_utils(remote_command_executor, scheduler_commands, cluster, region, mount_dirs, efs_ids):
+    compute_node_remote_command_executors = []
+    for compute_node_ip in get_compute_nodes_instance_ips(cluster.name, region):
+        compute_node_remote_command_executors(RemoteCommandExecutor(cluster, compute_node_ip=compute_node_ip))
+    for mount_dir in mount_dirs:
+        remote_command_executor.run_remote_command(f"sudo umount {mount_dir}")
+        for compute_node_remote_command_executor in compute_node_remote_command_executors:
+            compute_node_remote_command_executor.run_remote_command(f"sudo umount {mount_dir}")
+    assert_that(mount_dirs).is_length(len(efs_ids))
+    for mount_dir, efs_id in zip(mount_dirs, efs_ids):
+        remote_command_executor.run_remote_command(f"sudo mount -t efs -o tls {efs_id}:/ {mount_dir}")
+        for compute_node_remote_command_executor in compute_node_remote_command_executors:
+            compute_node_remote_command_executor.run_remote_command(f"sudo mount -t efs -o tls {efs_id}:/ {mount_dir}")
+        _test_efs_correctly_shared(remote_command_executor, mount_dir, scheduler_commands)
+    for mount_dir in mount_dirs:
+        test_efs_correctly_mounted(remote_command_executor, mount_dir)

tests/integration-tests/utils.py

@@ -214,6 +214,20 @@ def get_cluster_nodes_instance_ids(stack_name, region, instance_types=None, node
         raise
 
 
+def get_compute_nodes_instance_ips(stack_name, region):
+    """Return a list of compute Instances Ip's."""
+    try:
+        instances = _describe_cluster_instances(
+            stack_name,
+            region,
+            filter_by_node_type="Compute",
+        )
+        return [instance["PrivateIpAddress"] for instance in instances]
+    except Exception as e:
+        logging.error("Failed retrieving instance pds with exception: %s", e)
+        raise
+
+
 def _describe_cluster_instances(
     stack_name,
     region,