/
promoter_test.py
247 lines (211 loc) · 9.45 KB
/
promoter_test.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
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
#!/usr/bin/env python
"""
This script tests the steps of the promoter workflow.
- Checks the dlrn API that the hash under test has been promoted
to the promotion target
- Checks that containers with that hash are pushed to repo 2
- Checks that images are uploaded with that hash and linked to
promotion target
- Checks the promoter logs for expected strings
"""
import argparse
import dlrnapi_client
import docker
import logging
import os
import re
try:
import urllib2 as url_lib
except ImportError:
import urllib.request as url_lib
import yaml
def get_full_hash(commit_hash, distro_hash):
return "{}_{}".format(commit_hash, distro_hash[:8])
def check_dlrn_promoted_hash(stage_info):
''' Check that the commit, distro hash has been promoted to
promotion_target as recorded in DLRN. '''
dlrn_host = stage_info['dlrn_host']
promotion_target = stage_info['promotion_target']
commit_hash = stage_info['promotions']['promotion_candidate']['commit_hash']
distro_hash = stage_info['promotions']['promotion_candidate']['distro_hash']
logger = logging.getLogger('TestPromoter')
api_client = dlrnapi_client.ApiClient(host=dlrn_host)
dlrn = dlrnapi_client.DefaultApi(api_client=api_client)
params = dlrnapi_client.PromotionQuery()
params.commit_hash = commit_hash
params.distro_hash = distro_hash
try:
api_response = dlrn.api_promotions_get(params)
logger.debug(api_response)
except dlrnapi_client.rest.ApiException:
logger.error('Exception when calling api_promotions_get: %s',
dlrnapi_client.rest.ApiException)
raise
error_message = ("Expected commit hash: {}"
" has not been promoted to {}."
"".format(commit_hash, promotion_target))
conditions = [(promotion.promote_name == promotion_target)
for promotion in api_response]
assert any(conditions), error_message
def query_container_registry_promotion(stage_info):
''' Check that the hash containers have been pushed to the
promotion registry with the promotion_target tag. '''
# TODO(gcerami) Retain the possibility to specify custom values easily
registry_source = stage_info['registries']['source']['host']
registry_target = stage_info['registries']['targets'][0]['host']
promotion_target = stage_info['promotion_target']
full_hash = stage_info['promotions']['promotion_candidate']['full_hash']
missing_images = []
if 'localhost' in registry_source:
for line in stage_info['containers']:
# TODO(gcerami) we should check that manifests are there, and
# contain the proper information
name, tag = line.split(":")
reg_url = "http://{}/v2/{}/manifests/{}".format(
registry_target, name, tag
)
print("Checking for promoted container hash: " + reg_url)
try:
url_lib.urlopen(reg_url)
except url_lib.HTTPError:
print("Image not found - " + line)
missing_images.append(line)
# For the full_hash lines only, check that there is
# an equivalent promotion_target entry
if tag == full_hash:
reg_url = "http://{}/v2/{}/manifests/{}".format(
registry_target, name, promotion_target
)
print("Checking for promoted container tag: " + reg_url)
try:
url_lib.urlopen(reg_url)
except url_lib.HTTPError:
print("Image with named tag not found - " + line)
promo_tgt_line = line.replace(full_hash, promotion_target)
missing_images.append(promo_tgt_line)
else:
# TODO: how to verify promoter containers
print("Compare images tagged with hash and promotion target:")
assert missing_images == [], "Images are missing"
def compare_tagged_image_hash(stage_info):
''' Ensure that the promotion target images directory
is a soft link to the promoted full hash images directory. '''
images_base_dir = stage_info['overcloud_images']['base_dir']
user = stage_info['overcloud_images']['user']
key_path = stage_info['overcloud_images']['key_path']
distro = stage_info['distro']
distro_version = stage_info['distro_version']
release = stage_info['release']
promotion_target = stage_info['promotion_target']
full_hash = stage_info['promotions']['promotion_candidate']['full_hash']
if 'promoter-staging' not in images_base_dir:
print("Install required for nonstaging env")
import pysftp
sftp = pysftp.Connection(
host=images_base_dir,
username=user, private_key=key_path)
images_dir = os.path.join(
'/var/www/html/images',
release, 'rdo_trunk')
rl_module = sftp
else:
# Check that the promotion_target dir is a soft link
distro_full = distro + str(distro_version)
images_dir = os.path.join(
images_base_dir,
distro_full, release, 'rdo_trunk')
rl_module = os
error_message = "Promotion target dir is not a softlink"
full_hash_path = os.path.join(images_dir, full_hash)
print("Promotion target is: " + os.path.join(images_dir, promotion_target))
print("Full hash path is: " + full_hash_path)
promoted_hash_path = rl_module.readlink(
os.path.join(images_dir, promotion_target))
assert full_hash_path == promoted_hash_path, error_message
def parse_promotion_logs(stage_info):
''' Check that the promotion logs have the right
strings printed for the promotion status '''
logfile = stage_info['logfile']
release = stage_info['release']
promotion_target = stage_info['promotion_target']
full_hash = stage_info['promotions']['promotion_candidate']['full_hash']
commit_hash = stage_info['promotions']['promotion_candidate']['commit_hash']
distro_hash = stage_info['promotions']['promotion_candidate']['distro_hash']
candidate_name = stage_info['promotions']['promotion_candidate']['name']
logger = logging.getLogger('TestPromoter')
logger.debug("Open promoter file for reading")
full_hash = get_full_hash(commit_hash, distro_hash)
# Check if the logfile passed is web hosted
if 'http' in logfile:
from bs4 import BeautifulSoup
logger.debug("Reading web hosted log file")
url = url_lib.request.urlopen(logfile).read()
soup = BeautifulSoup(url, 'html.parser')
logfile_contents = soup.get_text()
else:
logger.debug("Reading local log file")
with open(logfile, 'r') as lf:
logfile_contents = lf.read()
# Check that the promoter process finished
error_message = "Promoter never finished"
assert 'promoter FINISHED' in logfile_contents, error_message
# We have a list of hashes at our disposal, we know which one
# will have to fail, and which one will have to pass
# We can do all in the same pass
success_pattern_container = re.compile(
r'promoter Promoting the container images for dlrn hash '
+ re.escape(commit_hash))
success_pattern_images = re.compile(
r'Promoting the qcow image for dlrn hash '
+ re.escape(full_hash) + r' on '
+ re.escape(release) + r' to '
+ re.escape(promotion_target))
success_pattern = re.compile(
r"Successful jobs for {'timestamp': (\d+), 'distro_hash': '"
+ re.escape(distro_hash)
+ r"', (.*) 'full_hash': '"
+ re.escape(full_hash)
+ r"', 'repo_hash': '"
+ re.escape(full_hash) + r"', 'commit_hash': '"
+ re.escape(commit_hash) + r"'}")
success_pattern_target = re.compile(
"promoter SUCCESS promoting "
+ re.escape(candidate_name) + r" as "
+ re.escape(promotion_target))
success_patterns = [
success_pattern,
success_pattern_images,
success_pattern_container,
success_pattern_target,
]
for commit in stage_info['commits']:
promotion_candidate = stage_info['promotions']['promotion_candidate']
if commit['full_hash'] == promotion_candidate['full_hash']:
# This commit is supposed succeed
# Check strings for passing hashes
print("Status Passing: " + commit['full_hash'])
# Build pattern for successful promotion
for check_pattern in success_patterns:
success_pattern_search = check_pattern.search(logfile_contents)
error_message = "Pattern not found- %s" % check_pattern.pattern
assert success_pattern_search is not None, error_message
def main():
logging.basicConfig(level=logging.DEBUG)
log = logging.getLogger("TestPromoter")
log.setLevel(logging.DEBUG)
parser = argparse.ArgumentParser(
description='Pass a config file.')
parser.add_argument('--stage-info-file', default="/tmp/stage-info.yaml")
args = parser.parse_args()
with open(args.stage_info_file) as si:
stage_info = yaml.safe_load(si)
print('Running test: check_dlrn_promoted_hash')
check_dlrn_promoted_hash(stage_info)
print('Running test: query_container_registry_promotion')
query_container_registry_promotion(stage_info)
print('Running test: compare_tagged_image_hash')
compare_tagged_image_hash(stage_info)
print('Running test: parse_promotion_logs')
parse_promotion_logs(stage_info)
if __name__ == "__main__":
main()