diff options
Diffstat (limited to 'cros_utils')
-rwxr-xr-x | cros_utils/bugs.py | 104 | ||||
-rwxr-xr-x | cros_utils/bugs_test.py | 124 |
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() |