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():
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 get_cib_status_lrm(self):
698 cib_dom = self.get_cib_empty()
699 new_status = xml.dom.minidom.parseString(
700 """
701 <status>
702 <node_state id="1" uname="rh70-node1">
703 <lrm id="1">
704 <lrm_resources>
705 <lrm_resource id="dummy" type="Dummy" class="ocf" provider="heartbeat">
706 <lrm_rsc_op id="dummy_monitor_30000" operation="monitor" call-id="34"
707 rc-code="1" on_node="Xrh70-node1X" exit-reason="test" />
708 <lrm_rsc_op id="dummy_stop_0" operation="stop" call-id="32"
709 rc-code="0" />
710 <lrm_rsc_op id="dummy_start_0" operation="start" call-id="33"
711 rc-code="0" />
712 </lrm_resource>
713 </lrm_resources>
714 </lrm>
715 </node_state>
716 <node_state id="2" uname="rh70-node2">
717 <lrm id="2">
718 <lrm_resources>
719 <lrm_resource id="dummy" type="Dummy" class="ocf" provider="heartbeat">
720 <lrm_rsc_op id="dummy_monitor_0" operation="monitor" call-id="5"
721 rc-code="1" />
722 </lrm_resource>
723 <lrm_resource id="dummy1" type="Dummy" class="ocf" provider="heartbeat">
724 <lrm_rsc_op id="dummy1_monitor_0" operation="monitor" call-id="3"
725 rc-code="0" />
726 </lrm_resource>
727 </lrm_resources>
728 </lrm>
729 </node_state>
730 </status>
731 """
732 ).documentElement
733 status = cib_dom.getElementsByTagName("status")[0]
734 status.parentNode.replaceChild(new_status, status)
735 return cib_dom
736
737 def test_resource_running_on(self):
738 status = xml.dom.minidom.parseString(
739 f"""
740 <crm_mon>
741 <summary />
742 <nodes />
743 <resources>
744 <resource id="myResource" role="Started">
745 <node name="rh70-node1" />
746 </resource>
747 <clone id="myClone">
748 <resource id="myClonedResource" role="Started">
749 <node name="rh70-node1" />
750 </resource>
751 <resource id="myClonedResource" role="Started">
752 <node name="rh70-node2" />
753 </resource>
754 <resource id="myClonedResource" role="Started">
755 <node name="rh70-node3" />
756 </resource>
757 </clone>
758 <clone id="myMaster">
759 <resource id="myMasteredResource:1" role="{const.PCMK_ROLE_UNPROMOTED}">
760 <node name="rh70-node2" />
761 </resource>
762 <resource id="myMasteredResource" role="{const.PCMK_ROLE_UNPROMOTED}">
763 <node name="rh70-node3" />
764 </resource>
765 <resource id="myMasteredResource" role="{const.PCMK_ROLE_PROMOTED}">
766 <node name="rh70-node1" />
767 </resource>
768 </clone>
769 <group id="myGroup">
770 <resource id="myGroupedResource" role="Started">
771 <node name="rh70-node2" />
772 </resource>
773 </group>
774 <clone id="myGroupClone">
775 <group id="myClonedGroup:0">
776 <resource id="myClonedGroupedResource" role="Started">
777 <node name="rh70-node1" />
778 </resource>
779 </group>
780 <group id="myClonedGroup:1">
781 <resource id="myClonedGroupedResource" role="Started">
782 <node name="rh70-node2" />
783 </resource>
784 </group>
785 <group id="myClonedGroup:2">
786 <resource id="myClonedGroupedResource" role="Started">
787 <node name="rh70-node3" />
788 </resource>
789 </group>
790 <group id="myClonedGroup:3">
791 <resource id="myClonedGroupedResource" role="Started">
792 <node name="rh70-node3" />
793 </resource>
794 </group>
795 </clone>
796 <clone id="myGroupMaster">
797 <group id="myMasteredGroup:0">
798 <resource id="myMasteredGroupedResource" role="{const.PCMK_ROLE_UNPROMOTED}">
799 <node name="rh70-node1" />
800 </resource>
801 </group>
802 <group id="myMasteredGroup:1">
803 <resource id="myMasteredGroupedResource" role="{const.PCMK_ROLE_PROMOTED}">
804 <node name="rh70-node2" />
805 </resource>
806 </group>
807 <group id="myMasteredGroup:2">
808 <resource id="myMasteredGroupedResource" role="{const.PCMK_ROLE_UNPROMOTED}">
809 <node name="rh70-node3" />
810 </resource>
811 </group>
812 </clone>
813 <resource id="myStoppedResource" role="Stopped">
814 </resource>
815 </resources>
816 </crm_mon>
817 """
818 ).documentElement
819
820 self.assertEqual(
821 utils.resource_running_on("myResource", status),
822 {
823 "message": "Resource 'myResource' is running on node rh70-node1.",
824 "is_running": True,
825 },
826 )
827 self.assertEqual(
828 utils.resource_running_on("myClonedResource", status),
829 {
830 "message": "Resource 'myClonedResource' is running on nodes "
831 "rh70-node1, rh70-node2, rh70-node3.",
832 "is_running": True,
833 },
834 )
835 self.assertEqual(
836 utils.resource_running_on("myClone", status),
837 {
838 "message": "Resource 'myClone' is running on nodes "
839 "rh70-node1, rh70-node2, rh70-node3.",
840 "is_running": True,
841 },
842 )
843 self.assertEqual(
844 utils.resource_running_on("myMasteredResource", status),
845 {
846 "message": (
847 "Resource 'myMasteredResource' is {promoted} on node "
848 "rh70-node1; {unpromoted} on nodes rh70-node2, rh70-node3."
849 ).format(
850 promoted=str(const.PCMK_ROLE_PROMOTED).lower(),
851 unpromoted=str(const.PCMK_ROLE_UNPROMOTED).lower(),
852 ),
853 "is_running": True,
854 },
855 )
856 self.assertEqual(
857 utils.resource_running_on("myMaster", status),
858 {
859 "message": (
860 "Resource 'myMaster' is {promoted} on node "
861 "rh70-node1; {unpromoted} on nodes rh70-node2, rh70-node3."
862 ).format(
863 promoted=str(const.PCMK_ROLE_PROMOTED).lower(),
864 unpromoted=str(const.PCMK_ROLE_UNPROMOTED).lower(),
865 ),
866 "is_running": True,
867 },
868 )
869 self.assertEqual(
870 utils.resource_running_on("myGroupedResource", status),
871 {
872 "message": "Resource 'myGroupedResource' is running on node "
873 "rh70-node2.",
874 "is_running": True,
875 },
876 )
877 self.assertEqual(
878 utils.resource_running_on("myGroup", status),
879 {
880 "message": "Resource 'myGroup' is running on node rh70-node2.",
881 "is_running": True,
882 },
883 )
884 self.assertEqual(
885 utils.resource_running_on("myClonedGroupedResource", status),
886 {
887 "message": "Resource 'myClonedGroupedResource' is running on nodes "
888 "rh70-node1, rh70-node2, rh70-node3, rh70-node3.",
889 "is_running": True,
890 },
891 )
892 self.assertEqual(
893 utils.resource_running_on("myClonedGroup", status),
894 {
895 "message": "Resource 'myClonedGroup' is running on nodes "
896 "rh70-node1, rh70-node2, rh70-node3, rh70-node3.",
897 "is_running": True,
898 },
899 )
900 self.assertEqual(
901 utils.resource_running_on("myGroupClone", status),
902 {
903 "message": "Resource 'myGroupClone' is running on nodes "
904 "rh70-node1, rh70-node2, rh70-node3, rh70-node3.",
905 "is_running": True,
906 },
907 )
908 self.assertEqual(
909 utils.resource_running_on("myMasteredGroupedResource", status),
910 {
911 "message": (
912 "Resource 'myMasteredGroupedResource' is {promoted} on node "
913 "rh70-node2; {unpromoted} on nodes rh70-node1, rh70-node3."
914 ).format(
915 promoted=str(const.PCMK_ROLE_PROMOTED).lower(),
916 unpromoted=str(const.PCMK_ROLE_UNPROMOTED).lower(),
917 ),
918 "is_running": True,
919 },
920 )
921 self.assertEqual(
922 utils.resource_running_on("myMasteredGroup", status),
923 {
924 "message": (
925 "Resource 'myMasteredGroup' is {promoted} on node "
926 "rh70-node2; {unpromoted} on nodes rh70-node1, rh70-node3."
927 ).format(
928 promoted=str(const.PCMK_ROLE_PROMOTED).lower(),
929 unpromoted=str(const.PCMK_ROLE_UNPROMOTED).lower(),
930 ),
931 "is_running": True,
932 },
933 )
934 self.assertEqual(
935 utils.resource_running_on("myGroupMaster", status),
936 {
937 "message": (
938 "Resource 'myGroupMaster' is {promoted} on node "
939 "rh70-node2; {unpromoted} on nodes rh70-node1, rh70-node3."
940 ).format(
941 promoted=str(const.PCMK_ROLE_PROMOTED).lower(),
942 unpromoted=str(const.PCMK_ROLE_UNPROMOTED).lower(),
943 ),
944 "is_running": True,
945 },
946 )
947 self.assertEqual(
948 utils.resource_running_on("notMyResource", status),
949 {
950 "message": "Resource 'notMyResource' is not running on any node",
951 "is_running": False,
952 },
953 )
954 self.assertEqual(
955 utils.resource_running_on("myStoppedResource", status),
956 {
957 "message": "Resource 'myStoppedResource' is not running on any node",
958 "is_running": False,
959 },
960 )
961
962 def test_get_operations_from_transitions(self):
963 transitions = xml.dom.minidom.parse(rc("transitions01.xml"))
964 self.assertEqual(
965 [
966 {
967 "id": "dummy",
968 "long_id": "dummy",
969 "operation": "stop",
970 "on_node": "rh7-3",
971 },
972 {
973 "id": "dummy",
974 "long_id": "dummy",
975 "operation": "start",
976 "on_node": "rh7-2",
977 },
978 {
979 "id": "d0",
980 "long_id": "d0:1",
981 "operation": "stop",
982 "on_node": "rh7-1",
983 },
984 {
985 "id": "d0",
986 "long_id": "d0:1",
987 "operation": "start",
988 "on_node": "rh7-2",
989 },
990 {
991 "id": "state",
992 "long_id": "state:0",
993 "operation": "stop",
994 "on_node": "rh7-3",
995 },
996 {
997 "id": "state",
998 "long_id": "state:0",
999 "operation": "start",
1000 "on_node": "rh7-2",
1001 },
1002 ],
1003 utils.get_operations_from_transitions(transitions),
1004 )
1005
1006 transitions = xml.dom.minidom.parse(rc("transitions02.xml"))
1007 self.assertEqual(
1008 [
1009 {
1010 "id": "RemoteNode",
1011 "long_id": "RemoteNode",
1012 "operation": "stop",
1013 "on_node": "virt-143",
1014 },
1015 {
1016 "id": "RemoteNode",
1017 "long_id": "RemoteNode",
1018 "operation": "migrate_to",
1019 "on_node": "virt-143",
1020 },
1021 {
1022 "id": "RemoteNode",
1023 "long_id": "RemoteNode",
1024 "operation": "migrate_from",
1025 "on_node": "virt-142",
1026 },
1027 {
1028 "id": "dummy8",
1029 "long_id": "dummy8",
1030 "operation": "stop",
1031 "on_node": "virt-143",
1032 },
1033 {
1034 "id": "dummy8",
1035 "long_id": "dummy8",
1036 "operation": "start",
1037 "on_node": "virt-142",
1038 },
1039 ],
1040 utils.get_operations_from_transitions(transitions),
1041 )
1042
1043 def test_get_resources_location_from_operations(self):
1044 cib_dom = self.get_cib_resources()
1045
1046 operations = []
1047 self.assertEqual(
1048 {},
1049 utils.get_resources_location_from_operations(cib_dom, operations),
1050 )
1051
1052 operations = [
1053 {
1054 "id": "myResource",
1055 "long_id": "myResource",
1056 "operation": "start",
1057 "on_node": "rh7-1",
1058 },
1059 ]
1060 self.assertEqual(
1061 {
1062 "myResource": {
1063 "id": "myResource",
1064 "id_for_constraint": "myResource",
1065 "long_id": "myResource",
1066 "start_on_node": "rh7-1",
1067 },
1068 },
1069 utils.get_resources_location_from_operations(cib_dom, operations),
1070 )
1071
1072 operations = [
1073 {
1074 "id": "myResource",
1075 "long_id": "myResource",
1076 "operation": "start",
1077 "on_node": "rh7-1",
1078 },
1079 {
1080 "id": "myResource",
1081 "long_id": "myResource",
1082 "operation": "start",
1083 "on_node": "rh7-2",
1084 },
1085 {
1086 "id": "myResource",
1087 "long_id": "myResource",
1088 "operation": "monitor",
1089 "on_node": "rh7-3",
1090 },
1091 {
1092 "id": "myResource",
1093 "long_id": "myResource",
1094 "operation": "stop",
1095 "on_node": "rh7-3",
1096 },
1097 ]
1098 self.assertEqual(
1099 {
1100 "myResource": {
1101 "id": "myResource",
1102 "id_for_constraint": "myResource",
1103 "long_id": "myResource",
1104 "start_on_node": "rh7-2",
1105 },
1106 },
1107 utils.get_resources_location_from_operations(cib_dom, operations),
1108 )
1109
1110 operations = [
1111 {
1112 "id": "myResource",
1113 "long_id": "myResource",
1114 "operation": "start",
1115 "on_node": "rh7-1",
1116 },
1117 {
1118 "id": "myClonedResource",
1119 "long_id": "myClonedResource:0",
1120 "operation": "start",
1121 "on_node": "rh7-1",
1122 },
1123 {
1124 "id": "myClonedResource",
1125 "long_id": "myClonedResource:0",
1126 "operation": "start",
1127 "on_node": "rh7-2",
1128 },
1129 {
1130 "id": "myClonedResource",
1131 "long_id": "myClonedResource:1",
1132 "operation": "start",
1133 "on_node": "rh7-3",
1134 },
1135 ]
1136 self.assertEqual(
1137 {
1138 "myResource": {
1139 "id": "myResource",
1140 "id_for_constraint": "myResource",
1141 "long_id": "myResource",
1142 "start_on_node": "rh7-1",
1143 },
1144 "myClonedResource:0": {
1145 "id": "myClonedResource",
1146 "id_for_constraint": "myClone",
1147 "long_id": "myClonedResource:0",
1148 "start_on_node": "rh7-2",
1149 },
1150 "myClonedResource:1": {
1151 "id": "myClonedResource",
1152 "id_for_constraint": "myClone",
1153 "long_id": "myClonedResource:1",
1154 "start_on_node": "rh7-3",
1155 },
1156 },
1157 utils.get_resources_location_from_operations(cib_dom, operations),
1158 )
1159
1160 operations = [
1161 {
1162 "id": "myUniqueClonedResource:0",
1163 "long_id": "myUniqueClonedResource:0",
1164 "operation": "start",
1165 "on_node": "rh7-1",
1166 },
1167 {
1168 "id": "myUniqueClonedResource:1",
1169 "long_id": "myUniqueClonedResource:1",
1170 "operation": "monitor",
1171 "on_node": "rh7-2",
1172 },
1173 {
1174 "id": "myUniqueClonedResource:2",
1175 "long_id": "myUniqueClonedResource:2",
1176 "operation": "start",
1177 "on_node": "rh7-3",
1178 },
1179 ]
1180 self.assertEqual(
1181 {
1182 "myUniqueClonedResource:0": {
1183 "id": "myUniqueClonedResource:0",
1184 "id_for_constraint": "myUniqueClone",
1185 "long_id": "myUniqueClonedResource:0",
1186 "start_on_node": "rh7-1",
1187 },
1188 "myUniqueClonedResource:2": {
1189 "id": "myUniqueClonedResource:2",
1190 "id_for_constraint": "myUniqueClone",
1191 "long_id": "myUniqueClonedResource:2",
1192 "start_on_node": "rh7-3",
1193 },
1194 },
1195 utils.get_resources_location_from_operations(cib_dom, operations),
1196 )
1197
1198 operations = [
1199 {
1200 "id": "myMasteredGroupedResource",
1201 "long_id": "myMasteredGroupedResource:0",
1202 "operation": "start",
1203 "on_node": "rh7-1",
1204 },
1205 {
1206 "id": "myMasteredGroupedResource",
1207 "long_id": "myMasteredGroupedResource:1",
1208 "operation": "demote",
1209 "on_node": "rh7-2",
1210 },
1211 {
1212 "id": "myMasteredGroupedResource",
1213 "long_id": "myMasteredGroupedResource:1",
1214 "operation": "promote",
1215 "on_node": "rh7-3",
1216 },
1217 ]
1218 self.assertEqual(
1219 {
1220 "myMasteredGroupedResource:0": {
1221 "id": "myMasteredGroupedResource",
1222 "id_for_constraint": "myGroupMaster",
1223 "long_id": "myMasteredGroupedResource:0",
1224 "start_on_node": "rh7-1",
1225 },
1226 "myMasteredGroupedResource:1": {
1227 "id": "myMasteredGroupedResource",
1228 "id_for_constraint": "myGroupMaster",
1229 "long_id": "myMasteredGroupedResource:1",
1230 "promote_on_node": "rh7-3",
1231 },
1232 },
1233 utils.get_resources_location_from_operations(cib_dom, operations),
1234 )
1235
1236 operations = [
1237 {
1238 "id": "myResource",
1239 "long_id": "myResource",
1240 "operation": "stop",
1241 "on_node": "rh7-1",
1242 },
1243 {
1244 "id": "myResource",
1245 "long_id": "myResource",
1246 "operation": "migrate_to",
1247 "on_node": "rh7-1",
1248 },
1249 {
1250 "id": "myResource",
1251 "long_id": "myResource",
1252 "operation": "migrate_from",
1253 "on_node": "rh7-2",
1254 },
1255 ]
1256 self.assertEqual(
1257 {
1258 "myResource": {
1259 "id": "myResource",
1260 "id_for_constraint": "myResource",
1261 "long_id": "myResource",
1262 "start_on_node": "rh7-2",
1263 },
1264 },
1265 utils.get_resources_location_from_operations(cib_dom, operations),
1266 )
1267
1268 def test_is_int(self):
1269 self.assertTrue(utils.is_int("-999"))
1270 self.assertTrue(utils.is_int("-1"))
1271 self.assertTrue(utils.is_int("0"))
1272 self.assertTrue(utils.is_int("1"))
1273 self.assertTrue(utils.is_int("99999"))
1274 self.assertTrue(utils.is_int(" 99999 "))
1275 self.assertFalse(utils.is_int("0.0"))
1276 self.assertFalse(utils.is_int("-1.0"))
1277 self.assertFalse(utils.is_int("-0.1"))
1278 self.assertFalse(utils.is_int("0.001"))
1279 self.assertFalse(utils.is_int("-999999.1"))
1280 self.assertFalse(utils.is_int("0.0001"))
1281 self.assertFalse(utils.is_int(""))
1282 self.assertFalse(utils.is_int(" "))
1283 self.assertFalse(utils.is_int("A"))
1284 self.assertFalse(utils.is_int("random 15 47 text "))
1285
1286 def test_dom_get_node(self):
1287 cib = self.get_cib_with_nodes_minidom()
1288 self.assertIsNone(utils.dom_get_node(cib, "non-existing-node"))
1289 node = utils.dom_get_node(cib, "rh7-1")
1290 self.assertEqual(node.getAttribute("uname"), "rh7-1")
1291 self.assertEqual(node.getAttribute("id"), "1")
1292
1293 def test_dom_prepare_child_element(self):
1294 cib = self.get_cib_with_nodes_minidom()
1295 node = cib.getElementsByTagName("node")[0]
1296 self.assertEqual(len(dom_get_child_elements(node)), 0)
1297 child = utils.dom_prepare_child_element(
1298 node, "utilization", "rh7-1-utilization"
1299 )
1300 self.assertEqual(len(dom_get_child_elements(node)), 1)
1301 self.assertEqual(child, dom_get_child_elements(node)[0])
1302 self.assertEqual(dom_get_child_elements(node)[0].tagName, "utilization")
1303 self.assertEqual(
1304 dom_get_child_elements(node)[0].getAttribute("id"),
1305 "rh7-1-utilization",
1306 )
1307 child2 = utils.dom_prepare_child_element(
1308 node, "utilization", "rh7-1-utilization"
1309 )
1310 self.assertEqual(len(dom_get_child_elements(node)), 1)
1311 self.assertEqual(child, child2)
1312
1313 def test_dom_update_nv_pair_add(self):
|
CID (unavailable; MK=1352971f7419d2e406aa2f4c36659464) (#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. |
1314 nv_set = xml.dom.minidom.parseString("<nvset/>").documentElement
1315 utils.dom_update_nv_pair(nv_set, "test_name", "test_val", "prefix-")
1316 self.assertEqual(len(dom_get_child_elements(nv_set)), 1)
1317 pair = dom_get_child_elements(nv_set)[0]
1318 self.assertEqual(pair.getAttribute("name"), "test_name")
1319 self.assertEqual(pair.getAttribute("value"), "test_val")
1320 self.assertEqual(pair.getAttribute("id"), "prefix-test_name")
1321 utils.dom_update_nv_pair(nv_set, "another_name", "value", "prefix2-")
1322 self.assertEqual(len(dom_get_child_elements(nv_set)), 2)
1323 self.assertEqual(pair, dom_get_child_elements(nv_set)[0])
1324 pair = dom_get_child_elements(nv_set)[1]
1325 self.assertEqual(pair.getAttribute("name"), "another_name")
1326 self.assertEqual(pair.getAttribute("value"), "value")
1327 self.assertEqual(pair.getAttribute("id"), "prefix2-another_name")
1328
1329 def test_dom_update_nv_pair_update(self):
1330 nv_set = xml.dom.minidom.parseString(
1331 """
1332 <nv_set>
1333 <nvpair id="prefix-test_name" name="test_name" value="test_val"/>
1334 <nvpair id="prefix2-another_name" name="another_name" value="value"/>
1335 </nv_set>
1336 """
1337 ).documentElement
1338 utils.dom_update_nv_pair(nv_set, "test_name", "new_value")
1339 self.assertEqual(len(dom_get_child_elements(nv_set)), 2)
1340 pair1 = dom_get_child_elements(nv_set)[0]
1341 pair2 = dom_get_child_elements(nv_set)[1]
1342 self.assertEqual(pair1.getAttribute("name"), "test_name")
1343 self.assertEqual(pair1.getAttribute("value"), "new_value")
1344 self.assertEqual(pair1.getAttribute("id"), "prefix-test_name")
1345 self.assertEqual(pair2.getAttribute("name"), "another_name")
1346 self.assertEqual(pair2.getAttribute("value"), "value")
1347 self.assertEqual(pair2.getAttribute("id"), "prefix2-another_name")
1348
1349 def test_dom_update_nv_pair_remove(self):
1350 nv_set = xml.dom.minidom.parseString(
1351 """
1352 <nv_set>
1353 <nvpair id="prefix-test_name" name="test_name" value="test_val"/>
1354 <nvpair id="prefix2-another_name" name="another_name" value="value"/>
1355 </nv_set>
1356 """
1357 ).documentElement
1358 utils.dom_update_nv_pair(nv_set, "non_existing_name", "")
1359 self.assertEqual(len(dom_get_child_elements(nv_set)), 2)
1360 utils.dom_update_nv_pair(nv_set, "another_name", "")
1361 self.assertEqual(len(dom_get_child_elements(nv_set)), 1)
1362 pair = dom_get_child_elements(nv_set)[0]
1363 self.assertEqual(pair.getAttribute("name"), "test_name")
1364 self.assertEqual(pair.getAttribute("value"), "test_val")
1365 self.assertEqual(pair.getAttribute("id"), "prefix-test_name")
1366 utils.dom_update_nv_pair(nv_set, "test_name", "")
1367 self.assertEqual(len(dom_get_child_elements(nv_set)), 0)
1368
1369 def test_convert_args_to_tuples(self):
1370 out = utils.convert_args_to_tuples(
1371 ["invalid_string", "key=value", "key2=val=ue", "k e y= v a l u e "]
1372 )
1373 self.assertEqual(
1374 out,
1375 [("key", "value"), ("key2", "val=ue"), ("k e y", " v a l u e ")],
1376 )
1377
1378 def test_dom_update_utilization_invalid(self):
1379 # commands writes to stderr
1380 # we want clean test output, so we capture it
1381 tmp_stderr = sys.stderr
1382 sys.stderr = StringIO()
1383
1384 el = xml.dom.minidom.parseString(
1385 """
1386 <resource id="test_id"/>
1387 """
1388 ).documentElement
1389 self.assertRaises(
1390 SystemExit,
1391 utils.dom_update_utilization,
1392 el,
1393 {"name": "invalid_val"},
1394 )
1395
1396 self.assertRaises(
1397 SystemExit, utils.dom_update_utilization, el, {"name": "0.01"}
1398 )
1399
1400 sys.stderr = tmp_stderr
1401
1402 def test_dom_update_utilization_add(self):
1403 el = xml.dom.minidom.parseString(
1404 """
1405 <resource id="test_id"/>
1406 """
1407 ).documentElement
1408 utils.dom_update_utilization(
1409 el,
1410 {
1411 "name": "",
1412 "key": "-1",
1413 "keys": "90",
1414 },
1415 )
1416
1417 self.assertEqual(len(dom_get_child_elements(el)), 1)
1418 u = dom_get_child_elements(el)[0]
1419 self.assertEqual(u.tagName, "utilization")
1420 self.assertEqual(u.getAttribute("id"), "test_id-utilization")
1421 self.assertEqual(len(dom_get_child_elements(u)), 2)
1422
1423 self.assertEqual(
1424 dom_get_child_elements(u)[0].getAttribute("id"),
1425 "test_id-utilization-key",
1426 )
1427 self.assertEqual(
1428 dom_get_child_elements(u)[0].getAttribute("name"), "key"
1429 )
1430 self.assertEqual(
1431 dom_get_child_elements(u)[0].getAttribute("value"), "-1"
1432 )
1433 self.assertEqual(
1434 dom_get_child_elements(u)[1].getAttribute("id"),
1435 "test_id-utilization-keys",
1436 )
1437 self.assertEqual(
1438 dom_get_child_elements(u)[1].getAttribute("name"), "keys"
1439 )
1440 self.assertEqual(
1441 dom_get_child_elements(u)[1].getAttribute("value"), "90"
1442 )
1443
1444 def test_dom_update_utilization_update_remove(self):
1445 el = xml.dom.minidom.parseString(
1446 """
1447 <resource id="test_id">
1448 <utilization id="test_id-utilization">
1449 <nvpair id="test_id-utilization-key" name="key" value="-1"/>
1450 <nvpair id="test_id-utilization-keys" name="keys" value="90"/>
1451 </utilization>
1452 </resource>
1453 """
1454 ).documentElement
1455 utils.dom_update_utilization(
1456 el,
1457 {
1458 "key": "100",
1459 "keys": "",
1460 },
1461 )
1462
1463 u = dom_get_child_elements(el)[0]
1464 self.assertEqual(len(dom_get_child_elements(u)), 1)
1465 self.assertEqual(
1466 dom_get_child_elements(u)[0].getAttribute("id"),
1467 "test_id-utilization-key",
1468 )
1469 self.assertEqual(
1470 dom_get_child_elements(u)[0].getAttribute("name"), "key"
1471 )
1472 self.assertEqual(
1473 dom_get_child_elements(u)[0].getAttribute("value"), "100"
1474 )
1475
1476 def test_dom_update_meta_attr_add(self):
1477 el = xml.dom.minidom.parseString(
1478 """
1479 <resource id="test_id"/>
1480 """
1481 ).documentElement
1482 utils.dom_update_meta_attr(
1483 el, [("name", ""), ("key", "test"), ("key2", "val")]
1484 )
1485
1486 self.assertEqual(len(dom_get_child_elements(el)), 1)
1487 u = dom_get_child_elements(el)[0]
1488 self.assertEqual(u.tagName, "meta_attributes")
1489 self.assertEqual(u.getAttribute("id"), "test_id-meta_attributes")
1490 self.assertEqual(len(dom_get_child_elements(u)), 2)
1491
1492 self.assertEqual(
1493 dom_get_child_elements(u)[0].getAttribute("id"),
1494 "test_id-meta_attributes-key",
1495 )
1496 self.assertEqual(
1497 dom_get_child_elements(u)[0].getAttribute("name"), "key"
1498 )
1499 self.assertEqual(
1500 dom_get_child_elements(u)[0].getAttribute("value"), "test"
1501 )
1502 self.assertEqual(
1503 dom_get_child_elements(u)[1].getAttribute("id"),
1504 "test_id-meta_attributes-key2",
1505 )
1506 self.assertEqual(
1507 dom_get_child_elements(u)[1].getAttribute("name"), "key2"
1508 )
1509 self.assertEqual(
1510 dom_get_child_elements(u)[1].getAttribute("value"), "val"
1511 )
1512
1513 def test_dom_update_meta_attr_update_remove(self):
1514 el = xml.dom.minidom.parseString(
1515 """
1516 <resource id="test_id">
1517 <meta_attributes id="test_id-utilization">
1518 <nvpair id="test_id-meta_attributes-key" name="key" value="test"/>
1519 <nvpair id="test_id-meta_attributes-key2" name="key2" value="val"/>
1520 </meta_attributes>
1521 </resource>
1522 """
1523 ).documentElement
1524 utils.dom_update_meta_attr(el, [("key", "another_val"), ("key2", "")])
1525
1526 u = dom_get_child_elements(el)[0]
1527 self.assertEqual(len(dom_get_child_elements(u)), 1)
1528 self.assertEqual(
1529 dom_get_child_elements(u)[0].getAttribute("id"),
1530 "test_id-meta_attributes-key",
1531 )
1532 self.assertEqual(
1533 dom_get_child_elements(u)[0].getAttribute("name"), "key"
1534 )
1535 self.assertEqual(
1536 dom_get_child_elements(u)[0].getAttribute("value"), "another_val"
1537 )
1538
1539 def test_get_utilization(self):
1540 el = xml.dom.minidom.parseString(
1541 """
1542 <resource id="test_id">
1543 <utilization id="test_id-utilization">
1544 <nvpair id="test_id-utilization-key" name="key" value="-1"/>
1545 <nvpair id="test_id-utilization-keys" name="keys" value="90"/>
1546 </utilization>
1547 </resource>
1548 """
1549 ).documentElement
1550 self.assertEqual({"key": "-1", "keys": "90"}, utils.get_utilization(el))
1551
1552 def test_get_utilization_str(self):
1553 el = xml.dom.minidom.parseString(
1554 """
1555 <resource id="test_id">
1556 <utilization id="test_id-utilization">
1557 <nvpair id="test_id-utilization-key" name="key" value="-1"/>
1558 <nvpair id="test_id-utilization-keys" name="keys" value="90"/>
1559 </utilization>
1560 </resource>
1561 """
1562 ).documentElement
1563 self.assertEqual("key=-1 keys=90", utils.get_utilization_str(el))
1564
1565 def assert_element_id(self, node, node_id, tag=None):
1566 self.assertTrue(
1567 isinstance(node, xml.dom.minidom.Element),
1568 "element with id '%s' not found" % node_id,
1569 )
1570 self.assertEqual(node.getAttribute("id"), node_id)
1571 if tag:
1572 self.assertEqual(node.tagName, tag)
1573
1574
1575 class RunParallelTest(TestCase):
1576 @staticmethod
1577 def fixture_create_worker(log, name, sleepSeconds=0):
1578 def worker():
1579 sleep(sleepSeconds)
1580 log.append(name)
1581
1582 return worker
1583
1584 def test_run_all_workers(self):
1585 log = []
1586 utils.run_parallel(
1587 [
1588 self.fixture_create_worker(log, "first"),
1589 self.fixture_create_worker(log, "second"),
1590 ],
1591 wait_seconds=0.1,
1592 )
1593
1594 self.assertEqual(sorted(log), sorted(["first", "second"]))
1595
1596
1597 class NodeActionTaskTest(TestCase):
1598 def test_can_run_action(self):
1599 def action(node, arg, kwarg=None):
1600 return (0, ":".join([node, arg, kwarg]))
1601
1602 report_list = []
1603
1604 def report(node, returncode, output):
1605 report_list.append("|".join([node, str(returncode), output]))
1606
1607 task = utils.create_task(report, action, "node", "arg", kwarg="kwarg")
1608 task()
1609
1610 self.assertEqual(["node|0|node:arg:kwarg"], report_list)
1611
1612
1613 class TouchCibFile(TestCase):
1614 @mock.patch("pcs.utils.os.path.isfile", mock.Mock(return_value=False))
1615 @mock.patch(
1616 "pcs.utils.write_empty_cib",
1617 mock.Mock(side_effect=OSError("some message")),
1618 )
1619 @mock.patch("pcs.utils.err")
1620 def test_exception_is_transformed_correctly(self, err):
1621 filename = "/fake/filename"
1622 utils.touch_cib_file(filename)
1623 err.assert_called_once_with(
1624 "Unable to write to file: '/fake/filename': 'some message'"
1625 )
1626
1627
1628 @mock.patch("pcs.utils.is_run_interactive", mock.Mock(return_value=False))
1629 class GetContinueConfirmation(TestCase):
1630 def setUp(self):
1631 self.text = "some warning text"
1632 patcher_output = mock.patch("pcs.cli.reports.output.print_to_stderr")
1633 self.addCleanup(patcher_output.stop)
1634 self.mock_output = patcher_output.start()
1635
1636 def test_yes(self):
1637 self.assertTrue(utils.get_continue_confirmation(self.text, True))
1638 self.mock_output.assert_called_once_with(f"Warning: {self.text}")
1639
1640 def test_nothing(self):
1641 with self.assertRaises(SystemExit) as cm:
1642 self.assertFalse(utils.get_continue_confirmation(self.text, False))
1643 self.assertEqual(cm.exception.code, 1)
1644 self.mock_output.assert_called_once_with(
1645 f"Error: {self.text}, use --yes to override"
1646 )
1647