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