pybash/test_pybash.py at main · jaykv/pybash · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
import pytest

from pybash.transformer import InvalidInterpolation
from pybash.transformer import transform as run_bash


def test_single_exec():
    assert run_bash('$ls -la') == 'subprocess.run(["ls","-la"])\n'
    assert run_bash('$python --version') == 'subprocess.run(["python","--version"])\n'
    assert run_bash('$ python --version') == 'subprocess.run(["python","--version"])\n'


def test_var_exec():
    assert run_bash('a = $echo "test 123"') == 'a = subprocess.check_output(["echo","test 123"])\n'
    assert run_bash('a = $ echo "test 123"') == 'a = subprocess.check_output(["echo","test 123"])\n'


def test_wrapped_exec():
    assert (
        run_bash('print(($ cat test.txt).decode("utf-8").strip())')
        == 'print((subprocess.check_output(["cat","test.txt"])).decode("utf-8").strip())\n'
    )


def test_inline_exec():
    assert run_bash('print(str($echo test 123))') == 'print(str(subprocess.check_output(["echo","test","123"])))\n'


def test_inline_dot_exec():
    assert (
        run_bash("print(($cat test.txt).decode('utf-8').strip())")
        == 'print((subprocess.check_output(["cat","test.txt"])).decode(\'utf-8\').strip())\n'
    )


def test_method_exec():
    src = '''def cp_test():
        $cp test.txt test_copy.txt
    '''
    assert (
        run_bash(src)
        == '''def cp_test():
        subprocess.run(["cp","test.txt","test_copy.txt"])
    '''
    )


def test_pipe_basic():
    assert (
        run_bash("$cat test.txt | sed 's/HELLO/HOWDY/g'")
        == 'cmd1 = subprocess.Popen(["cat","test.txt"], stdout=subprocess.PIPE); cmd2 = subprocess.run(["sed","s/HELLO/HOWDY/g"], stdin=cmd1.stdout)\n'
    )
    assert (
        run_bash("$cat test.txt > test2.txt")
        == 'fout = open("test2.txt", "wb"); cmd1 = subprocess.run(["cat","test.txt"], stdout=fout)\n'
    )


def test_pipe_redirect():
    assert (
        run_bash("$cat test.txt | sed 's/HELLO/HOWDY/g' > test2.txt")
        == 'cmd1 = subprocess.Popen(["cat","test.txt"], stdout=subprocess.PIPE);fout = open("test2.txt", "wb"); cmd1 = subprocess.run(["sed","s/HELLO/HOWDY/g"], stdout=fout, stdin=cmd1.stdout)\n'
    )


def test_pipe_pipe_pipe():
    assert (
        run_bash("$cat test.txt | sed 's/HELLO/HOWDY/g' | sed 's/HOW/WHY/g' | sed 's/WHY/WHEN/g'")
        == 'cmd1 = subprocess.Popen(["cat","test.txt"], stdout=subprocess.PIPE);cmd1 = subprocess.Popen(["sed","s/HELLO/HOWDY/g"], stdout=subprocess.PIPE, stdin=cmd1.stdout);cmd1 = subprocess.Popen(["sed","s/HOW/WHY/g"], stdout=subprocess.PIPE, stdin=cmd1.stdout); cmd2 = subprocess.run(["sed","s/WHY/WHEN/g"], stdin=cmd1.stdout)\n'
    )


def test_pipe_chained_redirect():
    assert (
        run_bash("$cat test.txt | sed 's/HELLO/HOWDY\\n/g' > test1.txt >> test2.txt > test3.txt")
        == 'cmd1 = subprocess.Popen(["cat","test.txt"], stdout=subprocess.PIPE);fout = open("test1.txt", "wb"); cmd1 = subprocess.run(["sed","s/HELLO/HOWDY\\n/g"], stdout=fout, stdin=cmd1.stdout);fout7 = open("test2.txt", "ab"); cmd1 = subprocess.run(["cat","test1.txt"], stdout=fout7);fout9 = open("test3.txt", "wb"); cmd1 = subprocess.run(["cat","test2.txt"], stdout=fout9)\n'
    )


def test_input_redirect():
    assert run_bash("$sort < test.txt") == 'fout = open("test.txt", "r"); cmd1 = subprocess.run(["sort"], stdin=fout)\n'
    assert (
        run_bash("$sort < test.txt | sed 's/SORT/WHAT/g'")
        == 'fout = open("test.txt", "r"); cmd1 = subprocess.Popen(["sort"], stdin=fout, stdout=subprocess.PIPE);cmd2 = subprocess.run(["sed","s/SORT/WHAT/g"], stdin=cmd1.stdout)\n'
    )
    assert (
        run_bash("$sort < test.txt | sed 's/SORT/WHAT/g' | sed 's/WHAT/WHY/g'")
        == 'fout = open("test.txt", "r"); cmd1 = subprocess.Popen(["sort"], stdin=fout, stdout=subprocess.PIPE);cmd1 = subprocess.Popen(["sed","s/SORT/WHAT/g"], stdout=subprocess.PIPE, stdin=cmd1.stdout); cmd2 = subprocess.run(["sed","s/WHAT/WHY/g"], stdin=cmd1.stdout)\n'
    )
    assert (
        run_bash("$sort < test.txt | sed 's/SORT/WHAT/g' | sed 's/WHAT/WHY/g' > iredirect_end.txt")
        == 'fout = open("test.txt", "r"); cmd1 = subprocess.Popen(["sort"], stdin=fout, stdout=subprocess.PIPE);cmd1 = subprocess.Popen(["sed","s/SORT/WHAT/g"], stdout=subprocess.PIPE, stdin=cmd1.stdout);fout = open("iredirect_end.txt", "wb"); cmd1 = subprocess.run(["sed","s/WHAT/WHY/g"], stdout=fout, stdin=cmd1.stdout)\n'
    )
    assert (
        run_bash("$sort < test.txt > test_wb_redirect.txt")
        == 'fout = open("test.txt", "r"); cmd1 = subprocess.Popen(["sort"], stdin=fout, stdout=subprocess.PIPE);fout = open("test_wb_redirect.txt", "wb"); fout.write(cmd1.stdout.read());\n'
    )
    assert (
        run_bash("$sort < test.txt >> test_ab_redirect.txt")
        == 'fout = open("test.txt", "r"); cmd1 = subprocess.Popen(["sort"], stdin=fout, stdout=subprocess.PIPE);fout = open("test_ab_redirect.txt", "ab"); fout.write(cmd1.stdout.read());\n'
    )


def test_shell_commands():
    assert run_bash(">ls .github/*") == 'subprocess.run("ls .github/*", shell=True)\n'


def test_direct_interpolate():
    assert run_bash("$git {{command}} {{option}}") == 'subprocess.run(["git",command,option])\n'
    assert run_bash("$git {{command}} {{process(option)}}") == 'subprocess.run(["git",command,process(option)])\n'
    assert (
        run_bash("$k get pods --show-{{display_type}}=true")
        == 'subprocess.run(["k","get","pods","--show-" + display_type + "=true"])\n'
    )


def test_fstring_interpolate():
    assert (
        run_bash("$kubectl get pods f{\"--\" + \"-\".join(['show', 'labels'])} -n f{ namespace  }")
        == 'subprocess.run(["kubectl","get","pods",f"""{"--" + "-".join([\'show\', \'labels\'])}""","-n",f"""{ namespace  }"""])\n'
    )
    assert run_bash("$git f{options['h']}") == 'subprocess.run(["git",f"""{options[\'h\']}"""])\n'
    assert run_bash("$f{HOME_DIR}/bin/python -sE") == 'subprocess.run([f"""{HOME_DIR}""" + "/bin/python","-sE"])\n'


def test_interpolate_combo():
    assert run_bash("$echo 'f{PODS} 123' {{ARGS}}") == 'subprocess.run(["echo",f"""{PODS}""" + " 123",ARGS])\n'
    assert run_bash("$git {{COMMAND}} f{ARGS} -v") == 'subprocess.run(["git",COMMAND,f"""{ARGS}""","-v"])\n'


def test_invalid_interpolate():
    with pytest.raises(InvalidInterpolation):
        assert run_bash("$git {{command}} {{ option }}")
        assert run_bash("$git {{command}} {{'\"'.join(option)}}")
        assert run_bash("$git {{command}} {{a['key']}}")


def test_no_parse():
    assert run_bash('if 5 > 4:') == 'if 5 > 4:'
    assert run_bash('if (pred1 and pred2) > 0:') == 'if (pred1 and pred2) > 0:'
    assert run_bash('print("$(echo hi)")') == 'print("$(echo hi)")'
    assert run_bash('a =$(echo "test 123")') == 'a =$(echo "test 123")'