File: test_eks_utils.py

package info (click to toggle)
python-moto 5.1.18-3
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 116,520 kB
  • sloc: python: 636,725; javascript: 181; makefile: 39; sh: 3
file content (135 lines) | stat: -rw-r--r-- 4,343 bytes parent folder | download | duplicates (2)
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