forked from mordechai/benchmark-runner
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_oc.py
226 lines (182 loc) · 5.89 KB
/
test_oc.py
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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
# Tests that are not required benchmark-operator pod
import time
import tempfile
import tarfile
import mock
from benchmark_runner.common.oc.oc import OC
from tests.integration.benchmark_runner.test_environment_variables import *
from benchmark_runner.common.prometheus.prometheus_snapshot import PrometheusSnapshot
def test_oc_get_ocp_server_version():
"""
This method gets ocp server version
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
assert oc.get_ocp_server_version()
def test_get_ocp_major_version():
"""
This method gets ocp major version
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
assert oc.get_ocp_major_version()
def test_get_ocp_minor_version():
"""
This method gets ocp minor version
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
assert oc.get_ocp_minor_version()
def test_oc_get_kata_operator_version():
"""
This method gets the sandboxed containers (kata) operator version
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
assert oc.get_kata_operator_version()
def test_oc_get_kata_rpm_version():
"""
This method gets the sandboxed containers (kata) rpm version
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
assert oc.get_kata_rpm_version(node=test_environment_variable['pin_node1'])
assert len(oc.get_kata_rpm_version(node=test_environment_variable['pin_node1']).split('.')) == 3
def test_oc_get_cnv_version():
"""
This method get cnv version
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
assert oc.get_cnv_version()
def test_oc_get_odf_version():
"""
This method get odf version
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
assert oc.get_odf_version()
def test_oc_get_pv_disk_ids():
"""
This method test get pv disk ids
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
assert len(oc.get_pv_disk_ids()) > 1
assert oc.get_pv_disk_ids()
def mock_get_worker_disk_ids(*args, **kwargs):
"""
This method mock method class get_worker_disk_ids
"""
return ['scsi-3600062b33333333333333333333333333']
def test_oc_get_free_disk_id():
"""
This method gets free_disk_id string
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
with mock.patch.object(OC, 'get_worker_disk_ids', new=mock_get_worker_disk_ids):
assert oc.get_free_disk_id()
def test_oc_get_odf_disk_count():
"""
This method gets odf disk count
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
odf_disk_count = oc.get_odf_disk_count()
assert (odf_disk_count is None or odf_disk_count > 0)
def test_oc_get_master_nodes():
"""
This method test get master nodes
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
assert oc.get_master_nodes()
def test_login():
"""
This method test login
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
assert oc.login()
def test_oc_get_pod_name():
"""
This test run oc get pod by name
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
assert oc._get_pod_name(pod_name='erererer', namespace=test_environment_variable['namespace']) == ''
def test_oc_get_pods():
"""
This test run oc get pods
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
assert oc.get_pods()
def test_get_prom_token():
"""
This method return prom token from cluster
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
assert oc.get_prom_token()
def test_is_cnv_installed():
"""
This method check if cnv operator is installed
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
assert oc.is_cnv_installed()
def test_is_kata_installed():
"""
This method checks if the sandboxed containers (kata) operator is installed
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
assert oc.is_kata_installed()
def test_is_odf_installed():
"""
This method check if odf operator is installed
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
assert oc.is_odf_installed()
def test_oc_exec():
"""
Test that oc exec works
:return:
"""
test_message = "I am here"
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
answer = oc.exec(pod_name="prometheus-k8s-0", namespace="openshift-monitoring", container='prometheus',
command=f'echo "{test_message}"')
assert answer == test_message
def test_collect_prometheus():
"""
Test that Prometheus data can be collected. TBD test that data is valid.
:return:
"""
oc = OC(kubeadmin_password=test_environment_variable['kubeadmin_password'])
oc.login()
with tempfile.TemporaryDirectory() as dirname:
snapshot = PrometheusSnapshot(oc=oc, artifacts_path=dirname, verbose=True)
snapshot.prepare_for_snapshot(pre_wait_time=1)
time.sleep(10)
tarball = snapshot.retrieve_snapshot(post_wait_time=1)
assert tarfile.is_tarfile(tarball)