Gecko [ tanishkshah.com ]

Name	Size	Permission	Date	Action
test_install_xray_plugin.py	29.14 KB	-rw-r--r--	2026-08-11 11:44
test_upload_reports_destination.py	7.35 KB	-rw-r--r--	2026-08-11 11:44

Rename

"""
Product-behaviour tests for the xray report-upload scripts' destination guard.

These tests pin the post-fix contract of ``upload-reports.sh`` and
``upload-reports-test.sh`` at the repo root: each script accepts exactly one
plain SSH destination and forwards *only* that destination to ``ssh`` ahead of
the fixed ``manage-files rollout`` remote command. An invocation that supplies
no destination, more than one positional, or an option-like destination (one
starting with ``-``) must be rejected before any ``ssh`` is spawned.

The scripts are exercised as black boxes. ``ssh``, ``tar`` and the
``build-utils/build-reports.sh`` build step are replaced with recording stubs
so the test has no real side effects (no network, no build): a stub ``ssh``
writes its argv to a file, letting the test assert positively on what the
script would have run.

Run from the repo root with:

python3 -m unittest discover -s plugins/tests -p 'test_*.py'

or directly:

python3 plugins/tests/test_upload_reports_destination.py
"""

import os
import shutil
import subprocess
import tempfile
import textwrap
import unittest

HERE = os.path.dirname(os.path.abspath(__file__))
REPO_ROOT = os.path.normpath(os.path.join(HERE, "..", ".."))

# Each script and the remote "subpath" argument it pins on the manage-files
# rollout command. The guard logic is byte-identical across both scripts; we
# parametrise so a regression in either sibling is caught.
SCRIPTS = {
    "upload-reports.sh": "xray/reports",
    "upload-reports-test.sh": "xray-test/reports",
}

def _write_exec(path, body):
    with open(path, "w") as fh:
        fh.write(textwrap.dedent(body))
    os.chmod(path, 0o755)

class UploadReportsDestinationGuardTests(unittest.TestCase):
    """The upload scripts must forward only a single validated destination."""

def setUp(self):
        # The stub interpreters must be executable, so the sandbox lives on the
        # repo filesystem (the system temp mount may be mounted noexec). It is
        # created beside this test file and removed in tearDown.
        self.sandbox = tempfile.mkdtemp(dir=HERE)
        self.bin_dir = os.path.join(self.sandbox, "bin")
        self.run_dir = os.path.join(self.sandbox, "run")
        os.makedirs(self.bin_dir)
        os.makedirs(os.path.join(self.run_dir, "dist"))
        os.makedirs(os.path.join(self.run_dir, "build-utils"))

# File where the stub ssh records the argv it was invoked with. Its
        # absence after a run means ssh was never called.
        self.ssh_argv_file = os.path.join(self.sandbox, "ssh_argv")

# Stub the local build step the scripts call by relative path.
        _write_exec(
            os.path.join(self.run_dir, "build-utils", "build-reports.sh"),
            """\
            #!/bin/sh
            exit 0
            """,
        )
        # Stub ssh: record argv, succeed. tar: succeed silently.
        _write_exec(
            os.path.join(self.bin_dir, "ssh"),
            """\
            #!/bin/sh
            printf '%s\\n' "$@" > "{argv_file}"
            exit 0
            """.format(argv_file=self.ssh_argv_file),
        )
        _write_exec(
            os.path.join(self.bin_dir, "tar"),
            """\
            #!/bin/sh
            exit 0
            """,
        )

def tearDown(self):
        shutil.rmtree(self.sandbox, ignore_errors=True)

def _run(self, script_name, *args):
        """Run a copy of the script in the sandbox with stubbed ssh/tar/build.

Returns the CompletedProcess. The script copy runs with the sandbox
        bin dir prepended to PATH so it resolves the stub ssh/tar.
        """
        script_copy = os.path.join(self.run_dir, script_name)
        shutil.copy(os.path.join(REPO_ROOT, script_name), script_copy)
        env = dict(os.environ)
        env["PATH"] = self.bin_dir + os.pathsep + env.get("PATH", "")
        return subprocess.run(
            ["bash", script_name, *args],
            cwd=self.run_dir,
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

def _ssh_argv(self):
        """Return the recorded ssh argv as a list, or None if ssh never ran."""
        if not os.path.exists(self.ssh_argv_file):
            return None
        with open(self.ssh_argv_file) as fh:
            return [line for line in fh.read().splitlines()]

def test_single_plain_destination_is_forwarded_to_ssh(self):
        """A lone plain destination reaches ssh as the only host token."""
        for script_name, subpath in SCRIPTS.items():
            with self.subTest(script=script_name):
                result = self._run(script_name, "files.example.com")
                self.assertEqual(
                    result.returncode,
                    0,
                    msg="stderr=%r" % result.stderr,
                )
                argv = self._ssh_argv()
                self.assertIsNotNone(argv, "ssh was not invoked")
                # ssh -T <dest> manage-files rollout --replace <subpath>
                self.assertEqual(
                    argv,
                    ["-T", "files.example.com", "manage-files", "rollout", "--replace", subpath],
                )

def test_destination_with_at_and_port_is_forwarded_verbatim(self):
        """A realistic user@host destination is accepted and passed through."""
        for script_name, subpath in SCRIPTS.items():
            with self.subTest(script=script_name):
                result = self._run(script_name, "deploy@cdn-host")
                self.assertEqual(
                    result.returncode,
                    0,
                    msg="stderr=%r" % result.stderr,
                )
                argv = self._ssh_argv()
                self.assertIsNotNone(argv, "ssh was not invoked")
                self.assertEqual(argv[0], "-T")
                self.assertEqual(argv[1], "deploy@cdn-host")

def test_option_like_destination_is_rejected_without_invoking_ssh(self):
        """A leading-dash destination is refused before ssh is spawned."""
        for script_name in SCRIPTS:
            with self.subTest(script=script_name):
                result = self._run(script_name, "-oProxyCommand=evil")
                self.assertNotEqual(result.returncode, 0)
                self.assertIsNone(
                    self._ssh_argv(),
                    "ssh must not be invoked for an option-like destination",
                )

def test_multiple_arguments_are_rejected_without_invoking_ssh(self):
        """More than one positional is refused before ssh is spawned."""
        for script_name in SCRIPTS:
            with self.subTest(script=script_name):
                result = self._run(script_name, "host", "extra-arg")
                self.assertNotEqual(result.returncode, 0)
                self.assertIsNone(
                    self._ssh_argv(),
                    "ssh must not be invoked for a multi-arg invocation",
                )

def test_no_arguments_are_rejected_without_invoking_ssh(self):
        """A missing destination is refused before ssh is spawned."""
        for script_name in SCRIPTS:
            with self.subTest(script=script_name):
                result = self._run(script_name)
                self.assertNotEqual(result.returncode, 0)
                self.assertIsNone(
                    self._ssh_argv(),
                    "ssh must not be invoked with no destination",
                )

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