How can I use a file as stdin in Python?

Question

I'm writing unit-tests for an application that uses Python's built-in cmd.Cmd class. I'm writing test-cases to test the shell program, which listens for user input on sys.stdin. In the constructor arguments for Cmd, there is an stdin parameter.

I have a Shell class that inherits from Cmd:

class Shell(cmd.Cmd):
    intro = "shell"
    prompt = "(shell) "

    def __init__(self, incoming_q, outgoing_q, completekey='tab', stdin=None, stdout=None):
        super().__init__(completekey=completekey, stdin=stdin, stdout=stdout)
        self.incoming_q = incoming_q
        self.outgoing_q = outgoing_q

    def parse(self, args):
        cm, args = args.split("+")
        ret = {
            "command": cm,
            "args": [],
            "flags": []
        }
        for arg in tuple(args.split()):
            if arg[0] == "-":
                ret["flags"].append(arg.strip("-"))
            else:
                ret["args"].append(arg)
        return ret

    def do_command(self, args):
        args = self.parse("command+" + args)
        self.outgoing_q.put(args)
        try:
            res = self.incoming_q.get(timeout=100)
            print(res)
        except Exception:
            print("Command timed out")

I want to create a Cmd instance and run the cmdloop in a separate process in the test setup.

class TestShellMethods(unittest.TestCase):

    def setUp(self):
        self.incoming_q = Queue()
        self.outgoing_q = Queue()
        # What I want to do is something like this
        self.stdin = open("test.txt", "w")
        self.shell = Shell(self.incoming_q, self.outgoing_q, stdin=open("test.txt", "r"))
        self.shell.use_rawinput = 0
        self.shell_p = Process(target=self.shell.cmdloop)
        self.shell_p.start()

    def test_command(self):
        self.stdin.write("command\r\n")
        while self.outgoing_q.empty():
            pass
        res = self.outgoing_q.get()
        self.incoming_q.put("RESPONSE RECEIVED")

    def tearDown(self):
        self.shell_p.terminate()

The built-in Cmd does the following to read from stdin when it is provided (using sys.stdin by default):

line = self.stdin.readline()
if not len(line):
    line = 'EOF'
else:
    line = line.rstrip('\r\n')

Since I am running the loop in a separate process, I'm trying to figure out the best way to implement this in Python. I could subclass Queue, and create a readline method for it and use that as stdin.

from multiprocessing import Queue

class FileQueue(Queue):

    def readline(self):
        if self.empty():
            return ""
        else:
            return self.get()

Is there a way to do this without resorting to trickery that involves taking advantage of duck-typing to make the program think that a Queue is a file object? Considering how cmd.Cmd has stdin as a parameter, I'm guessing that there is an intended way to do this, but the documentation does not have any example usage of passing in stdin.

You need self.stdin.flush() after you write to it so that Shell will see what you've written. — Barmar
– Barmar, Commented Mar 11, 2021 at 0:08
This probably isn't going to work, though. Cmd probably stops when it gets EOF on its stdin, which will happen if it reads faster than you write to the file. — Barmar
– Barmar, Commented Mar 11, 2021 at 0:11
@Barmar That is what's occurring, it's grabbing the EOF and trying to run an EOF command, since the line gets replaced with "EOF" if the readline return is empty. — Tristen
– Tristen, Commented Mar 11, 2021 at 0:16
What you need is a pipe. Pass the read end to Shell, and write to the write end in your test code. — Barmar
– Barmar, Commented Mar 11, 2021 at 0:26

Barmar · Accepted Answer · 2021-03-11 00:31:20Z

Use os.pipe().

Anything you write to the write end of the pipe will be read from the read end. Shell won't read EOF until your test code calls self.stdin.close().

Writing to a pipe is buffered, so you also need to flush after writing to it.

class TestShellMethods(unittest.TestCase):

    def setUp(self):
        self.incoming_q = Queue()
        self.outgoing_q = Queue()
        pipe = os.pipe
        self.stdin = pipe[1]
        self.shell = Shell(self.incoming_q, self.outgoing_q, stdin=pipe[0])
        self.shell.use_rawinput = 0
        self.shell_p = Process(target=self.shell.cmdloop)
        self.shell_p.start()

    def test_command(self):
        self.stdin.write("command\r\n")
        self.stdin.flush()
        while self.outgoing_q.empty():
            pass
        res = self.outgoing_q.get()
        self.incoming_q.put("RESPONSE RECEIVED")

    def tearDown(self):
        self.shell_p.terminate()
        self.stdin.close()

Collectives™ on Stack Overflow

How can I use a file as stdin in Python?

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related