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