The Assimilation Project  based on Assimilation version 1.1.7.1474836767
testcases.py
Go to the documentation of this file.
1 #!/usr/bin/env python
2 # vim: smartindent tabstop=4 shiftwidth=4 expandtab number
3 #
4 # This file is part of the Assimilation Project.
5 #
6 # Author: Alan Robertson <alanr@unix.sh>
7 # Copyright (C) 2014 - Assimilation Systems Limited
8 #
9 # Free support is available from the Assimilation Project community - http://assimproj.org
10 # Paid support is available from Assimilation Systems Limited - http://assimilationsystems.com
11 #
12 # The Assimilation software is free software: you can redistribute it and/or modify
13 # it under the terms of the GNU General Public License as published by
14 # the Free Software Foundation, either version 3 of the License, or
15 # (at your option) any later version.
16 #
17 # The Assimilation software is distributed in the hope that it will be useful,
18 # but WITHOUT ANY WARRANTY; without even the implied warranty of
19 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
20 # GNU General Public License for more details.
21 #
22 # You should have received a copy of the GNU General Public License
23 # along with the Assimilation Project software. If not, see http://www.gnu.org/licenses/
24 #
25 #
26 '''
27 This file defines classes which perform individual system tests.
28 '''
29 import sys, time, os
30 from py2neo import neo4j
31 sys.path.append('..')
32 sys.path.append('.')
33 from logwatcher import LogWatcher
34 from querytest import QueryTest
35 from docker import SystemTestEnvironment, TestSystem
36 # pylint: disable=E0401
37 import graphnodes as GN
38 from cmainit import CMAinit
39 from store import Store
40 
41 def logger(s, hardquote=True):
42  'Log our single argument to syslog'
43  print >> sys.stderr, ('LOGGER: %s' % str(s))
44  if hardquote:
45  s = s.replace("'", "'\\''")
46  os.system("logger -s '%s'" % s)
47  else:
48  s = s.replace('\\', '\\\\')
49  s = s.replace('"', '\\"')
50  os.system('logger -s "%s"' % s)
51 
52 class AssimSysTest(object):
53  '''AssimSysTest is an abstract base class for all our system-level tests.
54  '''
55  SUCCESS = 1
56  FAIL = 2
57  SKIPPED = 3
58 
59  testnames = {}
60  testset = []
61  stats = {}
62 
63 
64  @staticmethod
65  def register(ourclass):
66  'Decorator for registering a TestCase'
67  AssimSysTest.testset.append(ourclass)
68  AssimSysTest.testnames[ourclass.__name__] = ourclass
69  AssimSysTest.stats[ourclass.__name__] = {
70  AssimSysTest.SUCCESS:0, AssimSysTest.FAIL: 0, AssimSysTest.SKIPPED:0
71  }
72  return ourclass
73 
74  def nano_start_regexes(self, nano):
75  'Return a list of the expected regexes for a nanoprobe starting'
76  cma = self.testenviron.cma
77  return [
78  r' %s cma INFO: Drone %s registered from address \[::ffff:%s]'
79  % (cma.hostname, nano.hostname, nano.ipaddr),
80  r' (%s) nanoprobe\[.*]: NOTICE: Connected to CMA. Happiness :-D' % (nano.hostname),
81  r' (%s) nanoprobe\[.*]: INFO: .* Configuration from CMA is complete.' % (nano.hostname),
82  r' %s cma INFO: Processed u?n?changed tcpdiscovery JSON data from (%s) into graph.'
83  % (cma.hostname, nano.hostname),
84  ]
85  def nano_stop_regexes(self, nano):
86  'Return a list of the expected regexes for a nanoprobe stopping'
87  cma = self.testenviron.cma
88  return [
89  r' %s .*NOTICE: nanoprobe: exiting on SIGTERM' % (nano.hostname),
90  r' %s cma INFO: System %s at \[::ffff:%s]:1984 reports graceful shutdown'
91  % (cma.hostname, nano.hostname, nano.ipaddr),
92  r" %s nanoprobe.*: INFO: Count of 'other' pkts received: " % (nano.hostname),
93  ]
94 
95  def cma_start_regexes(self):
96  'Return a list of the expected regexes for the CMA starting'
97  cma = self.testenviron.cma
98  return [
99  (' %s .* INFO: Neo4j version .* // py2neo version .*'
100  ' // Python version .* // (java|openjdk) version.*' % cma.hostname),
101  ]
102 
103  def nano_startmonitor_regexes(self, nano, monitorname):
104  'Return a list of the expected regexes for starting the given service monitoring'
105  cma = self.testenviron.cma
106  return [
107  r' %s cma INFO: Monitoring of service %s:.*:%s::.* activated'
108  % (cma.hostname, nano.hostname, monitorname),
109  ]
110 
111  def nano_service_start_regexes(self, nano, monitorname):
112  'Return a list of regexes of messages expected when starting the given monitored service'
113  cma = self.testenviron.cma
114  return [
115  r' %s cma INFO: Service %s:.*:%s::.* is now operational'
116  % (cma.hostname, nano.hostname, monitorname)
117  ]
118 
119  def nano_service_stop_regexes(self, nano, monitorname):
120  'Return a list of regexes of messages expected when stopping the given monitored service'
121  cma = self.testenviron.cma
122  return [
123  r' %s cma INFO: Service %s:.*:%s::.* failed with'
124  % (cma.hostname, nano.hostname, monitorname),
125  ]
126 
127  # [R0201:AssimSysTest.cma_stop_regexes] Method could be a function
128  # pylint: disable=R0201
129  def cma_stop_regexes(self):
130  'Return a list of the expected regexes for the CMA stopping'
131  #cma = self.testenviron.cma
132  return []
133 
134  def __init__(self, store, logfilename, testenviron, debug=False):
135  'Initializer for the AssimSysTest class'
136  self.store = store
137  self.logfilename = logfilename
138  self.testenviron = testenviron
139  self.debug = debug
140  self.result = None
141 
142  def _record(self, result):
143  'Record results from a test -- success or failure'
144  AssimSysTest.stats[self.__class__.__name__][result] += 1
145  #print >> sys.stderr, '_RECORD RETURNING', result
146  self.result = result
147  return result
148 
149  def replace_result(self, newresult):
150  'Replace this test result with an updated result (usually failure)'
151  AssimSysTest.stats[self.__class__.__name__][self.result] -= 1
152  self.result = newresult
153  AssimSysTest.stats[self.__class__.__name__][newresult] += 1
154 
155  # pylint - R0913: too mary arguments
156  # pylint: disable=R0913
157  def checkresults(self, watcher, timeout, querystring, validator
158  , nano, service=None, allregexes=True, debug=False, minrows=1, maxrows=1):
159  '''
160  A utility function for checking the results of a test. It assumes
161  that you have already done a watcher.setwatch and initiated the
162  test. We then wait for the test results in the logs and
163  perform the query to validate the results.
164  '''
165 
166  query = QueryTest(self.store, querystring, GN.nodeconstructor, debug=debug)
167  if allregexes:
168  match = watcher.lookforall(timeout=timeout)
169  else:
170  match = watcher.look(timeout=timeout)
171  if debug:
172  print >> sys.stderr, ('DEBUG: Match returned %s' % match)
173  if match is None:
174  logger('ERROR: Test %s timed out waiting for %s [timeout:%s]'
175  % (self.__class__.__name__, str(watcher.unmatched), timeout))
176  return self._record(AssimSysTest.FAIL)
177  if debug:
178  print('DEBUG: Test %s found regex %s'
179  % (self.__class__.__name__, str(watcher.regexes)))
180  if query.check((nano, self.testenviron.cma, service), validator
181  , minrows=minrows, maxrows=maxrows):
182  if debug:
183  print('DEBUG: Test %s passed query %s'
184  % (self.__class__.__name__, querystring))
185  return self._record(AssimSysTest.SUCCESS)
186 
187  print >> sys.stderr, ('DEBUG: query.check() FAILED')
188  logger('ERROR: Test %s failed query %s' % (self.__class__.__name__, querystring))
189  return self._record(AssimSysTest.FAIL)
190 
191  def run(self, nano=None, debug=None, timeout=30):
192  'Abstract run method'
193  raise NotImplementedError('AssimSysTest.run is an abstract method')
194 
195  @staticmethod
196  # too many local variables
197  # pylint: disable=R0914
198  def initenviron(logname, maxdrones, debug=False, timeout=90, nanodebug=0, cmadebug=0):
199  'Initialize the test environment.'
200  logwatch = LogWatcher(logname, [], timeout, returnonlymatch=True, debug=debug)
201  logwatch.setwatch()
202  sysenv = SystemTestEnvironment(logname, maxdrones, nanodebug=nanodebug, cmadebug=cmadebug)
203  CMAinit(None, host=str(sysenv.cma.ipaddr), readonly=True,
204  neologin=SystemTestEnvironment.NEO4JLOGIN, neopass=SystemTestEnvironment.NEO4JPASS)
205  url = 'http://%s:%d/db/data/' % (sysenv.cma.ipaddr, 7474)
206  print >> sys.stderr, 'OPENING Neo4j at URL %s' % url
207  neo4j.authenticate('%s:7474' % sysenv.cma.ipaddr,
208  SystemTestEnvironment.NEO4JLOGIN,
209  SystemTestEnvironment.NEO4JPASS)
210  store = Store(neo4j.Graph(url), readonly=True)
211  for classname in GN.GraphNode.classmap:
212  GN.GraphNode.initclasstypeobj(store, classname)
213 
214  logger('$(grep MemFree: /proc/meminfo)', hardquote=False)
215  tq = QueryTest(store
216  , '''START drone=node:Drone('*:*') WHERE drone.status = "up" RETURN drone'''
217  , GN.nodeconstructor, debug=debug)
218 
219  if not tq.check([None,], minrows=maxdrones+1, maxrows=maxdrones+1
220  , delay=0.5, maxtries=20):
221  sysenv.cma.cleanupwhendone = False
222  raise RuntimeError('Query of "up" status failed. Weirdness')
223  return sysenv, store
224 
225 
226 @AssimSysTest.register
228  'A stop nanoprobe test'
229  def run(self, nano=None, debug=None, timeout=30):
230  'Actually stop the nanoprobe and see if it worked'
231  if debug is None:
232  debug = self.debug
233  if nano is None:
234  nanozero = self.testenviron.select_nano_service()
235  if len(nanozero) > 0:
236  nano = nanozero[0]
237  if (nano is None or nano.status != TestSystem.RUNNING or
238  SystemTestEnvironment.NANOSERVICE not in nano.runningservices):
239  return self._record(AssimSysTest.SKIPPED)
240  regexes = self.nano_stop_regexes(nano)
241  watch = LogWatcher(self.logfilename, regexes, timeout=timeout, debug=debug)
242  watch.setwatch()
243  qstr = ( '''START drone=node:Drone('*:*') '''
244  '''WHERE drone.designation = "{0.hostname}" and drone.status = "dead" '''
245  '''and drone.reason = "HBSHUTDOWN" RETURN drone''')
246  nano.stopservice(SystemTestEnvironment.NANOSERVICE)
247  return self.checkresults(watch, timeout, qstr, None, nano)
248 
249 @AssimSysTest.register
251  'A start nanoprobe test'
252  def run(self, nano=None, debug=None, timeout=240):
253  'Actually start the nanoprobe and see if it worked'
254  if debug is None:
255  debug = self.debug
256  if nano is None:
257  nanozero = self.testenviron.select_nano_noservice()
258  if len(nanozero) > 0:
259  nano = nanozero[0]
260  if (nano is None or nano.status != TestSystem.RUNNING
261  or SystemTestEnvironment.NANOSERVICE in nano.runningservices):
262  return self._record(AssimSysTest.SKIPPED)
263 
264  regexes = self.nano_start_regexes(nano)
265 
266  watch = LogWatcher(self.logfilename, regexes, timeout=timeout, debug=debug)
267  watch.setwatch()
268  qstr = ( '''START drone=node:Drone('*:*') '''
269  '''WHERE drone.designation = "{0.hostname}" and drone.status = "up" '''
270  '''RETURN drone''')
271  nano.startservice(SystemTestEnvironment.NANOSERVICE)
272  return self.checkresults(watch, timeout, qstr, None, nano)
273 
274 @AssimSysTest.register
276  '''A flip nanoprobe test - if it's up, bring it down -- and vice versa'''
277  def run(self, nano=None, debug=None, timeout=240):
278  'Actually flip the nanoprobe and see if it worked'
279  if debug is None:
280  debug = self.debug
281  if nano is None:
282  nanozero = self.testenviron.select_nanoprobe()
283  if len(nanozero) > 0:
284  nano = nanozero[0]
285  if nano is None:
286  return self._record(AssimSysTest.SKIPPED)
287  if SystemTestEnvironment.NANOSERVICE in nano.runningservices:
288  return self._record(
289  StopNanoprobe(self.store
290  , self.logfilename, self.testenviron).run
291  (nano, debug=self.debug, timeout=timeout))
292  return self._record(
293  StartNanoprobe(self.store, self.logfilename
294  , self.testenviron).run(nano, debug=self.debug, timeout=timeout))
295 
296 @AssimSysTest.register
298  'A restart nanoprobe test, stop then restart a nanoprobe'
299  def __init__(self, store, logfilename, testenviron, debug=False, delay=0):
300  AssimSysTest.__init__(self, store, logfilename, testenviron, debug)
301  self.delay = delay
302 
303  def run(self, nano=None, debug=None, timeout=240):
304  'Actually stop and start (restart) the nanoprobe and see if it worked'
305  if debug is None:
306  debug = self.debug
307  if nano is None:
308  try:
309  nano = self.testenviron.select_nano_service()[0]
310  except IndexError:
311  nano = None
312  if nano is None:
313  return self._record(AssimSysTest.SKIPPED)
314  rc = StopNanoprobe(self.store
315  , self.logfilename, self.testenviron).run(nano, debug=self.debug, timeout=timeout)
316  if rc != AssimSysTest.SUCCESS:
317  return self._record(rc)
318  if self.delay > 0:
319  time.sleep(self.delay)
320  return self._record(StartNanoprobe(self.store
321  , self.logfilename, self.testenviron).run(nano, debug=self.debug, timeout=timeout))
322 
323 
324 @AssimSysTest.register
326  'A restart CMA test: stop then restart the CMA. Scary stuff!'
327  def __init__(self, store, logfilename, testenviron, debug=False, delay=0):
328  AssimSysTest.__init__(self, store, logfilename, testenviron, debug)
329  self.delay = delay
330 
331  def run(self, nano=None, debug=None, timeout=60):
332  'Actually stop and start (restart) the CMA and see if it worked'
333  if debug is None:
334  debug = self.debug
335  cma = self.testenviron.cma
336  cma.stopservice(SystemTestEnvironment.CMASERVICE)
337  regexes = self.cma_start_regexes()
338  watch = LogWatcher(self.logfilename, regexes, timeout=timeout, debug=debug)
339  watch.setwatch()
340  if self.delay > 0:
341  time.sleep(self.delay)
342  cma.startservice(SystemTestEnvironment.CMASERVICE)
343  # This just makes sure the database is still up - which it should be...
344  # Once we receive the CMA update message, we really should already be good to go
345  qstr = '''START one=node(*) RETURN one LIMIT 1'''
346  return self.checkresults(watch, timeout, qstr, None, nano)
347 
348 @AssimSysTest.register
350  'A restart CMA+nanoprobe test: stop then restart the CMA followed by a nanoprobe reset'
351  def __init__(self, store, logfilename, testenviron, debug=False, delay=0):
352  AssimSysTest.__init__(self, store, logfilename, testenviron, debug)
353  self.delay = delay
354 
355  def run(self, nano=None, debug=None, timeout=240):
356  'Actually stop and start (restart) the CMA and see if it worked'
357  if debug is None:
358  debug = self.debug
359  rc = RestartCMA(self.store, self.logfilename, self.testenviron, debug=debug
360  , delay=self.delay).run(timeout=timeout)
361  if rc != AssimSysTest.SUCCESS:
362  return self._record(rc)
363  return self._record(RestartNanoprobe(self.store, self.logfilename, self.testenviron
364  , debug=debug, delay=self.delay).run(timeout=timeout))
365 
366 @AssimSysTest.register
368  'Simultaneously restart the CMA and a nanoprobe'
369  def __init__(self, store, logfilename, testenviron, debug=False, delay=0):
370  AssimSysTest.__init__(self, store, logfilename, testenviron, debug)
371  self.delay = delay
372 
373  def run(self, nano=None, debug=None, timeout=180):
374  '''Our default timeout is so long because we can take a while to give up shutting down
375  the nanoprobe - an ACK timeout might have to occur before it can shut down.
376  '''
377  if debug is None:
378  debug = self.debug
379  if nano is None:
380  nanozero = self.testenviron.select_nano_service()
381  if len(nanozero) < 1:
382  return self._record(AssimSysTest.SKIPPED)
383  nano = nanozero[0]
384  cma = self.testenviron.cma
385  regexes = self.nano_stop_regexes(nano)
386  regexes.extend(self.cma_stop_regexes())
387  regexes.extend(self.cma_start_regexes())
388  watch = LogWatcher(self.logfilename, regexes, timeout=timeout, debug=debug)
389  watch.setwatch()
390  cma.stopservice(SystemTestEnvironment.CMASERVICE)
391  nano.stopservice(SystemTestEnvironment.NANOSERVICE, async=True)
392  cma.startservice(SystemTestEnvironment.CMASERVICE)
393  if self.delay > 0:
394  time.sleep(self.delay)
395  qstr = ( '''START drone=node:Drone('*:*') '''
396  '''WHERE drone.designation = "{0.hostname}" and drone.status = "dead" '''
397  '''and drone.reason = "HBSHUTDOWN" RETURN drone''')
398  rc = self.checkresults(watch, timeout, qstr, None, nano)
399  if rc != AssimSysTest.SUCCESS:
400  return rc
401  # We have to do this in two parts because of the asynchronous shutdown above
402  regexes = self.nano_start_regexes(nano)
403  watch = LogWatcher(self.logfilename, regexes, timeout=timeout, debug=debug)
404  watch.setwatch()
405  nano.startservice(SystemTestEnvironment.NANOSERVICE)
406  qstr = ( '''START drone=node:Drone('*:*') '''
407  '''WHERE drone.designation = "{0.hostname}" and drone.status = "up" '''
408  '''RETURN drone''')
409  return self.checkresults(watch, timeout, qstr, None, nano)
410 
411 @AssimSysTest.register
413  '''We find a system not running some particular service, then we
414  start the service and restart the nanoprobe - forcing it to
415  discover the service pretty quickly.
416  '''
417  def __init__(self, store, logfilename, testenviron, debug=False
418  , service='bind9', monitorname=None):
419  'Initializer for the DiscoverService class'
420  AssimSysTest.__init__(self, store, logfilename, testenviron, debug)
421  self.service=service
422  if monitorname is None:
423  monitorname = service
424  self.monitorname = monitorname
425 
426  # W0221:Arguments number differs from overridden method
427  # pylint: disable=W0221
428  def run(self, nano=None, debug=None, timeout=240, service=None, monitorname=None):
429  if debug is None:
430  debug = self.debug
431  if service is None:
432  service = self.service
433  if monitorname is None:
434  monitorname = self.monitorname
435  if nano is None:
436  nanozero = self.testenviron.select_nano_noservice(service=service)
437  if nanozero is None or len(nanozero) < 1:
438  # bind doesn't seem to shut down properly - need to look into that...
439  return self._record(AssimSysTest.SKIPPED)
440  else:
441  nano = nanozero[0]
442  assert service not in nano.runningservices
443  if SystemTestEnvironment.NANOSERVICE not in nano.runningservices:
444  startregexes = self.nano_start_regexes(nano)
445  watch = LogWatcher(self.logfilename, startregexes, timeout=timeout, debug=debug)
446  watch.setwatch()
447  nano.startservice(SystemTestEnvironment.NANOSERVICE)
448  match = watch.look(timeout=timeout)
449  if match is None:
450  logger('ERROR: Test %s timed out waiting for any of %s [timeout:%s]'
451  % (self.__class__.__name__, str(watch.regexes), timeout))
452  return self._record(AssimSysTest.FAIL)
453  regexes = self.nano_stop_regexes(nano)
454  watch = LogWatcher(self.logfilename, regexes, timeout=timeout, debug=debug)
455  watch.setwatch()
456  nano.stopservice(SystemTestEnvironment.NANOSERVICE)
457  if watch.lookforall(timeout=timeout) is None:
458  logger('ERROR: Test %s timed out waiting for all of %s [timeout:%s]'
459  % (self.__class__.__name__, str(watch.unmatched), timeout))
460  return self._record(AssimSysTest.FAIL)
461  regexes = self.nano_start_regexes(nano)
462  regexes.extend(self.nano_startmonitor_regexes(nano, monitorname))
463  regexes.extend(self.nano_service_start_regexes(nano, monitorname))
464  watch = LogWatcher(self.logfilename, regexes, timeout=timeout, debug=debug)
465  watch.setwatch()
466  nano.startservice(service)
467  nano.startservice(SystemTestEnvironment.NANOSERVICE)
468  if watch.lookforall(timeout=timeout) is None:
469  logger('ERROR: Test %s timed out waiting for all of %s [timeout:%s]'
470  % (self.__class__.__name__, str(watch.unmatched), timeout))
471  return self._record(AssimSysTest.FAIL)
472  # @TODO make a better query
473  # but it should be enough to let us validate the rest
474  qstr = ( '''START drone=node:Drone('*:*') '''
475  '''WHERE drone.designation = "{0.hostname}" and drone.status = "up" '''
476  '''RETURN drone''')
477  return self.checkresults(watch, timeout, qstr, None, nano, debug=debug)
478 
479 
480 # A little test code...
481 if __name__ == "__main__":
482  # pylint: disable=R0914
483  def testmain(logname, maxdrones=3, debug=False):
484  'Test our test cases'
485  logger('Starting test of our test cases')
486  try:
487  sysenv, ourstore = AssimSysTest.initenviron(logname, maxdrones, debug
488  , cmadebug=5, nanodebug=3)
489  except AssertionError:
490  print 'FAILED initial startup - which is pretty basic'
491  print 'Any chance you have another CMA running??'
492  raise RuntimeError('Another CMA is running(?)')
493 
494  badregexes=(' ERROR: ', ' CRIT: ', ' CRITICAL: '
495  # 'HBDEAD'
496  #, r'Peer at address .* is dead'
497  , r'OUTALLDONE .* while in state NONE'
498  )
499  #for cls in [SimulCMAandNanoprobeRestart for j in range(0,20)]:
500  #for j in range(0,10):
501  #for cls in [DiscoverService for j in range(0,100)]:
502  for cls in AssimSysTest.testset:
503  badwatch = LogWatcher(logname, badregexes, timeout=1, debug=0)
504  logger('CREATED LOG WATCH with %s' % str(badregexes))
505  badwatch.setwatch()
506  logger('Starting test %s' % (cls.__name__))
507  if cls is DiscoverService:
508  ret = cls(ourstore, logname, sysenv, debug=debug
509  , service='bind9', monitorname='named').run()
510  else:
511  ret = cls(ourstore, logname, sysenv, debug=debug).run()
512  #print >> sys.stderr, 'Got return of %s from test %s' % (ret, cls.__name__)
513  badmatch = badwatch.look(timeout=1)
514  if badmatch is not None:
515  print 'OOPS! Got bad results!', badmatch
516  raise RuntimeError('Test %s said bad words! [%s]' % (cls.__name__, badmatch))
517  assert ret == AssimSysTest.SUCCESS or ret == AssimSysTest.SKIPPED
518  #assert ret == AssimSysTest.SUCCESS
519  logger('WOOT! All tests were successful!')
520 
521  if os.access('/var/log/syslog', os.R_OK):
522  sys.exit(testmain('/var/log/syslog', debug=False))
523  elif os.access('/var/log/messages', os.R_OK):
524  sys.exit(testmain('/var/log/messages', debug=False))
def run(self, nano=None, debug=None, timeout=240, service=None, monitorname=None)
Definition: testcases.py:428
def __init__(self, store, logfilename, testenviron, debug=False, delay=0)
Definition: testcases.py:299
def initenviron(logname, maxdrones, debug=False, timeout=90, nanodebug=0, cmadebug=0)
Definition: testcases.py:198
def run(self, nano=None, debug=None, timeout=240)
Definition: testcases.py:355
def register(ourclass)
Definition: testcases.py:65
def nano_service_start_regexes(self, nano, monitorname)
Definition: testcases.py:111
def run(self, nano=None, debug=None, timeout=30)
Definition: testcases.py:229
def __init__(self, store, logfilename, testenviron, debug=False, delay=0)
Definition: testcases.py:351
def _record(self, result)
Definition: testcases.py:142
def nano_startmonitor_regexes(self, nano, monitorname)
Definition: testcases.py:103
def nano_stop_regexes(self, nano)
Definition: testcases.py:85
def run(self, nano=None, debug=None, timeout=240)
Definition: testcases.py:303
def nano_service_stop_regexes(self, nano, monitorname)
Definition: testcases.py:119
def __init__(self, store, logfilename, testenviron, debug=False, delay=0)
Definition: testcases.py:327
def run(self, nano=None, debug=None, timeout=240)
Definition: testcases.py:252
def logger(s, hardquote=True)
Definition: testcases.py:41
def __init__(self, store, logfilename, testenviron, debug=False)
Definition: testcases.py:134
def cma_start_regexes(self)
Definition: testcases.py:95
def run(self, nano=None, debug=None, timeout=240)
Definition: testcases.py:277
def testmain(logname, maxdrones=3, debug=False)
Definition: testcases.py:483
def replace_result(self, newresult)
Definition: testcases.py:149
def __init__(self, store, logfilename, testenviron, debug=False, delay=0)
Definition: testcases.py:369
def __init__(self, store, logfilename, testenviron, debug=False, service='bind9', monitorname=None)
Definition: testcases.py:418
def run(self, nano=None, debug=None, timeout=30)
Definition: testcases.py:191
def checkresults(self, watcher, timeout, querystring, validator, nano, service=None, allregexes=True, debug=False, minrows=1, maxrows=1)
Definition: testcases.py:158
def run(self, nano=None, debug=None, timeout=60)
Definition: testcases.py:331
def nano_start_regexes(self, nano)
Definition: testcases.py:74
def run(self, nano=None, debug=None, timeout=180)
Definition: testcases.py:373
def cma_stop_regexes(self)
Definition: testcases.py:129