-
Notifications
You must be signed in to change notification settings - Fork 20
/
pc_helper.py
186 lines (163 loc) · 6.83 KB
/
pc_helper.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
import bs4
import logging
import re
from functools import lru_cache
import requests
logger = logging.getLogger("bot.openqabot.pc_helper")
def request_get(url):
"""
Do a HTTP get request. Return the request object for further processing
Raises a ValueError when the request fails
"""
req = requests.get(url)
if req.status_code != 200:
raise ValueError("http status code %d" % (req.status_code))
return req
def fetch_matching_link(url, regex):
"""
Apply odering by modification date (ascending) and return the first link that matches the given regex
"""
try:
# Note: ?C=M;O=A - C - compare , M - modify time , O - order , A - asc
# So, the first link matching the regex is the most recent one
req = requests.get(url + "/?C=M;O=A")
text = req.text
except (
requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
ValueError,
) as err:
logger.error("error fetching '%s': %s" % (url, err))
return None
getpage_soup = bs4.BeautifulSoup(text, "html.parser")
# Returns lazy iterator, so
links = getpage_soup.findAll("a", href=regex)
if links:
return f"{url}/{links[0].get('href')}"
raise ValueError("No matching links found")
# Gets the latest image from the given URL/regex
# image is of the format 'URL/regex'
def get_latest_pc_image(image):
basepath, _, regex = image.rpartition("/")
return fetch_matching_link(basepath, re.compile(regex))
def get_latest_tools_image(query):
# 'publiccloud_tools_<BUILD NUM>.qcow2' is generic name for image used by Public Cloud tests to run
# in openQA. query suppose to look like this "https://openqa.suse.de/group_overview/276.json" to get
# value for <BUILD NUM>
## Get the first not-failing item
build_results = request_get(query).json()["build_results"]
for build in build_results:
if build["failed"] == 0:
return "publiccloud_tools_{}.qcow2".format(build["build"])
return None
# Applies PUBLIC_CLOUD_IMAGE_LOCATION based on the given PUBLIC_CLOUD_IMAGE_REGEX
def apply_publiccloud_regex(settings):
try:
settings["PUBLIC_CLOUD_IMAGE_LOCATION"] = get_latest_pc_image(
settings["PUBLIC_CLOUD_IMAGE_REGEX"]
)
return settings
except (
requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
ValueError,
re.error,
) as e:
logger.warning(f"PUBLIC_CLOUD_IMAGE_REGEX handling failed: {e}")
settings["PUBLIC_CLOUD_IMAGE_LOCATION"] = None
return settings
def apply_pc_tools_image(settings):
try:
if "PUBLIC_CLOUD_TOOLS_IMAGE_QUERY" in settings:
settings["PUBLIC_CLOUD_TOOLS_IMAGE_BASE"] = get_latest_tools_image(
settings["PUBLIC_CLOUD_TOOLS_IMAGE_QUERY"]
)
del settings["PUBLIC_CLOUD_TOOLS_IMAGE_QUERY"]
return settings
except (
requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
ValueError,
re.error,
) as e:
logger.warning(f"PUBLIC_CLOUD_TOOLS_IMAGE_BASE handling failed: {e}")
return settings
# Perform a pint query. Sucessive queries are cached
@lru_cache(maxsize=32)
def pint_query(query):
return request_get(query).json()
# Applies PUBLIC_CLOUD_IMAGE_LOCATION based on the given PUBLIC_CLOUD_IMAGE_REGEX
def apply_publiccloud_pint_image(settings):
try:
images = pint_query(settings["PUBLIC_CLOUD_PINT_QUERY"])["images"]
region = (
settings["PUBLIC_CLOUD_PINT_REGION"]
if "PUBLIC_CLOUD_PINT_REGION" in settings
else None
)
# We need to include active and inactive images. Active images have precedence
# inactive images are maintained PC images which only receive security updates.
# See https://www.suse.com/c/suse-public-cloud-image-life-cycle/
image = None
for state in ["active", "inactive", "deprecated"]:
image = get_recent_pint_image(
images, settings["PUBLIC_CLOUD_PINT_NAME"], region, state=state
)
if image is not None:
break
if image is None:
raise ValueError("Cannot find matching image in pint")
settings["PUBLIC_CLOUD_IMAGE_ID"] = image[settings["PUBLIC_CLOUD_PINT_FIELD"]]
settings["PUBLIC_CLOUD_IMAGE_NAME"] = image["name"]
settings["PUBLIC_CLOUD_IMAGE_STATE"] = image["state"]
# Remove pint query settings. They are not required in the scheduled job
if "PUBLIC_CLOUD_PINT_QUERY" in settings:
del settings["PUBLIC_CLOUD_PINT_QUERY"]
if "PUBLIC_CLOUD_PINT_NAME" in settings:
del settings["PUBLIC_CLOUD_PINT_NAME"]
if "PUBLIC_CLOUD_PINT_REGION" in settings:
# If we define a region for the pint query, propagate this value
settings["PUBLIC_CLOUD_REGION"] = settings["PUBLIC_CLOUD_PINT_REGION"]
del settings["PUBLIC_CLOUD_PINT_REGION"]
if "PUBLIC_CLOUD_PINT_FIELD" in settings:
del settings["PUBLIC_CLOUD_PINT_FIELD"]
return settings
except (
requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
ValueError,
re.error,
) as e:
logger.warning(
f"PUBLIC_CLOUD_PINT_QUERY handling failed for {settings['PUBLIC_CLOUD_PINT_NAME']}: {e}"
)
settings["PUBLIC_CLOUD_IMAGE_ID"] = None
return settings
def get_recent_pint_image(images, name_regex, region=None, state="active"):
"""
From the given set of images (received json from pint), get the latest one that matches the given criteria (name given as regular expression, region given as string, and state given the state of the image)
Get the latest one based on 'publishedon'
"""
def is_newer(date1, date2):
# Checks if date1 is newer than date2. Expected date format: YYYYMMDD
# Because for the format, we can do a simple int comparison
return int(date1) > int(date2)
name = re.compile(name_regex)
if region == "":
region = None
recentimage = None
for image in images:
# Apply selection criteria. state and region criteria can be omitted by setting the corresponding variable to None
# This is required, because certain public cloud providers do not make a distinction on e.g. the region and thus this check is not needed there
if name.match(image["name"]) is None:
continue
if (state is not None) and (image["state"] != state):
continue
if (region is not None) and (region != image["region"]):
continue
# Get latest one based on 'publishedon'
if recentimage is None or is_newer(
image["publishedon"], recentimage["publishedon"]
):
recentimage = image
return recentimage