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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
|
import lib.test_const as consts
import re
import lib._metric_resolvers as mr
CANON_SB_VAULT_REGEX = re.compile(r"\w+=(value|file):[-\w]+:\w+")
CANON_YAV_REGEX = re.compile(r"\w+=(value|file):sec-[a-z0-9]+:\w+")
VALID_DNS_REQUIREMENTS = ("default", "local", "dns64")
VALID_NETWORK_REQUIREMENTS = ("full", "restricted")
def check_cpu(suite_cpu_requirements, test_size, is_kvm=False):
min_cpu_requirements = consts.TestRequirementsConstants.MinCpu
max_cpu_requirements = consts.TestSize.get_max_requirements(test_size).get(consts.TestRequirements.Cpu)
if isinstance(suite_cpu_requirements, str):
if all(
consts.TestRequirementsConstants.is_all_cpu(req) for req in (max_cpu_requirements, suite_cpu_requirements)
):
return None
return "Wrong 'cpu' requirements: {}, should be in [{}..{}] for {}-size tests".format(
suite_cpu_requirements, min_cpu_requirements, max_cpu_requirements, test_size
)
if not isinstance(suite_cpu_requirements, int):
return "Wrong 'cpu' requirements: {}, should be integer".format(suite_cpu_requirements)
if (
suite_cpu_requirements < min_cpu_requirements
or suite_cpu_requirements > consts.TestRequirementsConstants.get_cpu_value(max_cpu_requirements)
):
return "Wrong 'cpu' requirement: {}, should be in [{}..{}] for {}-size tests".format(
suite_cpu_requirements, min_cpu_requirements, max_cpu_requirements, test_size
)
return None
# TODO: Remove is_kvm param when there will be guarantees on RAM
def check_ram(suite_ram_requirements, test_size, is_kvm=False):
if not isinstance(suite_ram_requirements, int):
return "Wrong 'ram' requirements: {}, should be integer".format(suite_ram_requirements)
min_ram_requirements = consts.TestRequirementsConstants.MinRam
max_ram_requirements = (
consts.MAX_RAM_REQUIREMENTS_FOR_KVM
if is_kvm
else consts.TestSize.get_max_requirements(test_size).get(consts.TestRequirements.Ram)
)
if suite_ram_requirements < min_ram_requirements or suite_ram_requirements > max_ram_requirements:
err_msg = "Wrong 'ram' requirements: {}, should be in [{}..{}] for {}-size tests".format(
suite_ram_requirements, min_ram_requirements, max_ram_requirements, test_size
)
if is_kvm:
err_msg += ' with kvm requirements'
return err_msg
return None
def check_ram_disk(suite_ram_disk, test_size, is_kvm=False):
min_ram_disk = consts.TestRequirementsConstants.MinRamDisk
max_ram_disk = consts.TestSize.get_max_requirements(test_size).get(consts.TestRequirements.RamDisk)
if isinstance(suite_ram_disk, str):
if all(consts.TestRequirementsConstants.is_all_ram_disk(req) for req in (max_ram_disk, suite_ram_disk)):
return None
return "Wrong 'ram_disk' requirements: {}, should be in [{}..{}] for {}-size tests".format(
suite_ram_disk, 0, max_ram_disk, test_size
)
if not isinstance(suite_ram_disk, int):
return "Wrong 'ram_disk' requirements: {}, should be integer".format(suite_ram_disk)
if suite_ram_disk < min_ram_disk or suite_ram_disk > consts.TestRequirementsConstants.get_ram_disk_value(
max_ram_disk
):
return "Wrong 'ram_disk' requirement: {}, should be in [{}..{}] for {}-size tests".format(
suite_ram_disk, min_ram_disk, max_ram_disk, test_size
)
return None
def validate_sb_vault(name, value):
if not CANON_SB_VAULT_REGEX.match(value):
return "sb_vault value '{}' should follow pattern <ENV_NAME>=<value|file>:<owner>:<vault key>".format(value)
def validate_yav_vault(name, value):
if not CANON_YAV_REGEX.match(value):
return "yav value '{}' should follow pattern <ENV_NAME>=<value|file>:<sec-id>:<key>".format(value)
def validate_numerical_requirement(name, value):
if mr.resolve_value(value) is None:
return "Cannot convert [[imp]]{}[[rst]] to the proper [[imp]]{}[[rst]] requirement value".format(value, name)
def validate_choice_requirement(name, val, valid):
if val not in valid:
return "Unknown [[imp]]{}[[rst]] requirement: [[imp]]{}[[rst]], choose from [[imp]]{}[[rst]]".format(
name, val, ", ".join(valid)
)
def validate_force_sandbox_requirement(
name, value, test_size, is_force_sandbox, in_autocheck, is_fuzzing, is_kvm, is_ytexec_run, check_func
):
if is_force_sandbox or not in_autocheck or is_fuzzing or is_ytexec_run:
if value == 'all':
return
return validate_numerical_requirement(name, value)
error_msg = validate_numerical_requirement(name, value)
if error_msg:
return error_msg
return check_func(mr.resolve_value(value), test_size, is_kvm)
def validate_ram_disk_requirement(
name, value, test_size, is_force_sandbox, in_autocheck, is_fuzzing, is_kvm, is_ytexec_run, ram
):
error_msg = validate_force_sandbox_requirement(
name, value, test_size, is_force_sandbox, in_autocheck, is_fuzzing, is_kvm, is_ytexec_run, check_ram_disk
)
if error_msg:
return error_msg
if is_force_sandbox or not in_autocheck or test_size == consts.TestSize.Large:
return
if int(value) > int(ram):
return "Wrong 'ram_disk' value, 'ram_disk':{} should be no more than 'ram':{}".format(value, ram)
return None
# TODO: Remove is_kvm param when there will be guarantees on RAM
def validate_requirement(
req_name, value, test_size, is_force_sandbox, in_autocheck, is_fuzzing, is_kvm, is_ytexec_run, requirements
):
req_checks = {
'container': validate_numerical_requirement,
'cpu': lambda n, v: validate_force_sandbox_requirement(
n, v, test_size, is_force_sandbox, in_autocheck, is_fuzzing, is_kvm, is_ytexec_run, check_cpu
),
'disk_usage': validate_numerical_requirement,
'dns': lambda n, v: validate_choice_requirement(n, v, VALID_DNS_REQUIREMENTS),
'kvm': None,
'network': lambda n, v: validate_choice_requirement(n, v, VALID_NETWORK_REQUIREMENTS),
'ram': lambda n, v: validate_force_sandbox_requirement(
n, v, test_size, is_force_sandbox, in_autocheck, is_fuzzing, is_kvm, is_ytexec_run, check_ram
),
'ram_disk': lambda n, v: validate_ram_disk_requirement(
n,
v,
test_size,
is_force_sandbox,
in_autocheck,
is_fuzzing,
is_kvm,
is_ytexec_run,
requirements.get(
'ram', consts.TestSize.get_default_requirements(test_size).get(consts.TestRequirements.Ram)
),
),
'sb': None,
'sb_vault': validate_sb_vault,
'yav': validate_yav_vault,
}
if req_name not in req_checks:
return "Unknown requirement: [[imp]]{}[[rst]], choose from [[imp]]{}[[rst]]".format(
req_name, ", ".join(sorted(req_checks))
)
if req_name in ('container', 'disk') and not is_force_sandbox:
return "Only [[imp]]LARGE[[rst]] tests without [[imp]]ya:force_distbuild[[rst]] tag can have [[imp]]{}[[rst]] requirement".format(
req_name
)
check_func = req_checks[req_name]
if check_func:
return check_func(req_name, value)
|
