Python and Pytest calling a shell command containing tail fails

Question

Environment: platform qnx -- Python 2.7.12, pytest-4.6.9, py-1.8.1, pluggy-0.13.1 plugins: json-report-1.2.1, shell-0.2.3

Note: I know Python2.7 is old and unsupported, but there's no other version available for QNX at the moment.

Problem:

I'm running a tests that should kill a service when a certain keyword appears in its log. For this, I need it to be ran in the background. For this, I'm using the following shell command:

def test_kill_process():
    expected_output="XXXXXXXXX"
    expected_rc=0
    check_kill_process(expected_output, expected_rc)

import os
def check_kill_process(expected_output, expected_rc):
    test_log = File(r"/path/to/log")
    erase_log_entry = "Action"
    service=MyService()
    service.start()
    sleep(2)
    kill_command = "tail -f " + test_log.file_path + " | grep --line-buffered " + erase_log_entry + \
            " | while read ; do kill " + service.pid + " ; done &"
    os.popen(kill_command)
    service.action()
    f = open(test_log.file_path, "r")
    output = f.read()
    assert re.search(expected_output, output)

========================================================================

Without Pytest or even Python, works like a charm.

If I try using subprocess module to run the command, the test freezes indefinitely. If I try to use os.popen or os.system, the command ends in error:

tail:  read failed in '/path/to/logfile' (Invalid argument)

Moreover, if I try the same thing, with only a "cat" I get this:

--stdout--: Broken pipe

Thanks in advance if anyone has any ideea!

_If I try using subprocess_ - `subprocess` runs with an empty environment, so neither `LOGFILE` nor `KEYWORD` will be defined. Use the `env` arg to pass the environment. — hoefling, Jun 29 '20 at 12:11
Sorry, I wrote it not that good here. I pass the arguments directly, not as shell variables. I rewrote the question. Thanks for your prompt answer, but that's not it... — Razvan Badea, Jun 29 '20 at 12:19

score 1 · Answer 1 · 2020-06-29T13:40:19.270

1

From experience, subprocess and popen likes sequence of commands rather than a string. I think the technical reason is because of how different systems interpret a string path vs. arguments. Checkout the popen documetation for a more in-dept explanation.

But for example if I wanted to execute the command git commit -m "fixes a bug." I would have to split that string command into a list separated by space.

# command as a string
cmd = 'git commit -m "fixes a bug."'

# command as sequence of arguments. Consider using shlex.split() instead
cmd = cmd.split(' ')

# now call popen with the new formatted arguments
os.popen(cmd)

To make the your script even more robust. You can use a built in function shlex.split() to split the string into shell-like syntax for more complex cases.

Also it looks like you are also using | pipe in the command. You have to pass shell=True when calling popen for pipes to work.

edited Jun 29 '20 at 13:40

answered Jun 29 '20 at 13:16

Thanks for the input. Always used list for subprocess, but did not know about shlex. But still, this does not fix my problem. With subprocess.Popen(shlex.split(kill_command)) it just hangs. – Razvan Badea Jun 29 '20 at 13:34
Looks like you are also using `|` pipe in the command. You have to pass `shell=True` when calling popen for pipes to work. According to this post https://stackoverflow.com/questions/13332268/how-to-use-subprocess-command-with-pipes – Jun 29 '20 at 13:38
Tried, but same result. The test hangs indefinitely. – Razvan Badea Jun 29 '20 at 13:48

Z4-tier · Answer 2 · 2020-06-29T20:45:48.437

Instead of using subprocess (which is usually a symptom of unidiomatic python), why don't you just open(logfile, 'r') and use file.readline() in a loop to read the file? That is going to have the same effect as using tail without the overhead of a subprocess (the resulting code is also cleaner and easier to understand).

This should work pretty closely for what you're trying to do:

def wait_for_line(filename):
    f = open(filename, 'r')
    while 1:
        if line := f.readline():
            if re.match(some_regex, line):
                return True
        else:
            sleep(0.5)

Here is a complete working example:

import threading
from time import sleep
import re

FILENAME = 'some_file'
REGEX = r'^foo\s...!\s\d+$'
REGEX_MATCH = "foo bar! 123"

def perform_other_action(filename):
    f = open(filename, 'w')
    sleep(5) # just for effect
    f.write(REGEX_MATCH)


def wait_for_line(filename):
    f = open(filename, 'r')
    while 1:
        if line := f.readline():
            if re.match(REGEX, line):
                return True
        else:
            sleep(0.5) # throttle the read frequency if no new lines are found

with open(FILENAME, 'w') as f: # make sure file exists
    pass

t = threading.Thread(target=wait_for_line, args=(FILENAME,))
t.start() # returns immediately, wait_for_line runs in background
perform_other_action(FILENAME)
t.join()

The thing is I need this to run in the background so I can also issue an action to the service that will eventually match the condition in the shell line. — Razvan Badea, Jun 29 '20 at 13:08
@RazvanBadea Can you run that action before starting this loop? If that's not possible, how about creating a second thread to manage the action? — Z4-tier, Jun 29 '20 at 15:15
My first choice was threading. I ran into the same problem- hangs. I even made a small testscript that ran beautifully with python alone, but hanged with PyTest. I will try reversing the order though. — Razvan Badea, Jun 30 '20 at 05:39
Have you tried replacing the subprocess with the native `os.kill()` method? It might also be a good idea to place the `MyService()` setup in a test fixture. — Z4-tier, Jun 30 '20 at 18:24

Python and Pytest calling a shell command containing tail fails

2 Answers2