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>
diff --git a/cros_utils/bugs.py b/cros_utils/bugs.py
new file mode 100755
index 0000000..88fb767
--- /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 0000000..03dee64
--- /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()
