mirror of
https://github.com/xonsh/xonsh.git
synced 2025-03-04 16:34:47 +01:00
86e4f004e3
* Make job control task queue thread-local This allows callable aliases to maintain a separate task queue from the main thread. * Make job control dictionary thread-local Use XSH.all_jobs for the main thread and a separate dictionary for other threads. This allows callable aliases to keep track of their jobs separate from the main thread. * Implement use_main_jobs() context manager This allows threaded commands like jobs, disown, and bg to handle the main thread's job control * Run commands in the same process group if in a thread * Fix tests that use jobs and tasks * Add tests for subprocess call inside alias * Add news * Remove type declarations for _jobs_thread_local Fixes the mypy error: "Type cannot be declared in assignment to non-self attribute" Co-authored-by: Noorhteen Raja NJ <jnoortheen@gmail.com>
121 lines
3.5 KiB
Python
121 lines
3.5 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,
|
|
"aliases",
|
|
{
|
|
"echo": "cmd /c echo".split(),
|
|
"grep": "cmd /c findstr".split(),
|
|
},
|
|
raising=False,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"cmdline, stdout, stderr",
|
|
(
|
|
("!(echo hi)", "hi\n", ""),
|
|
("!(echo hi o>e)", "", "hi\n"),
|
|
pytest.param(
|
|
"![echo 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", "", 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", "", marks=skip_if_on_unix
|
|
),
|
|
("!(echo hi | grep h)", "hi\n", ""),
|
|
("!(echo hi | grep x)", "", ""),
|
|
),
|
|
)
|
|
def test_command_pipeline_capture(cmdline, stdout, stderr, xonsh_execer):
|
|
pipeline: CommandPipeline = xonsh_execer.eval(cmdline)
|
|
assert pipeline.out == stdout
|
|
assert pipeline.err == (stderr or None)
|
|
assert pipeline.raw_out == stdout.replace("\n", os.linesep).encode()
|
|
assert pipeline.raw_err == stderr.replace("\n", os.linesep).encode()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"cmdline, output",
|
|
(
|
|
("echo hi", "hi\n"),
|
|
("echo hi | grep h", "hi\n"),
|
|
("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!\n"
|
|
|
|
|
|
@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\n"),
|
|
("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
|