huangcm
2025-07-01 676035278781360996553c427a12bf358249ebf7
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
# Copyright (c) 2014 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
 
"""Mock out test results for puppylab.
"""
 
 
import logging
import os
import time
 
import common
from autotest_lib.client.common_lib import time_utils
 
# TODO(beeps): The right way to create these status logs is by creating a job
# object and invoking job.record on it. However we perform this template
# hack instead for the following reasons:
#   * The templates are a lot easier to understand at first glance, which
#     is really what one wants from a testing interface built for a
#     framework like autotest.
#   * Creating the job is wedged into core autotest code, so it has
#     unintended consequences like checking for hosts/labels etc.
#   * We are guaranteed to create the bare minimum by hand specifying the file
#     to write, and their contents. Job.record ends up checking and creating
#     several non-essential directoris in the process or recording status.
 
_SUCCESS_TEST_TEMPLATE = (
        "\tSTART\t%(test_name)s\t%(test_name)s"
        "\ttimestamp=%(timestamp)s\tlocaltime=%(date)s\n"
        "\t\tGOOD\t%(test_name)s\t%(test_name)s\ttimestamp=%(timestamp)s\t"
        "localtime=%(date)s\tcompleted successfully\n"
        "\tEND GOOD\t%(test_name)s\t%(test_name)s\ttimestamp=%(timestamp)s\t"
        "localtime=%(date)s")
 
 
_SUCCESS_JOB_TEMPLATE = (
        "START\t----\t----\ttimestamp=%(timestamp)s\tlocaltime=%(date)s"
        "\n\tSTART\t%(test_name)s\t%(test_name)s\ttimestamp=%(timestamp)s\t"
        "localtime=%(date)s\n\t\tGOOD\t%(test_name)s\t%(test_name)s"
        "\ttimestamp=%(timestamp)s\tlocaltime=%(date)s\tcompleted "
        "successfully\n\tEND GOOD\t%(test_name)s\t%(test_name)s"
        "\ttimestamp=%(timestamp)s\tlocaltime=%(date)s\n"
        "END GOOD\t----\t----\ttimestamp=%(timestamp)s\tlocaltime=%(date)s")
 
 
_JOB_KEYVALS_TEMPLATE = "hostname=%(hostname)s\nstatus_version=1\n"
 
 
class ResultsMocker(object):
    """Class to mock out the results of a test."""
 
    def _make_dirs(self):
        """Create essential directories needed for faking test results.
 
        @raises ValueError: If the directories crucial to reporting
            test status already exist.
        @raises OSError: If we cannot make one of the directories for
            an os related reason (eg: permissions).
        @raises AssertionError: If one of the directories silently failed
            creation.
        """
        logging.info("creating dir %s, %s, %s",
                self.results_dir, self.test_results, self.host_keyval_dir)
        if not os.path.exists(self.results_dir):
            os.makedirs(self.results_dir)
        if not os.path.exists(self.test_results):
            os.makedirs(self.test_results)
        if not os.path.exists(self.host_keyval_dir):
            os.makedirs(self.host_keyval_dir)
        assert(os.path.exists(self.test_results) and
               os.path.exists(self.results_dir) and
               os.path.exists(self.host_keyval_dir))
 
 
    def __init__(self, test_name, results_dir, machine_name):
        """Initialize a results mocker.
 
        @param test_name: The name of the test, eg: dummy_Pass.
        @param results_dir: The results directory this test will use.
        @param machine_name: A string representing the hostname the test will
            run on.
        """
        self.results_dir = results_dir
        self.test_results = os.path.join(results_dir, test_name)
        self.host_keyval_dir = os.path.join(self.results_dir, 'host_keyvals')
        self.machine_name = machine_name
        self.test_name = test_name
 
        self._make_dirs()
 
        # Status logs are used by the parser to declare a test as pass/fail.
        self.job_status = os.path.join(self.results_dir, 'status')
        self.job_status_log = os.path.join(self.results_dir, 'status.log')
        self.test_status = os.path.join(self.test_results, 'status')
 
        # keyvals are used by the parser to figure out fine grained information
        # about a test. Only job_keyvals are crucial to parsing.
        self.test_keyvals = os.path.join(self.test_results, 'keyval')
        self.job_keyvals = os.path.join(self.results_dir, 'keyval')
        self.host_keyvals = os.path.join(self.results_dir, machine_name)
 
 
    def _write(self, results_path, results):
        """Write the content in results to the file in results_path.
 
        @param results_path: The path to the results file.
        @param results: The content to write to the file.
        """
        logging.info('Writing results to %s', results_path)
        with open(results_path, 'w') as results_file:
            results_file.write(results)
 
 
    def generate_keyvals(self):
        """Apply templates to keyval files.
 
        There are 3 important keyvals files, only one of which is actually
        crucial to results parsing:
            host_keyvals - information about the DUT
            job_keyvals - information about the server_job
            test_keyvals - information about the test
 
        Parsing cannot complete without the job_keyvals. Everything else is
        optional. Keyvals are parsed into tko tables.
        """
        #TODO(beeps): Include other keyvals.
        self._write(
                self.job_keyvals,
                _JOB_KEYVALS_TEMPLATE %
                        {'hostname': self.machine_name})
 
 
    def generate_status(self):
        """Generate status logs.
 
        3 important status logs are required for successful parsing:
            test_name/status - core test status
            results_dir/status - server job status (has test status in it)
            status.log - compiled final status log
        """
        current_timestamp = int(time.time())
        test_info = {
            'test_name': self.test_name,
            'timestamp': current_timestamp,
            'date': time_utils.epoch_time_to_date_string(
                            current_timestamp, fmt_string='%b %d %H:%M:%S'),
        }
        self._write(
                self.job_status,
                _SUCCESS_JOB_TEMPLATE % test_info)
        self._write(
                self.job_status_log,
                _SUCCESS_JOB_TEMPLATE % test_info)
        self._write(
                self.test_status,
                _SUCCESS_TEST_TEMPLATE % test_info)
 
 
    def mock_results(self):
        """Create mock results in the directories used to init the instance."""
        self.generate_status()
        self.generate_keyvals()