cros_utils/bugs.py

#!/usr/bin/env python3
# Copyright 2021 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Utilities to file bugs."""

import base64
import datetime
import enum
import json
import os
from typing import Any, Dict, List, Optional


X20_PATH = "/google/data/rw/teams/c-compiler-chrome/prod_bugs"


class WellKnownComponents(enum.IntEnum):
    """A listing of "well-known" components recognized by our infra."""

    CrOSToolchainPublic = -1
    CrOSToolchainPrivate = -2


def _WriteBugJSONFile(object_type: str, json_object: Dict[str, Any]):
    """Writes a JSON file to X20_PATH with the given bug-ish object."""
    final_object = {
        "type": object_type,
        "value": json_object,
    }

    # The name of this has two parts:
    # - An easily sortable time, to provide uniqueness and let our service send
    #   things in the order they were put into the outbox.
    # - 64 bits of entropy, so two racing bug writes don't clobber the same file.
    now = datetime.datetime.utcnow().isoformat("T", "seconds") + "Z"
    entropy = base64.urlsafe_b64encode(os.getrandom(8))
    entropy_str = entropy.rstrip(b"=").decode("utf-8")
    file_path = os.path.join(X20_PATH, f"{now}_{entropy_str}.json")

    temp_path = file_path + ".in_progress"
    try:
        with open(temp_path, "w") as f:
            json.dump(final_object, f)
        os.rename(temp_path, file_path)
    except:
        os.remove(temp_path)
        raise
    return file_path


def AppendToExistingBug(bug_id: int, body: str):
    """Sends a reply to an existing bug."""
    _WriteBugJSONFile(
        "AppendToExistingBugRequest",
        {
            "body": body,
            "bug_id": bug_id,
        },
    )


def CreateNewBug(
    component_id: int,
    title: str,
    body: str,
    assignee: Optional[str] = None,
    cc: Optional[List[str]] = None,
):
    """Sends a request to create a new bug.

    Args:
      component_id: The component ID to add. Anything from WellKnownComponents
        also works.
      title: Title of the bug. Must be nonempty.
      body: Body of the bug. Must be nonempty.
      assignee: Assignee of the bug. Must be either an email address, or a
        "well-known" assignee (detective, mage).
      cc: A list of emails to add to the CC list. Must either be an email
        address, or a "well-known" individual (detective, mage).
    """
    obj = {
        "component_id": component_id,
        "subject": title,
        "body": body,
    }

    if assignee:
        obj["assignee"] = assignee

    if cc:
        obj["cc"] = cc

    _WriteBugJSONFile("FileNewBugRequest", obj)


def SendCronjobLog(cronjob_name: str, failed: bool, message: str):
    """Sends the record of a cronjob to our bug infra.

    cronjob_name: The name of the cronjob. Expected to remain consistent over
      time.
    failed: Whether the job failed or not.
    message: Any seemingly relevant context. This is pasted verbatim in a bug, if
      the cronjob infra deems it worthy.
    """
    _WriteBugJSONFile(
        "ChrotomationCronjobUpdate",
        {
            "name": cronjob_name,
            "message": message,
            "failed": failed,
        },
    )