Refactor parts of frr.py and add unittests

* Remove the bgp_as parameter from the _get_router_id() function as it
  wasn't used

* Run the run_vtysh_config() command using temporary files to avoid
  leaving the filesystem dirty when we run the OVN BGP Agent

* Fix logs to delay the interpolation

* Base test class to stop all mocks at the end of each test execution

Story: 2009165
Task: 43296

Signed-off-by: Lucas Alvares Gomes <lucasagomes@gmail.com>
Change-Id: Idf37e2fd3305c5b4d8f5a4d2f57f781852938e13

ovn_bgp_agent/drivers/openstack/utils/frr.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import json
+import tempfile
 
 from jinja2 import Template
 from oslo_log import log as logging
@@ -70,16 +71,35 @@ router bgp {{ bgp_as }} vrf {{ vrf_name }}
 '''
 
 
-def _get_router_id(bgp_as):
+def _get_router_id():
     output = ovn_bgp_agent.privileged.vtysh.run_vtysh_command(
         command='show ip bgp summary json')
     return json.loads(output).get('ipv4Unicast', {}).get('routerId')
 
 
+def _run_vtysh_config_with_tempfile(vrf_config):
+    try:
+        f = tempfile.NamedTemporaryFile(mode='w')
+        f.write(vrf_config)
+        f.flush()
+    except (IOError, OSError) as e:
+        LOG.error('Failed to create the VRF configuration '
+                  'file. Error: %s', e)
+        if f is not None:
+            f.close()
+        raise
+
+    try:
+        ovn_bgp_agent.privileged.vtysh.run_vtysh_config(f.name)
+    finally:
+        if f is not None:
+            f.close()
+
+
 def vrf_leak(vrf, bgp_as, bgp_router_id=None):
-    LOG.info("Add VRF leak for VRF {} on router bgp {}".format(vrf, bgp_as))
+    LOG.info("Add VRF leak for VRF %s on router bgp %s", vrf, bgp_as)
     if not bgp_router_id:
-        bgp_router_id = _get_router_id(bgp_as)
+        bgp_router_id = _get_router_id()
         if not bgp_router_id:
             LOG.error("Unknown router-id, needed for route leaking")
             return
@@ -87,17 +107,12 @@ def vrf_leak(vrf, bgp_as, bgp_router_id=None):
     vrf_template = Template(LEAK_VRF_TEMPLATE)
     vrf_config = vrf_template.render(vrf_name=vrf, bgp_as=bgp_as,
                                      bgp_router_id=bgp_router_id)
-    frr_config_file = "frr-config-vrf-leak-{}".format(vrf)
-    with open(frr_config_file, 'w') as vrf_config_file:
-        vrf_config_file.write(vrf_config)
-
-    ovn_bgp_agent.privileged.vtysh.run_vtysh_config(frr_config_file)
+    _run_vtysh_config_with_tempfile(vrf_config)
 
 
 def vrf_reconfigure(evpn_info, action):
-    LOG.info("FRR reconfiguration (action = {}) for evpn: {}".format(
-             action, evpn_info))
-    frr_config_file = None
+    LOG.info("FRR reconfiguration (action = %s) for evpn: %s",
+             action, evpn_info)
     if action == "add-vrf":
         vrf_template = Template(ADD_VRF_TEMPLATE)
         vrf_config = vrf_template.render(
@@ -105,18 +120,13 @@ def vrf_reconfigure(evpn_info, action):
                                    evpn_info['vni']),
             bgp_as=evpn_info['bgp_as'],
             vni=evpn_info['vni'])
-        frr_config_file = "frr-config-add-vrf-{}".format(evpn_info['vni'])
     elif action == "del-vrf":
         vrf_template = Template(DEL_VRF_TEMPLATE)
         vrf_config = vrf_template.render(
             vrf_name="{}{}".format(constants.OVN_EVPN_VRF_PREFIX,
                                    evpn_info['vni']),
             bgp_as=evpn_info['bgp_as'])
-        frr_config_file = "frr-config-del-vrf-{}".format(evpn_info['vni'])
     else:
         LOG.error("Unknown FRR reconfiguration action: %s", action)
         return
-    with open(frr_config_file, 'w') as vrf_config_file:
-        vrf_config_file.write(vrf_config)
-
-    ovn_bgp_agent.privileged.vtysh.run_vtysh_config(frr_config_file)
+    _run_vtysh_config_with_tempfile(vrf_config)

ovn_bgp_agent/tests/base.py

@@ -15,9 +15,15 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+from unittest import mock
+
 from oslotest import base
 
 
 class TestCase(base.BaseTestCase):
 
     """Test case base class for all unit tests."""
+
+    def setUp(self):
+        super(TestCase, self).setUp()
+        self.addCleanup(mock.patch.stopall)

ovn_bgp_agent/tests/unit/drivers/openstack/utils/test_frr.py

@@ -0,0 +1,98 @@
+# Copyright 2021 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import tempfile
+from unittest import mock
+
+from ovn_bgp_agent import constants
+from ovn_bgp_agent.drivers.openstack.utils import frr as frr_utils
+from ovn_bgp_agent.tests import base as test_base
+
+
+class TestFrr(test_base.TestCase):
+
+    def setUp(self):
+        super(TestFrr, self).setUp()
+        self.mock_vtysh = mock.patch('ovn_bgp_agent.privileged.vtysh').start()
+
+    def test__get_router_id(self):
+        router_id = 'fake-router'
+        self.mock_vtysh.run_vtysh_command.return_value = (
+            '{"ipv4Unicast": {"routerId": "%s"}}' % router_id)
+        ret = frr_utils._get_router_id()
+        self.assertEqual(router_id, ret)
+
+    def test__get_router_id_no_ipv4_settings(self):
+        self.mock_vtysh.run_vtysh_command.return_value = '{}'
+        ret = frr_utils._get_router_id()
+        self.assertIsNone(ret)
+
+    @mock.patch.object(frr_utils, '_get_router_id')
+    @mock.patch.object(tempfile, 'NamedTemporaryFile')
+    def test_vrf_leak(self, mock_tf, mock_gri):
+        vrf = 'fake-vrf'
+        bgp_as = 'fake-bgp-as'
+        router_id = 'fake-router-id'
+        mock_gri.return_value = router_id
+
+        frr_utils.vrf_leak(vrf, bgp_as)
+
+        write_arg = mock_tf.return_value.write.call_args_list[0][0][0]
+        self.assertIn(vrf, write_arg)
+        self.assertIn(bgp_as, write_arg)
+        # Assert the file was closed
+        mock_tf.return_value.close.assert_called_once_with()
+
+    @mock.patch.object(frr_utils, '_get_router_id')
+    @mock.patch.object(tempfile, 'NamedTemporaryFile')
+    def test_vrf_leak_no_router_id(self, mock_tf, mock_gri):
+        mock_gri.return_value = None
+        frr_utils.vrf_leak('fake-vrf', 'fake-bgp-as')
+        # Assert no file was created
+        self.assertFalse(mock_tf.called)
+
+    @mock.patch.object(tempfile, 'NamedTemporaryFile')
+    def _test_vrf_reconfigure(self, mock_tf, add_vrf=True):
+        action = 'add-vrf' if add_vrf else 'del-vrf'
+        evpn_info = {'vni': '1001', 'bgp_as': 'fake-bgp-as'}
+
+        frr_utils.vrf_reconfigure(evpn_info, action)
+
+        vrf_name = "{}{}".format(constants.OVN_EVPN_VRF_PREFIX,
+                                 evpn_info['vni'])
+        write_arg = mock_tf.return_value.write.call_args_list[0][0][0]
+
+        if add_vrf:
+            self.assertIn('\nvrf %s' % vrf_name, write_arg)
+            self.assertIn('\nrouter bgp %s' % evpn_info['bgp_as'], write_arg)
+        else:
+            self.assertIn("no vrf %s" % vrf_name, write_arg)
+            self.assertIn("no router bgp %s" % evpn_info['bgp_as'], write_arg)
+
+        self.mock_vtysh.run_vtysh_config.assert_called_once_with(
+            mock_tf.return_value.name)
+        # Assert the file was closed
+        mock_tf.return_value.close.assert_called_once_with()
+
+    def test_vrf_reconfigure_add_vrf(self):
+        self._test_vrf_reconfigure()
+
+    def test_vrf_reconfigure_del_vrf(self):
+        self._test_vrf_reconfigure(add_vrf=False)
+
+    def test_vrf_reconfigure_unknown_action(self):
+        frr_utils.vrf_reconfigure('fake-evpn-info', 'non-existing-action')
+        # Assert run_vtysh_command() wasn't called
+        self.assertFalse(self.mock_vtysh.run_vtysh_config.called)