tests/tools: extend vm.run_service() mockup

Handle stdout/stderr arguments and provide file-like objects when
requested.
This commit is contained in:
Marek Marczykowski-Górecki 2018-12-07 04:20:46 +01:00
parent 28067f70ac
commit 1d877742ae
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724

View File

@ -17,6 +17,7 @@
# #
# You should have received a copy of the GNU Lesser General Public License along # You should have received a copy of the GNU Lesser General Public License along
# with this program; if not, see <http://www.gnu.org/licenses/>. # with this program; if not, see <http://www.gnu.org/licenses/>.
import subprocess
import traceback import traceback
import unittest import unittest
@ -61,8 +62,14 @@ class TestProcess(object):
lambda: self.input_callback(self.stdin.getvalue())) lambda: self.input_callback(self.stdin.getvalue()))
else: else:
self.stdin.close = lambda: None self.stdin.close = lambda: None
self.stdout = stdout if stdout == subprocess.PIPE:
self.stderr = stderr self.stdout = io.BytesIO()
else:
self.stdout = stdout
if stderr == subprocess.PIPE:
self.stderr = io.BytesIO()
else:
self.stderr = stderr
self.returncode = 0 self.returncode = 0
def communicate(self, input=None): def communicate(self, input=None):
@ -146,7 +153,10 @@ class QubesTest(qubesadmin.app.QubesBase):
def run_service(self, dest, service, **kwargs): def run_service(self, dest, service, **kwargs):
self.service_calls.append((dest, service, kwargs)) self.service_calls.append((dest, service, kwargs))
return TestProcess(lambda input: self.service_calls.append((dest, return TestProcess(lambda input: self.service_calls.append((dest,
service, input))) service, input)),
stdout=kwargs.get('stdout', None),
stderr=kwargs.get('stderr', None),
)
class QubesTestCase(unittest.TestCase): class QubesTestCase(unittest.TestCase):