Dev: corosync: implement adding links (jsc#PED-8083)

crmsh/corosync.py

@@ -712,6 +712,13 @@ def update_node_addr(self, linknumber: int, node_addresses: typing.Mapping[int,
         links = self.links()
         if linknumber >= len(links):
             raise ValueError(f'Link {linknumber} does not exist.')
+        return self.__upsert_node_addr_impl(self._config, links, linknumber, node_addresses)
+
+    @staticmethod
+    def __upsert_node_addr_impl(
+            config: dict, links: typing.Sequence[Link],
+            linknumber: int, node_addresses: typing.Mapping[int, str],
+    ) -> dict:
         for nodeid, addr in node_addresses.items():
             try:
                 socket.getaddrinfo(addr, 0, flags=socket.AI_NUMERICHOST)
@@ -721,13 +728,23 @@ def update_node_addr(self, linknumber: int, node_addresses: typing.Mapping[int,
             if found is None:
                 raise ValueError(f'Unknown nodeid {nodeid}.')
             found.addr = addr
-        nodes = self._config['nodelist']['node']
+        nodes = config['nodelist']['node']
         assert isinstance(nodes, list)
         for node in nodes:
             updated_addr = node_addresses.get(int(node['nodeid']), None)
             if updated_addr is not None:
                 node[f'ring{linknumber}_addr'] = updated_addr
-        return self._config
+        return config
+
+    def add_link(self, node_addresses: typing.Mapping[int, str], options: dict[str, str|None]) -> dict:
+        links = self.links()
+        node_ids = {node.nodeid for node in links[0].nodes}
+        for nodeid in node_ids:
+            if nodeid not in node_addresses:
+                raise ValueError(f'The address of node {nodeid} is not specified.')
+        next_linknumber = len(links)
+        self.__upsert_node_addr_impl(self._config, next_linknumber, node_addresses)
+        return self.update_link(next_linknumber, options)
 
     def remove_link(self, linknumber: int) -> dict:
         """Remove the specified link.

test/unittests/test_corosync.py

@@ -591,5 +591,32 @@ def test_remove_last_link(self):
             self.lm.remove_link(0)
 
 
+@mock.patch('crmsh.corosync.LinkManager.update_link')
+@mock.patch('crmsh.corosync.LinkManager._LinkManager__upsert_node_addr_impl')
+@mock.patch('crmsh.corosync.LinkManager.links')
+class TestLinkManagerAddLink(unittest.TestCase):
+    def test_unspecified_node(self, mock_links, mock_upsert_node, mock_update_link):
+        mock_links.return_value = [corosync.Link(0, [
+            corosync.LinkNode(1, 'node1', '192.0.2.101'),
+            corosync.LinkNode(2, 'node2', '192.0.2.102'),
+        ])]
+        lm = corosync.LinkManager(dict())
+        with self.assertRaises(ValueError):
+            lm.add_link({1: '192.0.2.201'}, dict())
+        mock_upsert_node.assert_not_called()
+        mock_update_link.assert_not_called()
+
+    def test_unknown_node(self, mock_links, mock_upsert_node, mock_update_link):
+        mock_links.return_value = [corosync.Link(0, [
+            corosync.LinkNode(1, 'node1', '192.0.2.101'),
+        ])]
+        mock_upsert_node.side_effect = ValueError()
+        lm = corosync.LinkManager(dict())
+        with self.assertRaises(ValueError):
+            lm.add_link({1: '192.0.2.201', 2: '192.0.2.202'}, dict())
+        mock_upsert_node.assert_called_once()
+        mock_update_link.assert_not_called()
+
+
 if __name__ == '__main__':
     unittest.main()