Home | History | Annotate | Download | only in dynamic_suite 
      1 #pylint: disable-msg=C0111
      2 # Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
      3 # Use of this source code is governed by a BSD-style license that can be
      4 # found in the LICENSE file.
      5 
      6 """Fakes for dynamic_suite-related unit tests."""
      7 
      8 import common
      9 from autotest_lib.client.common_lib import control_data
     10 
     11 
     12 class FakeControlData(control_data.ControlData):
     13     """A fake parsed control file data structure."""
     14     def __init__(self, suite, attributes, data, time='LONG', expr=False,
     15                  dependencies=None, job_retries=0):
     16         self.string = 'text-' + data
     17         self.name = 'name-' + data
     18         self.path = None  # Will be set during 'parsing'.
     19         self.data = data
     20         self.suite = suite
     21         self.attributes = attributes
     22         self.test_type = 'Client'
     23         self.experimental = expr
     24         if not dependencies:
     25             dependencies=[]
     26         self.dependencies = dependencies
     27         self.time = time
     28         self.sync_count = 1
     29         self.job_retries = job_retries
     30         self.bug_template = {}
     31         self.require_ssp = None
     32         self.priority = 10
     33         self.fast = False
     34 
     35 
     36 class FakeJob(object):
     37     """Faked out RPC-client-side Job object."""
     38     def __init__(self, id=0, statuses=[], hostnames=[], parent_job_id=None):
     39         self.id = id
     40         self.hostnames = hostnames if hostnames else ['host%d' % id]
     41         self.owner = 'tester'
     42         self.name = 'Fake Job %d' % self.id
     43         self.statuses = statuses
     44         self.parent_job_id = parent_job_id
     45 
     46 
     47 class FakeHost(object):
     48     """Faked out RPC-client-side Host object."""
     49     def __init__(self, hostname='', status='Ready', locked=False, locked_by=''):
     50         self.hostname = hostname
     51         self.status = status
     52         self.locked = locked
     53         self.locked_by = locked_by
     54 
     55 
     56     def __str__(self):
     57         return '%s: %s.  %s%s' % (
     58             self.hostname, self.status,
     59             'Locked' if self.locked else 'Unlocked',
     60             ' by %s' % self.locked_by if self.locked else '')
     61 
     62 
     63 class FakeLabel(object):
     64     """Faked out RPC-client-side Label object."""
     65     def __init__(self, id=0):
     66         self.id = id
     67 
     68 
     69 class FakeStatus(object):
     70     """Fake replacement for server-side job status objects.
     71 
     72     @var status: 'GOOD', 'FAIL', 'ERROR', etc.
     73     @var test_name: name of the test this is status for
     74     @var reason: reason for failure, if any
     75     @var aborted: present and True if the job was aborted.  Optional.
     76     """
     77     def __init__(self, code, name, reason, aborted=None,
     78                  hostname=None, subdir='fake_Test.tag.subdir_tag',
     79                  job_tag='id-owner/hostname'):
     80         self.status = code
     81         self.test_name = name
     82         self.reason = reason
     83         self.hostname = hostname if hostname else 'hostless'
     84         self.entry = {}
     85         self.test_started_time = '2012-11-11 11:11:11'
     86         self.test_finished_time = '2012-11-11 12:12:12'
     87         self.job_tag=job_tag
     88         self.subdir=subdir
     89         if aborted:
     90             self.entry['aborted'] = True
     91         if hostname:
     92             self.entry['host'] = {'hostname': hostname}
     93 
     94 
     95     def __repr__(self):
     96         return '%s\t%s\t%s: %s' % (self.status, self.test_name, self.reason,
     97                                    self.hostname)
     98 
     99 
    100     def equals_record(self, status):
    101         """Compares this object to a recorded status."""
    102         if 'aborted' in self.entry and self.entry['aborted']:
    103             return status._status == 'ABORT'
    104         return (self.status == status._status and
    105                 status._test_name.endswith(self.test_name) and
    106                 self.reason == status._reason)
    107 
    108 
    109     def equals_hostname_record(self, status):
    110         """Compares this object to a recorded status.
    111 
    112         Expects the test name field of |status| to contain |self.hostname|.
    113         """
    114         return (self.status == status._status and
    115                 self.hostname in status._test_name and
    116                 self.reason == status._reason)
    117 
    118 
    119     def record_all(self, record):
    120         pass
    121 
    122 
    123     def is_good(self):
    124         pass
    125 
    126     def name(self):
    127         return self.test_name
    128 
    129 
    130 class FakeMultiprocessingPool(object):
    131     """Fake multiprocessing pool to mock out the map method."""
    132 
    133 
    134     def __init__(self, processes=None, initializer=None, initargs=(),
    135                  maxtasksperchild=None):
    136         pass
    137 
    138 
    139     def map(self, func, iterable, chunksize=None):
    140         """Use the standard map() built-in instead of Pool.map()"""
    141         return map(func, iterable)
    142 
    143 
    144     def close(self):
    145         pass
    146 
    147 
    148     def join(self):
    149         pass
    150