sonic-buildimage/dockers/docker-dhcp-relay/cli-plugin-tests/test_show_dhcp_relay.py

import pytest
import sys
import click
import os
sys.path.append('../cli/show/plugins/')
import show_dhcp_relay as show
import show.vlan as vlan
from swsscommon import swsscommon
from mock_config import COMMON_TEST_DATA, NEW_ADDED_TEST_DATA, MULTI_TEST_DATA
from parameterized import parameterized
from pyfakefs.fake_filesystem_unittest import patchfs
from unittest import mock

try:
    sys.path.insert(0, '../../../src/sonic-host-services/tests/common')
    from mock_configdb import MockConfigDb
    swsscommon.ConfigDBConnector = MockConfigDb
except KeyError:
    pass

expected_ipv6_table_with_header = """\
+-------------+----------------------+
|   Interface |   DHCP Relay Address |
+=============+======================+
|    Vlan1000 |         fc02:2000::1 |
|             |         fc02:2000::2 |
+-------------+----------------------+
"""

expected_ipv4_table_with_header = """\
+-------------+----------------------+
|   Interface |   DHCP Relay Address |
+=============+======================+
|    Vlan1000 |            192.0.0.1 |
|             |            192.0.0.2 |
+-------------+----------------------+
"""

expected_ipv4_table_with_enabled_dhcp_server_with_header = """\
+-------------+----------------------+
|   Interface |   DHCP Relay Address |
+=============+======================+
|    Vlan1000 |                  N/A |
+-------------+----------------------+
"""

expected_ipv6_table_without_header = """\
--------  ------------
Vlan1000  fc02:2000::1
          fc02:2000::2
--------  ------------
"""

expected_ipv6_table_multi_with_header = """\
+-------------+----------------------+
|   Interface |   DHCP Relay Address |
+=============+======================+
|    Vlan1000 |         fc02:2000::1 |
|             |         fc02:2000::2 |
+-------------+----------------------+
|    Vlan1001 |         fc02:2000::3 |
|             |         fc02:2000::4 |
+-------------+----------------------+
"""

expected_ipv4_table_multi_with_header = """\
+-------------+----------------------+
|   Interface |   DHCP Relay Address |
+=============+======================+
|    Vlan1000 |            192.0.0.1 |
|             |            192.0.0.2 |
+-------------+----------------------+
|    Vlan1001 |            192.0.0.3 |
|             |            192.0.0.4 |
+-------------+----------------------+
"""

DBCONFIG_PATH = '/var/run/redis/sonic-db/database_config.json'

IP_VER_TEST_PARAM_MAP = {
    "ipv4": {
        "entry": "dhcp_servers",
        "table": "VLAN"
    },
    "ipv6": {
        "entry": "dhcpv6_servers",
        "table": "DHCP_RELAY"
    }
}


def test_plugin_registration():
    cli = mock.MagicMock()
    show.register(cli)
    assert 'DHCP Helper Address' in dict(vlan.VlanBrief.COLUMNS)


@pytest.mark.parametrize("feature_table", [{}, {"dhcp_server": {"state": "disabled"}},
                                           {"dhcp_server": {"state": "enabled"}}, {"dhcp_server": {}}])
def test_dhcp_relay_column_output(feature_table):
    ctx = (
        ({'Vlan1001': {'dhcp_servers': ['192.0.0.1', '192.168.0.2']}}, {}, {}),
        (MockDb({"FEATURE": feature_table})),
    )
    if "dhcp_server" in feature_table and "state" in feature_table["dhcp_server"] and \
       feature_table["dhcp_server"]["state"] == "enabled":
        assert show.get_dhcp_helper_address(ctx, 'Vlan1001') == 'N/A'
    else:
        assert show.get_dhcp_helper_address(ctx, 'Vlan1001') == '192.0.0.1\n192.168.0.2'


@parameterized.expand(COMMON_TEST_DATA)
@patchfs
def test_show_dhcp_relay(test_name, test_data, fs):
    if not os.path.exists(DBCONFIG_PATH):
        fs.create_file(DBCONFIG_PATH)
    MockConfigDb.set_config_db(test_data["config_db"])
    config_db = MockConfigDb()
    ip_version = "ipv4" if "ipv4" in test_name else "ipv6"
    table = config_db.get_table(IP_VER_TEST_PARAM_MAP[ip_version]["table"])
    if test_name in ["ipv4_with_header", "ipv4_with_disabled_dhcp_server_with_header"]:
        result = show.get_dhcp_relay_data_with_header(table, IP_VER_TEST_PARAM_MAP[ip_version]["entry"])
        expected_output = expected_ipv4_table_with_header
    elif test_name == "ipv6_with_header":
        result = show.get_dhcp_relay_data_with_header(table, IP_VER_TEST_PARAM_MAP[ip_version]["entry"])
        expected_output = expected_ipv6_table_with_header
    elif test_name == "ipv6_without_header":
        result = show.get_data(table, "Vlan1000")
        expected_output = expected_ipv6_table_without_header
    elif test_name == "ipv4_with_enabled_dhcp_server_with_header":
        result = show.get_dhcp_relay_data_with_header(table, IP_VER_TEST_PARAM_MAP[ip_version]["entry"], True)
        expected_output = expected_ipv4_table_with_enabled_dhcp_server_with_header
    assert result == expected_output


@parameterized.expand(NEW_ADDED_TEST_DATA)
@patchfs
def test_show_new_added_dhcp_relay(test_name, test_data, fs):
    if not os.path.exists(DBCONFIG_PATH):
        fs.create_file(DBCONFIG_PATH)
    MockConfigDb.set_config_db(test_data["config_db"])
    config_db = MockConfigDb()
    ip_version = test_name
    table = config_db.get_table(IP_VER_TEST_PARAM_MAP[ip_version]["table"])
    if ip_version == "ipv4":
        result = show.get_dhcp_relay_data_with_header(table, IP_VER_TEST_PARAM_MAP[ip_version]["entry"])
        expected_output = expected_ipv4_table_with_header
        assert result == expected_output
    else:
        result = show.get_dhcp_relay_data_with_header(table, IP_VER_TEST_PARAM_MAP[ip_version]["entry"])
        expected_output = expected_ipv6_table_with_header
        assert result == expected_output

        result = show.get_data(table, "Vlan1001")
        expected_output = ""
        assert result == expected_output


@parameterized.expand(MULTI_TEST_DATA)
@patchfs
def test_show_multi_dhcp_relay(test_name, test_data, fs):
    if not os.path.exists(DBCONFIG_PATH):
        fs.create_file(DBCONFIG_PATH)
    MockConfigDb.set_config_db(test_data["config_db"])
    config_db = MockConfigDb()
    ip_version = test_name
    table = config_db.get_table(IP_VER_TEST_PARAM_MAP[ip_version]["table"])
    result = show.get_dhcp_relay_data_with_header(table, IP_VER_TEST_PARAM_MAP[ip_version]["entry"])
    if ip_version == "ipv4":
        expected_output = expected_ipv4_table_multi_with_header
    else:
        expected_output = expected_ipv6_table_multi_with_header
    assert result == expected_output


def test_show_dhcp_relay_ipv4_counter_with_enabled_dhcp_server():
    with mock.patch.object(show, "is_dhcp_server_enabled", return_value=True), \
         mock.patch.object(swsscommon.ConfigDBConnector, "connect", return_value=None), \
         mock.patch.object(swsscommon.ConfigDBConnector, "get_table", return_value=None), \
         mock.patch.object(click, "echo", return_value=None) as mock_echo:
        show.ipv4_counters("Etherner1")
        expected_param = "Unsupport to check dhcp_relay ipv4 counter when dhcp_server feature is enabled"
        mock_echo.assert_called_once_with(expected_param)


@pytest.mark.parametrize("enable_dhcp_server", [True, False])
def test_is_dhcp_server_enabled(enable_dhcp_server):
    result = show.is_dhcp_server_enabled({"dhcp_server": {"state": "enabled" if enable_dhcp_server else "disabled"}})
    assert result == enable_dhcp_server


class MockDb(object):
    class MockCfgDb(object):
        def __init__(self, mock_cfgdb):
            self.mock_cfgdb = mock_cfgdb

        def get_table(self, table_name):
            return self.mock_cfgdb.get(table_name, {})

    def __init__(self, mock_cfgdb):
        self.cfgdb = self.MockCfgDb(mock_cfgdb)