Commit 51cd4ae7 authored by jpic ∞'s avatar jpic ∞ 💾
Browse files

Add demo and command dump

parent cd2194f1
Pipeline #2218 passed with stages
in 30 seconds
......@@ -16,6 +16,13 @@ subprocesses at the same time. The purpose of this library is to:
This code was copy/pasted between projects and finally extracted on its own.
Demo
====
.. image:: https://yourlabs.io/oss/shlax/-/raw/master/demo.png
You will find the demo script in demo.py in this repository.
Usage
=====
......@@ -96,8 +103,9 @@ will be applied line by line:
Where is the rest?
==================
Shlax used to be the name of a much more ambitious poc-project that has been
extracted in two projects with clear boundaries, namely `sysplan
Shlax used to be the name of a much more ambitious poc-project, that you can
still find in the ``OLD`` branch of this repository. It has been extracted in
two projects with clear boundaries, namely `sysplan
<https://yourlabs.io/oss/sysplan>`_ and `podplan
<https://yourlabs.io/oss/podplan>`_ which are still in alpha state, but Shlax
as it is feature complete and stable.
<https://yourlabs.io/oss/podplan>`_ which are still in alpha state, although
Shlax as it is feature complete and stable.
demo.png

990 KB

import asyncio
from shlax import Subprocess
async def main():
colors = {
'^(.*).txt$': '{green}\\1.txt',
'^(.*).py$': '{bred}\\1.py',
}
await asyncio.gather(
Subprocess(
'for i in $(find .. | head); do echo $i; sleep .2; done',
regexps=colors,
prefix='parent',
).wait(),
Subprocess(
'for i in $(find . | head); do echo $i; sleep .3; done',
regexps=colors,
prefix='cwd',
).wait(),
)
asyncio.run(main())
import asyncio
import functools
import re
import shlex
import sys
from .colors import colors
......@@ -54,7 +55,6 @@ class Subprocess:
if len(args) == 1 and ' ' in args[0]:
args = ['sh', '-euc', args[0]]
self.cmd = ' '.join(args)
self.args = args
self.quiet = quiet if quiet is not None else False
self.prefix = prefix
......@@ -75,6 +75,18 @@ class Subprocess:
self.regexps[search] = replace
async def start(self, wait=True):
if not self.quiet:
self.output(
self.colors.bgray.encode()
+ b'+ '
+ shlex.join([
arg.replace('\n', '\\n')
for arg in self.args
]).encode()
+ self.colors.reset.encode(),
highlight=False
)
# Get a reference to the event loop as we plan to use
# low-level APIs.
loop = asyncio.get_running_loop()
......
......@@ -37,12 +37,10 @@ async def test_wait_unbound():
async def test_rc_1():
proc = await Proc(
'NON EXISTING COMMAND',
write=Mock(),
quiet=True,
).wait()
assert proc.rc != 0
proc.write.assert_called_once_with(
b'sh: line 1: NON: command not found\x1b[0m\n'
)
assert proc.err == 'sh: line 1: NON: command not found'
@pytest.mark.asyncio
......@@ -70,6 +68,16 @@ async def test_prefix():
).wait()
assert write.mock_calls == [
call(
Proc.prefix_colors[0].encode()
+ b'test_prefix '
+ Proc.colors.reset.encode()
+ b'| '
+ Proc.colors.bgray.encode()
+ b'+ sh -euc \'echo hi\''
+ Proc.colors.reset.encode()
+ b'\n'
),
call(
Proc.prefix_colors[0].encode()
+ b'test_prefix '
......@@ -78,6 +86,16 @@ async def test_prefix():
+ Proc.colors.reset.encode()
+ b'\n'
),
call(
Proc.prefix_colors[1].encode()
+ b'test_prefix_1 '
+ Proc.colors.reset.encode()
+ b'| '
+ Proc.colors.bgray.encode()
+ b'+ sh -euc \'echo hi\''
+ Proc.colors.reset.encode()
+ b'\n'
),
call(
Proc.prefix_colors[1].encode()
# padding has been added because of output1
......@@ -87,6 +105,17 @@ async def test_prefix():
+ Proc.colors.reset.encode()
+ b'\n'
),
call(
Proc.prefix_colors[0].encode()
# padding has been added because of output1
+ b' test_prefix '
+ Proc.colors.reset.encode()
+ b'| '
+ Proc.colors.bgray.encode()
+ b'+ sh -euc \'echo hi\''
+ Proc.colors.reset.encode()
+ b'\n'
),
call(
Proc.prefix_colors[0].encode()
# padding has been added because of output1
......@@ -108,6 +137,16 @@ async def test_prefix_multiline():
prefix='test_prefix',
).wait()
assert proc.write.mock_calls == [
call(
Proc.prefix_colors[0].encode()
+ b'test_prefix '
+ Proc.colors.reset.encode()
+ b'| '
+ Proc.colors.bgray.encode()
+ b'+ sh -euc \'echo -e "a\\nb"\''
+ Proc.colors.reset.encode()
+ b'\n'
),
call(
Proc.prefix_colors[0].encode()
+ b'test_prefix '
......@@ -140,7 +179,7 @@ async def test_highlight():
r'h([\w\d-]+)': 'h{cyan}\\1',
}
).wait()
proc.write.assert_called_once_with(b'h\x1b[38;5;51mi\x1b[0m\n')
proc.write.assert_called_with(b'h\x1b[38;5;51mi\x1b[0m\n')
@pytest.mark.asyncio
......@@ -155,4 +194,4 @@ async def test_highlight_if_not_colored():
r'h([\w\d-]+)': 'h{cyan}\\1',
}
).wait()
proc.write.assert_called_once_with(b'h\x1b[31mi\n')
proc.write.assert_called_with(b'h\x1b[31mi\n')
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment