Add more tests.

tests/test_shellous.py

@@ -15,6 +15,7 @@
 
 from shellous import Result, ResultError, sh
 from shellous.harvest import harvest_results
+from shellous.prompt import Prompt
 from shellous.runner import CANCELLED_EXIT_CODE
 
 # 4MB + 1: Much larger than necessary.
@@ -126,6 +127,20 @@ async def test_bulk(bulk_cmd):
  assert value == "462d6c497b393d2c9e1584a7b4636592da837ef66cf4ff871dc937f3fe309459"
 
 
+async def test_bulk_prompt(bulk_cmd):
+ "Test the Prompt class with bulk output."
+ cmd = bulk_cmd().encoding("latin1")
+
+ async with cmd.stdin(sh.CAPTURE).stdout(sh.CAPTURE) as run:
+ prompt = Prompt(run, ">>> ")
+ result = await prompt.receive(timeout=10.0)
+ assert len(result) == 4 * (1024 * 1024 + 1)
+ value = hashlib.sha256(result.encode("latin1")).hexdigest()
+ assert (
+ value == "462d6c497b393d2c9e1584a7b4636592da837ef66cf4ff871dc937f3fe309459"
+ )
+
+
 async def test_count(count_cmd):
  result = await count_cmd(5)
  assert result == "1\n2\n3\n4\n5\n"
@@ -535,6 +550,14 @@ async def test_redirect_stdin_unsupported_type(cat_cmd):
  await cat_cmd("abc").stdin(1 + 2j)
 
 
+async def test_redirect_sequence_stringio(echo_cmd, cat_cmd):
+ "Test reading/writing StringIO in sequence of commands."
+ buf = io.StringIO()
+ await echo_cmd("abc").stdout(buf)
+ result = await cat_cmd().stdin(buf)
+ assert result == "abc"
+
+
 async def test_broken_pipe():
  """Test broken pipe error for large data passed to stdin.