#4403

pending completion

Build # #4403

Build Type

push

coveralls-python

Committed by

suhaibmujahid

Commit Message

Replace the word "tool" with "rule"

Run Details

639 of 3205 branches covered (19.94%)

57 of 57 new or added lines in 6 files covered. (100.0%)

1816 of 8004 relevant lines covered (22.69%)

0.23 hits per line

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

70.93

/bugbot/multi_autofixers.py

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.

import argparse
from collections import ChainMap, defaultdict
from typing import Any, Callable, Counter, Dict, Iterable, Type

from bugbot import logger, utils
from bugbot.bzcleaner import BzCleaner
from bugbot.nag_me import Nag

RulesChanges = Dict[Type[BzCleaner], Any]


class UnexpectedRulesError(Exception):
    """Unexpected rules appear in merge function"""

    def __init__(self, rules: Iterable[Type[BzCleaner]]) -> None:
        """Constructor

        Args:
            rules: the rules that change the same field.
        """
        super().__init__()
        self.rules = rules

    def __str__(self):
        return (
            "Error: merge function does not support merge fields "
            f"from '{utils.english_list([rule.__name__ for rule in self.rules])}'"
        )


class MissingMergeFunctionError(Exception):
    """Merge function is missing"""

    def __init__(self, field: str) -> None:
        """Constructor

        Args:
            field: the field that is missing a merge function.
        """
        super().__init__()
        self.field = field

    def __str__(self):
        return f"Error: missing merge function for '{self.field}'"


class MultiAutoFixers:
    "Merge changes from multiple rules and apply them to Bugzilla at once"

    def __init__(
        self, *rules: BzCleaner, **merge_functions: Callable[[RulesChanges], dict]
    ):
        """Constructor

        Args:
            *rules: rules to merge their changes.
            **merge_functions: functions to merge field values when multiple
                rules are changing the same field. The key should be the field
                name.
        """
        super().__init__()
        for rule in rules:
            if isinstance(rule, Nag):
                logger.warning(
                    "%s implements Nag, however, nag emails will not be merged",
                    type(rule),
                )
        self.rules = rules
        self.merge_functions = merge_functions
        self.is_dryrun: bool = True

    def description(self):
        """Return a description of the rule"""
        return "Grouped changes for the following rules: " + self.name()

    def name(self):
        """Return the name of the rule"""
        return utils.english_list([rule.name() for rule in self.rules])

    def get_args_parser(self):
        """Get the arguments from the command line"""
        parser = argparse.ArgumentParser()
        parser.add_argument(
            "--production",
            dest="dryrun",
            action="store_false",
            help=(
                "If the flag is not passed, just do the query, and print the "
                "proposed changes and emails to console without emailing anyone"
            ),
        )

        return parser

    def run(self):
        """Run the rule"""
        args = self.get_args_parser().parse_args()
        self.is_dryrun = args.dryrun

        for rule in self.rules:
            rule.apply_autofix = False
            rule.run()

        new_changes = self._merge_changes_from_rules()
        no_bugmail = all(rule.no_bugmail for rule in self.rules)

        BzCleaner.apply_changes_on_bugzilla(
            self.name(), new_changes, no_bugmail, self.is_dryrun, db_extra={}
        )

    def _merge_changes_from_rules(self) -> Dict[str, dict]:
        all_changes: Dict[str, dict] = defaultdict(dict)
        for rule in self.rules:
            for bugid, changes in rule.autofix_changes.items():
                all_changes[bugid][rule.__class__] = changes

        for bugid, rules in all_changes.items():
            merged_changes = {}

            common_fields = (
                field
                for field, count in Counter(
                    field for changes in rules.values() for field in changes.keys()
                ).items()
                if count > 1
            )

            for field in common_fields:
                if field not in self.merge_functions:
                    raise MissingMergeFunctionError(field)

                rules_with_common_field = {
                    rule: changes for rule, changes in rules.items() if field in changes
                }
                merged_changes[field] = self.merge_functions[field](
                    rules_with_common_field
                )

            all_changes[bugid] = dict(ChainMap(merged_changes, *rules.values()))

        return all_changes

1	# This Source Code Form is subject to the terms of the Mozilla Public
2	# License, v. 2.0. If a copy of the MPL was not distributed with this file,
3	# You can obtain one at http://mozilla.org/MPL/2.0/.
4
5	import argparse	1✔
6	from collections import ChainMap, defaultdict	1✔
7	from typing import Any, Callable, Counter, Dict, Iterable, Type	1✔
8
9	from bugbot import logger, utils	1✔
10	from bugbot.bzcleaner import BzCleaner	1✔
11	from bugbot.nag_me import Nag	1✔
12
13	RulesChanges = Dict[Type[BzCleaner], Any]	1✔
14
15
16	class UnexpectedRulesError(Exception):	1✔
17	"""Unexpected rules appear in merge function"""
18
19	def __init__(self, rules: Iterable[Type[BzCleaner]]) -> None:	1✔
20	"""Constructor
21
22	Args:
23	rules: the rules that change the same field.
24	"""
25	super().__init__()	1✔
26	self.rules = rules	1✔
27
28	def __str__(self):	1✔
29	return (	×
30	"Error: merge function does not support merge fields "
31	f"from '{utils.english_list([rule.__name__ for rule in self.rules])}'"
32	)
33
34
35	class MissingMergeFunctionError(Exception):	1✔
36	"""Merge function is missing"""
37
38	def __init__(self, field: str) -> None:	1✔
39	"""Constructor
40
41	Args:
42	field: the field that is missing a merge function.
43	"""
44	super().__init__()	1✔
45	self.field = field	1✔
46
47	def __str__(self):	1✔
48	return f"Error: missing merge function for '{self.field}'"	×
49
50
51	class MultiAutoFixers:	1✔
52	"Merge changes from multiple rules and apply them to Bugzilla at once"
53
54	def __init__(	1✔
55	self, rules: BzCleaner, *merge_functions: Callable[[RulesChanges], dict]
56	):
57	"""Constructor
58
59	Args:
60	*rules: rules to merge their changes.
61	**merge_functions: functions to merge field values when multiple
62	rules are changing the same field. The key should be the field
63	name.
64	"""
65	super().__init__()	1✔
66	for rule in rules:	1✔
67	if isinstance(rule, Nag):	1!
68	logger.warning(	×
69	"%s implements Nag, however, nag emails will not be merged",
70	type(rule),
71	)
72	self.rules = rules	1✔
73	self.merge_functions = merge_functions	1✔
74	self.is_dryrun: bool = True	1✔
75
76	def description(self):	1✔
77	"""Return a description of the rule"""
78	return "Grouped changes for the following rules: " + self.name()	×
79
80	def name(self):	1✔
81	"""Return the name of the rule"""
82	return utils.english_list([rule.name() for rule in self.rules])	×
83
84	def get_args_parser(self):	1✔
85	"""Get the arguments from the command line"""
86	parser = argparse.ArgumentParser()	×
87	parser.add_argument(	×
88	"--production",
89	dest="dryrun",
90	action="store_false",
91	help=(
92	"If the flag is not passed, just do the query, and print the "
93	"proposed changes and emails to console without emailing anyone"
94	),
95	)
96
97	return parser	×
98
99	def run(self):	1✔
100	"""Run the rule"""
101	args = self.get_args_parser().parse_args()	×
102	self.is_dryrun = args.dryrun	×
103
104	for rule in self.rules:	×
105	rule.apply_autofix = False	×
106	rule.run()	×
107
108	new_changes = self._merge_changes_from_rules()	×
109	no_bugmail = all(rule.no_bugmail for rule in self.rules)	×
110
111	BzCleaner.apply_changes_on_bugzilla(	×
112	self.name(), new_changes, no_bugmail, self.is_dryrun, db_extra={}
113	)
114
115	def _merge_changes_from_rules(self) -> Dict[str, dict]:	1✔
116	all_changes: Dict[str, dict] = defaultdict(dict)	1✔
117	for rule in self.rules:	1✔
118	for bugid, changes in rule.autofix_changes.items():	1✔
119	all_changes[bugid][rule.__class__] = changes	1✔
120
121	for bugid, rules in all_changes.items():	1✔
122	merged_changes = {}	1✔
123
124	common_fields = (	1✔
125	field
126	for field, count in Counter(
127	field for changes in rules.values() for field in changes.keys()
128	).items()
129	if count > 1
130	)
131
132	for field in common_fields:	1✔
133	if field not in self.merge_functions:	1✔
134	raise MissingMergeFunctionError(field)	1✔
135
136	rules_with_common_field = {	1✔
137	rule: changes for rule, changes in rules.items() if field in changes
138	}
139	merged_changes[field] = self.merge_functions[field](	1✔
140	rules_with_common_field
141	)
142
143	all_changes[bugid] = dict(ChainMap(merged_changes, *rules.values()))	1✔
144
145	return all_changes	1✔

mozilla / relman-auto-nag / #4403

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