1 import sys
2 import xml.dom.minidom
3 from io import StringIO
4 from time import sleep
5 from unittest import (
6 TestCase,
7 mock,
8 )
9
10 from pcs import utils
11 from pcs.common import const
12
13 from pcs_test.tools.misc import get_test_resource as rc
14 from pcs_test.tools.xml import dom_get_child_elements
15
16 cib_with_nodes = rc("cib-empty-withnodes.xml")
17 empty_cib = rc("cib-empty.xml")
18
19 TestCase.maxDiff = None
20
21
22 class UtilsTest(TestCase):
23 @staticmethod
24 def get_cib_empty():
25 return xml.dom.minidom.parse(empty_cib)
26
27 @staticmethod
28 def get_cib_with_nodes_minidom():
29 return xml.dom.minidom.parse(cib_with_nodes)
30
31 def get_cib_resources(self):
32 cib_dom = self.get_cib_empty()
33 new_resources = xml.dom.minidom.parseString(
34 """
35 <resources>
36 <primitive id="myResource"
37 class="ocf" provider="heartbeat" type="Dummy">
38 </primitive>
39 <clone id="myClone">
40 <primitive id="myClonedResource"
41 class="ocf" provider="heartbeat" type="Dummy">
42 </primitive>
43 </clone>
44 <clone id="myUniqueClone">
45 <primitive id="myUniqueClonedResource"
46 class="ocf" provider="heartbeat" type="Dummy">
47 </primitive>
48 <meta-attributes>
49 <nvpair name="globally-unique" value="true" />
50 </meta-attributes>
51 </clone>
52 <master id="myMaster">
53 <primitive id="myMasteredResource"
54 class="ocf" provider="heartbeat" type="Dummy">
55 </primitive>
56 </master>
57 <group id="myGroup">
58 <primitive id="myGroupedResource"
59 class="ocf" provider="heartbeat" type="Dummy">
60 </primitive>
61 </group>
62 <clone id="myGroupClone">
63 <group id="myClonedGroup">
64 <primitive id="myClonedGroupedResource"
65 class="ocf" provider="heartbeat" type="Dummy">
66 </primitive>
67 </group>
68 </clone>
69 <master id="myGroupMaster">
70 <group id="myMasteredGroup">
71 <primitive id="myMasteredGroupedResource"
72 class="ocf" provider="heartbeat" type="Dummy">
73 </primitive>
74 </group>
75 </master>
76 <bundle id="myBundle">
77 <primitive id="myBundledResource"
78 class="ocf" provider="heartbeat" type="Dummy" />
79 </bundle>
80 <bundle id="myEmptyBundle"/>
81 </resources>
82 """
83 ).documentElement
84 resources = cib_dom.getElementsByTagName("resources")[0]
85 resources.parentNode.replaceChild(new_resources, resources)
86 return cib_dom
87
88 def test_dom_get_resources(self): # noqa: PLR0915
89 def test_dom_get(method, dom, ok_ids, bad_ids):
90 for element_id in ok_ids:
91 self.assert_element_id(method(dom, element_id), element_id)
92 for element_id in bad_ids:
93 self.assertFalse(method(dom, element_id))
94
95 cib_dom = self.get_cib_empty()
96 self.assertFalse(utils.dom_get_resource(cib_dom, "myResource"))
97 self.assertFalse(
98 utils.dom_get_resource_clone(cib_dom, "myClonedResource")
99 )
100 self.assertFalse(
101 utils.dom_get_resource_masterslave(cib_dom, "myMasteredResource")
102 )
103 self.assertFalse(utils.dom_get_group(cib_dom, "myGroup"))
104 self.assertFalse(utils.dom_get_group_clone(cib_dom, "myClonedGroup"))
105 self.assertFalse(
106 utils.dom_get_group_masterslave(cib_dom, "myMasteredGroup")
107 )
108 self.assertFalse(utils.dom_get_clone(cib_dom, "myClone"))
109 self.assertFalse(utils.dom_get_master(cib_dom, "myMaster"))
110 self.assertFalse(utils.dom_get_clone_ms_resource(cib_dom, "myClone"))
111 self.assertFalse(utils.dom_get_clone_ms_resource(cib_dom, "myMaster"))
112 self.assertFalse(
113 utils.dom_get_resource_clone_ms_parent(cib_dom, "myClonedResource")
114 )
115 self.assertFalse(
116 utils.dom_get_resource_clone_ms_parent(
117 cib_dom, "myMasteredResource"
118 )
119 )
120 self.assertIsNone(utils.dom_get_bundle(cib_dom, "myResource"))
121 self.assertIsNone(utils.dom_get_bundle(cib_dom, "notExisting"))
122 self.assertIsNone(
123 utils.dom_get_resource_bundle_parent(cib_dom, "myBundledResource")
124 )
125
126 cib_dom = self.get_cib_resources()
127 all_ids = {
128 "none",
129 "myResource",
130 "myClone",
131 "myClonedResource",
132 "myUniqueClone",
133 "myUniqueClonedResource",
134 "myMaster",
135 "myMasteredResource",
136 "myGroup",
137 "myGroupedResource",
138 "myGroupClone",
139 "myClonedGroup",
140 "myClonedGroupedResource",
141 "myGroupMaster",
142 "myMasteredGroup",
143 "myMasteredGroupedResource",
144 "myBundledResource",
145 "myBundle",
146 "myEmptyBundle",
147 }
148
149 resource_ids = {
150 "myResource",
151 "myClonedResource",
152 "myUniqueClonedResource",
153 "myGroupedResource",
154 "myMasteredResource",
155 "myClonedGroupedResource",
156 "myMasteredGroupedResource",
157 "myBundledResource",
158 }
159 test_dom_get(
160 utils.dom_get_resource,
161 cib_dom,
162 resource_ids,
163 all_ids - resource_ids,
164 )
165
166 cloned_ids = {
167 "myClonedResource",
168 "myUniqueClonedResource",
169 "myClonedGroupedResource",
170 }
171 test_dom_get(
172 utils.dom_get_resource_clone,
173 cib_dom,
174 cloned_ids,
175 all_ids - cloned_ids,
176 )
177
178 mastered_ids = {"myMasteredResource", "myMasteredGroupedResource"}
179 test_dom_get(
180 utils.dom_get_resource_masterslave,
181 cib_dom,
182 mastered_ids,
183 all_ids - mastered_ids,
184 )
185
186 group_ids = {"myGroup", "myClonedGroup", "myMasteredGroup"}
187 test_dom_get(
188 utils.dom_get_group, cib_dom, group_ids, all_ids - group_ids
189 )
190
191 cloned_group_ids = {"myClonedGroup"}
192 test_dom_get(
193 utils.dom_get_group_clone,
194 cib_dom,
195 cloned_group_ids,
196 all_ids - cloned_group_ids,
197 )
198
199 clone_ids = {"myClone", "myUniqueClone", "myGroupClone"}
200 test_dom_get(
201 utils.dom_get_clone, cib_dom, clone_ids, all_ids - clone_ids
202 )
203
204 mastered_group_ids = {"myMasteredGroup"}
205 test_dom_get(
206 utils.dom_get_group_masterslave,
207 cib_dom,
208 mastered_group_ids,
209 all_ids - mastered_group_ids,
210 )
211
212 master_ids = {"myMaster", "myGroupMaster"}
213 test_dom_get(
214 utils.dom_get_master, cib_dom, master_ids, all_ids - master_ids
215 )
216
217 bundle_ids = {"myBundle", "myEmptyBundle"}
218 test_dom_get(
219 utils.dom_get_bundle, cib_dom, bundle_ids, all_ids - bundle_ids
220 )
221
222 self.assert_element_id(
223 utils.dom_get_clone_ms_resource(cib_dom, "myClone"),
224 "myClonedResource",
225 )
226 self.assert_element_id(
227 utils.dom_get_clone_ms_resource(cib_dom, "myGroupClone"),
228 "myClonedGroup",
229 )
230 self.assert_element_id(
231 utils.dom_get_clone_ms_resource(cib_dom, "myMaster"),
232 "myMasteredResource",
233 )
234 self.assert_element_id(
235 utils.dom_get_clone_ms_resource(cib_dom, "myGroupMaster"),
236 "myMasteredGroup",
237 )
238
239 self.assert_element_id(
240 utils.dom_get_resource_clone_ms_parent(cib_dom, "myClonedResource"),
241 "myClone",
242 )
243 self.assert_element_id(
244 utils.dom_get_resource_clone_ms_parent(cib_dom, "myClonedGroup"),
245 "myGroupClone",
246 )
247 self.assert_element_id(
248 utils.dom_get_resource_clone_ms_parent(
249 cib_dom, "myClonedGroupedResource"
250 ),
251 "myGroupClone",
252 )
253 self.assert_element_id(
254 utils.dom_get_resource_clone_ms_parent(
255 cib_dom, "myMasteredResource"
256 ),
257 "myMaster",
258 )
259 self.assert_element_id(
260 utils.dom_get_resource_clone_ms_parent(cib_dom, "myMasteredGroup"),
261 "myGroupMaster",
262 )
263 self.assert_element_id(
264 utils.dom_get_resource_clone_ms_parent(
265 cib_dom, "myMasteredGroupedResource"
266 ),
267 "myGroupMaster",
268 )
269 self.assertEqual(
270 None, utils.dom_get_resource_clone_ms_parent(cib_dom, "myResource")
271 )
272 self.assertEqual(
273 None, utils.dom_get_resource_clone_ms_parent(cib_dom, "myGroup")
274 )
275 self.assertEqual(
276 None,
277 utils.dom_get_resource_clone_ms_parent(
278 cib_dom, "myGroupedResource"
279 ),
280 )
281
282 self.assertIsNone(
283 utils.dom_get_resource_bundle(
284 utils.dom_get_bundle(cib_dom, "myEmptyBundle")
285 )
286 )
287 self.assert_element_id(
288 utils.dom_get_resource_bundle(
289 utils.dom_get_bundle(cib_dom, "myBundle")
290 ),
291 "myBundledResource",
292 "primitive",
293 )
294
295 self.assert_element_id(
296 utils.dom_get_resource_bundle_parent(cib_dom, "myBundledResource"),
297 "myBundle",
298 )
299 self.assertIsNone(
300 utils.dom_get_resource_bundle_parent(cib_dom, "myResource")
301 )
302 self.assertIsNone(
303 utils.dom_get_resource_bundle_parent(cib_dom, "myClone")
304 )
305 self.assertIsNone(
306 utils.dom_get_resource_bundle_parent(cib_dom, "myClonedResource")
307 )
308 self.assertIsNone(
309 utils.dom_get_resource_bundle_parent(cib_dom, "myMaster")
310 )
311 self.assertIsNone(
312 utils.dom_get_resource_bundle_parent(cib_dom, "myMasteredGroup")
313 )
314 self.assertIsNone(
315 utils.dom_get_resource_bundle_parent(cib_dom, "myGroup")
316 )
317 self.assertIsNone(
318 utils.dom_get_resource_bundle_parent(cib_dom, "myGroupedResource")
319 )
320 self.assertIsNone(
321 utils.dom_get_resource_bundle_parent(cib_dom, "myGroupClone")
322 )
323 self.assertIsNone(
324 utils.dom_get_resource_bundle_parent(cib_dom, "myClonedGroup")
325 )
326 self.assertIsNone(
327 utils.dom_get_resource_bundle_parent(
328 cib_dom, "myClonedGroupedResource"
329 )
330 )
331
332 def test_dom_get_resource_remote_node_name(self):
333 dom = self.get_cib_empty()
334 new_resources = xml.dom.minidom.parseString(
335 """
336 <resources>
337 <primitive id="dummy1"
338 class="ocf" provider="heartbeat" type="Dummy">
339 </primitive>
340 <primitive class="ocf" id="vm-guest1" provider="heartbeat"
341 type="VirtualDomain">
342 <instance_attributes id="vm-guest1-instance_attributes">
343 <nvpair id="vm-guest1-instance_attributes-hypervisor"
344 name="hypervisor" value="qemu:///system"/>
345 <nvpair id="vm-guest1-instance_attributes-config"
346 name="config" value="/root/guest1.xml"/>
347 </instance_attributes>
348 <meta_attributes id="vm-guest1-meta_attributes">
349 <nvpair id="vm-guest1-meta_attributes-remote-node"
350 name="remote-node" value="guest1"/>
351 </meta_attributes>
352 </primitive>
353 <primitive id="dummy2"
354 class="ocf" provider="heartbeat" type="Dummy">
355 <instance_attributes id="vm-guest1-meta_attributes">
356 <nvpair id="dummy2-remote-node"
357 name="remote-node" value="guest2"/>
358 </instance_attributes>
359 </primitive>
360 <primitive id="dummy3"
361 class="ocf" provider="pacemaker" type="remote">
362 </primitive>
363 </resources>
364 """
365 ).documentElement
366 resources = dom.getElementsByTagName("resources")[0]
367 resources.parentNode.replaceChild(new_resources, resources)
368
369 self.assertEqual(
370 None,
371 utils.dom_get_resource_remote_node_name(
372 utils.dom_get_resource(dom, "dummy1")
373 ),
374 )
375 self.assertEqual(
376 None,
377 utils.dom_get_resource_remote_node_name(
378 utils.dom_get_resource(dom, "dummy2")
379 ),
380 )
381 self.assertEqual(
382 "guest1",
383 utils.dom_get_resource_remote_node_name(
384 utils.dom_get_resource(dom, "vm-guest1")
385 ),
386 )
387 self.assertEqual(
388 "dummy3",
389 utils.dom_get_resource_remote_node_name(
390 utils.dom_get_resource(dom, "dummy3")
391 ),
392 )
393
394 def test_dom_get_meta_attr_value(self):
395 dom = self.get_cib_empty()
396 new_resources = xml.dom.minidom.parseString(
397 """
398 <resources>
399 <primitive id="dummy1"
400 class="ocf" provider="heartbeat" type="Dummy">
401 </primitive>
402 <primitive class="ocf" id="vm-guest1" provider="heartbeat"
403 type="VirtualDomain">
404 <instance_attributes id="vm-guest1-instance_attributes">
405 <nvpair id="vm-guest1-instance_attributes-hypervisor"
406 name="hypervisor" value="qemu:///system"/>
407 <nvpair id="vm-guest1-instance_attributes-config"
408 name="config" value="/root/guest1.xml"/>
409 </instance_attributes>
410 <meta_attributes id="vm-guest1-meta_attributes">
411 <nvpair id="vm-guest1-meta_attributes-remote-node"
412 name="remote-node" value="guest1"/>
413 </meta_attributes>
414 </primitive>
415 <primitive id="dummy2"
416 class="ocf" provider="heartbeat" type="Dummy">
417 <instance_attributes id="vm-guest1-meta_attributes">
418 <nvpair id="dummy2-remote-node"
419 name="remote-node" value="guest2"/>
420 </instance_attributes>
421 </primitive>
422 </resources>
423 """
424 ).documentElement
425 resources = dom.getElementsByTagName("resources")[0]
426 resources.parentNode.replaceChild(new_resources, resources)
427
428 self.assertEqual(
429 None,
430 utils.dom_get_meta_attr_value(
431 utils.dom_get_resource(dom, "dummy1"), "foo"
432 ),
433 )
434 self.assertEqual(
435 None,
436 utils.dom_get_meta_attr_value(
437 utils.dom_get_resource(dom, "dummy2"), "remote-node"
438 ),
439 )
440 self.assertEqual(
441 "guest1",
442 utils.dom_get_meta_attr_value(
443 utils.dom_get_resource(dom, "vm-guest1"), "remote-node"
444 ),
445 )
446 self.assertEqual(
447 None,
448 utils.dom_get_meta_attr_value(
449 utils.dom_get_resource(dom, "vm-guest1"), "foo"
450 ),
451 )
452
453 def test_dom_get_parent_by_tag_name(self):
454 def dom_get_element_with_id(dom, tag_name, element_id):
455 for elem in dom.getElementsByTagName(tag_name):
456 if (
457 elem.hasAttribute("id")
458 and elem.getAttribute("id") == element_id
459 ):
460 return elem
461 return None
462
463 dom = xml.dom.minidom.parseString(
464 """
465 <aa id="aa1">
466 <bb id="bb1"/>
467 <bb id="bb2">
468 <cc id="cc1"/>
469 </bb>
470 <bb id="bb3">
471 <cc id="cc2"/>
472 </bb>
473 <dd id="dd1" />
474 </aa>
475 """
476 ).documentElement
477 bb1 = dom_get_element_with_id(dom, "bb", "bb1")
478 cc1 = dom_get_element_with_id(dom, "cc", "cc1")
479
480 self.assert_element_id(
481 utils.dom_get_parent_by_tag_names(bb1, ["aa"]), "aa1"
482 )
483 self.assert_element_id(
484 utils.dom_get_parent_by_tag_names(cc1, ["aa"]), "aa1"
485 )
486 self.assert_element_id(
487 utils.dom_get_parent_by_tag_names(cc1, ["bb"]), "bb2"
488 )
489
490 self.assertEqual(None, utils.dom_get_parent_by_tag_names(bb1, ["cc"]))
491 self.assertEqual(None, utils.dom_get_parent_by_tag_names(cc1, ["dd"]))
492 self.assertEqual(None, utils.dom_get_parent_by_tag_names(cc1, ["ee"]))
493
494 def test_validate_constraint_resource(self):
495 dom = self.get_cib_resources()
496 self.assertEqual(
497 (True, "", "myClone"),
498 utils.validate_constraint_resource(dom, "myClone"),
499 )
500 self.assertEqual(
501 (True, "", "myGroupClone"),
502 utils.validate_constraint_resource(dom, "myGroupClone"),
503 )
504 self.assertEqual(
505 (True, "", "myMaster"),
506 utils.validate_constraint_resource(dom, "myMaster"),
507 )
508 self.assertEqual(
509 (True, "", "myGroupMaster"),
510 utils.validate_constraint_resource(dom, "myGroupMaster"),
511 )
512 self.assertEqual(
513 (True, "", "myBundle"),
514 utils.validate_constraint_resource(dom, "myBundle"),
515 )
516 self.assertEqual(
517 (True, "", "myEmptyBundle"),
518 utils.validate_constraint_resource(dom, "myEmptyBundle"),
519 )
520 self.assertEqual(
521 (True, "", "myResource"),
522 utils.validate_constraint_resource(dom, "myResource"),
523 )
524 self.assertEqual(
525 (True, "", "myGroup"),
526 utils.validate_constraint_resource(dom, "myGroup"),
527 )
528 self.assertEqual(
529 (True, "", "myGroupedResource"),
530 utils.validate_constraint_resource(dom, "myGroupedResource"),
531 )
532
533 self.assertEqual(
534 (False, "Resource 'myNonexistent' does not exist", None),
535 utils.validate_constraint_resource(dom, "myNonexistent"),
536 )
537
538 message = (
539 "%s is a clone resource, you should use the clone id: "
540 "%s when adding constraints. Use --force to override."
541 )
542 self.assertEqual(
543 (False, message % ("myClonedResource", "myClone"), "myClone"),
544 utils.validate_constraint_resource(dom, "myClonedResource"),
545 )
546 self.assertEqual(
547 (
548 False,
549 message % ("myClonedGroup", "myGroupClone"),
550 "myGroupClone",
551 ),
552 utils.validate_constraint_resource(dom, "myClonedGroup"),
553 )
554 self.assertEqual(
555 (
556 False,
557 message % ("myClonedGroupedResource", "myGroupClone"),
558 "myGroupClone",
559 ),
560 utils.validate_constraint_resource(dom, "myClonedGroupedResource"),
561 )
562
563 message = (
564 "%s is a clone resource, you should use the clone id: "
565 "%s when adding constraints. Use --force to override."
566 )
567 self.assertEqual(
568 (False, message % ("myMasteredResource", "myMaster"), "myMaster"),
569 utils.validate_constraint_resource(dom, "myMasteredResource"),
570 )
571 self.assertEqual(
572 (
573 False,
574 message % ("myMasteredGroup", "myGroupMaster"),
575 "myGroupMaster",
576 ),
577 utils.validate_constraint_resource(dom, "myMasteredGroup"),
578 )
579 self.assertEqual(
580 (
581 False,
582 message % ("myMasteredGroupedResource", "myGroupMaster"),
583 "myGroupMaster",
584 ),
585 utils.validate_constraint_resource(
586 dom, "myMasteredGroupedResource"
587 ),
588 )
589
590 message = (
591 "%s is a bundle resource, you should use the bundle id: "
592 "%s when adding constraints. Use --force to override."
593 )
594 self.assertEqual(
595 (False, message % ("myBundledResource", "myBundle"), "myBundle"),
596 utils.validate_constraint_resource(dom, "myBundledResource"),
597 )
598
599 utils.pcs_options["--force"] = True
600 self.assertEqual(
601 (True, "", "myClone"),
602 utils.validate_constraint_resource(dom, "myClonedResource"),
603 )
604 self.assertEqual(
605 (True, "", "myGroupClone"),
606 utils.validate_constraint_resource(dom, "myClonedGroup"),
607 )
608 self.assertEqual(
609 (True, "", "myGroupClone"),
610 utils.validate_constraint_resource(dom, "myClonedGroupedResource"),
611 )
612 self.assertEqual(
613 (True, "", "myMaster"),
614 utils.validate_constraint_resource(dom, "myMasteredResource"),
615 )
616 self.assertEqual(
617 (True, "", "myGroupMaster"),
618 utils.validate_constraint_resource(dom, "myMasteredGroup"),
619 )
620 self.assertEqual(
621 (True, "", "myGroupMaster"),
622 utils.validate_constraint_resource(
623 dom, "myMasteredGroupedResource"
624 ),
625 )
626 self.assertEqual(
627 (True, "", "myBundle"),
628 utils.validate_constraint_resource(dom, "myBundledResource"),
629 )
630
631 def test_validate_xml_id(self):
632 self.assertEqual((True, ""), utils.validate_xml_id("dummy"))
633 self.assertEqual((True, ""), utils.validate_xml_id("DUMMY"))
634 self.assertEqual((True, ""), utils.validate_xml_id("dUmMy"))
635 self.assertEqual((True, ""), utils.validate_xml_id("dummy0"))
636 self.assertEqual((True, ""), utils.validate_xml_id("dum0my"))
637 self.assertEqual((True, ""), utils.validate_xml_id("dummy-"))
638 self.assertEqual((True, ""), utils.validate_xml_id("dum-my"))
639 self.assertEqual((True, ""), utils.validate_xml_id("dummy."))
640 self.assertEqual((True, ""), utils.validate_xml_id("dum.my"))
641 self.assertEqual((True, ""), utils.validate_xml_id("_dummy"))
642 self.assertEqual((True, ""), utils.validate_xml_id("dummy_"))
643 self.assertEqual((True, ""), utils.validate_xml_id("dum_my"))
644
645 self.assertEqual(
646 (False, "test id cannot be empty"),
647 utils.validate_xml_id("", "test id"),
648 )
649
650 msg = "invalid test id '%s', '%s' is not a valid first character for a test id"
651 self.assertEqual(
652 (False, msg % ("0", "0")), utils.validate_xml_id("0", "test id")
653 )
654 self.assertEqual(
655 (False, msg % ("-", "-")), utils.validate_xml_id("-", "test id")
656 )
657 self.assertEqual(
658 (False, msg % (".", ".")), utils.validate_xml_id(".", "test id")
659 )
660 self.assertEqual(
661 (False, msg % (":", ":")), utils.validate_xml_id(":", "test id")
662 )
663 self.assertEqual(
664 (False, msg % ("0dummy", "0")),
665 utils.validate_xml_id("0dummy", "test id"),
666 )
667 self.assertEqual(
668 (False, msg % ("-dummy", "-")),
669 utils.validate_xml_id("-dummy", "test id"),
670 )
671 self.assertEqual(
672 (False, msg % (".dummy", ".")),
673 utils.validate_xml_id(".dummy", "test id"),
674 )
675 self.assertEqual(
676 (False, msg % (":dummy", ":")),
677 utils.validate_xml_id(":dummy", "test id"),
678 )
679
680 msg = (
681 "invalid test id '%s', '%s' is not a valid character for a test id"
682 )
683 self.assertEqual(
684 (False, msg % ("dum:my", ":")),
685 utils.validate_xml_id("dum:my", "test id"),
686 )
687 self.assertEqual(
688 (False, msg % ("dummy:", ":")),
689 utils.validate_xml_id("dummy:", "test id"),
690 )
691 self.assertEqual(
692 (False, msg % ("dum?my", "?")),
693 utils.validate_xml_id("dum?my", "test id"),
694 )
695 self.assertEqual(
696 (False, msg % ("dummy?", "?")),
697 utils.validate_xml_id("dummy?", "test id"),
698 )
699
700 def test_is_score(self):
701 self.assertTrue(utils.is_score("INFINITY"))
702 self.assertTrue(utils.is_score("+INFINITY"))
703 self.assertTrue(utils.is_score("-INFINITY"))
704 self.assertTrue(utils.is_score("0"))
705 self.assertTrue(utils.is_score("+0"))
706 self.assertTrue(utils.is_score("-0"))
707 self.assertTrue(utils.is_score("123"))
708 self.assertTrue(utils.is_score("-123"))
709 self.assertTrue(utils.is_score("+123"))
710
711 self.assertFalse(utils.is_score(""))
712 self.assertFalse(utils.is_score("abc"))
713 self.assertFalse(utils.is_score("+abc"))
714 self.assertFalse(utils.is_score("-abc"))
715 self.assertFalse(utils.is_score("10a"))
716 self.assertFalse(utils.is_score("+10a"))
717 self.assertFalse(utils.is_score("-10a"))
718 self.assertFalse(utils.is_score("a10"))
719 self.assertFalse(utils.is_score("+a10"))
720 self.assertFalse(utils.is_score("a-10"))
721 self.assertFalse(utils.is_score("infinity"))
722 self.assertFalse(utils.is_score("+infinity"))
723 self.assertFalse(utils.is_score("-infinity"))
724 self.assertFalse(utils.is_score("+InFiNiTy"))
725 self.assertFalse(utils.is_score("INFINITY10"))
726 self.assertFalse(utils.is_score("INFINITY+10"))
727 self.assertFalse(utils.is_score("-INFINITY10"))
728 self.assertFalse(utils.is_score("+INFINITY+10"))
729 self.assertFalse(utils.is_score("10INFINITY"))
730 self.assertFalse(utils.is_score("+10+INFINITY"))
731
732 def get_cib_status_lrm(self):
733 cib_dom = self.get_cib_empty()
734 new_status = xml.dom.minidom.parseString(
735 """
736 <status>
737 <node_state id="1" uname="rh70-node1">
738 <lrm id="1">
739 <lrm_resources>
740 <lrm_resource id="dummy" type="Dummy" class="ocf" provider="heartbeat">
741 <lrm_rsc_op id="dummy_monitor_30000" operation="monitor" call-id="34"
742 rc-code="1" on_node="Xrh70-node1X" exit-reason="test" />
743 <lrm_rsc_op id="dummy_stop_0" operation="stop" call-id="32"
744 rc-code="0" />
745 <lrm_rsc_op id="dummy_start_0" operation="start" call-id="33"
746 rc-code="0" />
747 </lrm_resource>
748 </lrm_resources>
749 </lrm>
750 </node_state>
751 <node_state id="2" uname="rh70-node2">
752 <lrm id="2">
753 <lrm_resources>
754 <lrm_resource id="dummy" type="Dummy" class="ocf" provider="heartbeat">
755 <lrm_rsc_op id="dummy_monitor_0" operation="monitor" call-id="5"
756 rc-code="1" />
757 </lrm_resource>
758 <lrm_resource id="dummy1" type="Dummy" class="ocf" provider="heartbeat">
759 <lrm_rsc_op id="dummy1_monitor_0" operation="monitor" call-id="3"
760 rc-code="0" />
761 </lrm_resource>
762 </lrm_resources>
763 </lrm>
764 </node_state>
765 </status>
766 """
767 ).documentElement
768 status = cib_dom.getElementsByTagName("status")[0]
769 status.parentNode.replaceChild(new_status, status)
770 return cib_dom
771
772 def test_resource_running_on(self):
773 status = xml.dom.minidom.parseString(
774 f"""
775 <crm_mon>
776 <summary />
777 <nodes />
778 <resources>
779 <resource id="myResource" role="Started">
780 <node name="rh70-node1" />
781 </resource>
782 <clone id="myClone">
783 <resource id="myClonedResource" role="Started">
784 <node name="rh70-node1" />
785 </resource>
786 <resource id="myClonedResource" role="Started">
787 <node name="rh70-node2" />
788 </resource>
789 <resource id="myClonedResource" role="Started">
790 <node name="rh70-node3" />
791 </resource>
792 </clone>
793 <clone id="myMaster">
794 <resource id="myMasteredResource:1" role="{const.PCMK_ROLE_UNPROMOTED}">
795 <node name="rh70-node2" />
796 </resource>
797 <resource id="myMasteredResource" role="{const.PCMK_ROLE_UNPROMOTED}">
798 <node name="rh70-node3" />
799 </resource>
800 <resource id="myMasteredResource" role="{const.PCMK_ROLE_PROMOTED}">
801 <node name="rh70-node1" />
802 </resource>
803 </clone>
804 <group id="myGroup">
805 <resource id="myGroupedResource" role="Started">
806 <node name="rh70-node2" />
807 </resource>
808 </group>
809 <clone id="myGroupClone">
810 <group id="myClonedGroup:0">
811 <resource id="myClonedGroupedResource" role="Started">
812 <node name="rh70-node1" />
813 </resource>
814 </group>
815 <group id="myClonedGroup:1">
816 <resource id="myClonedGroupedResource" role="Started">
817 <node name="rh70-node2" />
818 </resource>
819 </group>
820 <group id="myClonedGroup:2">
821 <resource id="myClonedGroupedResource" role="Started">
822 <node name="rh70-node3" />
823 </resource>
824 </group>
825 <group id="myClonedGroup:3">
826 <resource id="myClonedGroupedResource" role="Started">
827 <node name="rh70-node3" />
828 </resource>
829 </group>
830 </clone>
831 <clone id="myGroupMaster">
832 <group id="myMasteredGroup:0">
833 <resource id="myMasteredGroupedResource" role="{const.PCMK_ROLE_UNPROMOTED}">
834 <node name="rh70-node1" />
835 </resource>
836 </group>
837 <group id="myMasteredGroup:1">
838 <resource id="myMasteredGroupedResource" role="{const.PCMK_ROLE_PROMOTED}">
839 <node name="rh70-node2" />
840 </resource>
841 </group>
842 <group id="myMasteredGroup:2">
843 <resource id="myMasteredGroupedResource" role="{const.PCMK_ROLE_UNPROMOTED}">
844 <node name="rh70-node3" />
845 </resource>
846 </group>
847 </clone>
848 <resource id="myStoppedResource" role="Stopped">
849 </resource>
850 </resources>
851 </crm_mon>
852 """
853 ).documentElement
854
855 self.assertEqual(
856 utils.resource_running_on("myResource", status),
857 {
858 "message": "Resource 'myResource' is running on node rh70-node1.",
859 "is_running": True,
860 },
861 )
862 self.assertEqual(
863 utils.resource_running_on("myClonedResource", status),
864 {
865 "message": "Resource 'myClonedResource' is running on nodes "
866 "rh70-node1, rh70-node2, rh70-node3.",
867 "is_running": True,
868 },
869 )
870 self.assertEqual(
871 utils.resource_running_on("myClone", status),
872 {
873 "message": "Resource 'myClone' is running on nodes "
874 "rh70-node1, rh70-node2, rh70-node3.",
875 "is_running": True,
876 },
877 )
878 self.assertEqual(
879 utils.resource_running_on("myMasteredResource", status),
880 {
881 "message": (
882 "Resource 'myMasteredResource' is {promoted} on node "
883 "rh70-node1; {unpromoted} on nodes rh70-node2, rh70-node3."
884 ).format(
885 promoted=str(const.PCMK_ROLE_PROMOTED).lower(),
886 unpromoted=str(const.PCMK_ROLE_UNPROMOTED).lower(),
887 ),
888 "is_running": True,
889 },
890 )
891 self.assertEqual(
892 utils.resource_running_on("myMaster", status),
893 {
894 "message": (
895 "Resource 'myMaster' is {promoted} on node "
896 "rh70-node1; {unpromoted} on nodes rh70-node2, rh70-node3."
897 ).format(
898 promoted=str(const.PCMK_ROLE_PROMOTED).lower(),
899 unpromoted=str(const.PCMK_ROLE_UNPROMOTED).lower(),
900 ),
901 "is_running": True,
902 },
903 )
904 self.assertEqual(
905 utils.resource_running_on("myGroupedResource", status),
906 {
907 "message": "Resource 'myGroupedResource' is running on node "
908 "rh70-node2.",
909 "is_running": True,
910 },
911 )
912 self.assertEqual(
913 utils.resource_running_on("myGroup", status),
914 {
915 "message": "Resource 'myGroup' is running on node rh70-node2.",
916 "is_running": True,
917 },
918 )
919 self.assertEqual(
920 utils.resource_running_on("myClonedGroupedResource", status),
921 {
922 "message": "Resource 'myClonedGroupedResource' is running on nodes "
923 "rh70-node1, rh70-node2, rh70-node3, rh70-node3.",
924 "is_running": True,
925 },
926 )
927 self.assertEqual(
928 utils.resource_running_on("myClonedGroup", status),
929 {
930 "message": "Resource 'myClonedGroup' is running on nodes "
931 "rh70-node1, rh70-node2, rh70-node3, rh70-node3.",
932 "is_running": True,
933 },
934 )
935 self.assertEqual(
936 utils.resource_running_on("myGroupClone", status),
937 {
938 "message": "Resource 'myGroupClone' is running on nodes "
939 "rh70-node1, rh70-node2, rh70-node3, rh70-node3.",
940 "is_running": True,
941 },
942 )
943 self.assertEqual(
944 utils.resource_running_on("myMasteredGroupedResource", status),
945 {
946 "message": (
947 "Resource 'myMasteredGroupedResource' is {promoted} on node "
948 "rh70-node2; {unpromoted} on nodes rh70-node1, rh70-node3."
949 ).format(
950 promoted=str(const.PCMK_ROLE_PROMOTED).lower(),
951 unpromoted=str(const.PCMK_ROLE_UNPROMOTED).lower(),
952 ),
953 "is_running": True,
954 },
955 )
956 self.assertEqual(
957 utils.resource_running_on("myMasteredGroup", status),
958 {
959 "message": (
960 "Resource 'myMasteredGroup' is {promoted} on node "
961 "rh70-node2; {unpromoted} on nodes rh70-node1, rh70-node3."
962 ).format(
963 promoted=str(const.PCMK_ROLE_PROMOTED).lower(),
964 unpromoted=str(const.PCMK_ROLE_UNPROMOTED).lower(),
965 ),
966 "is_running": True,
967 },
968 )
969 self.assertEqual(
970 utils.resource_running_on("myGroupMaster", status),
971 {
972 "message": (
973 "Resource 'myGroupMaster' is {promoted} on node "
974 "rh70-node2; {unpromoted} on nodes rh70-node1, rh70-node3."
975 ).format(
976 promoted=str(const.PCMK_ROLE_PROMOTED).lower(),
977 unpromoted=str(const.PCMK_ROLE_UNPROMOTED).lower(),
978 ),
979 "is_running": True,
980 },
981 )
982 self.assertEqual(
983 utils.resource_running_on("notMyResource", status),
984 {
985 "message": "Resource 'notMyResource' is not running on any node",
986 "is_running": False,
987 },
988 )
989 self.assertEqual(
990 utils.resource_running_on("myStoppedResource", status),
991 {
992 "message": "Resource 'myStoppedResource' is not running on any node",
993 "is_running": False,
994 },
995 )
996
997 def test_get_operations_from_transitions(self):
998 transitions = xml.dom.minidom.parse(rc("transitions01.xml"))
999 self.assertEqual(
1000 [
1001 {
1002 "id": "dummy",
1003 "long_id": "dummy",
1004 "operation": "stop",
1005 "on_node": "rh7-3",
1006 },
1007 {
1008 "id": "dummy",
1009 "long_id": "dummy",
1010 "operation": "start",
1011 "on_node": "rh7-2",
1012 },
1013 {
1014 "id": "d0",
1015 "long_id": "d0:1",
1016 "operation": "stop",
1017 "on_node": "rh7-1",
1018 },
1019 {
1020 "id": "d0",
1021 "long_id": "d0:1",
1022 "operation": "start",
1023 "on_node": "rh7-2",
1024 },
1025 {
1026 "id": "state",
1027 "long_id": "state:0",
1028 "operation": "stop",
1029 "on_node": "rh7-3",
1030 },
1031 {
1032 "id": "state",
1033 "long_id": "state:0",
1034 "operation": "start",
1035 "on_node": "rh7-2",
1036 },
1037 ],
1038 utils.get_operations_from_transitions(transitions),
1039 )
1040
1041 transitions = xml.dom.minidom.parse(rc("transitions02.xml"))
1042 self.assertEqual(
1043 [
1044 {
1045 "id": "RemoteNode",
1046 "long_id": "RemoteNode",
1047 "operation": "stop",
1048 "on_node": "virt-143",
1049 },
1050 {
1051 "id": "RemoteNode",
1052 "long_id": "RemoteNode",
1053 "operation": "migrate_to",
1054 "on_node": "virt-143",
1055 },
1056 {
1057 "id": "RemoteNode",
1058 "long_id": "RemoteNode",
1059 "operation": "migrate_from",
1060 "on_node": "virt-142",
1061 },
1062 {
1063 "id": "dummy8",
1064 "long_id": "dummy8",
1065 "operation": "stop",
1066 "on_node": "virt-143",
1067 },
1068 {
1069 "id": "dummy8",
1070 "long_id": "dummy8",
1071 "operation": "start",
1072 "on_node": "virt-142",
1073 },
1074 ],
1075 utils.get_operations_from_transitions(transitions),
1076 )
1077
1078 def test_get_resources_location_from_operations(self):
1079 cib_dom = self.get_cib_resources()
1080
1081 operations = []
1082 self.assertEqual(
1083 {},
1084 utils.get_resources_location_from_operations(cib_dom, operations),
1085 )
1086
1087 operations = [
1088 {
1089 "id": "myResource",
1090 "long_id": "myResource",
1091 "operation": "start",
1092 "on_node": "rh7-1",
1093 },
1094 ]
1095 self.assertEqual(
1096 {
1097 "myResource": {
1098 "id": "myResource",
1099 "id_for_constraint": "myResource",
1100 "long_id": "myResource",
1101 "start_on_node": "rh7-1",
1102 },
1103 },
1104 utils.get_resources_location_from_operations(cib_dom, operations),
1105 )
1106
1107 operations = [
1108 {
1109 "id": "myResource",
1110 "long_id": "myResource",
1111 "operation": "start",
1112 "on_node": "rh7-1",
1113 },
1114 {
1115 "id": "myResource",
1116 "long_id": "myResource",
1117 "operation": "start",
1118 "on_node": "rh7-2",
1119 },
1120 {
1121 "id": "myResource",
1122 "long_id": "myResource",
1123 "operation": "monitor",
1124 "on_node": "rh7-3",
1125 },
1126 {
1127 "id": "myResource",
1128 "long_id": "myResource",
1129 "operation": "stop",
1130 "on_node": "rh7-3",
1131 },
1132 ]
1133 self.assertEqual(
1134 {
1135 "myResource": {
1136 "id": "myResource",
1137 "id_for_constraint": "myResource",
1138 "long_id": "myResource",
1139 "start_on_node": "rh7-2",
1140 },
1141 },
1142 utils.get_resources_location_from_operations(cib_dom, operations),
1143 )
1144
1145 operations = [
1146 {
1147 "id": "myResource",
1148 "long_id": "myResource",
1149 "operation": "start",
1150 "on_node": "rh7-1",
1151 },
1152 {
1153 "id": "myClonedResource",
1154 "long_id": "myClonedResource:0",
1155 "operation": "start",
1156 "on_node": "rh7-1",
1157 },
1158 {
1159 "id": "myClonedResource",
1160 "long_id": "myClonedResource:0",
1161 "operation": "start",
1162 "on_node": "rh7-2",
1163 },
1164 {
1165 "id": "myClonedResource",
1166 "long_id": "myClonedResource:1",
1167 "operation": "start",
1168 "on_node": "rh7-3",
1169 },
1170 ]
1171 self.assertEqual(
1172 {
1173 "myResource": {
1174 "id": "myResource",
1175 "id_for_constraint": "myResource",
1176 "long_id": "myResource",
1177 "start_on_node": "rh7-1",
1178 },
1179 "myClonedResource:0": {
1180 "id": "myClonedResource",
1181 "id_for_constraint": "myClone",
1182 "long_id": "myClonedResource:0",
1183 "start_on_node": "rh7-2",
1184 },
1185 "myClonedResource:1": {
1186 "id": "myClonedResource",
1187 "id_for_constraint": "myClone",
1188 "long_id": "myClonedResource:1",
1189 "start_on_node": "rh7-3",
1190 },
1191 },
1192 utils.get_resources_location_from_operations(cib_dom, operations),
1193 )
1194
1195 operations = [
1196 {
1197 "id": "myUniqueClonedResource:0",
1198 "long_id": "myUniqueClonedResource:0",
1199 "operation": "start",
1200 "on_node": "rh7-1",
1201 },
1202 {
1203 "id": "myUniqueClonedResource:1",
1204 "long_id": "myUniqueClonedResource:1",
1205 "operation": "monitor",
1206 "on_node": "rh7-2",
1207 },
1208 {
1209 "id": "myUniqueClonedResource:2",
1210 "long_id": "myUniqueClonedResource:2",
1211 "operation": "start",
1212 "on_node": "rh7-3",
1213 },
1214 ]
1215 self.assertEqual(
1216 {
1217 "myUniqueClonedResource:0": {
1218 "id": "myUniqueClonedResource:0",
1219 "id_for_constraint": "myUniqueClone",
1220 "long_id": "myUniqueClonedResource:0",
1221 "start_on_node": "rh7-1",
1222 },
1223 "myUniqueClonedResource:2": {
1224 "id": "myUniqueClonedResource:2",
1225 "id_for_constraint": "myUniqueClone",
1226 "long_id": "myUniqueClonedResource:2",
1227 "start_on_node": "rh7-3",
1228 },
1229 },
1230 utils.get_resources_location_from_operations(cib_dom, operations),
1231 )
1232
1233 operations = [
1234 {
1235 "id": "myMasteredGroupedResource",
1236 "long_id": "myMasteredGroupedResource:0",
1237 "operation": "start",
1238 "on_node": "rh7-1",
1239 },
1240 {
1241 "id": "myMasteredGroupedResource",
1242 "long_id": "myMasteredGroupedResource:1",
1243 "operation": "demote",
1244 "on_node": "rh7-2",
1245 },
1246 {
1247 "id": "myMasteredGroupedResource",
1248 "long_id": "myMasteredGroupedResource:1",
1249 "operation": "promote",
1250 "on_node": "rh7-3",
1251 },
1252 ]
1253 self.assertEqual(
1254 {
1255 "myMasteredGroupedResource:0": {
1256 "id": "myMasteredGroupedResource",
1257 "id_for_constraint": "myGroupMaster",
1258 "long_id": "myMasteredGroupedResource:0",
1259 "start_on_node": "rh7-1",
1260 },
1261 "myMasteredGroupedResource:1": {
1262 "id": "myMasteredGroupedResource",
1263 "id_for_constraint": "myGroupMaster",
1264 "long_id": "myMasteredGroupedResource:1",
1265 "promote_on_node": "rh7-3",
1266 },
1267 },
1268 utils.get_resources_location_from_operations(cib_dom, operations),
1269 )
1270
1271 operations = [
1272 {
1273 "id": "myResource",
1274 "long_id": "myResource",
1275 "operation": "stop",
1276 "on_node": "rh7-1",
1277 },
1278 {
1279 "id": "myResource",
1280 "long_id": "myResource",
1281 "operation": "migrate_to",
1282 "on_node": "rh7-1",
1283 },
1284 {
1285 "id": "myResource",
1286 "long_id": "myResource",
1287 "operation": "migrate_from",
1288 "on_node": "rh7-2",
1289 },
1290 ]
1291 self.assertEqual(
1292 {
1293 "myResource": {
1294 "id": "myResource",
1295 "id_for_constraint": "myResource",
1296 "long_id": "myResource",
1297 "start_on_node": "rh7-2",
1298 },
1299 },
1300 utils.get_resources_location_from_operations(cib_dom, operations),
1301 )
1302
1303 def test_is_int(self):
1304 self.assertTrue(utils.is_int("-999"))
1305 self.assertTrue(utils.is_int("-1"))
1306 self.assertTrue(utils.is_int("0"))
1307 self.assertTrue(utils.is_int("1"))
1308 self.assertTrue(utils.is_int("99999"))
1309 self.assertTrue(utils.is_int(" 99999 "))
1310 self.assertFalse(utils.is_int("0.0"))
1311 self.assertFalse(utils.is_int("-1.0"))
1312 self.assertFalse(utils.is_int("-0.1"))
1313 self.assertFalse(utils.is_int("0.001"))
1314 self.assertFalse(utils.is_int("-999999.1"))
1315 self.assertFalse(utils.is_int("0.0001"))
1316 self.assertFalse(utils.is_int(""))
1317 self.assertFalse(utils.is_int(" "))
1318 self.assertFalse(utils.is_int("A"))
1319 self.assertFalse(utils.is_int("random 15 47 text "))
1320
1321 def test_dom_get_node(self):
1322 cib = self.get_cib_with_nodes_minidom()
1323 self.assertIsNone(utils.dom_get_node(cib, "non-existing-node"))
1324 node = utils.dom_get_node(cib, "rh7-1")
1325 self.assertEqual(node.getAttribute("uname"), "rh7-1")
1326 self.assertEqual(node.getAttribute("id"), "1")
1327
1328 def test_dom_prepare_child_element(self):
1329 cib = self.get_cib_with_nodes_minidom()
1330 node = cib.getElementsByTagName("node")[0]
1331 self.assertEqual(len(dom_get_child_elements(node)), 0)
1332 child = utils.dom_prepare_child_element(
1333 node, "utilization", "rh7-1-utilization"
1334 )
1335 self.assertEqual(len(dom_get_child_elements(node)), 1)
1336 self.assertEqual(child, dom_get_child_elements(node)[0])
1337 self.assertEqual(dom_get_child_elements(node)[0].tagName, "utilization")
1338 self.assertEqual(
1339 dom_get_child_elements(node)[0].getAttribute("id"),
1340 "rh7-1-utilization",
1341 )
1342 child2 = utils.dom_prepare_child_element(
1343 node, "utilization", "rh7-1-utilization"
1344 )
1345 self.assertEqual(len(dom_get_child_elements(node)), 1)
1346 self.assertEqual(child, child2)
1347
1348 def test_dom_update_nv_pair_add(self):
1349 nv_set = xml.dom.minidom.parseString("<nvset/>").documentElement
1350 utils.dom_update_nv_pair(nv_set, "test_name", "test_val", "prefix-")
1351 self.assertEqual(len(dom_get_child_elements(nv_set)), 1)
1352 pair = dom_get_child_elements(nv_set)[0]
1353 self.assertEqual(pair.getAttribute("name"), "test_name")
1354 self.assertEqual(pair.getAttribute("value"), "test_val")
1355 self.assertEqual(pair.getAttribute("id"), "prefix-test_name")
1356 utils.dom_update_nv_pair(nv_set, "another_name", "value", "prefix2-")
1357 self.assertEqual(len(dom_get_child_elements(nv_set)), 2)
1358 self.assertEqual(pair, dom_get_child_elements(nv_set)[0])
1359 pair = dom_get_child_elements(nv_set)[1]
1360 self.assertEqual(pair.getAttribute("name"), "another_name")
1361 self.assertEqual(pair.getAttribute("value"), "value")
1362 self.assertEqual(pair.getAttribute("id"), "prefix2-another_name")
1363
1364 def test_dom_update_nv_pair_update(self):
1365 nv_set = xml.dom.minidom.parseString(
1366 """
1367 <nv_set>
1368 <nvpair id="prefix-test_name" name="test_name" value="test_val"/>
1369 <nvpair id="prefix2-another_name" name="another_name" value="value"/>
1370 </nv_set>
1371 """
1372 ).documentElement
1373 utils.dom_update_nv_pair(nv_set, "test_name", "new_value")
1374 self.assertEqual(len(dom_get_child_elements(nv_set)), 2)
1375 pair1 = dom_get_child_elements(nv_set)[0]
1376 pair2 = dom_get_child_elements(nv_set)[1]
1377 self.assertEqual(pair1.getAttribute("name"), "test_name")
1378 self.assertEqual(pair1.getAttribute("value"), "new_value")
1379 self.assertEqual(pair1.getAttribute("id"), "prefix-test_name")
1380 self.assertEqual(pair2.getAttribute("name"), "another_name")
1381 self.assertEqual(pair2.getAttribute("value"), "value")
1382 self.assertEqual(pair2.getAttribute("id"), "prefix2-another_name")
1383
1384 def test_dom_update_nv_pair_remove(self):
1385 nv_set = xml.dom.minidom.parseString(
1386 """
1387 <nv_set>
1388 <nvpair id="prefix-test_name" name="test_name" value="test_val"/>
1389 <nvpair id="prefix2-another_name" name="another_name" value="value"/>
1390 </nv_set>
1391 """
1392 ).documentElement
1393 utils.dom_update_nv_pair(nv_set, "non_existing_name", "")
1394 self.assertEqual(len(dom_get_child_elements(nv_set)), 2)
1395 utils.dom_update_nv_pair(nv_set, "another_name", "")
1396 self.assertEqual(len(dom_get_child_elements(nv_set)), 1)
1397 pair = dom_get_child_elements(nv_set)[0]
1398 self.assertEqual(pair.getAttribute("name"), "test_name")
1399 self.assertEqual(pair.getAttribute("value"), "test_val")
1400 self.assertEqual(pair.getAttribute("id"), "prefix-test_name")
1401 utils.dom_update_nv_pair(nv_set, "test_name", "")
1402 self.assertEqual(len(dom_get_child_elements(nv_set)), 0)
1403
1404 def test_convert_args_to_tuples(self):
1405 out = utils.convert_args_to_tuples(
1406 ["invalid_string", "key=value", "key2=val=ue", "k e y= v a l u e "]
1407 )
1408 self.assertEqual(
1409 out,
1410 [("key", "value"), ("key2", "val=ue"), ("k e y", " v a l u e ")],
1411 )
1412
1413 def test_dom_update_utilization_invalid(self):
1414 # commands writes to stderr
1415 # we want clean test output, so we capture it
1416 tmp_stderr = sys.stderr
1417 sys.stderr = StringIO()
1418
1419 el = xml.dom.minidom.parseString(
1420 """
1421 <resource id="test_id"/>
1422 """
1423 ).documentElement
1424 self.assertRaises(
1425 SystemExit,
1426 utils.dom_update_utilization,
1427 el,
1428 {"name": "invalid_val"},
1429 )
1430
1431 self.assertRaises(
1432 SystemExit, utils.dom_update_utilization, el, {"name": "0.01"}
1433 )
1434
1435 sys.stderr = tmp_stderr
1436
1437 def test_dom_update_utilization_add(self):
1438 el = xml.dom.minidom.parseString(
1439 """
1440 <resource id="test_id"/>
1441 """
1442 ).documentElement
1443 utils.dom_update_utilization(
1444 el,
1445 {
1446 "name": "",
1447 "key": "-1",
1448 "keys": "90",
1449 },
1450 )
1451
1452 self.assertEqual(len(dom_get_child_elements(el)), 1)
1453 u = dom_get_child_elements(el)[0]
1454 self.assertEqual(u.tagName, "utilization")
1455 self.assertEqual(u.getAttribute("id"), "test_id-utilization")
1456 self.assertEqual(len(dom_get_child_elements(u)), 2)
1457
1458 self.assertEqual(
1459 dom_get_child_elements(u)[0].getAttribute("id"),
1460 "test_id-utilization-key",
1461 )
1462 self.assertEqual(
1463 dom_get_child_elements(u)[0].getAttribute("name"), "key"
1464 )
1465 self.assertEqual(
1466 dom_get_child_elements(u)[0].getAttribute("value"), "-1"
1467 )
1468 self.assertEqual(
1469 dom_get_child_elements(u)[1].getAttribute("id"),
1470 "test_id-utilization-keys",
1471 )
1472 self.assertEqual(
1473 dom_get_child_elements(u)[1].getAttribute("name"), "keys"
1474 )
1475 self.assertEqual(
1476 dom_get_child_elements(u)[1].getAttribute("value"), "90"
1477 )
1478
1479 def test_dom_update_utilization_update_remove(self):
1480 el = xml.dom.minidom.parseString(
1481 """
1482 <resource id="test_id">
1483 <utilization id="test_id-utilization">
1484 <nvpair id="test_id-utilization-key" name="key" value="-1"/>
1485 <nvpair id="test_id-utilization-keys" name="keys" value="90"/>
1486 </utilization>
1487 </resource>
1488 """
1489 ).documentElement
1490 utils.dom_update_utilization(
1491 el,
1492 {
1493 "key": "100",
1494 "keys": "",
1495 },
1496 )
1497
1498 u = dom_get_child_elements(el)[0]
1499 self.assertEqual(len(dom_get_child_elements(u)), 1)
1500 self.assertEqual(
1501 dom_get_child_elements(u)[0].getAttribute("id"),
1502 "test_id-utilization-key",
1503 )
1504 self.assertEqual(
1505 dom_get_child_elements(u)[0].getAttribute("name"), "key"
1506 )
1507 self.assertEqual(
1508 dom_get_child_elements(u)[0].getAttribute("value"), "100"
1509 )
1510
1511 def test_dom_update_meta_attr_add(self):
1512 el = xml.dom.minidom.parseString(
1513 """
1514 <resource id="test_id"/>
1515 """
1516 ).documentElement
1517 utils.dom_update_meta_attr(
1518 el, [("name", ""), ("key", "test"), ("key2", "val")]
1519 )
1520
1521 self.assertEqual(len(dom_get_child_elements(el)), 1)
1522 u = dom_get_child_elements(el)[0]
1523 self.assertEqual(u.tagName, "meta_attributes")
1524 self.assertEqual(u.getAttribute("id"), "test_id-meta_attributes")
1525 self.assertEqual(len(dom_get_child_elements(u)), 2)
1526
1527 self.assertEqual(
1528 dom_get_child_elements(u)[0].getAttribute("id"),
1529 "test_id-meta_attributes-key",
1530 )
1531 self.assertEqual(
1532 dom_get_child_elements(u)[0].getAttribute("name"), "key"
1533 )
1534 self.assertEqual(
1535 dom_get_child_elements(u)[0].getAttribute("value"), "test"
1536 )
1537 self.assertEqual(
1538 dom_get_child_elements(u)[1].getAttribute("id"),
1539 "test_id-meta_attributes-key2",
1540 )
1541 self.assertEqual(
1542 dom_get_child_elements(u)[1].getAttribute("name"), "key2"
1543 )
1544 self.assertEqual(
1545 dom_get_child_elements(u)[1].getAttribute("value"), "val"
1546 )
1547
1548 def test_dom_update_meta_attr_update_remove(self):
|
CID (unavailable; MK=ddf215f16526713f8be9ad03b4d4fe19) (#1 of 1): XML external entity processing enabled (SIGMA.xml_external_entity_enabled): |
|
(1) Event Sigma main event: |
The application uses Python's built in `xml` module which does not properly handle erroneous or maliciously constructed data, making the application vulnerable to one or more types of XML attacks. |
|
(2) Event remediation: |
Avoid using the `xml` module. Consider using the `defusedxml` module or similar which safely prevents all XML entity attacks. |
1549 el = xml.dom.minidom.parseString(
1550 """
1551 <resource id="test_id">
1552 <meta_attributes id="test_id-utilization">
1553 <nvpair id="test_id-meta_attributes-key" name="key" value="test"/>
1554 <nvpair id="test_id-meta_attributes-key2" name="key2" value="val"/>
1555 </meta_attributes>
1556 </resource>
1557 """
1558 ).documentElement
1559 utils.dom_update_meta_attr(el, [("key", "another_val"), ("key2", "")])
1560
1561 u = dom_get_child_elements(el)[0]
1562 self.assertEqual(len(dom_get_child_elements(u)), 1)
1563 self.assertEqual(
1564 dom_get_child_elements(u)[0].getAttribute("id"),
1565 "test_id-meta_attributes-key",
1566 )
1567 self.assertEqual(
1568 dom_get_child_elements(u)[0].getAttribute("name"), "key"
1569 )
1570 self.assertEqual(
1571 dom_get_child_elements(u)[0].getAttribute("value"), "another_val"
1572 )
1573
1574 def test_get_utilization(self):
1575 el = xml.dom.minidom.parseString(
1576 """
1577 <resource id="test_id">
1578 <utilization id="test_id-utilization">
1579 <nvpair id="test_id-utilization-key" name="key" value="-1"/>
1580 <nvpair id="test_id-utilization-keys" name="keys" value="90"/>
1581 </utilization>
1582 </resource>
1583 """
1584 ).documentElement
1585 self.assertEqual({"key": "-1", "keys": "90"}, utils.get_utilization(el))
1586
1587 def test_get_utilization_str(self):
1588 el = xml.dom.minidom.parseString(
1589 """
1590 <resource id="test_id">
1591 <utilization id="test_id-utilization">
1592 <nvpair id="test_id-utilization-key" name="key" value="-1"/>
1593 <nvpair id="test_id-utilization-keys" name="keys" value="90"/>
1594 </utilization>
1595 </resource>
1596 """
1597 ).documentElement
1598 self.assertEqual("key=-1 keys=90", utils.get_utilization_str(el))
1599
1600 def assert_element_id(self, node, node_id, tag=None):
1601 self.assertTrue(
1602 isinstance(node, xml.dom.minidom.Element),
1603 "element with id '%s' not found" % node_id,
1604 )
1605 self.assertEqual(node.getAttribute("id"), node_id)
1606 if tag:
1607 self.assertEqual(node.tagName, tag)
1608
1609
1610 class RunParallelTest(TestCase):
1611 @staticmethod
1612 def fixture_create_worker(log, name, sleepSeconds=0):
1613 def worker():
1614 sleep(sleepSeconds)
1615 log.append(name)
1616
1617 return worker
1618
1619 def test_run_all_workers(self):
1620 log = []
1621 utils.run_parallel(
1622 [
1623 self.fixture_create_worker(log, "first"),
1624 self.fixture_create_worker(log, "second"),
1625 ],
1626 wait_seconds=0.1,
1627 )
1628
1629 self.assertEqual(sorted(log), sorted(["first", "second"]))
1630
1631
1632 class NodeActionTaskTest(TestCase):
1633 def test_can_run_action(self):
1634 def action(node, arg, kwarg=None):
1635 return (0, ":".join([node, arg, kwarg]))
1636
1637 report_list = []
1638
1639 def report(node, returncode, output):
1640 report_list.append("|".join([node, str(returncode), output]))
1641
1642 task = utils.create_task(report, action, "node", "arg", kwarg="kwarg")
1643 task()
1644
1645 self.assertEqual(["node|0|node:arg:kwarg"], report_list)
1646
1647
1648 class TouchCibFile(TestCase):
1649 @mock.patch("pcs.utils.os.path.isfile", mock.Mock(return_value=False))
1650 @mock.patch(
1651 "pcs.utils.write_empty_cib",
1652 mock.Mock(side_effect=OSError("some message")),
1653 )
1654 @mock.patch("pcs.utils.err")
1655 def test_exception_is_transformed_correctly(self, err):
1656 filename = "/fake/filename"
1657 utils.touch_cib_file(filename)
1658 err.assert_called_once_with(
1659 "Unable to write to file: '/fake/filename': 'some message'"
1660 )
1661
1662
1663 @mock.patch("pcs.utils.is_run_interactive", mock.Mock(return_value=False))
1664 class GetContinueConfirmation(TestCase):
1665 def setUp(self):
1666 self.text = "some warning text"
1667 patcher_output = mock.patch("pcs.cli.reports.output.print_to_stderr")
1668 self.addCleanup(patcher_output.stop)
1669 self.mock_output = patcher_output.start()
1670
1671 def test_yes_force(self):
1672 self.assertTrue(utils.get_continue_confirmation(self.text, True, True))
1673 self.mock_output.assert_called_once_with(f"Warning: {self.text}")
1674
1675 def test_yes(self):
1676 self.assertTrue(utils.get_continue_confirmation(self.text, True, False))
1677 self.mock_output.assert_called_once_with(f"Warning: {self.text}")
1678
1679 def test_force(self):
1680 self.assertTrue(utils.get_continue_confirmation(self.text, False, True))
1681 self.assertEqual(
1682 self.mock_output.mock_calls,
1683 [
1684 mock.call(
1685 "Deprecation Warning: Using --force to confirm this action "
1686 "is deprecated and might be removed in a future release, "
1687 "use --yes instead"
1688 ),
1689 mock.call(f"Warning: {self.text}"),
1690 ],
1691 )
1692
1693 def test_nothing(self):
1694 with self.assertRaises(SystemExit) as cm:
1695 self.assertFalse(
1696 utils.get_continue_confirmation(self.text, False, False)
1697 )
1698 self.assertEqual(cm.exception.code, 1)
1699 self.mock_output.assert_called_once_with(
1700 f"Error: {self.text}, use --yes to override"
1701 )
1702