1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135
|
from copy import deepcopy
from random import randint
try:
from urllib.parse import urlparse
except ImportError:
from urlparse import urlparse
from moto.moto_api._internal import mock_random
from tests.test_eks.test_eks_constants import (
STATUS,
ClusterAttributes,
ClusterInputs,
FargateProfileAttributes,
FargateProfileInputs,
NodegroupAttributes,
NodegroupInputs,
ResponseAttributes,
)
generate_random_name = mock_random.get_random_string
def attributes_to_test(inputs, name):
"""
Assembles the list of tuples which will be used to validate test results.
"""
result = deepcopy(inputs.REQUIRED + inputs.OPTIONAL + [STATUS])
if isinstance(inputs, ClusterInputs):
result += [(ClusterAttributes.NAME, name)]
elif isinstance(inputs, NodegroupInputs):
result += [(NodegroupAttributes.NODEGROUP_NAME, name)]
return result
def generate_clusters(client, num_clusters, minimal):
"""
Generates 'num_clusters' number of clusters with randomized data and adds them to the mocked backend.
If 'minimal' is True, only the required values are generated; if False all values are generated.
Returns a list of the names of the generated clusters.
"""
return [
client.create_cluster(
name=generate_random_name(), **_input_builder(ClusterInputs, minimal)
)[ResponseAttributes.CLUSTER][ClusterAttributes.NAME]
for _ in range(num_clusters)
]
def generate_fargate_profiles(client, cluster_name, num_profiles, minimal):
"""
Generates 'num_profiles' number of fargate profiles with randomized data and adds them to
the mocked backend.
If 'minimal' is True, only the required values are generated; if False, all values are generated.
Returns a list of the names of the generated Fargate profiles.
"""
return [
client.create_fargate_profile(
fargateProfileName=generate_random_name(),
clusterName=cluster_name,
**_input_builder(FargateProfileInputs, minimal),
)[ResponseAttributes.FARGATE_PROFILE][
FargateProfileAttributes.FARGATE_PROFILE_NAME
]
for _ in range(num_profiles)
]
def generate_nodegroups(client, cluster_name, num_nodegroups, minimal):
"""
Generates 'num_nodegroups' number of nodegroups with randomized data and adds them to the mocked backend.
If 'minimal' is True, only the required values are generated; if False, all values are generated.
Returns a list of the names of the generated nodegroups.
"""
return [
client.create_nodegroup(
nodegroupName=generate_random_name(),
clusterName=cluster_name,
**_input_builder(NodegroupInputs, minimal),
)[ResponseAttributes.NODEGROUP][NodegroupAttributes.NODEGROUP_NAME]
for _ in range(num_nodegroups)
]
def generate_dict(prefix, count):
return {f"{prefix}_{_count}": str(_count) for _count in range(count)}
def is_valid_uri(value):
"""
Returns true if a provided string has the form of a valid uri.
"""
result = urlparse(value)
return all([result.scheme, result.netloc, result.path])
def region_matches_partition(region, partition):
"""
Returns True if the provided region and partition are a valid pair.
"""
valid_matches = [
("cn-", "aws-cn"),
("us-gov-", "aws-us-gov"),
("us-gov-iso-", "aws-iso"),
("us-gov-iso-b-", "aws-iso-b"),
]
for prefix, expected_partition in valid_matches:
if region.startswith(prefix):
return partition == expected_partition
return partition == "aws"
def _input_builder(options, minimal):
"""
Assembles the inputs which will be used to generate test object into a dictionary.
"""
values = deepcopy(options.REQUIRED)
if not minimal:
values.extend(deepcopy(options.OPTIONAL))
return dict(values)
def random_names(name_list):
"""
Returns one value picked at random a list, and one value guaranteed not to be on the list.
"""
name_on_list = name_list[randint(0, len(name_list) - 1)]
name_not_on_list = generate_random_name()
while name_not_on_list in name_list:
name_not_on_list = generate_random_name()
return name_on_list, name_not_on_list
|