feat: add command tail mode

README.md

@@ -35,7 +35,7 @@ pyallel "black --color --check --diff ." "MYPY_FORCE_COLOR=1 mypy ." "ruff check
       re-work of how we print command output as we currently just print output once the command
       finishes)
 - [x] Add CI checks to run the tests and linters against Python versions > 3.8
-- [ ] Add command mode arguments to support things like only tailing the last 10 lines
+- [x] Add command mode arguments to support things like only tailing the last 10 lines
       of a command whilst it is running e.g. `"tail=10 :: pytest ."`
 - [ ] Add visual examples of `pyallel` in action
 - [ ] Add custom parsing of command output to support filtering for errors (like vim's

src/pyallel/parser.py

@@ -20,6 +20,20 @@ def __repr__(self) -> str:
         return msg
 
 
+COMMANDS_HELP = """list of quoted commands to run e.g "mypy ." "black ."
+
+can provide environment variables to each command like so:
+
+     "MYPY_FORCE_COLOR=1 mypy ."
+
+command modes:
+
+can also provide modes to commands to do extra things:
+
+    "tail=10 :: pytest ." <-- only output the last 10 lines, only works in --stream mode
+"""
+
+
 def create_parser() -> ArgumentParser:
     parser = ArgumentParser(
         prog="pyallel",
@@ -28,9 +42,7 @@ def create_parser() -> ArgumentParser:
     )
     parser.add_argument(
         "commands",
-        help='list of quoted commands to run e.g "mypy ." "black ."\n\n'
-        "can provide environment variables to each command like so:\n\n"
-        '     "MYPY_FORCE_COLOR=1 mypy ."',
+        help=COMMANDS_HELP,
         nargs="*",
     )
     parser.add_argument(

src/pyallel/process.py

@@ -173,6 +173,10 @@ def stream(self) -> bool:
                 process_output = process.read().decode()
                 if process_output:
                     self.output[process.id] = process_output
+                    if process.tail_mode.enabled:
+                        process_output = "\n".join(
+                            process_output.splitlines()[-process.tail_mode.lines :]
+                        )
                     output += indent(process_output)
                     output += "\n"
                     if i != len(self.processes):
@@ -274,9 +278,15 @@ def from_commands(
         )
 
 
+@dataclass
+class TailMode:
+    enabled: bool = False
+    lines: int = 0
+
+
 @dataclass
 class Process:
-    id: UUID
+    id: UUID = field(repr=False, compare=False)
     name: str
     args: list[str]
     env: dict[str, str] = field(default_factory=dict)
@@ -287,6 +297,7 @@ class Process:
     fd_name: Path | None = None
     fd_read: BinaryIO | None = None
     fd: int | None = None
+    tail_mode: TailMode = field(default_factory=TailMode)
 
     def run(self) -> None:
         self.start = time.perf_counter()
@@ -330,9 +341,16 @@ def return_code(self) -> int | None:
 
     @classmethod
     def from_command(cls, command: str) -> Process:
+        tail_mode = TailMode()
         env = os.environ.copy()
         if " :: " in command:
-            _, _args = command.split(" :: ")
+            modes, _args = command.split(" :: ")
+            if modes:
+                for mode in modes.split():
+                    name, value = mode.split("=", maxsplit=1)
+                    if name == "tail":
+                        tail_mode.enabled = True
+                        tail_mode.lines = int(value)
             args = _args.split()
         else:
             args = command.split()
@@ -349,4 +367,6 @@ def from_command(cls, command: str) -> Process:
             raise InvalidExecutableError(parsed_args[0])
 
         str_args = shlex.split(" ".join(parsed_args[1:]))
-        return cls(id=uuid4(), name=parsed_args[0], args=str_args, env=env)
+        return cls(
+            id=uuid4(), name=parsed_args[0], args=str_args, env=env, tail_mode=tail_mode
+        )

tests/test_main.py

@@ -48,6 +48,20 @@ def test_run_single_command_with_env(capsys: CaptureFixture[str]) -> None:
     ]
 
 
+def test_run_single_command_with_tail_mode(capsys: CaptureFixture[str]) -> None:
+    exit_code = main.run("tail=10 :: sleep 0.1")
+    captured = capsys.readouterr()
+    out = captured.out.splitlines(keepends=True)
+    assert exit_code == 0, prettify_error(captured.out)
+    assert out == [
+        "Running commands...\n",
+        "\n",
+        "[sleep] done ✓\n",
+        "\n",
+        "Success!\n",
+    ]
+
+
 def test_run_multiple_commands(capsys: CaptureFixture[str]) -> None:
     exit_code = main.run("sh -c 'sleep 0.1; echo \"first\"'", "echo 'hi'")
     captured = capsys.readouterr()

tests/test_process.py

@@ -1,11 +1,17 @@
+import os
 import time
+from uuid import uuid4
 
 import pytest
-from pyallel.process import Process
+from pyallel.process import Process, TailMode
 
 
-def test_process_from_command() -> None:
-    Process.from_command("sleep 0.1")
+def test_from_command() -> None:
+    expected_process = Process(
+        id=uuid4(), name="sleep", args=["0.1"], env=os.environ.copy()
+    )
+    process = Process.from_command("sleep 0.1")
+    assert process == expected_process
 
 
 @pytest.mark.parametrize(
@@ -15,8 +21,40 @@ def test_process_from_command() -> None:
         pytest.param("TEST_VAR=1 OTHER_VAR=2", id="Multiple env vars"),
     ),
 )
-def test_process_from_command_with_env(env: str) -> None:
-    Process.from_command(f"{env} sleep 0.1")
+def test_from_command_with_env(env: str) -> None:
+    env_dict: dict[str, str] = {}
+    for t in env.split():
+        key, value = t.split("=")
+        env_dict[key] = value
+    expected_process = Process(
+        id=uuid4(), name="sleep", args=["0.1"], env=os.environ.copy() | env_dict
+    )
+    process = Process.from_command(f"{env} sleep 0.1")
+    assert process == expected_process
+
+
+def test_from_command_with_tail_mode() -> None:
+    expected_process = Process(
+        id=uuid4(),
+        name="sleep",
+        args=["0.1"],
+        env=os.environ.copy(),
+        tail_mode=TailMode(enabled=True, lines=10),
+    )
+    process = Process.from_command("tail=10 :: sleep 0.1")
+    assert process == expected_process
+
+
+def test_from_command_with_modes_and_env() -> None:
+    expected_process = Process(
+        id=uuid4(),
+        name="sleep",
+        args=["0.1"],
+        env=os.environ.copy() | {"TEST_VAR": "1"},
+        tail_mode=TailMode(enabled=True, lines=10),
+    )
+    process = Process.from_command("tail=10 :: TEST_VAR=1 sleep 0.1")
+    assert process == expected_process
 
 
 def test_read() -> None: