[chassis][multi asic] change acl_loader to use tcp socket for db communication

acl_loader/main.py

@@ -147,9 +147,9 @@ def __init__(self):
 
  namespaces = multi_asic.get_all_namespaces()
  for front_asic_namespaces in namespaces['front_ns']:
- self.per_npu_configdb[front_asic_namespaces] = ConfigDBConnector(use_unix_socket_path=True, namespace=front_asic_namespaces)
+ self.per_npu_configdb[front_asic_namespaces] = ConfigDBConnector(namespace=front_asic_namespaces)
  self.per_npu_configdb[front_asic_namespaces].connect()
- self.per_npu_statedb[front_asic_namespaces] = SonicV2Connector(use_unix_socket_path=True, namespace=front_asic_namespaces)
+ self.per_npu_statedb[front_asic_namespaces] = SonicV2Connector(namespace=front_asic_namespaces)
  self.per_npu_statedb[front_asic_namespaces].connect(self.per_npu_statedb[front_asic_namespaces].STATE_DB)
 
  self.read_tables_info()

tests/acl_loader_test.py

@@ -1,3 +1,4 @@
+import importlib
 import sys
 import os
 import pytest
@@ -217,3 +218,43 @@ def test_incremental_update(self, acl_loader):
  acl_loader.load_rules_from_file(os.path.join(test_path, 'acl_input/incremental_2.json'))
  acl_loader.incremental_update()
  assert acl_loader.rules_info[(('NTP_ACL', 'RULE_1'))]["PACKET_ACTION"] == "DROP"
+
+
+
+class TestMasicAclLoader(object):
+
+
+ @pytest.fixture(scope="class")
+ def acl_loader(self):
+ from .mock_tables import mock_multi_asic
+ importlib.reload(mock_multi_asic)
+ from .mock_tables import dbconnector
+ dbconnector.load_namespace_config()
+
+ with mock.patch("sonic_py_common.multi_asic.get_all_namespaces",
+ mock.MagicMock(return_value={'front_ns': ['asic0', 'asic1'], 'back_ns': '', 'fabric_ns': ''})):
+ yield AclLoader()
+
+ # mock single asic to avoid affecting other tests
+ from .mock_tables import mock_single_asic
+ importlib.reload(mock_single_asic)
+
+ def test_check_npu_db(self, acl_loader):
+ assert len(acl_loader.per_npu_configdb) == 2
+ assert len(acl_loader.per_npu_statedb) == 2
+
+ def test_incremental_update(self, acl_loader):
+ acl_loader.rules_info = {}
+ acl_loader.tables_db_info['NTP_ACL'] = {
+ "stage": "INGRESS",
+ "type": "CTRLPLANE"
+ }
+ acl_loader.load_rules_from_file(os.path.join(test_path, 'acl_input/incremental_1.json'))
+ acl_loader.rules_db_info = acl_loader.rules_info
+ assert acl_loader.rules_info[(('NTP_ACL', 'RULE_1'))]["PACKET_ACTION"] == "ACCEPT"
+ for configdb in acl_loader.per_npu_configdb.values():
+ configdb.mod_entry = mock.MagicMock(return_value=True)
+ configdb.set_entry = mock.MagicMock(return_value=True)
+ acl_loader.load_rules_from_file(os.path.join(test_path, 'acl_input/incremental_2.json'))
+ acl_loader.incremental_update()
+ assert acl_loader.rules_info[(('NTP_ACL', 'RULE_1'))]["PACKET_ACTION"] == "DROP"