• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
8import common
9from autotest_lib.client.common_lib import control_data
10
11
12class 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
36class 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
47class 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
63class FakeLabel(object):
64    """Faked out RPC-client-side Label object."""
65    def __init__(self, id=0):
66        self.id = id
67
68
69class 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
130class 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