[Bgpcfgd] Add unit tests

src/sonic-bgpcfgd/bgpcfgd/managers_intf.py

@@ -30,7 +30,7 @@ def set_handler(self, key, data):
             try:
                 network = netaddr.IPNetwork(str(network_str))
             except (netaddr.NotRegisteredError, netaddr.AddrFormatError, netaddr.AddrConversionError):
-                log_warn("Subnet '%s' format is wrong for interface '%s'" % (network_str, data["interface"]))
+                log_warn("Subnet '%s' format is wrong for interface '%s'" % (network_str, interface_name))
                 return True
             data["interface"] = interface_name
             data["prefixlen"] = str(network.prefixlen)

src/sonic-bgpcfgd/tests/test_bgp.py

@@ -0,0 +1,111 @@
+from unittest.mock import MagicMock, patch
+
+import os
+from bgpcfgd.directory import Directory
+from bgpcfgd.template import TemplateFabric
+from . import swsscommon_test
+from .util import load_constants
+from swsscommon import swsscommon
+import bgpcfgd.managers_bgp
+
+TEMPLATE_PATH = os.path.abspath('../../dockers/docker-fpm-frr/frr')
+
+def constructor():
+    cfg_mgr = MagicMock()
+    common_objs = {
+        'directory': Directory(),
+        'cfg_mgr':   cfg_mgr,
+        'tf':        TemplateFabric(TEMPLATE_PATH),
+        'constants': load_constants()['constants']
+    }
+
+    return_value_map = {
+        "['vtysh', '-c', 'show bgp vrfs json']": (0, "{\"vrfs\": {\"default\": {}}}", ""),
+        "['vtysh', '-c', 'show bgp vrf default neighbors json']": (0, "{\"10.10.10.1\": {}, \"20.20.20.1\": {}, \"fc00:10::1\": {}}", "")
+    }
+
+    bgpcfgd.managers_bgp.run_command = lambda cmd: return_value_map[str(cmd)]
+    m = bgpcfgd.managers_bgp.BGPPeerMgrBase(common_objs, "CONFIG_DB", swsscommon.CFG_BGP_NEIGHBOR_TABLE_NAME, "general", True)
+    assert m.peer_type == "general"
+    assert m.check_neig_meta == False # Because constants['bgp']['use_neighbors_meta'] is false in constants.yml
+
+    m.directory.put("CONFIG_DB", swsscommon.CFG_DEVICE_METADATA_TABLE_NAME, "localhost", {"bgp_asn": "65100"})
+    m.directory.put("CONFIG_DB", swsscommon.CFG_LOOPBACK_INTERFACE_TABLE_NAME, "Loopback0|11.11.11.11/32", {})
+    m.directory.put("CONFIG_DB", swsscommon.CFG_LOOPBACK_INTERFACE_TABLE_NAME, "Loopback0|FC00:1::32/128", {})
+    m.directory.put("LOCAL", "local_addresses", "30.30.30.30", {"interface": "Ethernet4|30.30.30.30/24"})
+    m.directory.put("LOCAL", "local_addresses", "fc00:20::20", {"interface": "Ethernet8|fc00:20::20/96"})
+    m.directory.put("LOCAL", "interfaces", "Ethernet4|30.30.30.30/24", {"anything": "anything"})
+    m.directory.put("LOCAL", "interfaces", "Ethernet8|fc00:20::20/96", {"anything": "anything"})
+
+    return m
+
+@patch('bgpcfgd.managers_bgp.log_info')
+def test_update_peer_up(mocked_log_info): 
+    m = constructor()
+    res = m.set_handler("10.10.10.1", {"admin_status": "up"})
+    assert res, "Expect True return value for peer update"
+    mocked_log_info.assert_called_with("Peer 'default|10.10.10.1' admin state is set to 'up'")
+
+@patch('bgpcfgd.managers_bgp.log_info')
+def test_update_peer_up_ipv6(mocked_log_info): 
+    m = constructor()
+    res = m.set_handler("fc00:10::1", {"admin_status": "up"})
+    assert res, "Expect True return value for peer update"
+    mocked_log_info.assert_called_with("Peer 'default|fc00:10::1' admin state is set to 'up'")
+
+@patch('bgpcfgd.managers_bgp.log_info')
+def test_update_peer_down(mocked_log_info): 
+    m = constructor()
+    res = m.set_handler("10.10.10.1", {"admin_status": "down"})
+    assert res, "Expect True return value for peer update"
+    mocked_log_info.assert_called_with("Peer 'default|10.10.10.1' admin state is set to 'down'")
+
+@patch('bgpcfgd.managers_bgp.log_err')
+def test_update_peer_no_admin_status(mocked_log_err):
+    m = constructor()
+    res = m.set_handler("10.10.10.1", {"anything": "anything"})
+    assert res, "Expect True return value for peer update"
+    mocked_log_err.assert_called_with("Peer '(default|10.10.10.1)': Can't update the peer. Only 'admin_status' attribute is supported")
+
+@patch('bgpcfgd.managers_bgp.log_err')
+def test_update_peer_invalid_admin_status(mocked_log_err):
+    m = constructor()
+    res = m.set_handler("10.10.10.1", {"admin_status": "invalid"})
+    assert res, "Expect True return value for peer update"
+    mocked_log_err.assert_called_with("Peer 'default|10.10.10.1': Can't update the peer. It has wrong attribute value attr['admin_status'] = 'invalid'")
+
+def test_add_peer():
+    m = constructor()
+    res = m.set_handler("30.30.30.1", {"local_addr": "30.30.30.30", "admin_status": "up"})
+    assert res, "Expect True return value"
+
+def test_add_peer_ipv6():
+    m = constructor()
+    res = m.set_handler("fc00:20::1", {"local_addr": "fc00:20::20", "admin_status": "up"})
+    assert res, "Expect True return value"
+
+@patch('bgpcfgd.managers_bgp.log_warn')
+def test_add_peer_no_local_addr(mocked_log_warn):
+    m = constructor()
+    res = m.set_handler("30.30.30.1", {"admin_status": "up"})
+    assert res, "Expect True return value"
+    mocked_log_warn.assert_called_with("Peer 30.30.30.1. Missing attribute 'local_addr'")
+
+@patch('bgpcfgd.managers_bgp.log_debug')
+def test_add_peer_invalid_local_addr(mocked_log_debug):
+    m = constructor()
+    res = m.set_handler("30.30.30.1", {"local_addr": "40.40.40.40", "admin_status": "up"})
+    assert not res, "Expect False return value"
+    mocked_log_debug.assert_called_with("Peer '30.30.30.1' with local address '40.40.40.40' wait for the corresponding interface to be set")
+
+@patch('bgpcfgd.managers_bgp.log_info')
+def test_del_handler(mocked_log_info):
+    m = constructor()
+    m.del_handler("10.10.10.1")
+    mocked_log_info.assert_called_with("Peer '(default|10.10.10.1)' has been removed")
+
+@patch('bgpcfgd.managers_bgp.log_warn')
+def test_del_handler_nonexist_peer(mocked_log_warn):
+    m = constructor()
+    m.del_handler("40.40.40.1")
+    mocked_log_warn.assert_called_with("Peer '(default|40.40.40.1)' has not been found")

src/sonic-bgpcfgd/tests/test_db.py

@@ -0,0 +1,40 @@
+from unittest.mock import MagicMock, patch
+
+from bgpcfgd.directory import Directory
+from bgpcfgd.template import TemplateFabric
+from . import swsscommon_test
+from swsscommon import swsscommon
+
+with patch.dict("sys.modules", swsscommon=swsscommon_test):
+    from bgpcfgd.managers_db import BGPDataBaseMgr
+
+def test_set_del_handler():
+    cfg_mgr = MagicMock()
+    common_objs = {
+        'directory': Directory(),
+        'cfg_mgr':   cfg_mgr,
+        'tf':        TemplateFabric(),
+        'constants': {},
+    }
+    m = BGPDataBaseMgr(common_objs, "CONFIG_DB", swsscommon.CFG_DEVICE_METADATA_TABLE_NAME)
+    assert m.constants == {}
+
+    # test set_handler
+    res = m.set_handler("test_key1", {"test_value1"})
+    assert res, "Returns always True"
+    assert "test_key1" in m.directory.get_slot(m.db_name, m.table_name)
+    assert m.directory.get(m.db_name, m.table_name, "test_key1") == {"test_value1"}
+
+    res = m.set_handler("test_key2", {})
+    assert res, "Returns always True"
+    assert "test_key2" in m.directory.get_slot(m.db_name, m.table_name)
+    assert m.directory.get(m.db_name, m.table_name, "test_key2") == {}
+
+    # test del_handler
+    m.del_handler("test_key")
+    assert "test_key" not in m.directory.get_slot(m.db_name, m.table_name)
+    assert "test_key2" in m.directory.get_slot(m.db_name, m.table_name)
+    assert m.directory.get(m.db_name, m.table_name, "test_key2") == {}
+
+    m.del_handler("test_key2")
+    assert "test_key2" not in m.directory.get_slot(m.db_name, m.table_name)

src/sonic-bgpcfgd/tests/test_directory.py

@@ -0,0 +1,57 @@
+from unittest.mock import MagicMock, patch
+from bgpcfgd.directory import Directory
+
+@patch('bgpcfgd.directory.log_err')
+def test_directory(mocked_log_err):
+    test_values = {
+        "key1": {
+            "key1_1": {
+                "key1_1_1": "value1_1_1",
+                "key1_1_2": "value1_1_2",
+                "key1_1_3": "value1_1_3"
+            }
+        },
+        "key2": {
+            "value2"
+        }
+    }
+
+    directory = Directory()
+
+    # Put test values
+    directory.put("db_name", "table", "key1", test_values["key1"])
+    directory.put("db_name", "table", "key2", test_values["key2"])
+
+    # Test get_path()
+    assert directory.get_path("db_name", "table", "") == test_values
+    assert directory.get_path("db_name", "table", "key1/key1_1/key1_1_1") == "value1_1_1"
+    assert directory.get_path("db_name", "table", "key1/key_nonexist") == None
+
+    # Test path_exist()
+    assert directory.path_exist("db_name", "table", "key1/key1_1/key1_1_1")
+    assert not directory.path_exist("db_name", "table_nonexist", "")
+    assert not directory.path_exist("db_name", "table", "key1/key_nonexist")
+
+    # Test get_slot()
+    assert directory.get_slot("db_name", "table") == test_values
+
+    # Test get()
+    assert directory.get("db_name", "table", "key2") == test_values["key2"]
+
+    # Test remove()
+    directory.remove("db_name", "table", "key2")
+    assert not directory.path_exist("db_name", "table", "key2")
+
+    # Test remove() with invalid input
+    directory.remove("db_name", "table_nonexist", "key2")
+    mocked_log_err.assert_called_with("Directory: Can't remove key 'key2' from slot 'db_name__table_nonexist'. The slot doesn't exist")
+    directory.remove("db_name", "table", "key_nonexist")
+    mocked_log_err.assert_called_with("Directory: Can't remove key 'key_nonexist' from slot 'db_name__table'. The key doesn't exist")
+
+    # Test remove_slot()
+    directory.remove_slot("db_name", "table")
+    assert not directory.available("db_name", "table")
+
+    # Test remove_slot() with nonexist table
+    directory.remove_slot("db_name", "table_nonexist")
+    mocked_log_err.assert_called_with("Directory: Can't remove slot 'db_name__table_nonexist'. The slot doesn't exist")

src/sonic-bgpcfgd/tests/test_frr.py

@@ -1,6 +1,68 @@
-from bgpcfgd.frr import FRR
-
+from unittest.mock import patch
+import bgpcfgd.frr
+import pytest
 
 def test_constructor():
-    f = FRR(["abc", "cde"])
+    f = bgpcfgd.frr.FRR(["abc", "cde"])
     assert f.daemons == ["abc", "cde"]
+
+def test_wait_for_daemons():
+    bgpcfgd.frr.run_command = lambda cmd, **kwargs: (0, ["abc", "cde"], "")
+    f = bgpcfgd.frr.FRR(["abc", "cde"])
+    f.wait_for_daemons(5)
+
+def test_wait_for_daemons_fail():
+    bgpcfgd.frr.run_command = lambda cmd, **kwargs: (0, ["abc", "non_expected"], "")
+    f = bgpcfgd.frr.FRR(["abc", "cde"])
+    with pytest.raises(Exception):
+        assert f.wait_for_daemons(5)
+
+def test_wait_for_daemons_error():
+    bgpcfgd.frr.run_command = lambda cmd, **kwargs: (1, ["abc", "cde"], "some error")
+    f = bgpcfgd.frr.FRR(["abc", "cde"])
+    with pytest.raises(Exception):
+        assert f.wait_for_daemons(5)
+
+def test_get_config():
+    bgpcfgd.frr.run_command = lambda cmd: (0, "expected config", "")
+    f = bgpcfgd.frr.FRR(["abc", "cde"])
+    out = f.get_config()
+    assert out == "expected config"
+
+@patch('bgpcfgd.frr.log_crit')
+def test_get_config_fail(mocked_log_crit):
+    bgpcfgd.frr.run_command = lambda cmd: (1, "some config", "some error")
+    f = bgpcfgd.frr.FRR(["abc", "cde"])
+    out = f.get_config()
+    assert out == ""
+    mocked_log_crit.assert_called_with("can't update running config: rc=1 out='some config' err='some error'")
+
+def test_write():
+    bgpcfgd.frr.run_command = lambda cmd: (0, "some output", "")
+    f = bgpcfgd.frr.FRR(["abc", "cde"])
+    res = f.write("config context")
+    assert res, "Expect True return value"
+
+def test_write_fail():
+    bgpcfgd.frr.run_command = lambda cmd: (1, "some output", "some error")
+    f = bgpcfgd.frr.FRR(["abc", "cde"])
+    res = f.write("config context")
+    assert not res, "Expect False return value"
+
+def test_restart_peer_groups():
+    bgpcfgd.frr.run_command = lambda cmd: (0, "some output", "")
+    f = bgpcfgd.frr.FRR(["abc", "cde"])
+    res = f.restart_peer_groups(["pg_1", "pg_2"])
+    assert res, "Expect True return value"
+
+@patch('bgpcfgd.frr.log_crit')
+def test_restart_peer_groups_fail(mocked_log_crit):
+    return_value_map = {
+        "['vtysh', '-c', 'clear bgp peer-group pg_1 soft in']": (0, "", ""),
+        "['vtysh', '-c', 'clear bgp peer-group pg_2 soft in']": (1, "some output", "some error")
+    }
+    bgpcfgd.frr.run_command = lambda cmd: return_value_map[str(cmd)]
+    f = bgpcfgd.frr.FRR(["abc", "cde"])
+    res = f.restart_peer_groups(["pg_1", "pg_2"])
+    assert not res, "Expect False return value"
+    mocked_log_crit.assert_called_with("Can't restart bgp peer-group 'pg_2'. rc='1', out='some output', err='some error'")

src/sonic-bgpcfgd/tests/test_intf.py

@@ -0,0 +1,52 @@
+from unittest.mock import MagicMock, patch
+
+from bgpcfgd.directory import Directory
+from bgpcfgd.template import TemplateFabric
+from swsscommon import swsscommon
+from bgpcfgd.managers_intf import InterfaceMgr
+
+def set_handler_test(manager, key, value):
+    res = manager.set_handler(key, value)
+    assert res, "Returns always True"
+    assert manager.directory.get(manager.db_name, manager.table_name, key) == value
+
+def del_handler_test(manager, key):
+    manager.del_handler(key)
+    assert manager.directory.get_path(manager.db_name, manager.table_name, key) == None
+
+@patch('bgpcfgd.managers_intf.log_warn')
+def test_intf(mocked_log_warn):
+    cfg_mgr = MagicMock()
+    common_objs = {
+        'directory': Directory(),
+        'cfg_mgr':   cfg_mgr,
+        'tf':        TemplateFabric(),
+        'constants': {},
+    }
+    m = InterfaceMgr(common_objs, "CONFIG_DB", swsscommon.CFG_VLAN_INTF_TABLE_NAME)
+
+    set_handler_test(m, "Vlan1000", {})
+    set_handler_test(m, "Vlan1000|192.168.0.1/21", {})
+
+    # test set handler with invalid ip network
+    res = m.set_handler("Vlan1000|invalid_netowrk", {})
+    assert res, "Returns always True"
+    mocked_log_warn.assert_called_with("Subnet 'invalid_netowrk' format is wrong for interface 'Vlan1000'")
+
+    del_handler_test(m, "Vlan1000")
+    del_handler_test(m, "Vlan1000|192.168.0.1/21")
+    del_handler_test(m, "Vlan1000|invalid_netowrk")
+    mocked_log_warn.assert_called_with("Subnet 'invalid_netowrk' format is wrong for interface 'Vlan1000'")
+
+def test_intf_ipv6():
+    cfg_mgr = MagicMock()
+    common_objs = {
+        'directory': Directory(),
+        'cfg_mgr':   cfg_mgr,
+        'tf':        TemplateFabric(),
+        'constants': {},
+    }
+    m = InterfaceMgr(common_objs, "CONFIG_DB", swsscommon.CFG_VLAN_INTF_TABLE_NAME)
+
+    set_handler_test(m, "Vlan1000|fc02:1000::1/64", {})
+    del_handler_test(m, "Vlan1000|fc02:1000::1/64")