-
Notifications
You must be signed in to change notification settings - Fork 21
/
openqa.py
132 lines (115 loc) · 4.28 KB
/
openqa.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
# Copyright SUSE LLC
# SPDX-License-Identifier: MIT
from functools import lru_cache
import logging
from pprint import pformat
from urllib.parse import ParseResult
from typing import Dict
from openqa_client.client import OpenQA_Client
from openqa_client.exceptions import RequestError
from . import DEVELOPMENT_PARENT_GROUP_ID, OPENQA_URL
from .loader.qem import update_job
from .errors import PostOpenQAError
from .types import Data
log = logging.getLogger("bot.openqa")
class openQAInterface:
def __init__(self, args) -> None:
self.url: ParseResult = args.openqa_instance
self.openqa = OpenQA_Client(server=self.url.netloc, scheme=self.url.scheme)
self.retries = 3
user_agent = {"User-Agent": "python-OpenQA_Client/qem-bot/1.0.0"}
self.openqa.session.headers.update(user_agent)
self.qem_token: Dict[str, str] = {"Authorization": f"Token {args.token}"}
def __bool__(self) -> bool:
"""Return True only for the configured openQA instance.
This is used for decide if the dashboard database should be updated or not
"""
return self.url.netloc == OPENQA_URL
def post_job(self, settings) -> None:
log.info(
"openqa-cli api --host %s -X post isos %s",
self.url.geturl(),
" ".join(["%s=%s" % (k, v) for k, v in settings.items()]),
)
try:
self.openqa.openqa_request(
"POST", "isos", data=settings, retries=self.retries
)
except RequestError as e:
log.error("openQA returned %s", e.args[-1])
log.error("Post failed with %s", pformat(settings))
raise PostOpenQAError
except Exception as e:
log.exception(e)
log.error("Post failed with %s", pformat(settings))
raise PostOpenQAError
def handle_job_not_found(self, job_id: int):
log.info("Job %s not found in openQA, marking as obsolete on dashboard", job_id)
update_job(self.qem_token, job_id, {"obsolete": True})
def get_jobs(self, data: Data):
log.info("Getting openQA tests results for %s", pformat(data))
param = {}
param["scope"] = "relevant"
param["latest"] = "1"
param["flavor"] = data.flavor
param["distri"] = data.distri
param["build"] = data.build
param["version"] = data.version
param["arch"] = data.arch
ret = None
try:
ret = self.openqa.openqa_request("GET", "jobs", param)["jobs"]
except Exception as e:
log.exception(e)
raise e
return ret
@lru_cache(maxsize=512)
def get_job_comments(self, job_id: int):
ret = []
try:
ret = self.openqa.openqa_request(
"GET", "jobs/%s/comments" % job_id, retries=self.retries
)
ret = list(map(lambda c: {"text": c.get("text", "")}, ret))
except Exception as e: # pylint: disable=broad-except
(_, _, status_code, *_) = e.args
if status_code == 404:
self.handle_job_not_found(job_id)
else:
log.exception(e)
return ret
@lru_cache(maxsize=256)
def is_devel_group(self, groupid: int) -> bool:
ret = None
try:
ret = self.openqa.openqa_request("GET", f"job_groups/{groupid}")
except Exception as e:
log.exception(e)
raise e
# return True as safe option if ret = None
return (
ret[0]["parent_id"] == DEVELOPMENT_PARENT_GROUP_ID if ret else True
) # ID of Development Group
@lru_cache(maxsize=256)
def get_single_job(self, job_id: int):
ret = None
try:
ret = self.openqa.openqa_request(
"GET",
"jobs/%s" % job_id,
)["job"]
except RequestError as e:
log.exception(e)
return ret
@lru_cache(maxsize=256)
def get_older_jobs(self, job_id: int, limit: int):
ret = []
try:
ret = self.openqa.openqa_request(
"GET",
"/tests/%s/ajax?previous_limit=%s&next_limit=0" % (job_id, limit),
retries=self.retries,
)
except RequestError as e:
log.exception(e)
return ret