metadata/test_common.star

# Copyright 2020 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.

load("@proto//chromiumos/config/api/test/harness/tnull/v1/tnull.proto",
    tnull_pb = "chromiumos.config.api.test.tnull.v1"
)

# Names from this lib: ArchiveArtifactRequest, ArchiveArtifactResponse,
#   ProgressSinkClientConfig, ReportLogRequest, ReportLogResponse,
#   ReportResultRequest, ReportResultResponse, Result
load("@proto//chromiumos/config/api/test/rtd/v1/progress.proto",
    progress_pb = "chromiumos.config.api.test.rtd.v1"
)

load("@proto//chromiumos/config/api/test/metadata/v1/metadata.proto",
    metadata_pb = "chromiumos.config.api.test.metadata.v1"
)

load("@proto//chromiumos/config/api/test/results/v2/result.proto",
    result_pb = "chromiumos.config.api.test.results.v2"
)

load("@proto//google/protobuf/struct.proto", google_pb = "google.protobuf")

_TEST_NAME_PREFIX = "remoteTestDrivers/tnull/tests/"


def _define_test(
    test_name,
    purpose,
    doc,
    attrs = None,
    owner_emails = None,
    owner_groups = None,
    setup = None,
    steps = None,
):
    if not setup:
      setup = _define_setup()
    contacts = (
        [metadata_pb.Contact(email = e) for e in (owner_emails or [])]
        + [metadata_pb.Contact(mdb_group = g) for g in (owner_groups or [])]
    )
    step_proto = tnull_pb.Steps(
        setup = setup,
        steps = [_define_step(step) for step in (steps or [])],
    )
    step_struct = proto.from_jsonpb(
        google_pb.Struct,
        proto.to_jsonpb(step_proto)
    )
    details = google_pb.Struct(fields = {
        "purpose": google_pb.Value(string_value = purpose),
        "doc": google_pb.Value(string_value = doc),
        "steps": google_pb.Value(struct_value=step_struct),
    })
    info = metadata_pb.Informational(
        authors = contacts,
        details = details,
    )
    return metadata_pb.Test(
        name = _TEST_NAME_PREFIX + test_name,
        attributes = [metadata_pb.Attribute(name = a) for a in (attrs or [])],
        informational = info,
    )


def _define_step(step):
  """
  Args:
    step, dict with the format
      {
        "method": <string naming the step type>,
        "args": {
            "common": { <dict of CommonArgs> },
            <other arguments to pass to the step proto's constructor>
        },
      }
      The method field is required; args is optional, and common is optional even
      if args is present.
  Returns:
    a tnull_pb.Step object
  """
  args = step.get("args", {})
  step_constructor = {
  # case step["method"]:
      "archive": tnull_pb.ArchiveStep,
      "log": tnull_pb.LogStep,
      "result": tnull_pb.ResultStep,
  }.get(step.get("method"),
  # default:
      None
  )
  if step_constructor == None:
      return tnull_pb.Step(other=tnull_pb.UnknownStep(args=args))

  common = tnull_pb.CommonArgSet(**args.pop("common_args", {}))
  substep = step_constructor(common_args=common, **args)
  return tnull_pb.Step(**{step["method"]: substep})


simple_success_result = result_pb.Result(state=result_pb.Result.SUCCEEDED)

def _define_setup(sink_config = None, setup_config = None):
  """
  Args:
    sink_config: progress_pb.ProgressSinkClientConfig object
    setup_config: dict with arguments for results/logs/artifacts to include
  Returns:
    tnull_pb.SetupStep object
  """
  if not sink_config:
    sink_config = progress_pb.ProgressSinkClientConfig()
  if not setup_config:
    setup_config = {}
  return tnull_pb.SetupStep(
      config = sink_config,
      result = simple_success_result,
  )


def _make_result(result):
  """
  Args:
    result, dict with the format:
    {
      "skipped": boolean, default false; if true, treat error-free result as
        SKIPPED rather than as SUCCEEDED.
      "errors": list of structs, each specifying an error with source
        and severity.
    }
  Returns:
    progress_pb.Result object
  """
  if result == None:
    return simple_success_result

  errs = _format_errors(result.get("errors", []))
  if result.get("skipped", False):
    state = result_pb.Result.SKIPPED
  elif len(errs) == 0:
    state = result_pb.Result.SUCCEEDED
  elif any([err.severity == result_pb.Result.Error.CRITICAL for err in errs]):
    state = result_pb.Result.FAILED
  else:
    state = result_pb.Result.SUCCEEDED
  return result_pb.Result(state=state, errors=errs)


def _format_errors(err_structs):
  """
  Args:
    err_structs: list of struct(source= <str>, severity= <str>)
  Returns:
    list of progress_pb.Result.Error objects
  """
  fixed_up_structs = [_fixup_error(s) for s in err_structs]
  return [_make_error(s.source, s.severity) for s in fixed_up_structs]


def _fixup_error(err_struct):
  if not getattr(err_struct, "source", None):
    fail("Invalid args for error: %s" % err_struct.to_json)
  sev = getattr(err_struct, "severity", None)
  return struct(source = err_struct.source, severity = sev)


def _make_error(source, severity):
  """
  Args:
   source: one of ["test", "rtd", "tls"]
   severity: one of ["error", "critical", "warn"]
  Returns:
    progress_pb.Error object
  """
  src = _get_source(source)
  if src == None:
    return fail("Error source unspecified")
  return result_pb.Result.Error(
      source = src,
      severity = _get_severity(severity)
  )


# map string or int to the appropriate constant. Default to None.
def _get_source(key):
  sources = {
      "rtd":  result_pb.Result.Error.RTD,
      "test": result_pb.Result.Error.TEST,
      "tls":  result_pb.Result.Error.TEST_LAB_SERVICES,
  }
  return sources.get(key)


# map string or int to the appropriate constant. Default to critical.
def _get_severity(key):
  severities = {
      "critical": result_pb.Result.Error.CRITICAL,
      "error":    result_pb.Result.Error.CRITICAL,
      "warn":     result_pb.Result.Error.WARNING,
      "warning":  result_pb.Result.Error.WARNING,
  }
  return severities.get(key, result_pb.Result.Error.WARNING)


test_common = struct(
    define_test = _define_test,
    define_setup = _define_setup,
)