Dev: ui_corosync: add subcommand 'crm corosync link update' (jsc#PED-…

…8083)

crmsh/corosync.py

@@ -603,6 +603,14 @@ def load_config_file(path=None):
         except (OSError, corosync_config_format.ParserException) as e:
             raise ValueError(str(e)) from None
 
+    @staticmethod
+    def write_config_file(dom, path=None, file_mode=0o644):
+        if not path:
+            path = conf()
+        with utils.open_atomic(path, 'w', fsync=True, encoding='utf-8') as f:
+            corosync_config_format.DomSerializer(dom, f)
+            os.fchmod(f.fileno(), file_mode)
+
     def totem_transport(self):
         try:
             return self._config['totem']['transport']

crmsh/ui_corosync.py

@@ -1,6 +1,10 @@
 # Copyright (C) 2013 Kristoffer Gronlund <[email protected]>
 # See COPYING for license information.
 import dataclasses
+import socket
+import sys
+import typing
+
 from . import command, sh
 from . import completers
 from . import utils
@@ -44,6 +48,71 @@ def _diff_nodes(args):
         return []
 
 
+@dataclasses.dataclass
+class LinkArgumentParser:
+    linknumber: int = -1
+    nodes: list[tuple[str, str]] = dataclasses.field(default_factory=list)
+    options: dict[str, str|None] = dataclasses.field(default_factory=dict)
+
+    class SyntaxException(Exception):
+        pass
+
+    def parse(self, args: typing.Sequence[str]):
+        if not args:
+            raise LinkArgumentParser.SyntaxException('linknumber is required')
+        i = 0
+        self.linknumber = self.__parse_linknumber(args, i)
+        i += 1
+        while i < len(args):
+            if args[i] == 'options':
+                i += 1
+                break
+            self.nodes.append(self.__parse_node_spec(args, i))
+            i += 1
+        if i == len(args):
+            if args[i-1] == 'options':
+                raise LinkArgumentParser.SyntaxException('no options are specified')
+            else:
+                return self
+        # else args[i-1] == 'options'
+        while i < len(args):
+            k, v = self.__parse_option_spec(args, i)
+            self.options[k] = v
+            i += 1
+        return self
+
+    @staticmethod
+    def __parse_linknumber(args: typing.Sequence[str], i: int):
+        if not args[i].isdecimal():
+            raise LinkArgumentParser.SyntaxException(f'expected linknumber, actual {args[i]}')
+        try:
+            return int(args[i])
+        except ValueError:
+            raise SyntaxError(f'expected linknumber, actual {args[i]}')
+
+    @staticmethod
+    def __parse_node_spec(args: typing.Sequence[str], i: int):
+        match args[i].split('=', 2):
+            case [name, addr]:
+                try:
+                    socket.getaddrinfo(addr, 0, flags=socket.AI_NUMERICHOST)
+                    return name, addr
+                except socket.gaierror:
+                    raise LinkArgumentParser.SyntaxException(f'invalid node address: {addr}')
+            case _:
+                raise LinkArgumentParser.SyntaxException(f'invalid node address specification: {args[i]}')
+
+    @staticmethod
+    def __parse_option_spec(args: typing.Sequence[str], i: int):
+        match args[i].split('=', 1):
+            case [k, '']:
+                return k, None
+            case [k, v]:
+                return k, v
+            case _:
+                raise LinkArgumentParser.SyntaxException(f'invalid option address specification: {args[i]}')
+
+
 class Link(command.UI):
     """This level provides subcommands for managing knet links."""
 
@@ -71,6 +140,23 @@ def do_show(self, context):
             print('')
         # TODO: show link status
 
+    def do_update(self, context, *argv):
+        # TODO: handle --help
+        lm = corosync.LinkManager.load_config_file()
+        if lm.totem_transport() != 'knet':
+            logger.error('Corosync is not using knet transport')
+            return False
+        try:
+            args = LinkArgumentParser().parse(argv)
+        except LinkArgumentParser.SyntaxException as e:
+            logger.error('%s', str(e))
+            print('Usage: link update <linknumber> [<node>=<addr> ...] [options <option>=<value> ...] ', file=sys.stderr)
+            return False
+        # TODO: update ringX_addr
+        lm.write_config_file(lm.update_link(args.linknumber, args.options))
+        logger.info("Use \"crm corosync diff\" to show the difference")
+        logger.info("Use \"crm corosync push\" to sync")
+
 
 class Corosync(command.UI):
     '''

test/unittests/test_ui_corosync.py

@@ -0,0 +1,59 @@
+import unittest
+
+from crmsh.ui_corosync import LinkArgumentParser
+
+
+class TestLinkArgumentParser(unittest.TestCase):
+    def test_parse_empty(self):
+        with self.assertRaises(LinkArgumentParser.SyntaxException):
+            LinkArgumentParser().parse(list())
+
+    def test_invalid_link_number(self):
+        with self.assertRaises(LinkArgumentParser.SyntaxException):
+            LinkArgumentParser().parse(['a0'])
+
+    def test_no_spec(self):
+        args = LinkArgumentParser().parse(['0'])
+        self.assertEqual(0, args.linknumber)
+        self.assertFalse(args.nodes)
+        self.assertFalse(args.options)
+
+    def test_addr_spec(self):
+        args = LinkArgumentParser().parse(['0', 'node1=192.0.2.100', 'node2=fd00:a0::10'])
+        self.assertEqual(0, args.linknumber)
+        self.assertFalse(args.options)
+        self.assertListEqual([('node1', '192.0.2.100'), ('node2', 'fd00:a0::10')], args.nodes)
+
+    def test_invalid_addr_spec(self):
+        with self.assertRaises(LinkArgumentParser.SyntaxException):
+            LinkArgumentParser().parse(['0', 'node1=192.0.2.300'])
+        with self.assertRaises(LinkArgumentParser.SyntaxException):
+            LinkArgumentParser().parse(['0', 'node1=fd00::a0::10'])
+        with self.assertRaises(LinkArgumentParser.SyntaxException):
+            LinkArgumentParser().parse(['0', 'node1=node1.example.com'])
+
+    def test_option_spec(self):
+        args = LinkArgumentParser().parse(['0', 'options', 'node1=192.0.2.100', 'node2=fd00:a0::10', 'foo='])
+        self.assertEqual(0, args.linknumber)
+        self.assertFalse(args.nodes)
+        self.assertDictEqual({'node1': '192.0.2.100', 'node2': 'fd00:a0::10', 'foo': None}, args.options)
+
+    def test_addrs_and_options(self):
+        args = LinkArgumentParser().parse(['0', 'node1=192.0.2.100', 'node2=fd00:a0::10', 'options', 'foo=bar=1'])
+        self.assertEqual(0, args.linknumber)
+        self.assertListEqual([('node1', '192.0.2.100'), ('node2', 'fd00:a0::10')], args.nodes)
+        self.assertDictEqual({'foo': 'bar=1'}, args.options)
+
+    def test_no_options(self):
+        with self.assertRaises(LinkArgumentParser.SyntaxException):
+            LinkArgumentParser().parse(['0', 'options'])
+
+    def test_garbage_inputs(self):
+        with self.assertRaises(LinkArgumentParser.SyntaxException):
+            LinkArgumentParser().parse(['0', 'foo'])
+        with self.assertRaises(LinkArgumentParser.SyntaxException):
+            LinkArgumentParser().parse(['0', 'node1=192.0.2.100', 'foo'])
+        with self.assertRaises(LinkArgumentParser.SyntaxException):
+            LinkArgumentParser().parse(['0', 'node1=192.0.2.100', 'options', 'foo'])
+        with self.assertRaises(LinkArgumentParser.SyntaxException):
+            LinkArgumentParser().parse(['0', 'node1=192.0.2.100', 'options', 'foo=bar', 'foo'])