[caclmgrd]: Tests for IP2ME rules generation

Signed-off-by: Christian Svensson <blue@cmd.nu>

tests/caclmgrd/caclmgrd_ip2me_test.py

@@ -0,0 +1,42 @@
+import os
+import sys
+
+from swsscommon import swsscommon
+from parameterized import parameterized
+from sonic_py_common.general import load_module_from_source
+from unittest import TestCase, mock
+from pyfakefs.fake_filesystem_unittest import patchfs
+
+from .test_ip2me_vectors import CACLMGRD_IP2ME_TEST_VECTOR
+from tests.common.mock_configdb import MockConfigDb
+
+
+DBCONFIG_PATH = '/var/run/redis/sonic-db/database_config.json'
+
+
+class TestCaclmgrdIP2Me(TestCase):
+    """
+        Test caclmgrd IP2Me
+    """
+    def setUp(self):
+        swsscommon.ConfigDBConnector = MockConfigDb
+        test_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+        modules_path = os.path.dirname(test_path)
+        scripts_path = os.path.join(modules_path, "scripts")
+        sys.path.insert(0, modules_path)
+        caclmgrd_path = os.path.join(scripts_path, 'caclmgrd')
+        self.caclmgrd = load_module_from_source('caclmgrd', caclmgrd_path)
+        self.maxDiff = None
+
+    @parameterized.expand(CACLMGRD_IP2ME_TEST_VECTOR)
+    @patchfs
+    def test_caclmgrd_ip2me(self, test_name, test_data, fs):
+        if not os.path.exists(DBCONFIG_PATH):
+            fs.create_file(DBCONFIG_PATH) # fake database_config.json
+
+        MockConfigDb.set_config_db(test_data["config_db"])
+        self.caclmgrd.ControlPlaneAclManager.get_namespace_mgmt_ip = mock.MagicMock()
+        self.caclmgrd.ControlPlaneAclManager.get_namespace_mgmt_ipv6 = mock.MagicMock()
+        caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
+        ret = caclmgrd_daemon.generate_block_ip2me_traffic_iptables_commands('')
+        self.assertListEqual(test_data["return"], ret)

tests/caclmgrd/test_ip2me_vectors.py

@@ -0,0 +1,127 @@
+from unittest.mock import call
+
+"""
+    caclmgrd ip2me block test vector
+"""
+CACLMGRD_IP2ME_TEST_VECTOR = [
+    [
+        "Only MGMT interface - default rules",
+        {
+            "config_db": {
+                "MGMT_INTERFACE": {
+                    "eth0|172.18.0.100/24": {
+                        "gwaddr": "172.18.0.1"
+                    }
+                },
+                "LOOPBACK_INTERFACE": {},
+                "VLAN_INTERFACE": {},
+                "PORTCHANNEL_INTERFACE": {},
+                "INTERFACE": {},
+                "DEVICE_METADATA": {
+                    "localhost": {
+                    }
+                },
+                "FEATURE": {},
+            },
+            "return": [
+                "iptables -A INPUT -d 172.18.0.0/32 -j DROP"
+            ],
+        },
+    ],
+    [
+        "Layer-3 loopback interfaces - block access",
+        {
+            "config_db": {
+                "LOOPBACK_INTERFACE": {
+                    "Loopback0|10.10.10.10/32": {},
+                },
+                "VLAN_INTERFACE": {},
+                "PORTCHANNEL_INTERFACE": {
+                    "PortChannel0001|10.10.11.10/32": {},
+                },
+                "INTERFACE": {
+                    "Ethernet0|10.10.12.10/32": {}
+                },
+                "MGMT_INTERFACE": {
+                    "eth0|172.18.0.100/24": {
+                        "gwaddr": "172.18.0.1"
+                    }
+                },
+                "DEVICE_METADATA": {
+                    "localhost": {
+                    }
+                },
+                "FEATURE": {},
+            },
+            "return": [
+                "iptables -A INPUT -d 10.10.10.10/32 -j DROP",
+                "iptables -A INPUT -d 172.18.0.0/32 -j DROP",
+                "iptables -A INPUT -d 10.10.11.10/32 -j DROP",
+                "iptables -A INPUT -d 10.10.12.10/32 -j DROP",
+            ],
+        },
+    ],
+    [
+        "One VLAN interface, /24, we are .1",
+        {
+            "config_db": {
+                "MGMT_INTERFACE": {
+                    "eth0|172.18.0.100/24": {
+                        "gwaddr": "172.18.0.1"
+                    }
+                },
+                "LOOPBACK_INTERFACE": {},
+                "VLAN_INTERFACE": {
+                    "Vlan110|10.10.11.1/24": {},
+                },
+                "PORTCHANNEL_INTERFACE": {},
+                "INTERFACE": {},
+                "DEVICE_METADATA": {
+                    "localhost": {
+                    }
+                },
+                "FEATURE": {},
+            },
+            "return": [
+                "iptables -A INPUT -d 172.18.0.0/32 -j DROP",
+                "iptables -A INPUT -d 10.10.11.1/32 -j DROP",
+            ],
+        },
+    ],
+    [
+        "One interface of each type, IPv6, /64 - block all interfaces but MGMT",
+        {
+            "config_db": {
+                "LOOPBACK_INTERFACE": {
+                    "Loopback0|2001:db8:10::/64": {},
+                },
+                "VLAN_INTERFACE": {
+                    "Vlan110|2001:db8:11::/64": {},
+                },
+                "PORTCHANNEL_INTERFACE": {
+                    "PortChannel0001|2001:db8:12::/64": {},
+                },
+                "INTERFACE": {
+                    "Ethernet0|2001:db8:13::/64": {}
+                },
+                "MGMT_INTERFACE": {
+                    "eth0|2001:db8:200::200/64": {
+                        "gwaddr": "2001:db8:200::100"
+                    }
+                },
+                "DEVICE_METADATA": {
+                    "localhost": {
+                    }
+                },
+                "FEATURE": {},
+            },
+            "return": [
+                "ip6tables -A INPUT -d 2001:db8:10::/128 -j DROP",
+                "ip6tables -A INPUT -d 2001:db8:200::/128 -j DROP",
+                "ip6tables -A INPUT -d 2001:db8:11::1/128 -j DROP",
+                "ip6tables -A INPUT -d 2001:db8:12::/128 -j DROP",
+                "ip6tables -A INPUT -d 2001:db8:13::/128 -j DROP"
+            ],
+        },
+    ]
+]

tests/common/mock_configdb.py

@@ -43,7 +43,10 @@ def set_entry(self, key, field, data):
         MockConfigDb.CONFIG_DB[key][field] = data
 
     def get_table(self, table_name):
-        return MockConfigDb.CONFIG_DB[table_name]
+        data = {}
+        for k, v in MockConfigDb.CONFIG_DB[table_name].items():
+            data[self.deserialize_key(k)] = v
+        return data
 
     def subscribe(self, table_name, callback):
         self.handlers[table_name] = callback