blob: 844017f99ef86d2f88f9eca4231395511b7119a9 [file] [log] [blame]
# -*- coding: utf-8 -*-
# Copyright 2017 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.
"""Helper script to classify the results of a special task."""
from __future__ import print_function
import json
import os
from chromite.lib import classifier
from chromite.lib import commandline
from chromite.lib import cros_logging as logging
from chromite.lib import gs
def GetParser():
"""Creates the argparse parser."""
parser = commandline.ArgumentParser(description=__doc__)
parser.add_argument('urls', type=str, nargs='*', action='store',
help='GS URL to special task logs')
parser.add_argument('--input_json', '-i', type=str, action='store',
help='JSON file to use as input.')
parser.add_argument('--key_path', '-k', type=str, action='append',
help='Key path to URLs, specifying all dict keys. '
'All lists are iterated, key path must end in '
'string of URL.')
parser.add_argument('--find_suite_tasks', action='store_true', default=False,
help='Set key path to match find_suite_tasks output. '
'ie (special_tasks, next_entry, gs_url)')
return parser
def ClassifySpecialTask(base_url):
"""Classify a single special task.
base_url: The GS URL to the special task directory.
ctx = gs.GSContext()
STATUS_LOG = 'status.log'
glob_url = os.path.join(base_url, '*', STATUS_LOG)
files = ctx.LS(glob_url)
content = ctx.Cat(files[0])
result = classifier.ClassifyLabJobStatusResult(content.splitlines())'%s: %s', base_url, result)
except gs.GSNoSuchKey:
logging.warning('%s: %s not found', base_url, STATUS_LOG)
def TraverseTree(root, path, func):
"""Traverse a JSON tree.
Allows traversing of the tree root, by iterating any lists encountered,
or following dictionaries according to the keys specified by path.
Leaf nodes execute a specified function.
See unittest for examples of traversal.
root: The list/dict/str to traverse.
path: A list of dictionary keys to traverse when encountered.
func: A function to execute with leaf nodes.
if isinstance(root, list):
for n in root:
TraverseTree(n, path, func)
elif path and len(path):
if not isinstance(root, dict):
logging.error('Unable to find key %s because not a dict, got %s',
path, type(root))
elif path[0] in root:
TraverseTree(root[path[0]], path[1:], func)
if not isinstance(root, basestring):
logging.error('Expecting string to classify, got: %s', type(root))
def main(argv):
parser = GetParser()
options = parser.parse_args(argv)
key_path = options.key_path
if options.find_suite_tasks:
key_path = ['special_tasks', 'next_entry', 'gs_url']
if options.input_json:
with open(options.input_json) as f:
json_file = json.load(f)
TraverseTree(json_file, key_path, ClassifySpecialTask)
for base_url in options.urls: