|
| 1 | +import importlib |
1 | 2 | import sys
|
2 | 3 | import os
|
3 | 4 | import pytest
|
@@ -217,3 +218,39 @@ def test_incremental_update(self, acl_loader):
|
217 | 218 | acl_loader.load_rules_from_file(os.path.join(test_path, 'acl_input/incremental_2.json'))
|
218 | 219 | acl_loader.incremental_update()
|
219 | 220 | assert acl_loader.rules_info[(('NTP_ACL', 'RULE_1'))]["PACKET_ACTION"] == "DROP"
|
| 221 | + |
| 222 | + |
| 223 | + |
| 224 | +class TestMasicAclLoader(object): |
| 225 | + |
| 226 | + |
| 227 | + @pytest.fixture(scope="class") |
| 228 | + def acl_loader(self): |
| 229 | + from .mock_tables import mock_multi_asic |
| 230 | + importlib.reload(mock_multi_asic) |
| 231 | + from .mock_tables import dbconnector |
| 232 | + dbconnector.load_namespace_config() |
| 233 | + |
| 234 | + with mock.patch("sonic_py_common.multi_asic.get_all_namespaces", |
| 235 | + mock.MagicMock(return_value={'front_ns': ['asic0', 'asic1'], 'back_ns': '', 'fabric_ns': ''})): |
| 236 | + yield AclLoader() |
| 237 | + |
| 238 | + def test_check_npu_db(self, acl_loader): |
| 239 | + assert len(acl_loader.per_npu_configdb) == 2 |
| 240 | + assert len(acl_loader.per_npu_statedb) == 2 |
| 241 | + |
| 242 | + def test_incremental_update(self, acl_loader): |
| 243 | + acl_loader.rules_info = {} |
| 244 | + acl_loader.tables_db_info['NTP_ACL'] = { |
| 245 | + "stage": "INGRESS", |
| 246 | + "type": "CTRLPLANE" |
| 247 | + } |
| 248 | + acl_loader.load_rules_from_file(os.path.join(test_path, 'acl_input/incremental_1.json')) |
| 249 | + acl_loader.rules_db_info = acl_loader.rules_info |
| 250 | + assert acl_loader.rules_info[(('NTP_ACL', 'RULE_1'))]["PACKET_ACTION"] == "ACCEPT" |
| 251 | + for configdb in acl_loader.per_npu_configdb.values(): |
| 252 | + configdb.mod_entry = mock.MagicMock(return_value=True) |
| 253 | + configdb.set_entry = mock.MagicMock(return_value=True) |
| 254 | + acl_loader.load_rules_from_file(os.path.join(test_path, 'acl_input/incremental_2.json')) |
| 255 | + acl_loader.incremental_update() |
| 256 | + assert acl_loader.rules_info[(('NTP_ACL', 'RULE_1'))]["PACKET_ACTION"] == "DROP" |
0 commit comments