"""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]) 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()