ExternalVendorCode/tahoe-lafs
1
0
Fork 0
mirror of https://github.com/tahoe-lafs/tahoe-lafs.git synced 2025-01-31 16:36:20 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

move run_cli() up to test/common_util.py

Browse Source
This commit is contained in:
Brian Warner 2016-09-09 10:25:11 -07:00
parent 442468f599
commit 07e4c491f5
2 changed files with 19 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
src/allmydata/test/cli/common.py
View File

@ -1,9 +1,6 @@
from cStringIO import StringIO
from twisted.internet import threads # CLI tests use deferToThread
from ...util.assertutil import precondition
from ...util.encodingutil import unicode_to_argv
from ...scripts import runner
from ..common_util import ReallyEqualMixin
from ..common_util import ReallyEqualMixin, run_cli
def parse_options(basedir, command, args):
o = runner.Options()
@ -20,18 +17,3 @@ class CLITestMixin(ReallyEqualMixin):
client_dir = unicode_to_argv(self.get_clientdir(i=client_num))
nodeargs = [ "--node-directory", client_dir ]
return run_cli(verb, nodeargs=nodeargs, *args, **kwargs)
def run_cli(verb, *args, **kwargs):
precondition(not [True for arg in args if not isinstance(arg, str)],
"arguments to do_cli must be strs -- convert using unicode_to_argv", args=args)
nodeargs = kwargs.get("nodeargs", [])
argv = nodeargs + [verb] + list(args)
stdin = kwargs.get("stdin", "")
stdout, stderr = StringIO(), StringIO()
d = threads.deferToThread(runner.runner, argv, run_by_human=False,
stdin=StringIO(stdin),
stdout=stdout, stderr=stderr)
def _done(rc):
return rc, stdout.getvalue(), stderr.getvalue()
d.addCallback(_done)
return d

19
src/allmydata/test/common_util.py
View File

@ -1,13 +1,15 @@
import os, signal, sys, time
from random import randrange
from cStringIO import StringIO
from twisted.internet import reactor, defer
from twisted.internet import reactor, defer, threads
from twisted.python import failure
from twisted.trial import unittest
from allmydata.util import fileutil, log
from ..util.assertutil import precondition
from allmydata.util.encodingutil import unicode_platform, get_filesystem_encoding
from ..scripts import runner
def skip_if_cannot_represent_filename(u):
precondition(isinstance(u, unicode))
@ -19,6 +21,21 @@ def skip_if_cannot_represent_filename(u):
except UnicodeEncodeError:
raise unittest.SkipTest("A non-ASCII filename could not be encoded on this platform.")
def run_cli(verb, *args, **kwargs):
precondition(not [True for arg in args if not isinstance(arg, str)],
"arguments to do_cli must be strs -- convert using unicode_to_argv", args=args)
nodeargs = kwargs.get("nodeargs", [])
argv = nodeargs + [verb] + list(args)
stdin = kwargs.get("stdin", "")
stdout, stderr = StringIO(), StringIO()
d = threads.deferToThread(runner.runner, argv, run_by_human=False,
stdin=StringIO(stdin),
stdout=stdout, stderr=stderr)
def _done(rc):
return rc, stdout.getvalue(), stderr.getvalue()
d.addCallback(_done)
return d
class DevNullDictionary(dict):
def __setitem__(self, key, value):
return