diff --git a/config/main.py b/config/main.py index 3958c6b760..0edee71969 100644 --- a/config/main.py +++ b/config/main.py @@ -1206,6 +1206,10 @@ def _restart_services(): reset_mgmt_interface_if_usb_not_running() def _per_namespace_swss_ready(service_name): + out, _ = clicommon.run_command(['systemctl', 'show', str(service_name), '--property', 'LoadState', '--value'], return_cmd=True) + if out.strip() in ("not-found", "masked"): + # swss not present on this platform (e.g. BMC): nothing to wait for. + return True out, _ = clicommon.run_command(['systemctl', 'show', str(service_name), '--property', 'ActiveState', '--value'], return_cmd=True) if out.strip() != "active": return False diff --git a/tests/config_test.py b/tests/config_test.py index 1cdc76c6b8..87e04d7b0a 100644 --- a/tests/config_test.py +++ b/tests/config_test.py @@ -5324,3 +5324,65 @@ def test_banner_motd(self): @classmethod def teardown_class(cls): print('TEARDOWN') + + +class TestSwssReady(object): + """Tests for the 'config reload' swss readiness gate on platforms without + swss (e.g. BMC), plus regression guards for normal switches.""" + + @classmethod + def setup_class(cls): + os.environ['UTILITIES_UNIT_TESTING'] = "1" + import config.main + importlib.reload(config.main) + + @classmethod + def teardown_class(cls): + os.environ['UTILITIES_UNIT_TESTING'] = "0" + + def test_swss_ready_when_service_not_found(self): + # BMC: swss not built in -> not-found -> ready, only LoadState queried. + with mock.patch('config.main.clicommon.run_command', + mock.MagicMock(return_value=("not-found", 0))) as mock_run: + assert config._per_namespace_swss_ready("swss.service") is True + assert mock_run.call_count == 1 + + def test_swss_ready_when_service_masked(self): + # Masked service -> ready, same as not-found. + with mock.patch('config.main.clicommon.run_command', + mock.MagicMock(return_value=("masked", 0))) as mock_run: + assert config._per_namespace_swss_ready("swss.service") is True + assert mock_run.call_count == 1 + + def test_swss_not_ready_when_loaded_but_inactive(self): + # Switch still booting: loaded but inactive -> not ready. + side_effect = [("loaded", 0), ("inactive", 0)] + with mock.patch('config.main.clicommon.run_command', + mock.MagicMock(side_effect=side_effect)): + assert config._per_namespace_swss_ready("swss.service") is False + + def test_swss_not_ready_when_active_but_not_settled(self): + # Active for < 120s -> not ready. + side_effect = [("loaded", 0), ("active", 0), ("1000000000", 0)] # up = 1000s + with mock.patch('config.main.clicommon.run_command', + mock.MagicMock(side_effect=side_effect)), \ + mock.patch('config.main.time.monotonic', + mock.MagicMock(return_value=1050.0)): # 50s < 120s + assert config._per_namespace_swss_ready("swss.service") is False + + def test_swss_ready_when_active_and_settled(self): + # Active for > 120s -> ready. + side_effect = [("loaded", 0), ("active", 0), ("1000000000", 0)] # up = 1000s + with mock.patch('config.main.clicommon.run_command', + mock.MagicMock(side_effect=side_effect)), \ + mock.patch('config.main.time.monotonic', + mock.MagicMock(return_value=2000.0)): # 1000s > 120s + assert config._per_namespace_swss_ready("swss.service") is True + + def test_swss_ready_single_asic_not_found(self): + # _swss_ready() end-to-end, single-ASIC, swss not-found. + with mock.patch('config.main.multi_asic.get_num_asics', + mock.MagicMock(return_value=1)), \ + mock.patch('config.main.clicommon.run_command', + mock.MagicMock(return_value=("not-found", 0))): + assert config._swss_ready() is True