cros_utils: Add a bug reporting module

This lets us conveniently report bugs/cronjob status/etc.

BUG=b:202424935, b:202425245
TEST=Unit tests

Change-Id: Ic276f6ecc69a4c8c0088143177b17836cee70f97
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/third_party/toolchain-utils/+/3319353
Reviewed-by: Jordan Abrahams <ajordanr@google.com>
Commit-Queue: George Burgess <gbiv@chromium.org>
Tested-by: George Burgess <gbiv@chromium.org>

2 files changed, 228 insertions, 0 deletions

diff --git a/cros_utils/bugs.py b/cros_utils/bugs.py
new file mode 100755
index 00000000..88fb7675
--- /dev/null
+++ b/cros_utils/bugs.py
@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+# Copyright 2021 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.
+
+"""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,
+  })
diff --git a/cros_utils/bugs_test.py b/cros_utils/bugs_test.py
new file mode 100755
index 00000000..03dee64d
--- /dev/null
+++ b/cros_utils/bugs_test.py
@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+# Copyright 2021 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.
+
+# We're testing protected methods, so allow protected access.
+# pylint: disable=protected-access
+
+"""Tests bug filing bits."""
+
+import json
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import bugs
+
+
+class Tests(unittest.TestCase):
+  """Tests for the bugs module."""
+  def testWritingJSONFileSeemsToWork(self):
+    """Tests JSON file writing."""
+    old_x20_path = bugs.X20_PATH
+
+    def restore_x20_path():
+      bugs.X20_PATH = old_x20_path
+
+    self.addCleanup(restore_x20_path)
+
+    with tempfile.TemporaryDirectory() as tempdir:
+      bugs.X20_PATH = tempdir
+      file_path = bugs._WriteBugJSONFile(
+          'ObjectType', {
+              'foo': 'bar',
+              'baz': bugs.WellKnownComponents.CrOSToolchainPublic,
+          })
+
+      self.assertTrue(file_path.startswith(tempdir),
+                      f'Expected {file_path} to start with {tempdir}')
+
+      with open(file_path) as f:
+        self.assertEqual(
+            json.load(f),
+            {
+                'type': 'ObjectType',
+                'value': {
+                    'foo': 'bar',
+                    'baz': int(bugs.WellKnownComponents.CrOSToolchainPublic),
+                },
+            },
+        )
+
+  @patch('bugs._WriteBugJSONFile')
+  def testAppendingToBugsSeemsToWork(self, mock_write_json_file):
+    """Tests AppendToExistingBug."""
+    bugs.AppendToExistingBug(1234, 'hello, world!')
+    mock_write_json_file.assert_called_once_with(
+        'AppendToExistingBugRequest',
+        {
+            'body': 'hello, world!',
+            'bug_id': 1234,
+        },
+    )
+
+  @patch('bugs._WriteBugJSONFile')
+  def testBugCreationSeemsToWork(self, mock_write_json_file):
+    """Tests CreateNewBug."""
+    test_case_additions = (
+        {},
+        {
+            'component_id': bugs.WellKnownComponents.CrOSToolchainPublic,
+        },
+        {
+            'assignee': 'foo@gbiv.com',
+            'cc': ['bar@baz.com'],
+        },
+    )
+
+    for additions in test_case_additions:
+      test_case = {
+          'component_id': 123,
+          'title': 'foo',
+          'body': 'bar',
+          **additions,
+      }
+
+      bugs.CreateNewBug(**test_case)
+
+      expected_output = {
+          'component_id': test_case['component_id'],
+          'subject': test_case['title'],
+          'body': test_case['body'],
+      }
+
+      assignee = test_case.get('assignee')
+      if assignee:
+        expected_output['assignee'] = assignee
+
+      cc = test_case.get('cc')
+      if cc:
+        expected_output['cc'] = cc
+
+      mock_write_json_file.assert_called_once_with(
+          'FileNewBugRequest',
+          expected_output,
+      )
+      mock_write_json_file.reset_mock()
+
+  @patch('bugs._WriteBugJSONFile')
+  def testCronjobLogSendingSeemsToWork(self, mock_write_json_file):
+    """Tests SendCronjobLog."""
+    bugs.SendCronjobLog('my_name', False, 'hello, world!')
+    mock_write_json_file.assert_called_once_with(
+        'ChrotomationCronjobUpdate',
+        {
+            'name': 'my_name',
+            'message': 'hello, world!',
+            'failed': False,
+        },
+    )
+
+
+if __name__ == '__main__':
+  unittest.main()