Add missing APIs to FakeProcess, making it match Popen (LP #1373224)

fixtures/_fixtures/popen.py

@@ -36,8 +36,23 @@ class FakeProcess(object):
         self._returncode = info.get('returncode', 0)
         self.returncode = None
 
-    def communicate(self):
+    @property
+    def args(self):
+        return self._args["args"]
+
+    def poll(self):
+        """Get the current value of FakeProcess.returncode.
+
+        The returncode is None before communicate() and/or wait() are called,
+        and it's set to the value provided by the 'info' dictionary otherwise
+        (or 0 in case 'info' doesn't specify a value).
+        """
+        return self.returncode
+
+    def communicate(self, input=None, timeout=None):
         self.returncode = self._returncode
+        if self.stdin and input:
+            self.stdin.write(input)
         if self.stdout:
             out = self.stdout.getvalue()
         else:
@@ -91,6 +106,11 @@ class FakePopen(Fixture):
 
             The default behaviour if no get_info is supplied is for the return
             process to have returncode of None, empty streams and a random pid.
+
+            After communicate() or wait() are called on the process object,
+            the returncode is set to whatever get_info returns (or 0 if
+            get_info is not supplied or doesn't return a dict with an explicit
+            'returncode' key).
         """
         super(FakePopen, self).__init__()
         self.get_info = get_info

fixtures/tests/_fixtures/test_popen.py

@@ -76,6 +76,7 @@ class TestFakePopen(testtools.TestCase, TestWithFixtures):
         fixture = self.useFixture(FakePopen())
         with subprocess.Popen(['ls -lh']) as proc:
             self.assertEqual(None, proc.returncode)
+            self.assertEqual(['ls -lh'], proc.args)
 
 
 class TestFakeProcess(testtools.TestCase):
@@ -95,10 +96,41 @@ class TestFakeProcess(testtools.TestCase):
         self.assertEqual((_b('foo'), ''), proc.communicate())
         self.assertEqual(0, proc.returncode)
 
+    def test_communicate_with_input(self):
+        proc = FakeProcess({}, {'stdout': BytesIO(_b('foo'))})
+        self.assertEqual((_b('foo'), ''), proc.communicate(input=_b("bar")))
+
+    def test_communicate_with_input_and_stdin(self):
+        stdin = BytesIO()
+        proc = FakeProcess({}, {'stdin': stdin})
+        proc.communicate(input=_b("hello"))
+        self.assertEqual(_b("hello"), stdin.getvalue())
+
+    def test_communicate_with_timeout(self):
+        proc = FakeProcess({}, {'stdout': BytesIO(_b('foo'))})
+        self.assertEqual((_b('foo'), ''), proc.communicate(timeout=10))
+
+    def test_args(self):
+        proc = FakeProcess({"args": ["ls", "-lh"]}, {})
+        proc.returncode = 45
+        self.assertEqual(45, proc.wait())
+        self.assertEqual(proc.args, ["ls", "-lh"])
+
     def test_kill(self):
         proc = FakeProcess({}, {})
         self.assertIs(None, proc.kill())
 
+    def test_poll(self):
+        proc = FakeProcess({}, {})
+        self.assertIs(None, proc.poll())
+        proc.communicate()
+        self.assertEqual(0, proc.poll())
+
+    def test_poll_with_returncode(self):
+        proc = FakeProcess({}, {})
+        proc.communicate()
+        self.assertEqual(0, proc.poll())
+
     def test_wait_with_timeout_and_endtime(self):
         proc = FakeProcess({}, {})
         self.assertEqual(0 , proc.wait(timeout=4, endtime=7))