1 from typing import (
2 Callable,
3 cast,
4 )
5
6 from lxml.etree import (
7 SubElement,
8 _Element,
9 )
10
11 from pcs.common import reports
12 from pcs.lib.cib.constraint import resource_set
13 from pcs.lib.cib.tools import (
14 ElementNotFound,
15 find_unique_id,
16 get_element_by_id,
17 )
18 from pcs.lib.errors import LibraryError
19 from pcs.lib.xml_tools import get_root
20
21 from .common import validate_constrainable_elements
22
23
24 # DEPRECATED
25 def _validate_attrib_names(attrib_names, options):
26 invalid_names = [name for name in options if name not in attrib_names]
27 if invalid_names:
28 raise LibraryError(
29 reports.ReportItem.error(
30 reports.messages.InvalidOptions(
31 sorted(invalid_names), sorted(attrib_names), None
32 )
33 )
34 )
35
36
37 # DEPRECATED, use pcs.lib.cib.constraint.common.validate_constrainable_elements
38 def find_valid_resource_id(
39 report_processor: reports.ReportProcessor, cib, in_clone_allowed, _id
40 ):
41 try:
42 report_processor.report_list(
43 validate_constrainable_elements(
44 [get_element_by_id(cib, _id)], in_clone_allowed
45 )
46 )
47 except ElementNotFound:
48 report_processor.report(
49 reports.ReportItem.error(reports.messages.IdNotFound(_id, []))
50 )
51 if report_processor.has_errors:
52 raise LibraryError()
53 return _id
54
55
56 # DEPRECATED
57 def prepare_options(attrib_names, options, create_id_fn, validate_id):
58 _validate_attrib_names(attrib_names + ("id",), options)
59 options = options.copy()
60
61 if "id" not in options:
62 options["id"] = create_id_fn()
63 else:
64 validate_id(options["id"])
65 return options
66
67
68 # DEPRECATED, use pcs.lib.cib.constraint.common._create_set_constraint_id
69 def create_id(cib, type_prefix, resource_set_list):
70 # Create a semi-random id. We need it to be predictable (for testing), short
71 # and somehow different than other ids so that we don't spend much time in
72 # find_unique_id.
73 # Avoid using actual resource names. It makes the id very long (consider 10
74 # or more resources in a set constraint). Also, if a resource is deleted
75 # and therefore removed from the constraint, the id no longer matches the
76 # constraint.
77 resource_ids = []
78 for _set in resource_set_list:
79 resource_ids.extend(_set["ids"])
80 id_part = "".join([_id[0] + _id[-1] for _id in resource_ids][:3])
81 return find_unique_id(cib, f"{type_prefix}_set_{id_part}")
82
83
84 # DEPRECATED, replace with pcs.lib.cib.constraint.common.DuplicatesChecker
85 def have_duplicate_resource_sets(element, other_element):
86 def get_id_set_list(element):
87 return [
88 resource_set.get_resource_id_set_list(resource_set_item)
89 for resource_set_item in element.findall(".//resource_set")
90 ]
91
92 return get_id_set_list(element) == get_id_set_list(other_element)
93
94
95 # DEPRECATED, replace with pcs.lib.cib.constraints.duplicates
96 def check_is_without_duplication(
97 report_processor: reports.ReportProcessor,
98 constraint_section: _Element,
99 element: _Element,
100 are_duplicate: Callable[[_Element, _Element], bool],
|
(1) Event identifier_typo: |
Using "duplication_allowed" appears to be a typo:
* Identifier "duplication_allowed" is only known to be referenced here, or in copies of this code.
* Identifier "duplication_alowed" is referenced elsewhere at least 2 times.
|
|
(3) Event remediation: |
Should identifier "duplication_allowed" be replaced by "duplication_alowed"? |
| Also see events: |
[identifier_use] |
101 duplication_allowed: bool = False,
102 ) -> None:
103 duplicate_element_list = [
104 duplicate_element
105 for duplicate_element in cast(
106 # The xpath method has a complicated return value, but we know our
107 # xpath expression returns only elements.
108 list[_Element],
109 constraint_section.xpath(
110 ".//*[local-name()=$tag_name]", tag_name=element.tag
111 ),
112 )
113 if (
114 element is not duplicate_element
115 and are_duplicate(element, duplicate_element)
116 )
117 ]
118 if not duplicate_element_list:
119 return
120
121 if report_processor.report_list(
122 [
123 reports.ReportItem(
124 severity=reports.item.get_severity(
125 reports.codes.FORCE,
126 duplication_allowed,
127 ),
128 message=reports.messages.DuplicateConstraintsExist(
129 [
130 str(duplicate.attrib["id"])
131 for duplicate in duplicate_element_list
132 ]
133 ),
134 ),
135 ]
136 ).has_errors:
137 raise LibraryError()
138
139
140 # DEPRECATED use pcs.lib.cib.constraint.common.create_constraint_with_set
141 def create_with_set(constraint_section, tag_name, options, resource_set_list):
142 if not resource_set_list:
143 raise LibraryError(
144 reports.ReportItem.error(reports.messages.EmptyResourceSetList())
145 )
146 element = SubElement(constraint_section, tag_name)
147 element.attrib.update(options)
148 if tag_name == "rsc_order":
149 all_resource_ids = []
150 for resource_set_item in resource_set_list:
151 all_resource_ids.extend(resource_set_item["ids"])
152 resource_set.is_resource_in_same_group(
153 get_root(constraint_section), all_resource_ids
154 )
155 for resource_set_item in resource_set_list:
156 resource_set.create_old(element, resource_set_item)
157 return element
158