1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
"""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, and an on_cancel
hook runs first.
"""
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 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 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()
|
