stdlib/test/system/test_subprocess.f90 at master · edish-github/stdlib · 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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
module test_subprocess
    use testdrive, only : new_unittest, unittest_type, error_type, check, skip_test
    use stdlib_system, only: process_type, run, runasync, is_running, wait, update, elapsed, is_windows, kill

    implicit none

contains

    !> Collect all exported unit tests
    subroutine collect_suite(testsuite)
        !> Collection of tests
        type(unittest_type), allocatable, intent(out) :: testsuite(:)

        testsuite = [ &
            new_unittest('test_run_synchronous', test_run_synchronous), &
            new_unittest('test_run_asynchronous', test_run_asynchronous), &
            new_unittest('test_process_kill', test_process_kill), &
            new_unittest('test_process_state', test_process_state), &
            new_unittest('test_input_redirection', test_input_redirection) &
        ]
    end subroutine collect_suite

    !> Test running a synchronous process
    subroutine test_run_synchronous(error)
        type(error_type), allocatable, intent(out) :: error
        type(process_type) :: process
        character(len=*), parameter :: command = "echo Hello"

        process = run(command, want_stdout=.true.)
        call check(error, process%completed)
        if (allocated(error)) return

        call check(error, trim(process%stdout) == "Hello", "stdout=<"//trim(process%stdout)//">, expected <Hello>")
    end subroutine test_run_synchronous

    !> Test running an asynchronous process
    subroutine test_run_asynchronous(error)
        type(error_type), allocatable, intent(out) :: error
        type(process_type) :: process
        logical :: running

        ! The closest possible to a cross-platform command that waits
        if (is_windows()) then
            process = runasync("ping -n 2 127.0.0.1")
        else
            process = runasync("ping -c 2 127.0.0.1")
        endif
        ! Should not be immediately completed
        call check(error, .not. process%completed, "ping process should not complete immediately")
        if (allocated(error)) return

        running = is_running(process)
        call check(error, running, "ping process should still be running immediately after started")
        if (allocated(error)) return

        call wait(process)
        call check(error, process%completed, "process should be complete after `call wait`")
        if (allocated(error)) return

        call check(error, elapsed(process)>1.0e-4, "There should be a non-zero elapsed time")

    end subroutine test_run_asynchronous

    !> Test killing an asynchronous process
    subroutine test_process_kill(error)
        type(error_type), allocatable, intent(out) :: error
        type(process_type) :: process
        logical :: running, success

        ! Start a long-running process asynchronously
        if (is_windows()) then
            process = runasync("ping -n 10 127.0.0.1")
        else
            process = runasync("ping -c 10 127.0.0.1")
        endif

        ! Ensure the process starts running
        call check(error, .not. process%completed, "Process should not be completed immediately after starting")
        if (allocated(error)) return

        running = is_running(process)
        call check(error, running, "Process should be running immediately after starting")
        if (allocated(error)) return

        ! Kill the process
        call kill(process, success)
        call check(error, success, "Failed to kill the process")
        if (allocated(error)) return

        ! Verify the process is no longer running
        call check(error, .not. is_running(process), "Process should not be running after being killed")
        if (allocated(error)) return

        ! Ensure process state updates correctly after killing
        call check(error, process%completed, "Process should be marked as completed after being killed")
    end subroutine test_process_kill

    !> Test updating and checking process state
    subroutine test_process_state(error)
        type(error_type), allocatable, intent(out) :: error
        type(process_type) :: process
        character(len=*), parameter :: command = "echo Testing"

        process = run(command, want_stdout=.true., want_stderr=.true.)

        call update(process)
        call check(error, process%completed)
        if (allocated(error)) return

        call check(error, process%exit_code == 0, "Check zero exit code")
        if (allocated(error)) return

        call check(error, len_trim(process%stderr) == 0, "Check no stderr output")
        if (allocated(error)) return

        call check(error, trim(process%stdout) == "Testing", "stdout=<"//trim(process%stdout)//">, expected <Testing>")
        if (allocated(error)) return
    end subroutine test_process_state

    !> Test input redirection
    subroutine test_input_redirection(error)
        type(error_type), allocatable, intent(out) :: error
        type(process_type) :: process
        character(len=*), parameter :: input_string = "Hello Stdin"

        if (is_windows()) then
             ! findstr "^" echoes input lines.
             ! Note: We need complex quoting because of how arguments are parsed.
             ! Actually, sticking to something simpler if possible.
             ! "more" implies paging which might hang. "sort" is usually safe.
             process = run("sort", stdin=input_string, want_stdout=.true.)
        else
             process = run("cat", stdin=input_string, want_stdout=.true.)
        endif

        call check(error, process%completed, "Process did not complete")
        if (allocated(error)) return

        call check(error, process%exit_code == 0, "Process failed with non-zero exit code")
        if (allocated(error)) return

        ! Check if output matches input (sort of "Hello Stdin" is "Hello Stdin")
        call check(error, index(process%stdout, input_string) > 0, &
                   "Output <"//trim(process%stdout)//"> should contain <"//input_string//">")

    end subroutine test_input_redirection

end module test_subprocess

program tester
    use, intrinsic :: iso_fortran_env, only : error_unit
    use testdrive, only : run_testsuite, new_testsuite, testsuite_type
    use test_subprocess, only : collect_suite

    implicit none

    integer :: stat, is
    type(testsuite_type), allocatable :: testsuites(:)
    character(len=*), parameter :: fmt = '("#", *(1x, a))'

    stat = 0

    testsuites = [ &
        new_testsuite("subprocess", collect_suite) &
    ]

    do is = 1, size(testsuites)
        write(error_unit, fmt) "Testing:", testsuites(is)%name
        call run_testsuite(testsuites(is)%collect, error_unit, stat)
    end do

    if (stat > 0) then
        write(error_unit, '(i0, 1x, a)') stat, "test(s) failed!"
        error stop
    end if
end program