Package pybaz :: Package backends :: Module commandline
[frames] | no frames]

Source Code for Module pybaz.backends.commandline

  1  # arch-tag: cb40743b-1a3d-4c56-9647-512721976ad2 
  2  # Copyright (C) 2004 Canonical Ltd. 
  3  #               Author: David Allouche <david@canonical.com> 
  4  # 
  5  #    This program is free software; you can redistribute it and/or modify 
  6  #    it under the terms of the GNU General Public License as published by 
  7  #    the Free Software Foundation; either version 2 of the License, or 
  8  #    (at your option) any later version. 
  9  # 
 10  #    This program is distributed in the hope that it will be useful, 
 11  #    but WITHOUT ANY WARRANTY; without even the implied warranty of 
 12  #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the 
 13  #    GNU General Public License for more details. 
 14  # 
 15  #    You should have received a copy of the GNU General Public License 
 16  #    along with this program; if not, write to the Free Software 
 17  #    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA 
 18   
 19  """Command-line back-end glue.""" 
 20   
 21  import sys 
 22  from pybaz import compat 
 23   
 24  ### Argument Checking ### 
 25   
26 -def _check_expected_param(expected):
27 msg_format = ("'expected' param must be a sequence of positive" 28 " integers but was: %r") 29 try: 30 iterator = iter(expected) 31 except TypeError: 32 raise TypeError, msg_format % expected 33 for status in iterator: 34 if not isinstance(status, int): 35 raise TypeError, msg_format % expected 36 if status < 0: 37 raise ValueError, msg_format % expected
38 39 40 ### Default Backend ### 41
42 -def default_backend():
43 """Default command-line backend. 44 45 :rtype: `CommandLineBackend` 46 """ 47 import baz 48 import logger 49 backend = CommandLineBackend( 50 command = 'baz', 51 module = baz, 52 spawning = DelayedGuessedSpawningStrategy, 53 logger = logger.Logger()) 54 return backend
55 56
57 -class CommandLineBackend(object):
58 59 """Facade for command-line back-end options.""" 60
61 - def __init__(self, command, module, spawning, logger):
62 self._command = command 63 self._module = module 64 self._spawning = spawning 65 self._logger = logger 66 self._spawner = None 67 self._version = None
68
69 - def _get_command(self):
70 return self._command
71
72 - def _set_command(self, command):
73 assert command 74 self._command = command 75 self._spawner = None 76 self._version = None
77 78 command = property( 79 _get_command, _set_command, None, 80 """Name of the command line program. 81 82 :type: str""") 83
84 - def _get_spawning(self):
85 return self._spawning
86
87 - def _set_spawning(self, spawning):
88 assert spawning 89 self._spawning = spawning 90 self._spawner = None
91 92 spawning_strategy = property( 93 _get_spawning, _set_spawning, None, 94 """`SpawningStrategy` factory used to create the spawner. 95 96 :type: factory of `SpawningStrategy`""") 97
98 - def _get_logger(self):
99 return self._logger
100
101 - def _set_logger(self, logger):
102 assert logger 103 self._logger = logger 104 self._spawner = None
105 106 logger = property( 107 _get_logger, _set_logger, None, 108 """Command line logger. 109 110 :type: `logger.Logger`""") 111
112 - def _get_spawner(self):
113 if self._spawner is None: 114 assert self._spawning 115 assert self._command 116 assert self._logger 117 self._spawner = self._spawning(self._command, self._logger) 118 return self._spawner
119
120 - def null_cmd(self, args, chdir=None):
121 status = self._get_spawner().status_cmd(args, chdir, (0,)) 122 assert status == 0
123
124 - def one_cmd(self, args, chdir=None):
125 status, line = self.status_one_cmd(args, (0,), chdir) 126 assert status == 0 127 return line
128
129 - def sequence_cmd(self, args, expected=(0,), chdir=None, stderr_too=False):
130 # XXX stderr_too=True will fail with the TwistedSpawningStrategy. That 131 # XXX will be fixed when the process handling subsystem is replaced by 132 # XXX Gnarly. -- David Allouche 2005-05-27 133 _check_expected_param(expected) 134 sequence_cmd = self._get_spawner().sequence_cmd 135 if stderr_too: 136 return sequence_cmd(args, chdir, expected, stderr_too=stderr_too) 137 else: 138 return sequence_cmd(args, chdir, expected)
139
140 - def text_cmd(self, args, chdir=None):
141 status, text = self._get_spawner().status_text_cmd(args, chdir, (0,)) 142 assert status == 0 143 return text
144
145 - def status_cmd(self, args, expected, chdir=None):
146 _check_expected_param(expected) 147 return self._get_spawner().status_cmd(args, chdir, expected)
148
149 - def status_one_cmd(self, args, expected, chdir=None):
150 _check_expected_param(expected) 151 seq = self.sequence_cmd(args, expected, chdir) 152 try: 153 out = seq.next() 154 except StopIteration: 155 return seq.status, None 156 tail = list(seq) 157 if not tail: 158 return seq.status, out 159 raise AssertionError, ( 160 'one liner actually produced multiline output:\n%s' 161 % '\n'.join([out] + tail))
162
163 - def status_text_cmd(self, args, expected):
164 _check_expected_param(expected) 165 return self._get_spawner().status_text_cmd(args, None, expected)
166
167 - def _get_version(self):
168 if self._version is None: 169 line = list(self.sequence_cmd(('--version',)))[0] 170 self._version = compat.BazaarCommandVersion(line) 171 return self._version
172 173 version = property(_get_version, doc=""" 174 The current command version. 175 176 Instance of pybaz.compat.BazaarCommandVersion.""") 177 178 __pybaz_version = compat.BazaarCommandVersion( 179 'baz Bazaar version 1.4~200504281027') 180
181 - def _get_compat(self):
182 return compat._get_status(self.__pybaz_version, self.version)
183 184 compatibility = property(_get_compat)
185 186
187 -class SpawningStrategy(object):
188 189 """Abstract class for process spawning strategies.""" 190
191 - def status_cmd(self, args, chdir, expected):
192 raise NotImplementedError
193
194 - def status_text_cmd(self, args, chdir, expected):
195 raise NotImplementedError
196
197 - def sequence_cmd(self, args, chdir=None, expected=(0,)):
198 raise NotImplementedError
199 200
201 -class DelayedGuessedSpawningStrategy(SpawningStrategy):
202 """SpawningStrategy that uses Twisted if it is present in ``sys.modules``. 203 204 This SpawningStrategy tries to do the right thing my looking at 205 the contents of ``sys.modules``: if "twisted" or a module of that 206 package is loaded when the first spawning method is run, it will 207 use the Twisted spawning strategy. Otherwise, it will use a simple 208 fork/exec spawning strategy which does not depend on Twisted. 209 """ 210 211 __pychecker__ = 'no-override' 212
213 - def __init__(self, *args, **kwargs):
214 self._init_args = args 215 self._init_kwargs = kwargs 216 self._cached_guess = None
217
218 - def _guess(self):
219 if self._cached_guess is not None: 220 return self._cached_guess 221 strategy = None 222 for name in sys.modules: 223 if name == 'twisted' or name.startswith('twisted.'): 224 import knotted 225 strategy = knotted.TwistedSpawningStrategy 226 break 227 else: 228 import forkexec 229 strategy = forkexec.PyArchSpawningStrategy 230 self._cached_guess = strategy(*self._init_args, **self._init_kwargs) 231 return self._cached_guess
232
233 - def sequence_cmd(self, *args, **kwargs):
234 return self._guess().sequence_cmd(*args, **kwargs)
235
236 - def status_cmd(self, *args, **kwargs):
237 return self._guess().status_cmd(*args, **kwargs)
238
239 - def status_text_cmd(self, *args, **kwargs):
240 return self._guess().status_text_cmd(*args, **kwargs)
241