[Meta] Add GitHub workflow for run_scripts pytest unit tests

.github/workflows/run-scripts-unittests.yml

@@ -0,0 +1,42 @@
+name: run_scripts unit tests
+
+on:
+  pull_request:
+    paths:
+      - 'fboss/oss/scripts/run_scripts/**'
+      - '.github/workflows/run-scripts-unittests.yml'
+  push:
+    branches: [ main ]
+    paths:
+      - 'fboss/oss/scripts/run_scripts/**'
+      - '.github/workflows/run-scripts-unittests.yml'
+  workflow_dispatch:
+
+# Cancel superseded runs on the same ref to avoid wasting CI minutes on
+# rapid PR pushes — hermetic tests, safe to cancel mid-run.
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+# Read-only test workflow; no need for the default broad write scope.
+permissions:
+  contents: read
+
+jobs:
+  Run-Scripts-Unit-Tests:
+    runs-on: ubuntu-latest
+    # Pure-Python pytest suite finishes in seconds; 10 min is a generous
+    # upper bound that catches genuine hangs without sitting on the GHA
+    # default 6h timeout.
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v6
+      - uses: actions/setup-python@v6
+        with:
+          python-version: '3.12'
+      - name: Install pytest and parameterized
+        # Pin major versions: hermetic tests should be reproducible across
+        # CI runs even when upstream releases breaking-API majors.
+        run: pip install 'pytest>=8,<9' 'parameterized>=0.9,<1'
+      - name: Run pytest
+        run: pytest fboss/oss/scripts/run_scripts/tests/ -v

fboss/oss/scripts/run_scripts/tests/conftest.py

@@ -14,8 +14,16 @@
 
 from argparse import ArgumentParser
 
+import run_test
 from run_test import TestRunner
 
+# `run_test.args` is populated at runtime by argparse inside main(), so the
+# attribute does not exist at import time. Declare it here so unit tests can
+# patch it without `create=True` — which means a rename of `args` in run_test.py
+# will surface as a loud AttributeError at patch time instead of being silently
+# fabricated by mock.
+run_test.args = None
+
 
 class StubTestRunner(TestRunner):
     """Concrete TestRunner with all abstract methods stubbed for testing."""

fboss/oss/scripts/run_scripts/tests/test_agent_service.py

@@ -0,0 +1,127 @@
+# Copyright Meta Platforms, Inc. and affiliates.
+# @noautodeps
+"""Tests for fboss_agent_utils HW agent systemd service lifecycle."""
+
+import os
+import sys
+from unittest.mock import ANY, MagicMock, patch
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+import fboss_agent_utils
+from fboss_agent_utils import (
+    agent_can_warm_boot_file_path,
+    cleanup_hw_agent_service,
+    setup_and_start_hw_agent_service,
+    warm_boot_hw_agent,
+)
+
+
+class TestAgentCanWarmBootFilePath:
+    def test_none_returns_base_path(self):
+        path = agent_can_warm_boot_file_path(switch_index=None)
+        assert path == fboss_agent_utils.FBOSS_AGENT_WB_FLAG_FILE
+        assert not path.endswith("_0")
+
+    def test_index_appends_underscore_suffix(self):
+        assert agent_can_warm_boot_file_path(switch_index=0).endswith("_0")
+        assert agent_can_warm_boot_file_path(switch_index=3).endswith("_3")
+
+
+class TestCleanupHwAgentService:
+    def test_stops_all_three_service_variants_and_pkills(self):
+        with patch.object(fboss_agent_utils.subprocess, "run") as mock_run:
+            cleanup_hw_agent_service([0])
+
+        joined = "\n".join(c.args[0] for c in mock_run.call_args_list)
+        assert "systemctl stop fboss_hw_agent@0" in joined
+        assert "systemctl stop fboss_hw_agent_for_testing_0" in joined
+        assert "systemctl stop fboss_hw_agent_oss@0" in joined
+        assert "systemctl disable fboss_hw_agent_oss@0" in joined
+        assert "systemctl daemon-reload" in joined
+        assert "pkill -f fboss_hw_agent@0" in joined
+        assert "pkill -f fboss_hw_agent_for_testing_0" in joined
+        assert "pkill -f fboss_hw_agent_oss@0" in joined
+
+
+class TestSetupHwAgentServicePreconditions:
+    def test_raises_when_binary_missing(self, tmp_path):
+        config_path = tmp_path / "agent.conf"
+        config_path.write_text("")
+        with pytest.raises(Exception, match="HW Agent Service binary"):
+            setup_and_start_hw_agent_service(
+                switch_indexes=[0],
+                fboss_agent_config_path=str(config_path),
+                hw_agent_service_bin_path=str(tmp_path / "does_not_exist"),
+            )
+
+    def test_raises_when_config_missing(self, tmp_path):
+        with pytest.raises(Exception, match="Agent config path"):
+            setup_and_start_hw_agent_service(
+                switch_indexes=[0],
+                fboss_agent_config_path=str(tmp_path / "missing.conf"),
+                hw_agent_service_bin_path=sys.executable,
+            )
+
+
+class TestSetupAndStartDispatch:
+    # _setup_hw_agent_service writes systemd unit files to /tmp and rsyslog
+    # configs to /etc/rsyslog.d (which fails for non-root and is a real-system
+    # side effect). The tests here only verify the warm/cold dispatch and
+    # failure propagation in setup_and_start_hw_agent_service, so we stub
+    # _setup_hw_agent_service out entirely.
+    def test_warm_boot_dispatch(self):
+        with (
+            patch.object(fboss_agent_utils, "_setup_hw_agent_service"),
+            patch.object(
+                fboss_agent_utils, "warm_boot_hw_agent", return_value=[0]
+            ) as mock_warm,
+            patch.object(
+                fboss_agent_utils, "cold_boot_hw_agent", return_value=[0]
+            ) as mock_cold,
+        ):
+            setup_and_start_hw_agent_service(
+                switch_indexes=[0],
+                fboss_agent_config_path="ignored.conf",
+                hw_agent_service_bin_path=sys.executable,
+                is_warm_boot=True,
+            )
+        mock_warm.assert_called_once_with([0], service_name=ANY)
+        mock_cold.assert_not_called()
+
+    def test_cold_boot_dispatch_and_failure_raises(self):
+        with (
+            patch.object(fboss_agent_utils, "_setup_hw_agent_service"),
+            patch.object(fboss_agent_utils, "warm_boot_hw_agent", return_value=[0]),
+            patch.object(
+                fboss_agent_utils, "cold_boot_hw_agent", return_value=[1]
+            ) as mock_cold,
+            pytest.raises(Exception, match="cold-boot"),
+        ):
+            setup_and_start_hw_agent_service(
+                switch_indexes=[0],
+                fboss_agent_config_path="ignored.conf",
+                hw_agent_service_bin_path=sys.executable,
+                is_warm_boot=False,
+            )
+        mock_cold.assert_called_once_with([0], service_name=ANY)
+
+
+class TestWarmBootMissingFileBehavior:
+    def test_warmboot_proceeds_to_start_when_warmboot_file_missing(self):
+        def fake_run(cmd, **_):
+            rc = 1 if cmd.startswith("stat ") else 0
+            return MagicMock(returncode=rc)
+
+        with patch.object(
+            fboss_agent_utils.subprocess, "run", side_effect=fake_run
+        ) as mock_run:
+            return_codes = warm_boot_hw_agent(switch_indexes=[0])
+
+        commands = [c.args[0] for c in mock_run.call_args_list]
+        assert any(
+            c.startswith("systemctl start fboss_hw_agent_oss@0") for c in commands
+        )
+        assert return_codes == [0]

fboss/oss/scripts/run_scripts/tests/test_bcm_runner.py

@@ -0,0 +1,31 @@
+# Copyright Meta Platforms, Inc. and affiliates.
+# @noautodeps
+"""Tests for BcmTestRunner."""
+
+import pytest
+from run_test import BcmTestRunner
+
+
+@pytest.fixture
+def bcm_runner():
+    return BcmTestRunner()
+
+
+class TestConstants:
+    def test_default_paths_and_binary(self, bcm_runner):
+        assert bcm_runner._get_config_path() == "/etc/coop/bcm.conf"
+        assert bcm_runner._get_test_binary_name() == "/opt/fboss/bin/bcm_test"
+        assert bcm_runner._get_known_bad_tests_file() == ""
+        assert bcm_runner._get_unsupported_tests_file() == ""
+        assert bcm_runner._get_sai_replayer_logging_flags("/any/path") == []
+        assert bcm_runner._get_sai_replayer_logging_flags(None) == []
+        assert bcm_runner._get_test_run_args("dummy.conf") == []
+
+
+class TestSaiLoggingFlagsSignature:
+    def test_no_arg_signature_returns_empty(self, bcm_runner):
+        assert bcm_runner._get_sai_logging_flags() == []
+
+    def test_calling_with_sai_logging_arg_raises_typeerror(self, bcm_runner):
+        with pytest.raises(TypeError):
+            bcm_runner._get_sai_logging_flags("WARN")

fboss/oss/scripts/run_scripts/tests/test_fboss_agent_utils.py

@@ -64,7 +64,7 @@ def test_multi_npu(self, mock_run):
 
     @patch("fboss_agent_utils.subprocess.run")
     def test_raises_on_stop_failure(self, mock_run):
-        def side_effect(cmd, **kwargs):
+        def side_effect(cmd, **_kwargs):
             if "systemctl stop fboss_sw_agent" in cmd:
                 return _mock_run(1)
             return _mock_run(0)
@@ -102,7 +102,7 @@ def test_sw_down(self, mock_run):
 
     @patch("fboss_agent_utils.subprocess.run")
     def test_hw_down(self, mock_run):
-        def side_effect(cmd, **kwargs):
+        def side_effect(cmd, **_kwargs):
             if "fboss_sw_agent" in cmd:
                 return _mock_run(0)
             return _mock_run(3)

fboss/oss/scripts/run_scripts/tests/test_fsdb_service.py

@@ -0,0 +1,90 @@
+# Copyright Meta Platforms, Inc. and affiliates.
+# @noautodeps
+"""Tests for fsdb_service_utils FSDB systemd service lifecycle."""
+
+import os
+import sys
+from unittest.mock import patch
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+import fsdb_service_utils
+from fsdb_service_utils import (
+    _start_fsdb_service,
+    cleanup_fsdb_service,
+    setup_and_start_fsdb_service,
+)
+
+
+class TestCleanupFsdbService:
+    def test_stops_all_three_variants_and_pkills(self):
+        with patch.object(fsdb_service_utils.subprocess, "run") as mock_run:
+            cleanup_fsdb_service()
+        commands = "\n".join(c.args[0] for c in mock_run.call_args_list)
+        assert "systemctl stop fsdb.service" in commands
+        assert "systemctl stop fsdb_service_for_testing" in commands
+        assert "systemctl stop fsdb_service_oss" in commands
+        assert "systemctl disable fsdb_service_oss" in commands
+        assert "systemctl daemon-reload" in commands
+        assert "pkill -f fsdb_service_oss" in commands
+
+
+class TestSetupPreconditions:
+    def _patches(self, monkeypatch, tmp_path):
+        monkeypatch.setattr(
+            fsdb_service_utils,
+            "_FSDB_SERVICE_UNIT_FILE_PATH",
+            str(tmp_path / "fsdb_oss.service"),
+        )
+        monkeypatch.setattr(
+            fsdb_service_utils,
+            "_FSDB_SERVICE_RSYSLOG_CONF_PATH",
+            str(tmp_path / "rsyslog.conf"),
+        )
+
+    def test_raises_when_binary_missing(self, monkeypatch, tmp_path):
+        self._patches(monkeypatch, tmp_path)
+        monkeypatch.setattr(
+            fsdb_service_utils,
+            "_DEFAULT_OSS_FSDB_SERVICE_PATH",
+            str(tmp_path / "missing_fsdb_bin"),
+        )
+        with pytest.raises(Exception, match="fsdb_service binary"):
+            setup_and_start_fsdb_service()
+
+    def test_raises_when_config_missing(self, monkeypatch, tmp_path):
+        self._patches(monkeypatch, tmp_path)
+        monkeypatch.setattr(
+            fsdb_service_utils, "_DEFAULT_OSS_FSDB_SERVICE_PATH", sys.executable
+        )
+        with pytest.raises(Exception, match="fsdb_service config path"):
+            setup_and_start_fsdb_service(
+                fsdb_service_config_path=str(tmp_path / "missing.conf")
+            )
+
+
+class TestStartColdVsWarmBoot:
+    def test_cold_boot_creates_marker(self):
+        with (
+            patch.object(fsdb_service_utils.subprocess, "run") as mock_run,
+            patch.object(
+                fsdb_service_utils, "_setup_fsdb_service_coldboot"
+            ) as mock_marker,
+        ):
+            _start_fsdb_service(is_warm_boot=False)
+            mock_marker.assert_called_once()
+            commands = "\n".join(c.args[0] for c in mock_run.call_args_list)
+            assert "systemctl enable" in commands
+            assert "systemctl start fsdb_service_oss" in commands
+
+    def test_warm_boot_skips_marker(self):
+        with (
+            patch.object(fsdb_service_utils.subprocess, "run"),
+            patch.object(
+                fsdb_service_utils, "_setup_fsdb_service_coldboot"
+            ) as mock_marker,
+        ):
+            _start_fsdb_service(is_warm_boot=True)
+        mock_marker.assert_not_called()