mirror of
https://github.com/xonsh/xonsh.git
synced 2025-03-04 08:24:40 +01:00
c5cb7044b5
### Motivation
* To have an ability to manage the output format added ``$XONSH_SUBPROC_OUTPUT_FORMAT`` to switch the way to return the output lines. Default ``stream_lines`` to return text. Alternative ``list_lines`` to return the list of lines. Also supported custom lambda function.
* Additionally the [proposal to change default behavior](https://github.com/xonsh/xonsh/pull/5377#discussion_r1587627131) for a single line case.
* Closes #3924 as soft solution.
### Before
```xsh
mkdir -p /tmp/tst && cd /tmp/tst && touch 1 2 3
$(ls)
# '1\n2\n3\n'
id $(whoami)
# id: ‘pc\n’: no such user: Invalid argument
du $(ls)
# du: cannot access '1'$'\n''2'$'\n''3'$'\n': No such file or directory
ls $(fzf)
# ls: cannot access 'FUNDING.yml'$'\n': No such file or directory
```
### After
```xsh
mkdir -p /tmp/tst && cd /tmp/tst && touch 1 2 3
$XONSH_SUBPROC_OUTPUT_FORMAT = 'list_lines'
$(ls)
# ['1', '2', '3']
[f for f in $(ls)]
# ['1', '2', '3']
id $(whoami)
# uid=501(user) gid=20(staff)
du $(ls)
# 0 1
# 0 2
# 0 3
ls $(fzf)
# FUNDING.yml
# etc
mkdir -p /tmp/@($(whoami))/dir
cat /etc/passwd | grep $(whoami)
```
### Notes
* It will be good to improve parser for cases like `mkdir -p /tmp/$(whoami)/dir`. PR is welcome!
* I named the default mode as `stream_lines` (instead of just `stream` or `raw`) because in fact we transform raw output into stream of lines and possibly reduce the length of output ([replacing `\r\n` to `\n`](c3a12b2a9c/xonsh/procs/pipelines.py (L380-L383))). May be some day we need to add raw "stream" output format.
* Now anybody can implement bash `IFS` behavior in [bashisms](https://github.com/xonsh/xontrib-bashisms).
---------
Co-authored-by: a <1@1.1>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
127 lines
3.7 KiB
Python
127 lines
3.7 KiB
Python
"""
|
|
Tests for command pipelines.
|
|
"""
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from xonsh.platform import ON_WINDOWS
|
|
from xonsh.procs.pipelines import CommandPipeline
|
|
from xonsh.pytest.tools import skip_if_on_unix, skip_if_on_windows
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def patched_events(monkeypatch, xonsh_events, xonsh_session):
|
|
from xonsh.jobs import get_tasks
|
|
|
|
get_tasks().clear()
|
|
# needed for ci tests
|
|
monkeypatch.setitem(
|
|
xonsh_session.env, "RAISE_SUBPROC_ERROR", False
|
|
) # for the failing `grep` commands
|
|
monkeypatch.setitem(
|
|
xonsh_session.env, "XONSH_CAPTURE_ALWAYS", True
|
|
) # capture output of ![]
|
|
if ON_WINDOWS:
|
|
monkeypatch.setattr(
|
|
xonsh_session.commands_cache,
|
|
"aliases",
|
|
{
|
|
"echo": "cmd /c echo".split(),
|
|
"grep": "cmd /c findstr".split(),
|
|
},
|
|
raising=False,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"cmdline, stdout, stderr, raw_stdout",
|
|
(
|
|
("!(echo hi)", "hi", "", "hi\n"),
|
|
("!(echo hi o>e)", "", "hi\n", ""),
|
|
pytest.param(
|
|
"![echo hi]",
|
|
"hi",
|
|
"",
|
|
"hi\n",
|
|
marks=pytest.mark.xfail(
|
|
ON_WINDOWS,
|
|
reason="ConsoleParallelReader doesn't work without a real console",
|
|
),
|
|
),
|
|
pytest.param(
|
|
"![echo hi o>e]",
|
|
"",
|
|
"hi\n",
|
|
"",
|
|
marks=pytest.mark.xfail(
|
|
ON_WINDOWS, reason="stderr isn't captured in ![] on windows"
|
|
),
|
|
),
|
|
pytest.param(
|
|
r"!(echo 'hi\nho')", "hi\nho\n", "", "hi\nho\n", marks=skip_if_on_windows
|
|
), # won't work with cmd
|
|
# for some reason cmd's echo adds an extra space:
|
|
pytest.param(
|
|
r"!(cmd /c 'echo hi && echo ho')",
|
|
"hi \nho\n",
|
|
"",
|
|
"hi \nho\n",
|
|
marks=skip_if_on_unix,
|
|
),
|
|
("!(echo hi | grep h)", "hi", "", "hi\n"),
|
|
("!(echo hi | grep x)", "", "", ""),
|
|
),
|
|
)
|
|
def test_command_pipeline_capture(cmdline, stdout, stderr, raw_stdout, xonsh_execer):
|
|
pipeline: CommandPipeline = xonsh_execer.eval(cmdline)
|
|
assert pipeline.out == stdout
|
|
assert pipeline.err == (stderr or None)
|
|
assert pipeline.raw_out == raw_stdout.replace("\n", os.linesep).encode()
|
|
assert pipeline.raw_err == stderr.replace("\n", os.linesep).encode()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"cmdline, output",
|
|
(
|
|
("echo hi", "hi"),
|
|
("echo hi | grep h", "hi"),
|
|
("echo hi | grep x", ""),
|
|
pytest.param("echo -n hi", "hi", marks=skip_if_on_windows),
|
|
),
|
|
)
|
|
def test_simple_capture(cmdline, output, xonsh_execer):
|
|
assert xonsh_execer.eval(f"$({cmdline})") == output
|
|
|
|
|
|
def test_raw_substitution(xonsh_execer):
|
|
assert xonsh_execer.eval("$(echo @(b'bytes!'))") == "bytes!"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"cmdline, result",
|
|
(
|
|
("bool(!(echo 1))", True),
|
|
("bool(!(nocommand))", False),
|
|
("int(!(echo 1))", 0),
|
|
("int(!(nocommand))", 1),
|
|
("hash(!(echo 1))", 0),
|
|
("hash(!(nocommand))", 1),
|
|
("str(!(echo 1))", "1"),
|
|
("str(!(nocommand))", ""),
|
|
("!(echo 1) == 0", True),
|
|
("!(nocommand) == 1", True),
|
|
pytest.param("!(echo -n str) == 'str'", True, marks=skip_if_on_windows),
|
|
("!(nocommand) == ''", True),
|
|
),
|
|
)
|
|
def test_casting(cmdline, result, xonsh_execer):
|
|
assert xonsh_execer.eval(f"{cmdline}") == result
|
|
|
|
|
|
@skip_if_on_windows
|
|
def test_background_pgid(xonsh_session, monkeypatch):
|
|
monkeypatch.setitem(xonsh_session.env, "XONSH_INTERACTIVE", True)
|
|
pipeline = xonsh_session.execer.eval("![echo hi &]")
|
|
assert pipeline.term_pgid is not None
|