Source Code for Module epyunit.unittest.subprocess

  1  # -*- coding: utf-8 -*- 
  2  """The module 'epyunit.unittest.subprocess' provides classes derived from 'unittest' for tests of arbitrary subprocesses. 
  3   
  4  """ 
  5  from __future__ import absolute_import 
  6   
  7  __author__ = 'Arno-Can Uestuensoez' 
  8  __license__ = "Artistic-License-2.0 + Forced-Fairplay-Constraints" 
  9  __copyright__ = "Copyright (C) 2010-2016 Arno-Can Uestuensoez @Ingenieurbuero Arno-Can Uestuensoez" 
 10  __version__ = '0.2.1' 
 11  __uuid__='9de52399-7752-4633-9fdc-66c87a9200b8' 
 12   
 13  __docformat__ = "restructuredtext en" 
 14   
 15  import sys,os 
 16  from types import NoneType 
 17   
 18  import unittest 
 19   
 20  version = '{0}.{1}'.format(*sys.version_info[:2]) 
 21  if not version in ('2.6', '2.7',): # pragma: no cover 
 22      raise Exception("Requires Python-2.6.* or higher") 
 23   
 24   
 25  from epyunit.SubprocUnit import SubprocessUnit,SProcUnitRules 
 26   
 27 -class TestExecutableException(Exception): 
 28      """Application error for called executable. 
 29      """ 
 30      pass 
 31   
 32 -class FileNodeDoesNotExists(Exception): 
 33      """Filesystem node does not exist. 
 34      """ 
 35      pass 
 36   
 37 -class TestExecutable(unittest.TestCase): 
 38      """Extends TestCase for subprocesses. 
 39      """ 
 40   
 41 -    def __init__(self,*args, **kargs): 
 42          """Initializes the test case, passes parameters to 'unittest.TestCase'.  
 43   
 44          Args: 
 45              **kargs: Parameter specific for the operation, 
 46                  passed through to **setkargs**. 
 47   
 48          Returns: 
 49              When successful returns 'True'. 
 50   
 51          Raises: 
 52              passed through exceptions: 
 53   
 54          """ 
 55          super(TestExecutable,self).__init__(*args,**kargs) 
 56   
 57          self.spunit = None 
 58          """Subprocess unit assigned to this test case.""" 
 59          if not kargs.get('spunit'): # else shift setting into setkargs() 
 60              self.spunit = SubprocessUnit() 
 61   
 62          self.spcache = [0,None,None] 
 63          """Subprocess output caches. 
 64          """ 
 65       
 66          self.setkargs(**kargs) # sets parameters 
 67          pass 
 68   
 69 -    def setkargs(self,**kargs): 
 70          """Sets provided parameters for the subprocess call context. 
 71           
 72          Applicable for the initial call of self.__init__(),  
 73          and later modification. Called for each start of 
 74          a subprocess in order to update optional the specific  
 75          call context modification. 
 76   
 77          Args: 
 78              **kargs: Parameters specific for the operations. 
 79   
 80                  cache:  
 81                      Sets caching of the results for subprocess call.  
 82   
 83                  noparent:  
 84                      Suppress call of parent class.  
 85   
 86                  rules:  
 87                      Sets the rules object to be used. 
 88                      See: epyunit.SubprocUnit.SProcUnitRules 
 89   
 90                  spunit:  
 91                      Sets the subprocess unit. 
 92                      See: epyunit.SubprocUnit.SProcUnit 
 93   
 94          Returns: 
 95              When successful returns 'True', else returns either 'False', or 
 96              raises an exception. 
 97   
 98          Raises: 
 99              passed through exceptions: 
100               
101          """ 
102          self.noparent = kargs.get('noparent',False) 
103          for k,v in kargs.items(): 
104              if k == 'rules': 
105                  self.spunit.setruleset(v) 
106              elif k == 'spunit': 
107                  self.spunit = v 
108              elif k == 'cache': 
109                  self.spcache = v 
110   
111   
112 -    def callSubprocess(self,callstr,**kargs): 
113          """Calls a subprocess and fetches the result data. 
114   
115          Args: 
116              callstr: Complete call string for subprocess. 
117               
118              **kargs:  
119                  Pass-through parameters for 'SystemCalls.callit()'. 
120               
121                  cache:  
122                      Caches results. 
123   
124          Returns: 
125              When successful returns the tupel: 
126   
127                  returnvalue := (exit-value, stdout-value, stderr-value) 
128   
129          Raises: 
130              passed through exceptions: 
131           
132          """ 
133          _sargs = kargs.copy() 
134          _cache = self.spcache 
135          for k,v in _sargs.items(): 
136              if k == 'cache': 
137                  _cache = v 
138                  _sargs.pop(k) 
139   
140          if _cache: 
141              self.spcache = self.spunit.callit(callstr,**_sargs) 
142              return self.spcache 
143          return self.spunit.callit(callstr,**_sargs) 
144   
145 -    def assertEqual(self,exp,cur=None): 
146          """Asserts tuple result value of the called process. 
147   
148          Args: 
149              exp: Expected value: 
150                  (exit-value, stdout-value, stderr-value) 
151               
152              cur: Current value: 
153                  (exit-value, stdout-value, stderr-value) 
154   
155                  default:=self.spcache 
156   
157          Returns: 
158              When successful returns the True, else raises an exception. 
159   
160          Raises: 
161              Raises result of assertEqueal. 
162   
163          """ 
164          if type(cur) == NoneType: 
165              cur = self.spcache 
166          super(TestExecutable,self).assertEqual(exp, cur) 
167          return True 
168       
169 -    def assertExit(self,exp,cur=None): 
170          """Asserts on the exit value of the called process. 
171   
172          Args: 
173              exp: Expected value: 
174                  exit-value 
175               
176              cur: Current value: 
177                  exit-value 
178   
179                  default:=self.spcache[0] 
180   
181          Returns: 
182              When successful returns the True, else raises an exception. 
183   
184          Raises: 
185              Raises result of assertEqual. 
186   
187          """ 
188          if type(cur) == NoneType: 
189              cur = self.spcache[0] 
190          assert exp[0] == cur[0] 
191          pass 
192       
193 -    def assertExists(self,fpname): 
194          """Asserts on the existance of a filesystem node. 
195   
196          Args: 
197              fpname: Filesystem node. 
198               
199   
200          Returns: 
201              When successful returns the True, else raises an exception. 
202   
203          Raises: 
204              Raises result FileNodeDoesNotExists. 
205   
206          """ 
207          self.assertTrue(os.path.exists(fpname)) 
208          return True 
209   
210 -    def assertStdout(self,exp,cur=None): 
211          """Asserts a list of provided regexpr on the STDOUT of the called process. 
212   
213          Args: 
214              exp: Expected value: 
215                  stdout-value 
216               
217              cur: Current value: 
218                  stdout-value 
219   
220                  default:=self.spcache[1] 
221   
222          Returns: 
223              When successful returns the True, else raises an exception. 
224   
225          Raises: 
226              Raises result of assertEqual. 
227   
228          """ 
229          if type(cur) == NoneType: 
230              cur = self.spcache[1] 
231          return super(TestExecutable,self).assertEqual(exp[1], cur[1]) 
232          pass 
233   
234 -    def assertStderr(self,exp,cur=None): 
235          """Asserts a list of provided regexpr on the STDERR of the called process. 
236   
237          Args: 
238              exp: Expected value: 
239                  stderr-value 
240               
241              cur: Current value: 
242                  stderr-value 
243   
244                  default:=self.spcache[2] 
245   
246          Returns: 
247              When successful returns the True, else raises an exception. 
248   
249          Raises: 
250              Raises result of assertEqual. 
251   
252          """ 
253          if type(cur) == NoneType: 
254              cur = self.spcache[2] 
255          return super(TestExecutable,self).assertEqual(exp[2], cur[2]) 
256          pass 
257   
258 -    def assertSubprocess(self,callstr, res, **kargs): 
259          """Calls a subprocess and asserts the result data. 
260   
261          Args: 
262              callstr:  
263                  Complete call string for subprocess. 
264               
265              res:  
266                  Result from 'TestExecutable.callSubprocess()' 
267   
268                      returnvalue := (exit-value, stdout-value, stderr-value) 
269   
270              **kargs:  
271                  Pass-through parameters for 'SystemCalls.callit()'. 
272               
273                  cache:  
274                      Caches results. 
275   
276          Returns: 
277              When successful returns True, else raises exception. 
278   
279          Raises: 
280              passed through exceptions: 
281           
282          """ 
283          ret = self.callSubprocess(callstr) 
284          return self.assertEqual(ret, res) 
285   
286   
287 -    def __str__(self): 
288          """Prints the current test state. 
289          """ 
290          res  = "" 
291          res += str(self.spcache)  
292          return res 
293