test_docker_lint.py

#!/usr/bin/env python3
"""Test bounded and observable Docker lint execution."""

from __future__ import annotations

import contextlib
import io
import os
import re
import signal
import stat
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock

from host_capability import requires_symlink

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
import docker_lint

MOUNT_SOURCE = re.compile(r'^type=bind,"src=(.*)",dst=')


class FakeRunner:
    """Record commands and return configured Docker results."""

    def __init__(self, failure: int | None = None, timeout: bool = False) -> None:
        self.commands: list[tuple[list[str], int]] = []
        self.mounted: list[dict[str, bool]] = []
        self.sources: list[Path] = []
        self.modes: list[int] = []
        self.failure = failure
        self.timeout = timeout

    def __call__(
        self, command: list[str], timeout: int, *, capture_output: bool = False
    ) -> subprocess.CompletedProcess[str]:
        self.commands.append((command, timeout))
        for argument in command:
            source = MOUNT_SOURCE.search(argument)
            if source:
                tree = Path(source.group(1).replace('""', '"'))
                self.mounted.append(
                    {
                        path.relative_to(tree).as_posix(): path.is_symlink()
                        for path in tree.rglob("*")
                    }
                )
                self.sources.append(tree)
                self.modes.append(stat.S_IMODE(tree.stat().st_mode))
        if self.timeout and command[:2] == ["docker", "run"]:
            raise docker_lint.CommandTimedOut(f"timed out after {timeout}s")
        if command[:3] == ["docker", "image", "inspect"]:
            return subprocess.CompletedProcess(command, 0, "example@sha256:123\n", "")
        returncode = (
            self.failure if self.failure is not None and command[:2] == ["docker", "run"] else 0
        )
        return subprocess.CompletedProcess(command, returncode, "", "")


class DockerLintCase(unittest.TestCase):
    """Build a small tracked tree for each runner scenario."""

    def setUp(self) -> None:
        self.root = Path(self.enterContext(tempfile.TemporaryDirectory()))
        subprocess.run(["git", "init", "-q", str(self.root)], check=True)
        # A separate temp dir, not self.root's own parent, which is the shared system temp root.
        self.outside = Path(self.enterContext(tempfile.TemporaryDirectory()))

    def track(self, name: str, body: str = "content\n") -> None:
        path = self.root / name
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(body, encoding="utf-8")
        subprocess.run(["git", "-C", str(self.root), "add", "--", name], check=True)

    def track_symlink(self, name: str, target: Path) -> None:
        path = self.root / name
        path.parent.mkdir(parents=True, exist_ok=True)
        path.symlink_to(target)
        subprocess.run(["git", "-C", str(self.root), "add", "--", name], check=True)

    def invoke(self, selected: set[str], runner: FakeRunner) -> tuple[int, str]:
        output = io.StringIO()
        with contextlib.redirect_stdout(output):
            result = docker_lint.lint(self.root, 17, selected, runner)
        return result, output.getvalue()

    def test_quiet_success_reports_targets_phases_and_completion(self) -> None:
        self.track("README.md")
        runner = FakeRunner()
        result, output = self.invoke({"markdownlint"}, runner)
        self.assertEqual(0, result)
        self.assertIn("TARGETS markdownlint: 1 file(s)", output)
        self.assertIn("PHASE pull", output)
        self.assertIn("START inspect markdownlint (timeout 17s)", output)
        self.assertIn("COMPLETE inspect markdownlint", output)
        self.assertIn("PHASE execution: pulls complete, repository mounts begin", output)
        self.assertIn("START lint markdownlint (1 file(s)) (timeout 17s)", output)
        self.assertIn("COMPLETE lint markdownlint (1 file(s))", output)
        self.assertIn("RESULT success: 1 linter(s) completed", output)
        run = next(command for command, _ in runner.commands if command[:2] == ["docker", "run"])
        self.assertIn("--network=none", run)
        self.assertEqual(1, len(runner.sources))
        self.assertNotEqual(self.root, runner.sources[0])
        self.assertIn(docker_lint.docker_mount(runner.sources[0], "/workdir"), run)
        self.assertIn("example@sha256:123", run)
        self.assertEqual("--", run[-2])
        self.assertEqual("README.md", run[-1])
        self.assertTrue(all(timeout == 17 for _, timeout in runner.commands))

    def test_zero_targets_skips_pull_and_execution(self) -> None:
        runner = FakeRunner()
        result, output = self.invoke({"shellcheck"}, runner)
        self.assertEqual(0, result)
        self.assertIn("SKIP shellcheck: zero targets", output)
        self.assertIn("COMPLETE lint: zero applicable linters", output)
        self.assertEqual([], runner.commands)

    def test_timeout_is_distinct_from_container_failure(self) -> None:
        self.track("README.md")
        result, output = self.invoke({"markdownlint"}, FakeRunner(timeout=True))
        self.assertEqual(1, result)
        self.assertIn("TIMEOUT lint markdownlint", output)
        self.assertIn("RESULT failed: timed out after 17s", output)
        self.assertNotIn("FAILED lint markdownlint", output)

    def test_container_failure_reports_exit_code(self) -> None:
        self.track("README.md")
        result, output = self.invoke({"markdownlint"}, FakeRunner(failure=9))
        self.assertEqual(1, result)
        self.assertIn("FAILED lint markdownlint (1 file(s)) (exit 9)", output)
        self.assertIn("RESULT failed: container failure (exit 9)", output)

    def test_process_start_failure_is_not_reported_as_timeout(self) -> None:
        output = io.StringIO()
        with (
            mock.patch.object(subprocess, "run", side_effect=FileNotFoundError("docker")),
            contextlib.redirect_stdout(output),
            self.assertRaisesRegex(docker_lint.CommandFailed, "could not start command"),
        ):
            docker_lint.run_step(
                "missing Docker", ["docker", "pull", "image"], 17, docker_lint.run_command
            )
        self.assertIn("FAILED missing Docker: could not start command:", output.getvalue())
        self.assertNotIn("TIMEOUT", output.getvalue())

    def test_shellcheck_receives_each_tracked_file_as_one_argument(self) -> None:
        self.track("scripts/a shell.sh")
        runner = FakeRunner()
        result, _ = self.invoke({"shellcheck"}, runner)
        self.assertEqual(0, result)
        run = next(command for command, _ in runner.commands if command[:2] == ["docker", "run"])
        self.assertEqual("scripts/a shell.sh", run[-1])

    def test_shellcheck_literal_marker_precedes_option_shaped_filename(self) -> None:
        linter = next(linter for linter in docker_lint.LINTERS if linter.name == "shellcheck")
        command = docker_lint.container_command(
            self.root, linter, "example@sha256:123", ["-release.sh"]
        )
        self.assertEqual(["--", "-release.sh"], command[-2:])

    def test_shfmt_literal_marker_precedes_option_shaped_filename(self) -> None:
        linter = next(linter for linter in docker_lint.LINTERS if linter.name == "shfmt")
        command = docker_lint.container_command(
            self.root, linter, "example@sha256:123", ["-release.sh"]
        )
        self.assertEqual(["-d", "--", "-release.sh"], command[-3:])

    def test_shellcheck_and_shfmt_pick_up_an_extensionless_shebang_script(self) -> None:
        self.track("ops/vps-backup-pull", "#!/usr/bin/env bash\nset -Eeuo pipefail\necho hi\n")
        for name in ("shellcheck", "shfmt"):
            linter = next(linter for linter in docker_lint.LINTERS if linter.name == name)
            self.assertEqual(["ops/vps-backup-pull"], docker_lint.tracked_files(self.root, linter))

    def test_extensionless_non_shell_file_is_not_picked_up(self) -> None:
        self.track("ops/README", "not a script\n")
        self.track("ops/run-me", "#!/usr/bin/env python3\nprint('hi')\n")
        linter = next(linter for linter in docker_lint.LINTERS if linter.name == "shellcheck")
        self.assertEqual([], docker_lint.tracked_files(self.root, linter))

    def test_extensionless_shebang_script_merges_with_glob_matched_scripts(self) -> None:
        self.track("regular.sh", "#!/usr/bin/env bash\necho hi\n")
        self.track("ops/vps-backup-pull", "#!/bin/sh\necho hi\n")
        linter = next(linter for linter in docker_lint.LINTERS if linter.name == "shellcheck")
        self.assertEqual(
            ["ops/vps-backup-pull", "regular.sh"],
            docker_lint.tracked_files(self.root, linter),
        )

    @requires_symlink
    def test_extensionless_symlink_is_never_followed(self) -> None:
        secret = self.outside / "secret"
        secret.write_text("#!/usr/bin/env bash\n", encoding="utf-8")
        self.track_symlink("ops/evil-symlink", secret)
        linter = next(linter for linter in docker_lint.LINTERS if linter.name == "shellcheck")
        with mock.patch.object(Path, "open", side_effect=AssertionError("symlink target opened")):
            self.assertEqual([], docker_lint.tracked_files(self.root, linter))

    @requires_symlink
    def test_has_shell_shebang_reports_false_for_a_symlink_without_reading_it(self) -> None:
        secret = self.outside / "secret"
        secret.write_text("#!/usr/bin/env bash\n", encoding="utf-8")
        self.track_symlink("ops/evil-symlink", secret)
        with mock.patch.object(Path, "open", side_effect=AssertionError("symlink target opened")):
            self.assertFalse(docker_lint.has_shell_shebang(self.root, "ops/evil-symlink"))

    def test_has_shell_shebang_raises_rather_than_swallowing_a_read_failure(self) -> None:
        self.track("ops/unreadable")
        with (
            mock.patch.object(Path, "open", side_effect=PermissionError("denied")),
            self.assertRaisesRegex(docker_lint.CommandFailed, "could not read"),
        ):
            docker_lint.has_shell_shebang(self.root, "ops/unreadable")

    def test_extensionless_untracked_shebang_script_is_not_picked_up(self) -> None:
        path = self.root / "ops" / "vps-backup-pull"
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text("#!/usr/bin/env bash\necho hi\n", encoding="utf-8")
        linter = next(linter for linter in docker_lint.LINTERS if linter.name == "shellcheck")
        self.assertEqual([], docker_lint.tracked_files(self.root, linter))

    def test_node_modules_is_not_a_target_at_any_depth(self) -> None:
        for name in ("node_modules/pkg/README.md", "web/node_modules/pkg/README.md"):
            path = self.root / name
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text("# Third party\n", encoding="utf-8")
        self.track("README.md", "# Ours\n")
        markdownlint = next(
            linter for linter in docker_lint.LINTERS if linter.name == "markdownlint"
        )
        self.assertEqual(["README.md"], docker_lint.tracked_files(self.root, markdownlint))

    def test_pattern_free_linter_still_gets_targets_without_node_modules(self) -> None:
        self.track("README.md")
        path = self.root / "node_modules" / "pkg" / "index.js"
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text("module.exports = 1;\n", encoding="utf-8")
        editorconfig = next(
            linter for linter in docker_lint.LINTERS if linter.name == "editorconfig-checker"
        )
        self.assertEqual(["README.md"], docker_lint.tracked_files(self.root, editorconfig))

    def test_node_modules_shell_script_is_not_a_shellcheck_target(self) -> None:
        path = self.root / "node_modules" / "pkg" / "bin" / "run.sh"
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text("#!/bin/sh\necho hi\n", encoding="utf-8")
        shellcheck = next(linter for linter in docker_lint.LINTERS if linter.name == "shellcheck")
        self.assertEqual([], docker_lint.tracked_files(self.root, shellcheck))

    def test_shell_shebang_interpreter_rejects_bash_as_a_plain_argument(self) -> None:
        cases = {
            "#!/bin/bash": "bash",
            "#!/bin/sh": "sh",
            "#!/usr/bin/env bash": "bash",
            "#!/usr/bin/env\tbash": "bash",
            "#!/usr/bin/env -S bash -e": "bash",
            "#!/usr/bin/python bash": None,
            "#!/usr/bin/env python sh": None,
            "#!/usr/bin/env -S python -m sh": None,
            "#!/usr/bin/env": None,
            "not a shebang": None,
        }
        for line, expected in cases.items():
            with self.subTest(line=line):
                self.assertEqual(expected, docker_lint.shell_shebang_interpreter(line))

    def test_extensionless_script_naming_bash_only_as_an_argument_is_excluded(self) -> None:
        self.track("ops/run-me", "#!/usr/bin/python bash\nprint('hi')\n")
        linter = next(linter for linter in docker_lint.LINTERS if linter.name == "shellcheck")
        self.assertEqual([], docker_lint.tracked_files(self.root, linter))

    def test_extensionless_shebang_script_with_no_trailing_newline_is_picked_up(self) -> None:
        self.track("ops/vps-backup-pull", "#!/usr/bin/env bash")
        linter = next(linter for linter in docker_lint.LINTERS if linter.name == "shellcheck")
        self.assertEqual(["ops/vps-backup-pull"], docker_lint.tracked_files(self.root, linter))

    def test_shell_shebang_interpreter_walks_past_env_grammar(self) -> None:
        cases = {
            "#!/usr/bin/env FOO=1 bash": "bash",
            "#!/usr/bin/env -u bash python": None,
            "#!/usr/bin/env -i FOO=1 bash": "bash",
            "#!/usr/bin/env --unset=FOO bash": "bash",
            "#!/usr/bin/env -C /tmp bash": "bash",
            "#!/usr/bin/env FOO=1 BAR=2 sh": "sh",
        }
        for line, expected in cases.items():
            with self.subTest(line=line):
                self.assertEqual(expected, docker_lint.shell_shebang_interpreter(line))

    def test_cspell_literal_marker_precedes_option_shaped_filename(self) -> None:
        linter = next(linter for linter in docker_lint.LINTERS if linter.name == "cspell")
        command = docker_lint.container_command(
            self.root, linter, "example@sha256:123", ["-release.md"]
        )
        self.assertEqual(["--", "-release.md"], command[-2:])

    def test_invalid_git_root_reports_failed_result(self) -> None:
        root = Path(self.enterContext(tempfile.TemporaryDirectory()))
        output = io.StringIO()
        with contextlib.redirect_stdout(output):
            result = docker_lint.lint(root, 17, {"markdownlint"}, FakeRunner())
        self.assertEqual(1, result)
        self.assertIn("RESULT failed: target discovery failed:", output.getvalue())

    def test_mount_quotes_commas_and_embedded_quotes(self) -> None:
        root = Path('/tmp/docker-lint,"root')
        self.assertEqual(
            f'type=bind,"src={root.parent}{os.sep}docker-lint,""root",dst=/workdir,readonly',
            docker_lint.docker_mount(root, "/workdir"),
        )

    def test_git_ignored_files_never_reach_the_mounted_tree(self) -> None:
        self.track(".gitignore", "secrets.yaml\n.artifacts/\n")
        self.track("README.md")
        self.track("docs/guide.md")
        (self.root / "secrets.yaml").write_text("password: example\n", encoding="utf-8")
        (self.root / ".artifacts" / "bin").mkdir(parents=True)
        (self.root / ".artifacts" / "bin" / "Example.pdb").write_bytes(b"\r\n")
        (self.root / "notes.md").write_text("untracked, unignored\n", encoding="utf-8")
        runner = FakeRunner()
        result, output = self.invoke({"editorconfig-checker", "markdownlint"}, runner)
        self.assertEqual(0, result)
        self.assertIn("SNAPSHOT 4 tracked or unignored path(s) staged to mount", output)
        self.assertEqual(2, len(runner.mounted))
        for mounted in runner.mounted:
            self.assertEqual(
                {".git", ".gitignore", "README.md", "docs", "docs/guide.md", "notes.md"},
                set(mounted),
            )
        git_dir = (self.root / ".git").resolve()
        self.assertTrue(all(source.parent.resolve() == git_dir for source in runner.sources))
        self.assertFalse(any(source.exists() for source in runner.sources))

    @requires_symlink
    def test_snapshot_copies_a_symlink_without_following_it(self) -> None:
        outside = self.outside / "outside.md"
        outside.write_text("outside the checkout\n", encoding="utf-8")
        self.track("README.md")
        self.track_symlink("linked.md", outside)
        runner = FakeRunner()
        result, _ = self.invoke({"markdownlint"}, runner)
        self.assertEqual(0, result)
        self.assertEqual({".git": False, "README.md": False, "linked.md": True}, runner.mounted[0])

    def test_snapshot_skips_a_tracked_file_deleted_from_disk(self) -> None:
        self.track("README.md")
        self.track("gone.md")
        (self.root / "gone.md").unlink()
        runner = FakeRunner()
        result, _ = self.invoke({"editorconfig-checker"}, runner)
        self.assertEqual(0, result)
        self.assertEqual({".git": False, "README.md": False}, runner.mounted[0])

    def test_a_tracked_file_deleted_from_disk_is_not_a_lint_target(self) -> None:
        self.track("README.md")
        self.track("gone.md")
        (self.root / "gone.md").unlink()
        markdownlint = next(
            linter for linter in docker_lint.LINTERS if linter.name == "markdownlint"
        )
        self.assertEqual(["README.md"], docker_lint.tracked_files(self.root, markdownlint))

    def test_a_deleted_extensionless_script_does_not_fail_discovery(self) -> None:
        self.track("run", "#!/bin/bash\necho run\n")
        self.track("gone", "#!/bin/sh\necho gone\n")
        (self.root / "gone").unlink()
        shellcheck = next(linter for linter in docker_lint.LINTERS if linter.name == "shellcheck")
        self.assertEqual(["run"], docker_lint.tracked_files(self.root, shellcheck))

    def test_sigterm_during_execution_still_removes_the_snapshot(self) -> None:
        self.track("README.md")
        sources: list[Path] = []

        def terminated(
            command: list[str], timeout: int, *, capture_output: bool = False
        ) -> subprocess.CompletedProcess[str]:
            for argument in command:
                source = MOUNT_SOURCE.search(argument)
                if source:
                    sources.append(Path(source.group(1)))
                    docker_lint.exit_on_sigterm(15, None)
            return FakeRunner()(command, timeout, capture_output=capture_output)

        with (
            contextlib.redirect_stdout(io.StringIO()),
            self.assertRaises(SystemExit) as raised,
        ):
            docker_lint.lint(self.root, 17, {"markdownlint"}, terminated)
        self.assertEqual(143, raised.exception.code)
        self.assertEqual(1, len(sources))
        self.assertFalse(sources[0].exists())

    def test_main_registers_the_sigterm_handler(self) -> None:
        previous = signal.getsignal(signal.SIGTERM)
        self.addCleanup(signal.signal, signal.SIGTERM, previous)
        with mock.patch.object(docker_lint, "lint", return_value=0) as lint:
            self.assertEqual(0, docker_lint.main(["--root", str(self.root)]))
        lint.assert_called_once()
        self.assertIs(docker_lint.exit_on_sigterm, signal.getsignal(signal.SIGTERM))

    def test_snapshot_of_a_read_only_root_still_stages_and_takes_its_mode(self) -> None:
        self.track("README.md")
        self.root.chmod(0o555)
        self.addCleanup(self.root.chmod, 0o755)
        runner = FakeRunner()
        result, output = self.invoke({"markdownlint"}, runner)
        self.assertEqual(0, result, output)
        self.assertEqual({".git": False, "README.md": False}, runner.mounted[0])
        self.assertEqual(0o555, runner.modes[0])

    def test_markdown_literal_marker_precedes_negated_filename(self) -> None:
        linter = next(linter for linter in docker_lint.LINTERS if linter.name == "markdownlint")
        command = docker_lint.container_command(
            self.root, linter, "example@sha256:123", ["!release.md"]
        )
        self.assertEqual(["--", "!release.md"], command[-2:])

    def test_long_file_lists_are_split_into_multiple_commands(self) -> None:
        linter = next(linter for linter in docker_lint.LINTERS if linter.name == "shellcheck")
        files = [f"scripts/{index}-{'x' * 200}.sh" for index in range(400)]
        batches = docker_lint.file_batches(linter, files)
        self.assertGreater(len(batches), 1)
        self.assertEqual(files, [file for batch in batches for file in batch])
        self.assertTrue(
            all(
                sum(len(file.encode()) + 4 for file in batch) <= docker_lint.MAX_FILE_ARGUMENT_BYTES
                for batch in batches
            )
        )

        runner = FakeRunner()
        with mock.patch.object(docker_lint, "tracked_files", return_value=files):
            result, output = self.invoke({"shellcheck"}, runner)
        self.assertEqual(0, result)
        runs = [command for command, _ in runner.commands if command[:2] == ["docker", "run"]]
        self.assertEqual(len(batches), len(runs))
        self.assertIn(f"batch 1/{len(batches)}", output)

    def test_timed_out_container_is_removed_with_a_shorter_bound(self) -> None:
        command = ["docker", "run", "--name", "bounded-lint", "image"]
        timed_out = subprocess.TimeoutExpired(command, 90)
        cleanup: subprocess.CompletedProcess[str] = subprocess.CompletedProcess(["docker", "rm"], 0)
        with (
            mock.patch.object(subprocess, "run", side_effect=[timed_out, cleanup]) as run,
            self.assertRaisesRegex(docker_lint.CommandFailed, "timed out after 90s"),
        ):
            docker_lint.run_command(command, 90)
        self.assertEqual(90, run.call_args_list[0].kwargs["timeout"])
        self.assertEqual(["docker", "rm", "--force", "bounded-lint"], run.call_args_list[1].args[0])
        self.assertEqual(30, run.call_args_list[1].kwargs["timeout"])

    def test_cleanup_start_failure_remains_a_timeout_result(self) -> None:
        command = ["docker", "run", "--name", "bounded-lint", "image"]
        timed_out = subprocess.TimeoutExpired(command, 90)
        with (
            mock.patch.object(subprocess, "run", side_effect=[timed_out, OSError("no docker")]),
            self.assertRaisesRegex(
                docker_lint.CommandTimedOut, "cleanup could not start for bounded-lint"
            ),
        ):
            docker_lint.run_command(command, 90)


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