2262239407

Build Type

push

github

Committed by Lukas Holecek

Commit Message

Add stomp consumers

Run Details

541 of 664 branches covered (81.48%)

Branch coverage included in aggregate %.

218 of 261 new or added lines in 5 files covered. (83.52%)

3 existing lines in 1 file now uncovered.

1682 of 1989 relevant lines covered (84.57%)

0.85 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.14

/greenwave/resultsdb_listener.py

# SPDX-License-Identifier: GPL-2.0+
from greenwave.base_listener import BaseListener
from greenwave.product_versions import subject_product_version
from greenwave.subjects.factory import (
    create_subject_from_data,
    UnknownSubjectDataError,
)


def _unpack_value(value):
    """
    If value is list with single element, returns the element, otherwise
    returns the value.
    """
    if isinstance(value, list) and len(value) == 1:
        value = value[0]
    return value


def _get_brew_task_id(msg):
    data = msg.get("data")
    if not data:
        return None

    task_id = _unpack_value(data.get("brew_task_id"))
    try:
        return int(task_id)
    except (ValueError, TypeError):
        return None


class ResultsDBListener(BaseListener):
    monitor_labels = {"handler": "resultsdb"}

    def __init__(self, config_obj=None):
        super().__init__(uid_suffix="resultsdb", config_obj=config_obj)
        self.topic = self.app.config["LISTENER_RESULTSDB_QUEUE"]
        self.koji_base_url = self.app.config["KOJI_BASE_URL"]

    @staticmethod
    def announcement_subject(msg):
        """
        Returns pairs of (subject type, subject identifier) for announcement
        consideration from the message.
        """

        try:
            data = msg["data"]  # New format
        except KeyError:
            data = msg["task"]  # Old format

        unpacked = {k: _unpack_value(v) for k, v in data.items()}

        try:
            subject = create_subject_from_data(unpacked)
        except UnknownSubjectDataError:
            return None

        # note: it is *intentional* that we do not handle old format
        # compose-type messages, because it is impossible to reliably
        # produce a decision from these. compose decisions can only be
        # reliably made from new format messages, where we can rely on
        # productmd.compose.id being available. See:
        # https://pagure.io/greenwave/issue/122
        # https://pagure.io/taskotron/resultsdb/issue/92
        # https://pagure.io/taskotron/resultsdb/pull-request/101
        # https://pagure.io/greenwave/pull-request/262#comment-70350
        if subject.type == "compose" and "productmd.compose.id" not in data:
            return None

        return subject

    def _consume_message(self, msg):
        try:
            testcase = msg["testcase"]["name"]
        except KeyError:
            testcase = msg["task"]["name"]

        try:
            submit_time = msg["submit_time"]
        except KeyError:
            submit_time = msg["result"]["submit_time"]

        outcome = msg.get("outcome")
        if outcome in self.app.config["OUTCOMES_INCOMPLETE"]:
            self.app.logger.debug("Assuming no decision change on outcome %r", outcome)
            return False

        brew_task_id = _get_brew_task_id(msg)

        subject = self.announcement_subject(msg)
        if subject is None:
            return False

        self.app.logger.debug("Considering subject: %r", subject)

        product_version = subject_product_version(
            subject,
            self.koji_base_url,
            brew_task_id,
        )

        self.app.logger.debug("Guessed product version: %r", product_version)

        self._publish_decision_change(
            submit_time=submit_time,
            subject=subject,
            testcase=testcase,
            product_version=product_version,
            publish_testcase=False,
        )
        return True


if __name__ == "__main__":
    listener = ResultsDBListener()
    listener.listen()
    app = listener.app

1	# SPDX-License-Identifier: GPL-2.0+
2	from greenwave.base_listener import BaseListener	1✔
3	from greenwave.product_versions import subject_product_version	1✔
4	from greenwave.subjects.factory import (	1✔
5	create_subject_from_data,
6	UnknownSubjectDataError,
7	)
8
9
10	def _unpack_value(value):	1✔
11	"""
12	If value is list with single element, returns the element, otherwise
13	returns the value.
14	"""
15	if isinstance(value, list) and len(value) == 1:	1✔
16	value = value[0]	1✔
17	return value	1✔
18
19
20	def _get_brew_task_id(msg):	1✔
21	data = msg.get("data")	1✔
22	if not data:	1✔
23	return None	1✔
24
25	task_id = _unpack_value(data.get("brew_task_id"))	1✔
26	try:	1✔
27	return int(task_id)	1✔
28	except (ValueError, TypeError):	1✔
29	return None	1✔
30
31
32	class ResultsDBListener(BaseListener):	1✔
33	monitor_labels = {"handler": "resultsdb"}	1✔
34
35	def __init__(self, config_obj=None):	1✔
36	super().__init__(uid_suffix="resultsdb", config_obj=config_obj)	1✔
37	self.topic = self.app.config["LISTENER_RESULTSDB_QUEUE"]	1✔
38	self.koji_base_url = self.app.config["KOJI_BASE_URL"]	1✔
39
40	@staticmethod	1✔
41	def announcement_subject(msg):	1✔
42	"""
43	Returns pairs of (subject type, subject identifier) for announcement
44	consideration from the message.
45	"""
46
47	try:	1✔
48	data = msg["data"] # New format	1✔
49	except KeyError:	1✔
50	data = msg["task"] # Old format	1✔
51
52	unpacked = {k: _unpack_value(v) for k, v in data.items()}	1✔
53
54	try:	1✔
55	subject = create_subject_from_data(unpacked)	1✔
56	except UnknownSubjectDataError:	1✔
57	return None	1✔
58
59	# note: it is intentional that we do not handle old format
60	# compose-type messages, because it is impossible to reliably
61	# produce a decision from these. compose decisions can only be
62	# reliably made from new format messages, where we can rely on
63	# productmd.compose.id being available. See:
64	# https://pagure.io/greenwave/issue/122
65	# https://pagure.io/taskotron/resultsdb/issue/92
66	# https://pagure.io/taskotron/resultsdb/pull-request/101
67	# https://pagure.io/greenwave/pull-request/262#comment-70350
68	if subject.type == "compose" and "productmd.compose.id" not in data:	1✔
69	return None	1✔
70
71	return subject	1✔
72
73	def _consume_message(self, msg):	1✔
74	try:	1✔
75	testcase = msg["testcase"]["name"]	1✔
76	except KeyError:	1✔
77	testcase = msg["task"]["name"]	1✔
78
79	try:	1✔
80	submit_time = msg["submit_time"]	1✔
81	except KeyError:	1✔
82	submit_time = msg["result"]["submit_time"]	1✔
83
84	outcome = msg.get("outcome")	1✔
85	if outcome in self.app.config["OUTCOMES_INCOMPLETE"]:	1✔
86	self.app.logger.debug("Assuming no decision change on outcome %r", outcome)	1✔
87	return False	1✔
88
89	brew_task_id = _get_brew_task_id(msg)	1✔
90
91	subject = self.announcement_subject(msg)	1✔
92	if subject is None:	1!
NEW 93	return False	×
94
95	self.app.logger.debug("Considering subject: %r", subject)	1✔
96
97	product_version = subject_product_version(	1✔
98	subject,
99	self.koji_base_url,
100	brew_task_id,
101	)
102
103	self.app.logger.debug("Guessed product version: %r", product_version)	1✔
104
105	self._publish_decision_change(	1✔
106	submit_time=submit_time,
107	subject=subject,
108	testcase=testcase,
109	product_version=product_version,
110	publish_testcase=False,
111	)
112	return True	1✔
113
114
115	if __name__ == "__main__":
116	listener = ResultsDBListener()
117	listener.listen()
118	app = listener.app

release-engineering / greenwave / 2262239407

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous