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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
|
import argparse
from copy import copy
import os
from snakeoil.demandload import demandload
from snakeoil.osutils import pjoin
demandload(
'grp',
'pwd',
)
class IpcCommandError(Exception):
"""IPC errors related to parsing arguments or running the command."""
class ArgumentParser(argparse.ArgumentParser):
"""Raise IPC exception for argparse errors.
Otherwise standard argparse prints the parser usage then outputs the error
message to stderr.
"""
def error(self, message):
raise IpcCommandError(message)
class IpcCommand(object):
"""Commands sent from the bash side of the ebuild daemon to run."""
def __init__(self, op, async=False):
self.op = op
self.async = async
def __call__(self, ebd, numargs_str=None):
self.ebd = ebd
cmdargs = {}
if numargs_str is not None:
numargs = int(numargs_str.split('=', 1)[1])
for x in range(numargs):
k, v = self.read().rstrip('\n').split('=', 1)
cmdargs[k] = v
try:
self.parse_args(**cmdargs)
ret = self.run()
except IpcCommandError as e:
ret = e
# return completion status to the bash side if running in synchronous mode
if not self.async:
self.write(ret)
def parse_args(self, **kwargs):
"""Parse the args passed from the bash side."""
pass
def run(self):
"""Run the requested IPC command."""
raise NotImplementedError()
def read(self):
"""Read from the ebuild daemon."""
return self.ebd.read()
def write(self, s):
"""Write data to the ebuild daemon."""
self.ebd.write(s)
def _parse_group(group):
try:
return grp.getgrnam(group).gr_gid
except KeyError:
pass
return int(group)
def _parse_user(user):
try:
return pwd.getpwnam(user).pw_uid
except KeyError:
pass
return int(user)
def _parse_mode(mode):
try:
return int(mode, 8)
except ValueError:
return None
class Doins(IpcCommand):
"""Python wrapper for doins."""
parser = ArgumentParser(add_help=False)
parser.add_argument('-r', action='store_true', dest='recursive')
# supported install options
parser.add_argument('-g', '--group', default=-1, type=_parse_group)
parser.add_argument('-o', '--owner', default=-1, type=_parse_user)
parser.add_argument('-m', '--mode', default=0o755, type=_parse_mode)
# specified targets to install
parser.add_argument('targets', nargs=argparse.REMAINDER)
def parse_args(self, dest='', args=''):
self.opts, unknown = self.parser.parse_known_args([x for x in args.split('\x07') if x])
# add args with hyphen prefixes that aren't known options to target list
self.opts.targets.extend(unknown)
if not self.opts.targets:
raise IpcCommandError('missing targets')
self.dest = pjoin(self.op.ED, dest).rstrip(os.path.sep)
def run(self):
# dirs = set()
# for target in targets:
# if os.path.isdir(pjoin(self.op.env["WORKDIR"], target)):
# os.mkdir(pjoin(self.op.ED, target))
return 0
class Dodoc(Doins):
"""Python wrapper for dodoc."""
_base_parser = ArgumentParser(add_help=False)
_base_parser.add_argument('targets', nargs=argparse.REMAINDER)
def __init__(self, *args):
super(Dodoc, self).__init__(*args)
self.parser = copy(self._base_parser)
if self.op.pkg.eapi.options.dodoc_allow_recursive:
self.parser.add_argument('-r', action='store_true', dest='recursive')
def parse_args(self, **kwargs):
super(Dodoc, self).parse_args(**kwargs)
self.opts.mode = 0o644
class Dohtml(Doins):
"""Python wrapper for dohtml."""
parser = ArgumentParser(add_help=False)
parser.add_argument('-r', action='store_true', dest='recursive')
parser.add_argument('-V', action='store_true', dest='verbose')
parser.add_argument('-A')
parser.add_argument('-a')
parser.add_argument('-f')
parser.add_argument('-x')
parser.add_argument('-p')
parser.add_argument('targets', nargs=argparse.REMAINDER)
def run(self):
return 0
class Compress(IpcCommand):
"""Compress files tagged on the bash side."""
def parse_args(self, targets='', skip=''):
self.targets = set(x for x in targets.split('\x07') if x)
self.skip = set(x for x in skip.split('\x07') if x)
def run(self):
return 0
|
GitWeb