Display complete pinpoint server error message
This CL adds a new module `requests` for wrapping authed session
requests. In case of a server response error, this module tries to get
an error message from `response.json()` and add it to the
`raise_for_status()` error message. Other modules and unit tests were
simplified as a result.
Bug: b/463582730
Change-Id: I94ec0ffb31917b1e7906a0905f7d9fc23d2e729a
Reviewed-on: https://chromium-review.googlesource.com/c/crossbench/+/7209260
Reviewed-by: Patrick Thier <pthier@chromium.org>
Commit-Queue: Maxim Sheshukov <maximsheshukov@google.com>
diff --git a/crossbench/pinpoint/cancel_job.py b/crossbench/pinpoint/cancel_job.py
index 90cbccc..037a605 100644
--- a/crossbench/pinpoint/cancel_job.py
+++ b/crossbench/pinpoint/cancel_job.py
@@ -6,19 +6,18 @@
import json
+from crossbench.pinpoint import requests
from crossbench.pinpoint.api import PINPOINT_CANCEL_JOB_API_URL
-from crossbench.pinpoint.auth import get_auth_session
from crossbench.pinpoint.helper import annotate
def cancel_job(job_id: str, reason: str) -> None:
"""Cancels a Pinpoint job."""
- authed_session = get_auth_session()
payload = {
"job_id": job_id,
"reason": reason,
}
with annotate("Cancelling Pinpoint job"):
- response = authed_session.post(PINPOINT_CANCEL_JOB_API_URL, data=payload)
+ response = requests.post(PINPOINT_CANCEL_JOB_API_URL, data=payload)
response.raise_for_status()
- print(json.dumps(response.json(), indent=2))
+ print(json.dumps(response.json(), indent=2))
diff --git a/crossbench/pinpoint/job_config.py b/crossbench/pinpoint/job_config.py
index 6e552fa..0f8673d 100644
--- a/crossbench/pinpoint/job_config.py
+++ b/crossbench/pinpoint/job_config.py
@@ -7,21 +7,20 @@
import json
from typing import Any
+from crossbench.pinpoint import requests
from crossbench.pinpoint.api import PINPOINT_JOB_API_URL_TEMPLATE
-from crossbench.pinpoint.auth import get_auth_session
from crossbench.pinpoint.config import PinpointTryJobConfig
from crossbench.pinpoint.helper import annotate
def fetch_job_config(job_id: str, full: bool = False) -> dict[str, Any]:
"""Fetches the configuration for a specific Pinpoint job."""
- authed_session = get_auth_session()
url = PINPOINT_JOB_API_URL_TEMPLATE.format(job_id=job_id)
params = {}
if full:
params["o"] = ["STATE", "ESTIMATE"]
with annotate("Fetching Job Config"):
- response = authed_session.get(url, params=params)
+ response = requests.get(url, params=params)
response.raise_for_status()
return response.json()
diff --git a/crossbench/pinpoint/list_benchmarks.py b/crossbench/pinpoint/list_benchmarks.py
index fdcc8c0..8781357 100644
--- a/crossbench/pinpoint/list_benchmarks.py
+++ b/crossbench/pinpoint/list_benchmarks.py
@@ -4,15 +4,14 @@
from __future__ import annotations
+from crossbench.pinpoint import requests
from crossbench.pinpoint.api import CHROMEPERF_TEST_SUITES_API_URL
-from crossbench.pinpoint.auth import get_auth_session
from crossbench.pinpoint.helper import annotate
def fetch_benchmarks() -> list[str]:
"""Fetches the list of available benchmarks from the Chromeperf API."""
- authed_session = get_auth_session()
with annotate("Fetching benchmarks"):
- response = authed_session.post(CHROMEPERF_TEST_SUITES_API_URL)
+ response = requests.post(CHROMEPERF_TEST_SUITES_API_URL)
response.raise_for_status()
return response.json()
diff --git a/crossbench/pinpoint/list_bots.py b/crossbench/pinpoint/list_bots.py
index a03818b..78f09b5 100644
--- a/crossbench/pinpoint/list_bots.py
+++ b/crossbench/pinpoint/list_bots.py
@@ -4,15 +4,14 @@
from __future__ import annotations
+from crossbench.pinpoint import requests
from crossbench.pinpoint.api import PINPOINT_CONFIG_API_URL
-from crossbench.pinpoint.auth import get_auth_session
from crossbench.pinpoint.helper import annotate
def fetch_bots() -> list[str]:
"""Fetches the list of available Pinpoint bots from the Pinpoint API."""
- authed_session = get_auth_session()
with annotate("Fetching available bots"):
- response = authed_session.post(PINPOINT_CONFIG_API_URL)
+ response = requests.post(PINPOINT_CONFIG_API_URL)
response.raise_for_status()
return response.json()["configurations"]
diff --git a/crossbench/pinpoint/list_builds.py b/crossbench/pinpoint/list_builds.py
index 26a1e15..db532fb 100644
--- a/crossbench/pinpoint/list_builds.py
+++ b/crossbench/pinpoint/list_builds.py
@@ -11,8 +11,8 @@
from tabulate import tabulate
+from crossbench.pinpoint import requests
from crossbench.pinpoint.api import PINPOINT_BUILDS_API_URL_TEMPLATE
-from crossbench.pinpoint.auth import get_auth_session
from crossbench.pinpoint.format_time import DATETIME_FORMAT, format_time
from crossbench.pinpoint.helper import annotate
@@ -37,11 +37,10 @@
def fetch_builds(bot: str) -> list[Build]:
- authed_session = get_auth_session()
url = PINPOINT_BUILDS_API_URL_TEMPLATE.format(bot=bot)
with annotate(f"Fetching recent builds for '{bot}'"):
- response = authed_session.get(url)
+ response = requests.get(url)
response.raise_for_status()
return _convert_json_to_builds(response.json())
diff --git a/crossbench/pinpoint/list_jobs.py b/crossbench/pinpoint/list_jobs.py
index 48598d8..3b0bad6 100644
--- a/crossbench/pinpoint/list_jobs.py
+++ b/crossbench/pinpoint/list_jobs.py
@@ -10,54 +10,42 @@
import json
import sys
from concurrent.futures import ThreadPoolExecutor
-from typing import TYPE_CHECKING, Any, Final, Mapping
+from typing import Any, Final, Mapping
-import requests
import yaml
from tabulate import tabulate
+from crossbench.pinpoint import requests
from crossbench.pinpoint.api import JOB_SHORTEN_URL_TEMPLATE, \
PINPOINT_JOBS_API_URL, USERINFO_API_URL
-from crossbench.pinpoint.auth import get_auth_session
from crossbench.pinpoint.format_time import DATETIME_FORMAT, format_time
from crossbench.pinpoint.helper import annotate
from crossbench.pinpoint.list_format import ListFormatEnum
from crossbench.pinpoint.user import UserEnum
-if TYPE_CHECKING:
- from google.auth.transport import requests as auth_requests
-
def list_jobs(user: UserEnum | str, number: int, truncate: int | None,
output_format: ListFormatEnum) -> None:
- authed_session = get_auth_session()
+ emails_to_query = _fetch_user_emails(user)
- try:
- emails_to_query = _fetch_user_emails(authed_session, user)
+ jobs = []
+ with annotate("Fetching jobs"), ThreadPoolExecutor() as executor:
+ results = executor.map(lambda email: _fetch_jobs(number, email),
+ emails_to_query)
+ jobs = list(itertools.chain.from_iterable(results))
- jobs = []
- with annotate("fetching jobs"), ThreadPoolExecutor() as executor:
- results = executor.map(
- lambda email: _fetch_jobs(authed_session, number, email),
- emails_to_query)
- jobs = list(itertools.chain.from_iterable(results))
+ jobs.sort(key=lambda job: job.get("created", ""), reverse=True)
- jobs.sort(key=lambda job: job.get("created", ""), reverse=True)
+ if not jobs:
+ print("No jobs found.")
+ return
- if not jobs:
- print("No jobs found.")
- return
-
- _display_jobs(jobs[:number], output_format, user == UserEnum.ALL, truncate)
-
- except requests.exceptions.RequestException as e:
- print(f"An error occurred while fetching jobs: {e}")
+ _display_jobs(jobs[:number], output_format, user == UserEnum.ALL, truncate)
-def _fetch_user_emails(authed_session: auth_requests.AuthorizedSession,
- user: UserEnum | str) -> set[str | None]:
+def _fetch_user_emails(user: UserEnum | str) -> set[str | None]:
if user == UserEnum.ME:
- email = _get_user_email(authed_session)
+ email = _get_user_email()
username = email.split("@")[0]
return {email, f"{username}@google.com", f"{username}@chromium.org"}
if user == UserEnum.ALL:
@@ -65,16 +53,14 @@
return {user}
-def _get_user_email(authed_session: auth_requests.AuthorizedSession) -> str:
- with annotate("fetching user-email"):
- response = authed_session.get(USERINFO_API_URL)
+def _get_user_email() -> str:
+ with annotate("Fetching user-email"):
+ response = requests.get(USERINFO_API_URL)
response.raise_for_status()
return response.json()["email"]
-def _fetch_jobs(authed_session: auth_requests.AuthorizedSession,
- number: int,
- email: str | None = None) -> list[dict[str, Any]]:
+def _fetch_jobs(number: int, email: str | None = None) -> list[dict[str, Any]]:
jobs = []
next_cursor = None
params = {}
@@ -85,7 +71,7 @@
if next_cursor:
params["next_cursor"] = next_cursor
- response = authed_session.get(PINPOINT_JOBS_API_URL, params=params)
+ response = requests.get(PINPOINT_JOBS_API_URL, params=params)
response.raise_for_status()
data = response.json()
jobs.extend(data.get("jobs", []))
diff --git a/crossbench/pinpoint/list_stories.py b/crossbench/pinpoint/list_stories.py
index 0702833..84ff115 100644
--- a/crossbench/pinpoint/list_stories.py
+++ b/crossbench/pinpoint/list_stories.py
@@ -4,17 +4,16 @@
from __future__ import annotations
+from crossbench.pinpoint import requests
from crossbench.pinpoint.api import CHROMEPERF_DESCRIBE_API_URL
-from crossbench.pinpoint.auth import get_auth_session
from crossbench.pinpoint.helper import annotate
def fetch_stories(benchmark_name: str) -> list[str]:
"""Fetches the list of available stories for a given benchmark."""
- authed_session = get_auth_session()
params = {"test_suite": benchmark_name, "master": "ChromiumPerf"}
with annotate(f"Fetching stories for benchmark '{benchmark_name}'"):
- response = authed_session.post(CHROMEPERF_DESCRIBE_API_URL, params=params)
+ response = requests.post(CHROMEPERF_DESCRIBE_API_URL, params=params)
response.raise_for_status()
data = response.json()
return data.get("cases", [])
diff --git a/crossbench/pinpoint/patch_resolver.py b/crossbench/pinpoint/patch_resolver.py
index b4c712d..091f12f 100644
--- a/crossbench/pinpoint/patch_resolver.py
+++ b/crossbench/pinpoint/patch_resolver.py
@@ -8,7 +8,7 @@
import re
from urllib.parse import urlparse
-from crossbench.pinpoint.auth import get_auth_session
+from crossbench.pinpoint import requests
def resolve_patch(patch: str) -> str:
@@ -43,9 +43,7 @@
def _revision_to_url(change_id: str, patchset: str | None = None) -> str:
chromium_review_url = "https://chromium-review.googlesource.com"
- authed_session = get_auth_session()
- api_response = authed_session.get(chromium_review_url +
- f"/changes/{change_id}")
+ api_response = requests.get(chromium_review_url + f"/changes/{change_id}")
api_response.raise_for_status()
# Skip the protective prefix ")]}'"
diff --git a/crossbench/pinpoint/requests.py b/crossbench/pinpoint/requests.py
new file mode 100644
index 0000000..6eb344e
--- /dev/null
+++ b/crossbench/pinpoint/requests.py
@@ -0,0 +1,47 @@
+# Copyright 2025 The Chromium Authors
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+from __future__ import annotations
+
+import json
+
+import requests
+
+from crossbench.pinpoint import auth
+
+
+def get(url: str, **kwargs) -> requests.Response:
+ return _method("GET", url, **kwargs)
+
+
+def post(url: str, **kwargs) -> requests.Response:
+ return _method("POST", url, **kwargs)
+
+
+class ServerError(requests.exceptions.HTTPError):
+
+ def __init__(self, error: requests.exceptions.HTTPError) -> None:
+ error_message = ""
+ try:
+ data = error.response.json()
+ if error_text := data.get("error"):
+ error_message = f"\n{error_text}"
+ else:
+ error_message = f"\n{data}"
+ except json.JSONDecodeError:
+ pass
+ super().__init__(str(error) + error_message, response=error.response)
+
+
+def _method(method: str, url: str, **kwargs) -> requests.Response:
+ response = auth.get_auth_session().request(method, url, **kwargs)
+ _raise_for_status(response)
+ return response
+
+
+def _raise_for_status(response: requests.Response) -> None:
+ try:
+ response.raise_for_status()
+ except requests.exceptions.HTTPError as e:
+ raise ServerError(e) from e
diff --git a/crossbench/pinpoint/start_job.py b/crossbench/pinpoint/start_job.py
index 87e9cf3..e108178 100644
--- a/crossbench/pinpoint/start_job.py
+++ b/crossbench/pinpoint/start_job.py
@@ -7,8 +7,8 @@
import json
from typing import TYPE_CHECKING
+from crossbench.pinpoint import requests
from crossbench.pinpoint.api import PINPOINT_START_JOB_API_URL
-from crossbench.pinpoint.auth import get_auth_session
from crossbench.pinpoint.helper import annotate
if TYPE_CHECKING:
@@ -17,9 +17,8 @@
def start_job(config: PinpointTryJobConfig) -> None:
"""Starts a new Pinpoint job."""
- authed_session = get_auth_session()
with annotate("Starting Pinpoint job"):
- response = authed_session.post(
+ response = requests.post(
PINPOINT_START_JOB_API_URL, data=config.to_request_dict())
response.raise_for_status()
print(json.dumps(response.json(), indent=2))
diff --git a/tests/crossbench/pinpoint/auth_session_mixin.py b/tests/crossbench/pinpoint/auth_session_mixin.py
deleted file mode 100644
index f647625..0000000
--- a/tests/crossbench/pinpoint/auth_session_mixin.py
+++ /dev/null
@@ -1,25 +0,0 @@
-# Copyright 2025 The Chromium Authors
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-from __future__ import annotations
-
-import unittest
-from typing import Final
-from unittest import mock
-
-import requests
-
-
-class MockAuthSessionMixin(unittest.TestCase):
- """Mixin to mock the get_auth_session function."""
-
- _get_auth_session_patch_target: Final[
- str] = "crossbench.pinpoint.auth.get_auth_session"
-
- def setUp(self):
- super().setUp()
- self.mock_get_auth_session = self.enterContext(
- mock.patch(self._get_auth_session_patch_target))
- self.mock_session = mock.Mock(spec=requests.Session)
- self.mock_get_auth_session.return_value = self.mock_session
diff --git a/tests/crossbench/pinpoint/requests_mixin.py b/tests/crossbench/pinpoint/requests_mixin.py
new file mode 100644
index 0000000..9b95d74
--- /dev/null
+++ b/tests/crossbench/pinpoint/requests_mixin.py
@@ -0,0 +1,19 @@
+# Copyright 2025 The Chromium Authors
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+from __future__ import annotations
+
+import unittest
+from unittest import mock
+
+
+class MockRequestsMixin(unittest.TestCase):
+ """Mixin to mock the requests.get and requests.post functions."""
+
+ def setUp(self):
+ super().setUp()
+ self.mock_get = self.enterContext(
+ mock.patch("crossbench.pinpoint.requests.get"))
+ self.mock_post = self.enterContext(
+ mock.patch("crossbench.pinpoint.requests.post"))
diff --git a/tests/crossbench/pinpoint/test_config.py b/tests/crossbench/pinpoint/test_config.py
index 2df38e3..d9e8fc1 100644
--- a/tests/crossbench/pinpoint/test_config.py
+++ b/tests/crossbench/pinpoint/test_config.py
@@ -12,7 +12,7 @@
from crossbench.pinpoint.config import PinpointTryJobConfig, VariantConfig
from crossbench.pinpoint.list_builds import Build
from tests import test_helper
-from tests.crossbench.pinpoint.auth_session_mixin import MockAuthSessionMixin
+from tests.crossbench.pinpoint.requests_mixin import MockRequestsMixin
_TEST_PATCH: Final[
str] = "https://chromium-review.googlesource.com/c/crossbench/+/12345"
@@ -20,7 +20,7 @@
str] = "https://chromium-review.googlesource.com/c/crossbench/+/111/2"
-class VariantConfigTest(MockAuthSessionMixin):
+class VariantConfigTest(MockRequestsMixin):
_get_auth_session_patch_target = "crossbench.pinpoint.auth.get_auth_session"
def setUp(self):
@@ -135,7 +135,7 @@
})
-class PinpointTryJobConfigTest(MockAuthSessionMixin):
+class PinpointTryJobConfigTest(MockRequestsMixin):
_get_auth_session_patch_target = "crossbench.pinpoint.auth.get_auth_session"
def setUp(self):
diff --git a/tests/crossbench/pinpoint/test_job_config.py b/tests/crossbench/pinpoint/test_job_config.py
index 2bd9975..d8d4de3 100644
--- a/tests/crossbench/pinpoint/test_job_config.py
+++ b/tests/crossbench/pinpoint/test_job_config.py
@@ -4,7 +4,6 @@
from __future__ import annotations
-from typing import Final
from unittest import mock
import requests
@@ -15,7 +14,7 @@
from crossbench.pinpoint.job_config import convert_job_config, \
fetch_job_config, print_job_config
from tests import test_helper
-from tests.crossbench.pinpoint.auth_session_mixin import MockAuthSessionMixin
+from tests.crossbench.pinpoint.requests_mixin import MockRequestsMixin
_TEST_JOB_ID = "1234567890"
_TEST_JOB_RESPONSE = {
@@ -40,9 +39,7 @@
}
-class JobConfigTest(MockAuthSessionMixin):
- _get_auth_session_patch_target: Final[
- str] = "crossbench.pinpoint.job_config.get_auth_session"
+class JobConfigTest(MockRequestsMixin):
def setUp(self):
super().setUp()
@@ -56,7 +53,7 @@
mock_response.raise_for_status.return_value = None
return mock_response
- self.mock_session.get.side_effect = mock_post_side_effect
+ self.mock_get.side_effect = mock_post_side_effect
def test_fetch_job_config(self):
config_dict = fetch_job_config(_TEST_JOB_ID)
@@ -64,18 +61,15 @@
def test_fetch_job_config_full(self):
fetch_job_config(_TEST_JOB_ID, full=True)
- _, kwargs = self.mock_session.get.call_args
+ _, kwargs = self.mock_get.call_args
self.assertEqual(kwargs, {"params": {"o": ["STATE", "ESTIMATE"]}})
def test_fetch_job_config_api_error(self):
- self.mock_session.get.side_effect = requests.exceptions.HTTPError(
- "API Error")
+ self.mock_get.side_effect = requests.exceptions.HTTPError("API Error")
with self.assertRaises(MultiException):
fetch_job_config(_TEST_JOB_ID)
- self.mock_get_auth_session.assert_called_once()
-
def test_convert_job_config(self):
config_dict = fetch_job_config(job_id=_TEST_JOB_ID)
converted_config_dict = convert_job_config(config_dict)
diff --git a/tests/crossbench/pinpoint/test_list_benchmarks.py b/tests/crossbench/pinpoint/test_list_benchmarks.py
index 9920a56..954a603 100644
--- a/tests/crossbench/pinpoint/test_list_benchmarks.py
+++ b/tests/crossbench/pinpoint/test_list_benchmarks.py
@@ -4,7 +4,6 @@
from __future__ import annotations
-from typing import Final
from unittest import mock
import requests
@@ -13,12 +12,10 @@
from crossbench.pinpoint.api import CHROMEPERF_TEST_SUITES_API_URL
from crossbench.pinpoint.list_benchmarks import fetch_benchmarks
from tests import test_helper
-from tests.crossbench.pinpoint.auth_session_mixin import MockAuthSessionMixin
+from tests.crossbench.pinpoint.requests_mixin import MockRequestsMixin
-class ListBenchmarksTest(MockAuthSessionMixin):
- _get_auth_session_patch_target: Final[
- str] = "crossbench.pinpoint.list_benchmarks.get_auth_session"
+class ListBenchmarksTest(MockRequestsMixin):
def setUp(self):
super().setUp()
@@ -34,26 +31,21 @@
mock_response.raise_for_status.return_value = None
return mock_response
- self.mock_session.post.side_effect = mock_post_side_effect
+ self.mock_post.side_effect = mock_post_side_effect
def test_fetch_benchmarks(self):
benchmarks = fetch_benchmarks()
self.assertEqual(benchmarks, ["benchmark1", "benchmark2", "benchmark3"])
- self.mock_get_auth_session.assert_called_once()
- self.mock_session.post.assert_called_once_with(
- CHROMEPERF_TEST_SUITES_API_URL)
+ self.mock_post.assert_called_once_with(CHROMEPERF_TEST_SUITES_API_URL)
def test_fetch_benchmarks_api_error(self):
- self.mock_session.post.side_effect = requests.exceptions.HTTPError(
- "API Error")
+ self.mock_post.side_effect = requests.exceptions.HTTPError("API Error")
with self.assertRaises(MultiException):
fetch_benchmarks()
- self.mock_get_auth_session.assert_called_once()
- self.mock_session.post.assert_called_once_with(
- CHROMEPERF_TEST_SUITES_API_URL)
+ self.mock_post.assert_called_once_with(CHROMEPERF_TEST_SUITES_API_URL)
if __name__ == "__main__":
diff --git a/tests/crossbench/pinpoint/test_list_bots.py b/tests/crossbench/pinpoint/test_list_bots.py
index a4275af..733dfad 100644
--- a/tests/crossbench/pinpoint/test_list_bots.py
+++ b/tests/crossbench/pinpoint/test_list_bots.py
@@ -4,7 +4,6 @@
from __future__ import annotations
-from typing import Final
from unittest import mock
import requests
@@ -13,12 +12,10 @@
from crossbench.pinpoint.api import PINPOINT_CONFIG_API_URL
from crossbench.pinpoint.list_bots import fetch_bots
from tests import test_helper
-from tests.crossbench.pinpoint.auth_session_mixin import MockAuthSessionMixin
+from tests.crossbench.pinpoint.requests_mixin import MockRequestsMixin
-class ListBotsTest(MockAuthSessionMixin):
- _get_auth_session_patch_target: Final[
- str] = "crossbench.pinpoint.list_bots.get_auth_session"
+class ListBotsTest(MockRequestsMixin):
def setUp(self):
super().setUp()
@@ -34,24 +31,21 @@
mock_response.raise_for_status.return_value = None
return mock_response
- self.mock_session.post.side_effect = mock_post_side_effect
+ self.mock_post.side_effect = mock_post_side_effect
def test_fetch_bots(self):
bots = fetch_bots()
self.assertEqual(bots, ["bot1", "bot2", "bot3"])
- self.mock_get_auth_session.assert_called_once()
- self.mock_session.post.assert_called_once_with(PINPOINT_CONFIG_API_URL)
+ self.mock_post.assert_called_once_with(PINPOINT_CONFIG_API_URL)
def test_fetch_bots_api_error(self):
- self.mock_session.post.side_effect = requests.exceptions.HTTPError(
- "API Error")
+ self.mock_post.side_effect = requests.exceptions.HTTPError("API Error")
with self.assertRaises(MultiException):
fetch_bots()
- self.mock_get_auth_session.assert_called_once()
- self.mock_session.post.assert_called_once_with(PINPOINT_CONFIG_API_URL)
+ self.mock_post.assert_called_once_with(PINPOINT_CONFIG_API_URL)
if __name__ == "__main__":
diff --git a/tests/crossbench/pinpoint/test_list_builds.py b/tests/crossbench/pinpoint/test_list_builds.py
index fdc8e43..5cae17f 100644
--- a/tests/crossbench/pinpoint/test_list_builds.py
+++ b/tests/crossbench/pinpoint/test_list_builds.py
@@ -4,7 +4,6 @@
from __future__ import annotations
-from typing import Final
from unittest import mock
import requests
@@ -13,12 +12,10 @@
from crossbench.pinpoint.api import PINPOINT_BUILDS_API_URL_TEMPLATE
from crossbench.pinpoint.list_builds import Build, fetch_builds, list_builds
from tests import test_helper
-from tests.crossbench.pinpoint.auth_session_mixin import MockAuthSessionMixin
+from tests.crossbench.pinpoint.requests_mixin import MockRequestsMixin
-class ListBuildsTest(MockAuthSessionMixin):
- _get_auth_session_patch_target: Final[
- str] = "crossbench.pinpoint.list_builds.get_auth_session"
+class ListBuildsTest(MockRequestsMixin):
def setUp(self):
super().setUp()
@@ -47,7 +44,7 @@
mock_response.raise_for_status.return_value = None
return mock_response
- self.mock_session.get.side_effect = mock_get_side_effect
+ self.mock_get.side_effect = mock_get_side_effect
def test_fetch_builds_contain_correct_builds(self):
builds = fetch_builds("test-bot")
@@ -66,8 +63,7 @@
self.assertIn("commit0", output)
def test_list_builds_api_error(self):
- self.mock_session.get.side_effect = requests.exceptions.HTTPError(
- "API Error")
+ self.mock_get.side_effect = requests.exceptions.HTTPError("API Error")
with self.assertRaises(MultiException):
list_builds("test-bot", 1)
diff --git a/tests/crossbench/pinpoint/test_list_stories.py b/tests/crossbench/pinpoint/test_list_stories.py
index 0893d14..5cda422 100644
--- a/tests/crossbench/pinpoint/test_list_stories.py
+++ b/tests/crossbench/pinpoint/test_list_stories.py
@@ -4,7 +4,6 @@
from __future__ import annotations
-from typing import Final
from unittest import mock
import requests
@@ -13,12 +12,10 @@
from crossbench.pinpoint.api import CHROMEPERF_DESCRIBE_API_URL
from crossbench.pinpoint.list_stories import fetch_stories
from tests import test_helper
-from tests.crossbench.pinpoint.auth_session_mixin import MockAuthSessionMixin
+from tests.crossbench.pinpoint.requests_mixin import MockRequestsMixin
-class ListStoriesTest(MockAuthSessionMixin):
- _get_auth_session_patch_target: Final[
- str] = "crossbench.pinpoint.list_stories.get_auth_session"
+class ListStoriesTest(MockRequestsMixin):
def setUp(self):
super().setUp()
@@ -40,13 +37,12 @@
mock_response.raise_for_status.return_value = None
return mock_response
- self.mock_session.post.side_effect = mock_post_side_effect
+ self.mock_post.side_effect = mock_post_side_effect
def test_fetch_stories(self):
stories = fetch_stories("benchmark1")
self.assertEqual(stories, ["story1", "story2"])
- self.mock_get_auth_session.assert_called_once()
- self.mock_session.post.assert_called_once_with(
+ self.mock_post.assert_called_once_with(
CHROMEPERF_DESCRIBE_API_URL,
params={
"test_suite": "benchmark1",
@@ -58,8 +54,7 @@
self.assertEqual(stories, [])
def test_fetch_stories_api_error(self):
- self.mock_session.post.side_effect = requests.exceptions.HTTPError(
- "API Error")
+ self.mock_post.side_effect = requests.exceptions.HTTPError("API Error")
with self.assertRaises(MultiException):
fetch_stories("benchmark1")
diff --git a/tests/crossbench/pinpoint/test_patch_resolver.py b/tests/crossbench/pinpoint/test_patch_resolver.py
index 706be97..36796c4 100644
--- a/tests/crossbench/pinpoint/test_patch_resolver.py
+++ b/tests/crossbench/pinpoint/test_patch_resolver.py
@@ -9,7 +9,7 @@
from crossbench.pinpoint import patch_resolver
from tests import test_helper
-from tests.crossbench.pinpoint.auth_session_mixin import MockAuthSessionMixin
+from tests.crossbench.pinpoint.requests_mixin import MockRequestsMixin
_TEST_PATCH: Final[
str] = "https://chromium-review.googlesource.com/c/chromium/src/+/12345"
@@ -17,10 +17,7 @@
_TEST_PATCH_WITH_PATCHSET: Final[str] = _TEST_PATCH + "/6"
-class PatchResolverTest(MockAuthSessionMixin):
-
- _get_auth_session_patch_target = (
- "crossbench.pinpoint.patch_resolver.get_auth_session")
+class PatchResolverTest(MockRequestsMixin):
def setUp(self):
super().setUp()
@@ -35,7 +32,7 @@
mock_response.raise_for_status.return_value = None
return mock_response
- self.mock_session.get.side_effect = mock_get_side_effect
+ self.mock_get.side_effect = mock_get_side_effect
def test_resolve_patch(self):
for protocol in ["", "https://"]:
diff --git a/tests/crossbench/pinpoint/test_requests.py b/tests/crossbench/pinpoint/test_requests.py
new file mode 100644
index 0000000..f1cbdf6
--- /dev/null
+++ b/tests/crossbench/pinpoint/test_requests.py
@@ -0,0 +1,78 @@
+# Copyright 2025 The Chromium Authors
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+from __future__ import annotations
+
+import json
+import unittest
+from unittest import mock
+
+import requests
+
+from crossbench.pinpoint import requests as pinpoint_requests
+from tests import test_helper
+
+
+class RequestsTest(unittest.TestCase):
+
+ def setUp(self):
+ super().setUp()
+ self.mock_response = mock.Mock()
+ self.mock_response.raise_for_status.return_value = None
+ error = requests.exceptions.HTTPError(
+ "Base Error", response=self.mock_response)
+ self.mock_response.raise_for_status.side_effect = error
+
+ self.mock_session = mock.Mock(spec=requests.Session)
+ self.mock_session.request.return_value = self.mock_response
+
+ self.mock_get_auth_session = self.enterContext(
+ mock.patch("crossbench.pinpoint.requests.auth.get_auth_session"))
+ self.mock_get_auth_session.return_value = self.mock_session
+
+ def test_get_success(self):
+ self.mock_response.raise_for_status.side_effect = None
+ response = pinpoint_requests.get("http://example.com", params={"a": 1})
+
+ self.mock_session.request.assert_called_once_with(
+ "GET", "http://example.com", params={"a": 1})
+ self.assertEqual(response, self.mock_response)
+
+ def test_post_success(self):
+ self.mock_response.raise_for_status.side_effect = None
+ response = pinpoint_requests.post("http://example.com", json={"a": 1})
+
+ self.mock_session.request.assert_called_once_with(
+ "POST", "http://example.com", json={"a": 1})
+ self.assertEqual(response, self.mock_response)
+
+ def test_error_with_message(self):
+ self.mock_response.json.return_value = {"error": "Detailed error message"}
+
+ with self.assertRaises(pinpoint_requests.ServerError) as cm:
+ pinpoint_requests.get("http://example.com")
+
+ self.assertIn("Base Error", str(cm.exception))
+ self.assertIn("Detailed error message", str(cm.exception))
+
+ def test_error_with_data(self):
+ self.mock_response.json.return_value = {"other": "data"}
+
+ with self.assertRaises(pinpoint_requests.ServerError) as cm:
+ pinpoint_requests.get("http://example.com")
+
+ self.assertIn("Base Error", str(cm.exception))
+ self.assertIn("{'other': 'data'}", str(cm.exception))
+
+ def test_error_non_json(self):
+ self.mock_response.json.side_effect = json.JSONDecodeError("msg", "doc", 0)
+
+ with self.assertRaises(pinpoint_requests.ServerError) as cm:
+ pinpoint_requests.get("http://example.com")
+
+ self.assertEqual(str(cm.exception), "Base Error")
+
+
+if __name__ == "__main__":
+ test_helper.run_pytest(__file__)
diff --git a/tests/crossbench/pinpoint/test_start_job.py b/tests/crossbench/pinpoint/test_start_job.py
index b3b1c31..a2f16b6 100644
--- a/tests/crossbench/pinpoint/test_start_job.py
+++ b/tests/crossbench/pinpoint/test_start_job.py
@@ -5,7 +5,6 @@
from __future__ import annotations
import json
-from typing import Final
from unittest import mock
import requests
@@ -15,12 +14,10 @@
from crossbench.pinpoint.config import PinpointTryJobConfig
from crossbench.pinpoint.start_job import start_job
from tests import test_helper
-from tests.crossbench.pinpoint.auth_session_mixin import MockAuthSessionMixin
+from tests.crossbench.pinpoint.requests_mixin import MockRequestsMixin
-class StartJobTest(MockAuthSessionMixin):
- _get_auth_session_patch_target: Final[
- str] = "crossbench.pinpoint.start_job.get_auth_session"
+class StartJobTest(MockRequestsMixin):
def setUp(self):
super().setUp()
@@ -36,7 +33,7 @@
mock_response.raise_for_status.return_value = None
return mock_response
- self.mock_session.post.side_effect = mock_post_side_effect
+ self.mock_post.side_effect = mock_post_side_effect
def test_start_job_correct_parameters(self):
start_job(
@@ -87,12 +84,11 @@
"experiment_extra_args":
'--extra-browser-args="--js-flags=--exp-js-flag"',
}
- self.mock_session.post.assert_called_with(
+ self.mock_post.assert_called_with(
PINPOINT_START_JOB_API_URL, data=expected_payload)
def test_start_job_api_error(self):
- self.mock_session.post.side_effect = requests.exceptions.HTTPError(
- "API Error")
+ self.mock_post.side_effect = requests.exceptions.HTTPError("API Error")
with self.assertRaises(MultiException):
start_job(
PinpointTryJobConfig(
