| # Copyright 2014 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unit tests for perf_uploader module.""" |
| |
| import json |
| import os |
| import tempfile |
| import urllib.parse |
| import urllib.request |
| |
| from chromite.lib import cros_test_lib |
| from chromite.lib import osutils |
| from chromite.lib import perf_uploader |
| |
| |
| class PerfUploadTestCase(cros_test_lib.MockTestCase): |
| """Base utility class to setup mock objects and temp file for tests.""" |
| |
| def setUp(self): |
| presentation_info = perf_uploader.PresentationInfo( |
| main_name="ChromeOSPerf", |
| test_name="TestName", |
| ) |
| self.PatchObject( |
| perf_uploader, |
| "_GetPresentationInfo", |
| return_value=presentation_info, |
| ) |
| # TODO(b/236161656): Fix. |
| # pylint: disable-next=consider-using-with |
| self.file_name = tempfile.NamedTemporaryFile().name |
| |
| def tearDown(self): |
| osutils.SafeUnlink(self.file_name) |
| |
| |
| class OutputPerfValueTest(PerfUploadTestCase): |
| """Test function OutputPerfValue.""" |
| |
| def testInvalidDescription(self): |
| perf_uploader.OutputPerfValue(self.file_name, "a" * 256, 0, "ignored") |
| self.assertRaises( |
| ValueError, |
| perf_uploader.OutputPerfValue, |
| "ignored", |
| "a" * 257, |
| 0, |
| "ignored", |
| ) |
| |
| perf_uploader.OutputPerfValue( |
| self.file_name, "abcXYZ09-_.", 0, "ignored" |
| ) |
| self.assertRaises( |
| ValueError, |
| perf_uploader.OutputPerfValue, |
| "ignored", |
| "a\x00c", |
| 0, |
| "ignored", |
| ) |
| |
| def testInvalidUnits(self): |
| self.assertRaises( |
| ValueError, |
| perf_uploader.OutputPerfValue, |
| "ignored", |
| "ignored", |
| 0, |
| "a" * 257, |
| ) |
| self.assertRaises( |
| ValueError, |
| perf_uploader.OutputPerfValue, |
| "ignored", |
| "ignored", |
| 0, |
| "a\x00c", |
| ) |
| |
| def testValidJson(self): |
| perf_uploader.OutputPerfValue(self.file_name, "desc", 42, "units") |
| data = osutils.ReadFile(self.file_name) |
| entry = json.loads(data) |
| self.assertIsInstance(entry, dict) |
| |
| |
| class LoadPerfValuesTest(PerfUploadTestCase): |
| """Test function LoadPerfValues.""" |
| |
| def testEmptyFile(self): |
| osutils.WriteFile(self.file_name, "") |
| entries = perf_uploader.LoadPerfValues(self.file_name) |
| self.assertEqual(0, len(entries)) |
| |
| def testLoadOneValue(self): |
| perf_uploader.OutputPerfValue(self.file_name, "desc", 41, "units") |
| entries = perf_uploader.LoadPerfValues(self.file_name) |
| self.assertEqual(1, len(entries)) |
| self.assertEqual(41, entries[0].value) |
| self.assertEqual("desc", entries[0].description) |
| self.assertEqual(True, entries[0].higher_is_better) |
| |
| def testLoadTwoValues(self): |
| perf_uploader.OutputPerfValue(self.file_name, "desc", 41, "units") |
| perf_uploader.OutputPerfValue(self.file_name, "desc2", 42, "units2") |
| entries = perf_uploader.LoadPerfValues(self.file_name) |
| self.assertEqual(2, len(entries)) |
| self.assertEqual(41, entries[0].value) |
| self.assertEqual(42, entries[1].value) |
| self.assertEqual("desc2", entries[1].description) |
| self.assertEqual(None, entries[1].graph) |
| |
| |
| class SendToDashboardTest(PerfUploadTestCase): |
| """Ensure perf values are sent to chromeperf via HTTP.""" |
| |
| def setUp(self): |
| self.urlopen = self.PatchObject(urllib.request, "urlopen") |
| |
| def testOneEntry(self): |
| perf_uploader.OutputPerfValue(self.file_name, "desc1", 42, "unit") |
| perf_values = perf_uploader.LoadPerfValues(self.file_name) |
| perf_uploader.UploadPerfValues( |
| perf_values, |
| "platform", |
| "TestName", |
| cros_version="cros", |
| chrome_version="chrome", |
| ) |
| request = self.urlopen.call_args[0][0] |
| self.assertEqual( |
| os.path.join(perf_uploader.DASHBOARD_URL, "add_point"), |
| request.get_full_url(), |
| ) |
| data = request.data |
| data = urllib.parse.parse_qs(data)[b"data"] |
| entries = [json.loads(x) for x in data] |
| entry = entries[0][0] |
| self.assertEqual( |
| "cros", entry["supplemental_columns"]["r_cros_version"] |
| ) |
| self.assertEqual(42, entry["value"]) |
| self.assertEqual("cbuildbot.TestName/desc1", entry["test"]) |
| self.assertEqual("unit", entry["units"]) |
| |
| def testCustomDashboard(self): |
| """Verify we can set data to different dashboards.""" |
| perf_uploader.OutputPerfValue(self.file_name, "desc1", 42, "unit") |
| perf_values = perf_uploader.LoadPerfValues(self.file_name) |
| perf_uploader.UploadPerfValues( |
| perf_values, |
| "platform", |
| "TestName", |
| cros_version="cros", |
| chrome_version="chrome", |
| dashboard="http://localhost", |
| ) |
| request = self.urlopen.call_args[0][0] |
| self.assertEqual("http://localhost/add_point", request.get_full_url()) |
| |
| |
| class UploadPerfValuesTest(PerfUploadTestCase): |
| """Test UploadPerfValues function.""" |
| |
| def setUp(self): |
| self.send_func = self.PatchObject(perf_uploader, "_SendToDashboard") |
| |
| def testOneEntry(self): |
| """Upload one perf value.""" |
| perf_uploader.OutputPerfValue(self.file_name, "desc1", 42, "unit") |
| perf_values = perf_uploader.LoadPerfValues(self.file_name) |
| perf_uploader.UploadPerfValues( |
| perf_values, |
| "platform", |
| "TestName", |
| cros_version="cros", |
| chrome_version="chrome", |
| ) |
| positional_args, _ = self.send_func.call_args |
| first_param = positional_args[0] |
| data = json.loads(first_param["data"]) |
| self.assertEqual(1, len(data)) |
| entry = data[0] |
| self.assertEqual("unit", entry["units"]) |
| self.assertEqual( |
| "cros", entry["supplemental_columns"]["r_cros_version"] |
| ) |
| self.assertEqual( |
| "chrome", entry["supplemental_columns"]["r_chrome_version"] |
| ) |
| self.assertEqual("cros-platform", entry["bot"]) |
| self.assertEqual(42, entry["value"]) |
| self.assertEqual(0, entry["error"]) |
| |
| def testRevision(self): |
| """Verify revision is accepted over cros/chrome version.""" |
| perf_uploader.OutputPerfValue(self.file_name, "desc1", 42, "unit") |
| perf_values = perf_uploader.LoadPerfValues(self.file_name) |
| perf_uploader.UploadPerfValues( |
| perf_values, "platform", "TestName", revision=12345 |
| ) |
| positional_args, _ = self.send_func.call_args |
| first_param = positional_args[0] |
| data = json.loads(first_param["data"]) |
| entry = data[0] |
| self.assertEqual(12345, entry["revision"]) |
| |
| def testTwoEntriesOfSameTest(self): |
| """Upload one test, two perf values.""" |
| perf_uploader.OutputPerfValue(self.file_name, "desc1", 40, "unit") |
| perf_uploader.OutputPerfValue(self.file_name, "desc1", 42, "unit") |
| perf_values = perf_uploader.LoadPerfValues(self.file_name) |
| perf_uploader.UploadPerfValues( |
| perf_values, |
| "platform", |
| "TestName", |
| cros_version="cros", |
| chrome_version="chrome", |
| ) |
| positional_args, _ = self.send_func.call_args |
| first_param = positional_args[0] |
| data = json.loads(first_param["data"]) |
| self.assertEqual(1, len(data)) |
| entry = data[0] |
| self.assertEqual("unit", entry["units"]) |
| # Average of 40 and 42 |
| self.assertEqual(41, entry["value"]) |
| # Standard deviation sqrt(2) |
| self.assertEqual(1.4142, entry["error"]) |
| |
| def testTwoTests(self): |
| """Upload two tests, one perf value each.""" |
| perf_uploader.OutputPerfValue(self.file_name, "desc1", 40, "unit") |
| perf_uploader.OutputPerfValue(self.file_name, "desc2", 42, "unit") |
| perf_values = perf_uploader.LoadPerfValues(self.file_name) |
| perf_uploader.UploadPerfValues( |
| perf_values, |
| "platform", |
| "TestName", |
| cros_version="cros", |
| chrome_version="chrome", |
| ) |
| positional_args, _ = self.send_func.call_args |
| first_param = positional_args[0] |
| data = json.loads(first_param["data"]) |
| self.assertEqual(2, len(data)) |
| data = sorted(data, key=lambda x: x["test"]) |
| entry = data[0] |
| self.assertEqual(40, entry["value"]) |
| self.assertEqual(0, entry["error"]) |
| entry = data[1] |
| self.assertEqual(42, entry["value"]) |
| self.assertEqual(0, entry["error"]) |
| |
| def testTwoTestsThreeEntries(self): |
| """Upload two tests, one perf value each.""" |
| perf_uploader.OutputPerfValue(self.file_name, "desc1", 40, "unit") |
| perf_uploader.OutputPerfValue(self.file_name, "desc1", 42, "unit") |
| perf_uploader.OutputPerfValue(self.file_name, "desc2", 42, "unit") |
| perf_values = perf_uploader.LoadPerfValues(self.file_name) |
| perf_uploader.UploadPerfValues( |
| perf_values, |
| "platform", |
| "TestName", |
| cros_version="cros", |
| chrome_version="chrome", |
| ) |
| positional_args, _ = self.send_func.call_args |
| first_param = positional_args[0] |
| data = json.loads(first_param["data"]) |
| self.assertEqual(2, len(data)) |
| data = sorted(data, key=lambda x: x["test"]) |
| entry = data[0] |
| self.assertEqual(41, entry["value"]) |
| self.assertEqual(1.4142, entry["error"]) |
| entry = data[1] |
| self.assertEqual(42, entry["value"]) |
| self.assertEqual(0, entry["error"]) |
| |
| def testDryRun(self): |
| """Make sure dryrun mode doesn't upload.""" |
| self.send_func.side_effect = AssertionError("dryrun should not upload") |
| perf_uploader.OutputPerfValue(self.file_name, "desc1", 40, "unit") |
| perf_values = perf_uploader.LoadPerfValues(self.file_name) |
| perf_uploader.UploadPerfValues( |
| perf_values, |
| "platform", |
| "TestName", |
| cros_version="cros", |
| chrome_version="chrome", |
| dry_run=True, |
| ) |
