cli.py
4.31 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
"""
Digress's CLI interface.
"""
import inspect
import sys
from optparse import OptionParser
import textwrap
from types import MethodType
from digress import __version__ as version
def dispatchable(func):
"""
Mark a method as dispatchable.
"""
func.digress_dispatchable = True
return func
class Dispatcher(object):
"""
Dispatcher for CLI commands.
"""
def __init__(self, fixture):
self.fixture = fixture
fixture.dispatcher = self
def _monkey_print_help(self, optparse, *args, **kwargs):
# monkey patches OptionParser._print_help
OptionParser.print_help(optparse, *args, **kwargs)
print >>sys.stderr, "\nAvailable commands:"
maxlen = max([ len(command_name) for command_name in self.commands ])
descwidth = 80 - maxlen - 4
for command_name, command_meth in self.commands.iteritems():
print >>sys.stderr, " %s %s\n" % (
command_name.ljust(maxlen + 1),
("\n" + (maxlen + 4) * " ").join(
textwrap.wrap(" ".join(filter(
None,
command_meth.__doc__.strip().replace("\n", " ").split(" ")
)),
descwidth
)
)
)
def _enable_flush(self):
self.fixture.flush_before = True
def _populate_parser(self):
self.commands = self._get_commands()
self.optparse = OptionParser(
usage = "usage: %prog [options] command [args]",
description = "Digress CLI frontend for %s." % self.fixture.__class__.__name__,
version = "Digress %s" % version
)
self.optparse.print_help = MethodType(self._monkey_print_help, self.optparse, OptionParser)
self.optparse.add_option(
"-f",
"--flush",
action="callback",
callback=lambda option, opt, value, parser: self._enable_flush(),
help="flush existing data for a revision before testing"
)
self.optparse.add_option(
"-c",
"--cases",
metavar="FOO,BAR",
action="callback",
dest="cases",
type=str,
callback=lambda option, opt, value, parser: self._select_cases(*value.split(",")),
help="test cases to run, run with command list to see full list"
)
def _select_cases(self, *cases):
self.fixture.cases = filter(lambda case: case.__name__ in cases, self.fixture.cases)
def _get_commands(self):
commands = {}
for name, member in inspect.getmembers(self.fixture):
if hasattr(member, "digress_dispatchable"):
commands[name] = member
return commands
def _run_command(self, name, *args):
if name not in self.commands:
print >>sys.stderr, "error: %s is not a valid command\n" % name
self.optparse.print_help()
return
command = self.commands[name]
argspec = inspect.getargspec(command)
max_arg_len = len(argspec.args) - 1
min_arg_len = max_arg_len - ((argspec.defaults is not None) and len(argspec.defaults) or 0)
if len(args) < min_arg_len:
print >>sys.stderr, "error: %s takes at least %d arguments\n" % (
name,
min_arg_len
)
print >>sys.stderr, "%s\n" % command.__doc__
self.optparse.print_help()
return
if len(args) > max_arg_len:
print >>sys.stderr, "error: %s takes at most %d arguments\n" % (
name,
max_arg_len
)
print >>sys.stderr, "%s\n" % command.__doc__
self.optparse.print_help()
return
command(*args)
def pre_dispatch(self):
pass
def dispatch(self):
self._populate_parser()
self.optparse.parse_args()
self.pre_dispatch()
args = self.optparse.parse_args()[1] # arguments may require reparsing after pre_dispatch; see test_x264.py
if len(args) == 0:
print >>sys.stderr, "error: no command specified\n"
self.optparse.print_help()
return
command = args[0]
addenda = args[1:]
self._run_command(command, *addenda)