bssh 3.0.1

Parallel SSH command execution tool for cluster management
Documentation
#!/usr/bin/env python3
"""Focused unit tests for the OpenSSH compatibility harness."""

from __future__ import annotations

import importlib.util
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock


MODULE_PATH = Path(__file__).with_name("run.py")
SPEC = importlib.util.spec_from_file_location("openssh_regress", MODULE_PATH)
if SPEC is None or SPEC.loader is None:
    raise RuntimeError(f"cannot load {MODULE_PATH}")
openssh_regress = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = openssh_regress
SPEC.loader.exec_module(openssh_regress)


class ManifestTests(unittest.TestCase):
    def test_committed_manifest_is_valid(self) -> None:
        selection = openssh_regress.read_selection(openssh_regress.DEFAULT_SELECTION)

        self.assertEqual(sum(row.disposition == "run" for row in selection), 79)
        self.assertEqual(sum(row.disposition == "skip" for row in selection), 11)
        self.assertTrue(all(row.reason for row in selection if row.disposition != "run"))
        self.assertEqual(
            next(row.disposition for row in selection if row.test == "forwarding"), "run"
        )
        timeout_overrides = {
            row.test: row.timeout_seconds
            for row in selection
            if row.timeout_seconds is not None
        }
        self.assertEqual(
            timeout_overrides,
            {"forward-control": 180, "sshsig": 180},
        )
        self.assertNotIn("pubkey-priority", {row.test for row in selection})

    def test_pin_includes_an_immutable_commit(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            pin_file = Path(directory) / "openssh-version"
            pin_file.write_text(
                "V_10_3_P1\n2d98db98331803cbb820211b2fb0d31a6e71e58e\n",
                encoding="utf-8",
            )
            with mock.patch.object(openssh_regress, "PIN_FILE", pin_file):
                pin = openssh_regress.read_pin()
            self.assertEqual(pin.tag, "V_10_3_P1")
            self.assertEqual(
                pin.commit, "2d98db98331803cbb820211b2fb0d31a6e71e58e"
            )

    def test_manifest_rejects_duplicate_names(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "selection.tsv"
            rows = ["test\tdisposition\treason\ttimeout_seconds"]
            rows.extend(f"test-{index}\trun\t\t" for index in range(88))
            rows.extend(["duplicate\trun\t\t", "duplicate\texclude\treason\t"])
            path.write_text("\n".join(rows) + "\n", encoding="utf-8")

            with self.assertRaisesRegex(ValueError, "duplicate"):
                openssh_regress.read_selection(path)

    def test_manifest_rejects_invalid_timeout(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "selection.tsv"
            rows = ["test\tdisposition\treason\ttimeout_seconds"]
            rows.extend(f"test-{index}\trun\t\t" for index in range(89))
            rows.append("invalid\trun\t\tzero")
            path.write_text("\n".join(rows) + "\n", encoding="utf-8")

            with self.assertRaisesRegex(ValueError, "invalid timeout"):
                openssh_regress.read_selection(path)

    def test_tree_inventory_reports_drift(self) -> None:
        selection = [
            openssh_regress.Selection("present", "run", ""),
            openssh_regress.Selection("missing", "skip", "reason"),
        ]
        with tempfile.TemporaryDirectory() as directory:
            regress = Path(directory) / "regress"
            regress.mkdir()
            (regress / "present.sh").touch()
            (regress / "unexpected.sh").touch()

            with self.assertRaisesRegex(ValueError, "missing.*unexpected"):
                openssh_regress.validate_tree_inventory(Path(directory), selection)


class EnvironmentTests(unittest.TestCase):
    def test_reference_environment_names_all_t_exec_helpers(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            tree = Path(directory)
            (tree / "Makefile").write_text(
                "TEST_SHELL = /bin/sh\nTEST_MALLOC_OPTIONS = CFGJRSUX\n", encoding="utf-8"
            )
            client = tree / "candidate bssh"

            env = openssh_regress.reference_environment(tree, client)

            wrapper = Path(env["TEST_SSH_SSH"])
            self.assertEqual(wrapper.stat().st_size, openssh_regress.CLIENT_WRAPPER_BYTES)
            self.assertTrue(os.access(wrapper, os.X_OK))
            self.assertTrue(
                wrapper.read_text(encoding="utf-8").startswith(
                    f'#!/bin/sh\nexec \'{client}\' "$@"\n#'
                )
            )

        expected = {
            "TEST_SSH_SCP",
            "TEST_SSH_SSH",
            "TEST_SSH_SSHD",
            "TEST_SSH_SSHD_SESSION",
            "TEST_SSH_SSHD_AUTH",
            "TEST_SSH_SSHAGENT",
            "TEST_SSH_SSHADD",
            "TEST_SSH_SSHKEYGEN",
            "TEST_SSH_SSHPKCS11HELPER",
            "TEST_SSH_SSHKEYSCAN",
            "TEST_SSH_SFTP",
            "TEST_SSH_PKCS11_HELPER",
            "TEST_SSH_SK_HELPER",
            "TEST_SSH_SFTPSERVER",
        }
        self.assertTrue(expected.issubset(env))
        self.assertEqual(env["MALLOC_OPTIONS"], "CFGJRSUX")

    def test_process_timeout_terminates_the_process_group(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            result = openssh_regress.run_process(
                [sys.executable, "-c", "import time; time.sleep(5)"],
                Path(directory),
                {},
                1,
            )

        self.assertTrue(result.timed_out)
        self.assertLess(result.duration_ms, 4_000)

    def test_process_output_and_log_are_bounded(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            log_path = Path(directory) / "output.log"
            with (
                mock.patch.object(openssh_regress, "MAX_CAPTURE_BYTES", 128),
                mock.patch.object(openssh_regress, "MAX_LOG_BYTES", 256),
            ):
                result = openssh_regress.run_process(
                    [
                        sys.executable,
                        "-c",
                        "import sys; sys.stdout.write(chr(120) * 2048)",
                    ],
                    Path(directory),
                    {},
                    5,
                    log_path,
                )
            log_size = log_path.stat().st_size

        self.assertIn("output truncated by harness", result.output)
        self.assertLessEqual(len(result.output.encode()), 168)
        self.assertLessEqual(log_size, 256)


    def test_detached_output_holder_cannot_defeat_timeout(self) -> None:
        child_pid = None
        try:
            with tempfile.TemporaryDirectory() as directory:
                result = openssh_regress.run_process(
                    [
                        sys.executable,
                        "-c",
                        "import subprocess,time; p=subprocess.Popen([chr(47)+chr(98)+chr(105)+chr(110)+chr(47)+chr(115)+chr(108)+chr(101)+chr(101)+chr(112), str(10)], start_new_session=True); print(p.pid, flush=True); time.sleep(10)",
                    ],
                    Path(directory),
                    {},
                    1,
                )
                child_pid = int(result.output.splitlines()[0])
            self.assertTrue(result.timed_out)
            self.assertLess(result.duration_ms, 5_000)
        finally:
            if child_pid is not None:
                try:
                    openssh_regress.os.killpg(
                        child_pid, openssh_regress.signal.SIGKILL
                    )
                except ProcessLookupError:
                    pass

    def test_log_cleanup_requires_ownership_marker(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            unowned = root / "unowned-logs"
            unowned.mkdir()
            important = unowned / "important"
            important.write_text("keep", encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "unowned"):
                openssh_regress.reset_log_directory(unowned)
            self.assertEqual(important.read_text(encoding="utf-8"), "keep")

            owned = root / "owned-logs"
            openssh_regress.reset_log_directory(owned)
            transient = owned / "old.log"
            transient.write_text("stale", encoding="utf-8")
            openssh_regress.reset_log_directory(owned)
            self.assertFalse(transient.exists())
            self.assertTrue((owned / ".bssh-openssh-regress").is_file())


class ClassificationTests(unittest.TestCase):
    def test_candidate_failure_is_genuine_only_when_reference_passes(self) -> None:
        failed = openssh_regress.ProcessResult(1, 10, "FAIL: candidate\n", False)
        passed = openssh_regress.ProcessResult(0, 12, "ok\n", False)
        with mock.patch.object(openssh_regress, "run_one", side_effect=[failed, passed]):
            result = openssh_regress.classify(
                Path("tree"), Path("bssh"), Path("ssh"), "sample", 1, Path("logs")
            )

        self.assertEqual(result.verdict, "fail")
        self.assertEqual(result.first_failure_line, "FAIL: candidate")

    def test_candidate_failure_is_environmental_when_reference_fails(self) -> None:
        candidate = openssh_regress.ProcessResult(1, 10, "FAIL: candidate\n", False)
        reference = openssh_regress.ProcessResult(1, 12, "FATAL: environment\n", False)
        with mock.patch.object(openssh_regress, "run_one", side_effect=[candidate, reference]):
            result = openssh_regress.classify(
                Path("tree"), Path("bssh"), Path("ssh"), "sample", 1, Path("logs")
            )

        self.assertEqual(result.verdict, "environmental")
        self.assertEqual(result.reference_first_failure_line, "FATAL: environment")

    def test_upstream_skip_is_not_retried(self) -> None:
        skipped = openssh_regress.ProcessResult(0, 10, "SKIPPED: unavailable\n", False)
        with mock.patch.object(openssh_regress, "run_one", return_value=skipped) as run_one:
            result = openssh_regress.classify(
                Path("tree"), Path("bssh"), Path("ssh"), "sample", 1, Path("logs")
            )

        self.assertEqual(result.verdict, "skip")
        run_one.assert_called_once()


    def test_failed_skip_marker_is_retried(self) -> None:
        failed = openssh_regress.ProcessResult(1, 10, "SKIPPED: partial\nFAIL\n", False)
        passed = openssh_regress.ProcessResult(0, 12, "ok\n", False)
        with mock.patch.object(
            openssh_regress, "run_one", side_effect=[failed, passed]
        ) as run_one:
            result = openssh_regress.classify(
                Path("tree"), Path("bssh"), Path("ssh"), "sample", 1, Path("logs")
            )

        self.assertEqual(result.verdict, "fail")
        self.assertEqual(run_one.call_count, 2)


if __name__ == "__main__":
    unittest.main()