path: root/app/tests/test_backends_common.py

"""Tests for backends.common subprocess/git helpers.

The streaming mode of ``run_console_subprocess`` (used by the in-TUI task
view) is exercised with a real child process: output lines are captured and
forwarded, cancellation kills the child and returns 130, an on_cancel hook
runs first, and the no-output stall watchdog kills a wedged child and
returns 124.
"""

import sys
import threading
import unittest
from unittest import mock

from backends import common


class RunConsoleSubprocessStreamingTests(unittest.TestCase):
    def test_streaming_emits_merged_lines(self):
        lines = []
        rc = common.run_console_subprocess(
            [sys.executable, "-c",
             "import sys; sys.stdout.write('hello\\nworld\\n'); "
             "sys.stderr.write('oops\\n')"],
            emit=lines.append)
        self.assertEqual(rc, 0)
        # stdout/stderr are merged, in arrival order.
        self.assertEqual(sorted(lines), ["hello", "oops", "world"])

    def test_streaming_returns_the_exit_code(self):
        rc = common.run_console_subprocess(
            [sys.executable, "-c", "import sys; sys.exit(3)"],
            emit=lambda line: None)
        self.assertEqual(rc, 3)

    def test_cancel_kills_the_process_and_returns_130(self):
        cancel = threading.Event()
        cancel.set()
        rc = common.run_console_subprocess(
            [sys.executable, "-c", "import time; time.sleep(60)"],
            emit=lambda line: None, cancel=cancel)
        self.assertEqual(rc, 130)

    def test_on_cancel_hook_runs_before_kill(self):
        cancel = threading.Event()
        cancel.set()
        touched = []
        common.run_console_subprocess(
            [sys.executable, "-c", "import time; time.sleep(60)"],
            emit=lambda line: None, cancel=cancel,
            on_cancel=lambda: touched.append(True))
        self.assertEqual(touched, [True])

    def test_carriage_return_progress_streams_incrementally(self):
        # tqdm/HuggingFace-style \r-only progress: a readline-based reader
        # blocked until the next \n, so the updates arrived in one burst
        # (or the stall watchdog fired first). Each \r segment must be
        # emitted as its own line.
        lines = []
        rc = common.run_console_subprocess(
            [sys.executable, "-c",
             "import sys, time\n"
             "for i in range(4):\n"
             "    sys.stdout.write(f'pct {i}\\r'); sys.stdout.flush()\n"
             "    time.sleep(0.2)\n"
             "sys.stdout.write('done\\n'); sys.stdout.flush()\n"],
            emit=lines.append, stall_timeout=1.0)
        self.assertEqual(rc, 0)
        self.assertEqual(lines, [f"pct {i}" for i in range(4)] + ["done"])

    def test_url_with_port_preserves_userinfo_and_ipv6(self):
        self.assertEqual(
            common.url_with_port("http://user:pass@host:8000", 8080),
            "http://user:pass@host:8080")
        self.assertEqual(
            common.url_with_port("http://[::1]:8000", 8080),
            "http://[::1]:8080")
        self.assertEqual(
            common.url_with_port("http://host:8000/path", 8080),
            "http://host:8080/path")


class RunConsoleSubprocessStallTests(unittest.TestCase):
    """The no-output watchdog: a silent child is killed and reported 124."""

    def test_stall_kills_a_silent_child_and_returns_124(self):
        lines = []
        rc = common.run_console_subprocess(
            [sys.executable, "-c",
             "import sys, time; print('start', flush=True); "
             "time.sleep(60)"],
            emit=lines.append, stall_timeout=0.5)
        self.assertEqual(rc, 124)
        self.assertEqual(lines[0], "start")
        # The stall is announced to the view before the kill.
        self.assertTrue(any("[ERROR]" in line and "No output" in line
                            for line in lines), lines)

    def test_no_stall_while_output_keeps_flowing(self):
        lines = []
        rc = common.run_console_subprocess(
            [sys.executable, "-c",
             "import sys, time\n"
             "for _ in range(6):\n"
             "    print('tick', flush=True)\n"
             "    time.sleep(0.2)\n"],
            emit=lines.append, stall_timeout=1.0)
        self.assertEqual(rc, 0)
        self.assertEqual(lines, ["tick"] * 6)

    def test_console_path_has_no_watchdog(self):
        # Without emit the child inherits the terminal; stall_timeout is
        # a no-op there (the caller sees raw output and can Ctrl-C).
        rc = common.run_console_subprocess(
            [sys.executable, "-c", "print('hi')"], stall_timeout=0.001)
        self.assertEqual(rc, 0)


class RunConsoleSubprocessQuietTimeoutTests(unittest.TestCase):
    """A timed-out quiet probe returns a failed result, never raises."""

    def test_timeout_returns_failed_result(self):
        proc = common.run_console_subprocess_quiet(
            [sys.executable, "-c", "import time; time.sleep(30)"],
            timeout=0.5)
        self.assertIsNotNone(proc)
        self.assertEqual(proc.returncode, -1)

    def test_untimed_probe_still_reports_the_exit_code(self):
        proc = common.run_console_subprocess_quiet(
            [sys.executable, "-c", "import sys; sys.exit(5)"])
        self.assertEqual(proc.returncode, 5)


class GitCloneTests(unittest.TestCase):
    def test_git_clone_console_passes_through(self):
        with mock.patch.object(common, "run_console_subprocess",
                                        return_value=0) as run:
            self.assertEqual(common.git_clone("url", common.Path("/t")), 0)
        # Console mode: no --progress flag, plain git clone.
        self.assertEqual(run.call_args[0][0],
                         ["git", "clone", "url", "/t"])

    def test_git_clone_streaming_adds_progress(self):
        def emit(line):
            pass
        with mock.patch.object(common, "run_console_subprocess",
                                        return_value=0) as run:
            self.assertEqual(common.git_clone("url", common.Path("/t"),
                                              emit=emit), 0)
        argv = run.call_args[0][0]
        self.assertEqual(argv[:3], ["git", "clone", "--progress"])
        self.assertIn("url", argv)
        self.assertEqual(run.call_args[1]["emit"], emit)


class GitUpdateTests(unittest.TestCase):
    """git_update: fetch, then hard reset to origin's default branch."""

    def _patched(self, fetch_rc=0, branch="main"):
        """Patch run_console_subprocess (fetch/reset) and the branch probe."""
        run = mock.patch.object(common, "run_console_subprocess",
                                return_value=fetch_rc).start()
        mock.patch.object(common, "origin_default_branch",
                          return_value=branch).start()
        return run

    def tearDown(self):
        mock.patch.stopall()

    def test_fetch_then_hard_reset_to_origin_head(self):
        run = self._patched()
        self.assertEqual(common.git_update(common.Path("/co")), 0)
        self.assertEqual(
            run.call_args_list[0][0][0],
            ["git", "-C", "/co", "fetch", "origin"])
        self.assertEqual(
            run.call_args_list[1][0][0],
            ["git", "-C", "/co", "reset", "--hard", "origin/main"])

    def test_streaming_adds_progress_and_passes_emit(self):
        emit = lambda line: None  # noqa: E731
        run = self._patched()
        self.assertEqual(common.git_update(common.Path("/co"), emit=emit), 0)
        self.assertEqual(
            run.call_args_list[0][0][0],
            ["git", "-C", "/co", "fetch", "--progress", "origin"])
        self.assertEqual(run.call_args_list[0][1]["emit"], emit)
        self.assertEqual(run.call_args_list[1][1]["emit"], emit)

    def test_fetch_failure_short_circuits_the_reset(self):
        run = self._patched(fetch_rc=128)
        self.assertEqual(common.git_update(common.Path("/co")), 128)
        self.assertEqual(run.call_count, 1)

    def test_reset_uses_the_remote_default_branch(self):
        run = self._patched(branch="trunk")
        self.assertEqual(common.git_update(common.Path("/co")), 0)
        self.assertEqual(
            run.call_args_list[1][0][0],
            ["git", "-C", "/co", "reset", "--hard", "origin/trunk"])


class OriginDefaultBranchTests(unittest.TestCase):
    def test_symbolic_ref_name_is_returned(self):
        proc = mock.Mock(returncode=0,
                         stdout=b"refs/remotes/origin/master\n")
        with mock.patch.object(common, "run_console_subprocess_quiet",
                               return_value=proc):
            self.assertEqual(common.origin_default_branch(
                common.Path("/co")), "master")

    def test_missing_ref_falls_back_to_main(self):
        proc = mock.Mock(returncode=128, stdout=b"")
        with mock.patch.object(common, "run_console_subprocess_quiet",
                               return_value=proc):
            self.assertEqual(common.origin_default_branch(
                common.Path("/co")), "main")

    def test_unstartable_probe_falls_back_to_main(self):
        with mock.patch.object(common, "run_console_subprocess_quiet",
                               return_value=None):
            self.assertEqual(common.origin_default_branch(
                common.Path("/co")), "main")


class GitHeadTests(unittest.TestCase):
    def test_head_sha_is_returned(self):
        proc = mock.Mock(returncode=0, stdout=b"abc123\n")
        with mock.patch.object(common, "run_console_subprocess_quiet",
                               return_value=proc) as run:
            self.assertEqual(common.git_head(common.Path("/co")), "abc123")
        self.assertEqual(run.call_args[0][0],
                         ["git", "-C", "/co", "rev-parse", "HEAD"])

    def test_not_a_repo_yields_none(self):
        proc = mock.Mock(returncode=128, stdout=b"")
        with mock.patch.object(common, "run_console_subprocess_quiet",
                               return_value=proc):
            self.assertIsNone(common.git_head(common.Path("/co")))

    def test_unstartable_probe_yields_none(self):
        with mock.patch.object(common, "run_console_subprocess_quiet",
                               return_value=None):
            self.assertIsNone(common.git_head(common.Path("/co")))


class GitCommitTimeTests(unittest.TestCase):
    def test_committer_time_is_parsed(self):
        proc = mock.Mock(returncode=0, stdout=b"1756300000\n")
        with mock.patch.object(common, "run_console_subprocess_quiet",
                               return_value=proc) as run:
            self.assertEqual(common.git_commit_time(common.Path("/co")),
                             1756300000)
        self.assertEqual(run.call_args[0][0],
                         ["git", "-C", "/co", "show", "-s",
                          "--format=%ct", "HEAD"])

    def test_not_a_repo_yields_none(self):
        proc = mock.Mock(returncode=128, stdout=b"")
        with mock.patch.object(common, "run_console_subprocess_quiet",
                               return_value=proc):
            self.assertIsNone(common.git_commit_time(common.Path("/co")))

    def test_unparsable_output_yields_none(self):
        proc = mock.Mock(returncode=0, stdout=b"not-a-number\n")
        with mock.patch.object(common, "run_console_subprocess_quiet",
                               return_value=proc):
            self.assertIsNone(common.git_commit_time(common.Path("/co")))

    def test_unstartable_probe_yields_none(self):
        with mock.patch.object(common, "run_console_subprocess_quiet",
                               return_value=None):
            self.assertIsNone(common.git_commit_time(common.Path("/co")))


class ParseRequestOptionsTests(unittest.TestCase):
    """parse_request_options: the shared --option / TUI-field parser."""

    def test_single_item(self):
        self.assertEqual(common.parse_request_options("speed=1.1"),
                         {"speed": "1.1"})

    def test_comma_and_whitespace_separators_mix(self):
        self.assertEqual(
            common.parse_request_options("emotion=neutral, speed=1.1"),
            {"emotion": "neutral", "speed": "1.1"})
        self.assertEqual(
            common.parse_request_options("a=1 b=2\tc=3"),
            {"a": "1", "b": "2", "c": "3"})

    def test_keys_are_stripped_and_blank_text_is_empty(self):
        self.assertEqual(common.parse_request_options("  "), {})
        self.assertEqual(common.parse_request_options(""), {})
        # Tokens cannot contain whitespace (items split on it), so a lone
        # "=" with a blank key is the malformed case, caught below.
        self.assertEqual(common.parse_request_options("speed=1"),
                         {"speed": "1"})

    def test_value_is_kept_verbatim(self):
        self.assertEqual(
            common.parse_request_options("url=http://x:8080/path?a=1"),
            {"url": "http://x:8080/path?a=1"})

    def test_later_duplicates_override_earlier_ones(self):
        self.assertEqual(common.parse_request_options("a=1,a=2"),
                         {"a": "2"})

    def test_item_without_equals_is_rejected(self):
        with self.assertRaises(ValueError):
            common.parse_request_options("emotion=neutral nonsense")

    def test_item_with_a_blank_key_is_rejected(self):
        with self.assertRaises(ValueError):
            common.parse_request_options("=value")


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