blob: f92de54a979be4fa8ca4e1a1f4c2ce453edb7819 [file] [log] [blame]
# Copyright (c) 2011 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.
"""Retrieve and process the test summary results.json files.
Specifically, this code enables access to information about crashes
identified during job runs.
__author__ = [' (Mike Truty)',
' (Dale Curtis)']
import commands
import json
import os
import dash_util
# String resources.
from dash_strings import AUTOTEST_ARCHIVE
from dash_strings import AUTOTEST_PATH
from dash_strings import AUTOTEST_SERVER
from dash_strings import CGI_RETRIEVE_LOGS_CMD
from dash_strings import GSUTIL_GET_CMD
from dash_strings import WGET_CMD
LOG_BASE_PATH = '%s/%s/results.json'
class TestSummaryInfo(object):
"""Class to enable retrieval of test summary info files."""
def __init__(self, job_cache_dir):
"""Initialize some job status caches.
There are two caches for job result artifacts: an in-memory one and one
on disk.
_job_cache_dir: contains the file-based on-disk cache of job results files.
_job_summary_cache: an in-memory dictionary of job results for test lookups.
job_cache_dir: base location for the file-based job result cache.
self._job_cache_dir = job_cache_dir
self._job_summary_cache = {}
def _GetJsonFromFileOrString(file_or_string, is_file=True):
"""Helper to convert text retrieved to Json.
file_or_string: filename or string to consume.
is_file: flag to inform logic if open is needed.
Json version of the string (file text).
if is_file:
with open(file_or_string) as f:
summary_json = json.load(f)
summary_json = json.loads(file_or_string)
except ValueError:
# This ValueError raised when json.load(s) finds improper Json text.
summary_json = {}
return summary_json
def _LocalResultFile(self, job_name, base_dir=None, use_json=True):
"""Helper to find and retrieve the results file on a local machine.
The file may be located in a result dir or a cache dir.
job_name: a key to finding the job data under autotest results.
base_dir: used to find the job result file cache. If None, look in
the dashboard job file cache.
use_json: flag to suggest if Json-parsing is needed.
If the file is located:
and use_json=True, then return Json valid version of the file.
and not use_json=True, then return raw file contents.
If file not found, return None.
base_dir = base_dir or self._job_cache_dir
log_file_path = os.path.abspath(LOG_BASE_PATH % (base_dir, job_name))
if os.path.isfile(log_file_path):
if use_json:
return self._GetJsonFromFileOrString(log_file_path)
with open(log_file_path) as f:
return None
def _RetrieveResultsJson(self, job_name):
"""Helper to retrieve the results.json file from a result server.
The tko/retrieve_logs.cgi script handles finding the results server
and/or retrieving results from gs using gsutil.
job_name: used to locate the job-specific result.json.
results_base = os.path.join(AUTOTEST_SERVER, CGI_RETRIEVE_LOGS_CMD)
log_file_path = LOG_BASE_PATH % (results_base, job_name)
return commands.getoutput('%s %s' % (WGET_CMD, log_file_path))
def _UpdateFileCache(self, job_name):
"""Helper to update a job file cache with a results Json file.
This is complicated by the fact that results files may be located
on the local machine, a local autotest server or remotely on a
results server or in Google Storage.
job_name: a key to finding the job data under autotest results.
Json valid version of the file content or None.
summary_text = self._RetrieveResultsJson(job_name)
cache_path = os.path.abspath(LOG_BASE_PATH % (self._job_cache_dir,
dash_util.SaveHTML(cache_path, summary_text)
return self._GetJsonFromFileOrString(summary_text, is_file=False)
def RetrieveTestSummary(self, job_tag, test_name):
"""Retrieves test artifacts from the Autotest server for a given test.
Autotest drops a Json file which contains failed tests, crashes, and log
file snippets in each job results directory. We use this information to
reduce wget usage and find crashes for a given job.
Requests are cached to reduce round-trip time to the server which can be
very substantial.
Extract path to results from tag. Sometimes the test['tag'] is:
<job_name>/<group name>/<host name(s)>
Other times it's just:
<job_name>/<host name>
It depends on how tests were scheduled. Usually, if present, group name
indicates that the job was spread across multiple hosts.
The results.json is always in sub directory under <job_name>.
job_tag: Path under Autotest results to find test result file.
test_name: Used to find previously cached test results.
Json test artifact if it can be loaded from the Autotest server, None
job_name = '/'.join(job_tag.split('/')[0:2])
# Get job summary from in-memory cache, then the actual file or file cache.
if job_name not in self._job_summary_cache:
# Now ensure the file cache is updated since job entry not in memory.
summary_json = self._LocalResultFile(job_name)
if summary_json is None:
summary_json = self._UpdateFileCache(job_name)
self._job_summary_cache[job_name] = summary_json
# Return the results for this test if we have them in the cache.
return self._job_summary_cache[job_name].get(test_name)