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