CODEWITHSUNDEEP
pythonfilemultiprocessing
Tristen
Tristen

Reputation: 382

How can I use a file as stdin in Python?

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.

Upvotes: 0

Views: 609

Answers (1)

Barmar
Barmar

Reputation: 781068

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()

Upvotes: 1

Related Questions