如何在Python中使用文件作为stdin ?

Question

我正在为一个使用Python内置cmd的应用程序编写单元测试。Cmd类。我正在编写测试用例来测试shell程序，该程序监听sys.stdin上的用户输入。在Cmd的构造函数参数中，有一个stdin参数。

我有一个Shell类继承自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")

我想创建一个Cmd实例并在测试设置的单独进程中运行cmdloop。

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("commandrn")
while self.outgoing_q.empty():
pass
res = self.outgoing_q.get()
self.incoming_q.put("RESPONSE RECEIVED")
def tearDown(self):
self.shell_p.terminate()

内置Cmd在提供stdin时执行以下操作从stdin中读取(使用sys。(默认):

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

因为我在一个单独的进程中运行循环，所以我试图找出在Python中实现它的最佳方法。我可以继承Queue，并为它创建一个readline方法，并将其用作stdin。

from multiprocessing import Queue
class FileQueue(Queue):
def readline(self):
if self.empty():
return ""
else:
return self.get()

是否有一种方法可以做到这一点，而不需要利用duck-typing来使程序认为Queue是一个文件对象?考虑到cmd。Cmd有stdin作为参数，我猜有一种预期的方式来做到这一点，但文档没有任何传入stdin的示例用法。

Answer 1

使用os.pipe()

您写入到管道写端的任何内容都将从读端读取。Shell不会读取EOF，直到您的测试代码调用self.stdin.close()。

对管道的写入是缓冲的，因此在写入后还需要刷新。

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