/usr/share/crmsh/tests/unittests/test_corosync.py is in crmsh 2.2.0-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | # Copyright (C) 2013 Kristoffer Gronlund <kgronlund@suse.com>
# See COPYING for license information.
#
# unit tests for parse.py
import os
import unittest
from crmsh import corosync
from crmsh.corosync import Parser, make_section, make_value
F1 = open(os.path.join(os.path.dirname(__file__), 'corosync.conf.1')).read()
F2 = open(os.path.join(os.path.dirname(__file__), 'corosync.conf.2')).read()
F3 = open(os.path.join(os.path.dirname(__file__), 'bug-862577_corosync.conf')).read()
def _valid(parser):
depth = 0
for t in parser._tokens:
if t.token not in (corosync._tCOMMENT,
corosync._tBEGIN,
corosync._tEND,
corosync._tVALUE):
raise AssertionError("illegal token " + str(t))
if t.token == corosync._tBEGIN:
depth += 1
if t.token == corosync._tEND:
depth -= 1
if depth != 0:
raise AssertionError("Unbalanced sections")
def _print(parser):
print parser.to_string()
class TestCorosyncParser(unittest.TestCase):
def test_parse(self):
p = Parser(F1)
_valid(p)
self.assertEqual(p.get('logging.logfile'), '/var/log/cluster/corosync.log')
self.assertEqual(p.get('totem.interface.ttl'), '1')
p.set('totem.interface.ttl', '2')
_valid(p)
self.assertEqual(p.get('totem.interface.ttl'), '2')
p.remove('quorum')
_valid(p)
self.assertEqual(p.count('quorum'), 0)
p.add('', make_section('quorum', []))
_valid(p)
self.assertEqual(p.count('quorum'), 1)
p.set('quorum.votequorum', '2')
_valid(p)
self.assertEqual(p.get('quorum.votequorum'), '2')
p.set('bananas', '5')
_valid(p)
self.assertEqual(p.get('bananas'), '5')
def test_logfile(self):
self.assertEqual(corosync.logfile(F1), '/var/log/cluster/corosync.log')
self.assertEqual(corosync.logfile('# nothing\n'), None)
def test_udpu(self):
p = Parser(F2)
_valid(p)
self.assertEqual(p.count('nodelist.node'), 5)
p.add('nodelist',
make_section('nodelist.node',
make_value('nodelist.node.ring0_addr', '10.10.10.10') +
make_value('nodelist.node.nodeid', str(corosync.next_nodeid(p)))))
_valid(p)
self.assertEqual(p.count('nodelist.node'), 6)
self.assertEqual(p.get_all('nodelist.node.nodeid'),
['1', '2', '3'])
def test_add_node_no_nodelist(self):
"test checks that if there is no nodelist, no node is added"
from crmsh.corosync import make_section, make_value, next_nodeid
p = Parser(F1)
_valid(p)
nid = next_nodeid(p)
self.assertEqual(p.count('nodelist.node'), nid - 1)
p.add('nodelist',
make_section('nodelist.node',
make_value('nodelist.node.ring0_addr', 'foo') +
make_value('nodelist.node.nodeid', str(nid))))
_valid(p)
self.assertEqual(p.count('nodelist.node'), nid - 1)
def test_add_node_nodelist(self):
from crmsh.corosync import make_section, make_value, next_nodeid
p = Parser(F2)
_valid(p)
nid = next_nodeid(p)
c = p.count('nodelist.node')
p.add('nodelist',
make_section('nodelist.node',
make_value('nodelist.node.ring0_addr', 'foo') +
make_value('nodelist.node.nodeid', str(nid))))
_valid(p)
self.assertEqual(p.count('nodelist.node'), c + 1)
self.assertEqual(next_nodeid(p), nid + 1)
def test_remove_node(self):
p = Parser(F2)
_valid(p)
self.assertEqual(p.count('nodelist.node'), 5)
p.remove_section_where('nodelist.node', 'nodeid', '2')
_valid(p)
self.assertEqual(p.count('nodelist.node'), 4)
self.assertEqual(p.get_all('nodelist.node.nodeid'),
['1'])
def test_bnc862577(self):
p = Parser(F3)
_valid(p)
self.assertEqual(p.count('service.ver'), 1)
if __name__ == '__main__':
unittest.main()
|