blob: 74c923de82aeec267ef80fb92375205e0a4a071a [file] [log] [blame]
# Copyright 2020 The LUCI Authors. All rights reserved.
# Use of this source code is governed under the Apache License, Version 2.0
# that can be found in the LICENSE file.
import errno
import re
from datetime import datetime
from io import open
from google.protobuf import text_format as textpb
# The relative path to the file where all warnings are defined under recipe
# path (where the "recipes" and/or "recipe_modules" directories sit)
def parse_warning_definitions(file_path):
"""Parse the warning definition file at the given absolute path. The file
content is expected to be in text proto format of warning.DefinitionCollection
proto message. Duplicate warning names will be raised. Each warning definition
will be validated. The conditions are documented in warning.proto.
* file_path (str) - Absolute path to warning definition file
Returns a dict of warning name to warning.Definition proto message instance
raw_text = ''
with open(file_path, encoding='utf-8') as f:
raw_text =
except IOError as ex:
if ex.errno == errno.ENOENT:
# No warning defined
return {}
raise ex
from PB.recipe_engine.warning import DefinitionCollection
definition_collection = textpb.Parse(raw_text, DefinitionCollection())
definitions = list(definition_collection.warning)
if definition_collection.HasField('monorail_bug_default'):
ret = {}
for definition in definitions:
if in ret:
raise ValueError(
'Found warning definitions with duplicate name: %s' %
ret[] = definition
return ret
def _populate_bug_issue_fields(definitions, monorail_bug_default,
"""If default field value has been declared for bugs/issues, run through all
bugs/issues declared in all warning definitions and assign default value to
fields which are unset.
* definitions (list of warning.Definition)
* monorail_bug_default (warning.MonorailBugDefault): contains the default
value for some fields (namely, host and project) in warning.MonorailBug
for definition in definitions:
for bug in definition.monorail_bug: = or
bug.project = bug.project or monorail_bug_default.project
for iss in definition.google_issue: = or
def _validate(definition):
"""Ensure the given warning definition is valid. ValueError will be
raised otherwise. All conditions are documented in warning.proto.
* definition (warning.Definition proto)
if not re.match(r"^[A-Z][A-Z0-9]*(\_[A-Z0-9]+)*$",
raise ValueError(
'Expect warning name to be all CAPS or num snake case. Actual: %s' % (
if definition.deadline:
datetime.strptime(definition.deadline, '%Y-%m-%d')
except ValueError:
raise ValueError(
'The deadline should be in YYYY-MM-DD format. Actual: %s' % (
for bug in definition.monorail_bug:
err_msg_template = 'Field: %s is required; Got empty value'
_require_non_zero_value(, err_msg_template % 'host')
_require_non_zero_value(bug.project, err_msg_template % 'project')
_require_non_zero_value(, err_msg_template % 'id')
for iss in definition.google_issue:
err_msg_template = 'Field: %s is required; Got empty value'
_require_non_zero_value(, err_msg_template % 'host')
_require_non_zero_value(, err_msg_template % 'id')
def _require_non_zero_value(value, message):
"""Raise ValueError with message if the supplied value is a zero value
if not value:
raise ValueError(message)