# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
"""Methods for testing interactions with mozharness."""
import json
import os
import sys
from mozbuild.base
import MozbuildObject
from six
import string_types
here = os.path.abspath(os.path.dirname(__file__))
build = MozbuildObject.from_environment(cwd=here)
sys.path.insert(0, os.path.join(build.topsrcdir,
"testing",
"mozharness"))
from mozharness.base.errors
import BaseErrorList
from mozharness.base.log
import INFO
from mozharness.mozilla.structuredlog
import StructuredOutputParser
from mozharness.mozilla.testing.errors
import HarnessErrorList
def get_mozharness_status(suite, lines, status, formatter=
None, buf=
None):
"""Given list of log lines, determine what the mozharness status would be."""
parser = StructuredOutputParser(
config={
"log_level": INFO},
error_list=BaseErrorList + HarnessErrorList,
strict=
False,
suite_category=suite,
)
if formatter:
parser.formatter = formatter
# Processing the log with mozharness will re-print all the output to stdout
# Since this exact same output has already been printed by the actual test
# run, temporarily redirect stdout to devnull.
buf = buf
or open(os.devnull,
"w")
orig = sys.stdout
sys.stdout = buf
for line
in lines:
parser.parse_single_line(json.dumps(line))
sys.stdout = orig
return parser.evaluate_parser(status)
def filter_action(actions, lines):
if isinstance(actions, string_types):
actions = (actions,)
# pylint --py3k: W1639
return list(filter(
lambda x: x[
"action"]
in actions, lines))
