crashdb_impl: improve GitHub API robustness

apport/crashdb_impl/github.py

@@ -37,37 +37,52 @@ def __init__(
         self.__expiry = 0.0
         self.message_callback = message_callback
 
-    def _post(self, url: str, data: str) -> Any:
-        """Posts the given data to the given URL.
-        Uses auth token if available"""
-        headers = {"Accept": "application/vnd.github.v3+json"}
+    def _post(
+        self,
+        url: str,
+        data: str,
+        content_type: str,
+        accept: str = "application/vnd.github.v3+json",
+    ) -> Any:
+        """Post data to the given URL.
+
+        Uses the authentication token if available.
+        """
+        headers = {"Accept": accept, "Content-Type": content_type}
         if self.__access_token:
             headers["Authorization"] = f"token {self.__access_token}"
+
         request = urllib.request.Request(
             url, data=data.encode("utf-8"), headers=headers, method="POST"
         )
+
         try:
             with urllib.request.urlopen(request, timeout=5.0) as response:
-                return json.loads(response.read())
-        except urllib.error.URLError as err:
+                return json.loads(response.read().decode("utf-8"))
+        except urllib.error.URLError:
             if self.message_callback:
                 self.message_callback(
                     "Failed connection",
                     f"Failed connection to {url}.\n"
                     f"Please check your internet connection and try again.",
                 )
-            raise err
+            raise
         finally:
             self.__last_request = time.time()
 
-    def api_authentication(self, url: str, data: dict) -> Any:
-        """Authenticate against the GitHub API."""
-        return self._post(url, urllib.parse.urlencode(data))
+    def api_authentication(self, url: str, data: Mapping[str, Any]) -> Any:
+        """Authentication against the GitHub API."""
+        return self._post(
+            url,
+            urllib.parse.urlencode(data),
+            "application/x-www-form-urlencoded",
+            "application/json",
+        )
 
-    def api_open_issue(self, owner: str, repo: str, data: dict) -> Any:
+    def api_open_issue(self, owner: str, repo: str, data: Mapping[str, Any]) -> Any:
         """Open a new issue on the GitHub project."""
         url = f"https://api.github.com/repos/{owner}/{repo}/issues"
-        return self._post(url, json.dumps(data))
+        return self._post(url, json.dumps(data), "application/json")
 
     def __enter__(self) -> Self:
         """Enters login process. At exit, login process ends."""
@@ -93,10 +108,10 @@ def __enter__(self) -> Self:
 
         self.__authentication_data = {
             "client_id": self.__client_id,
-            "device_code": f'{response["device_code"]}',
+            "device_code": f"{response['device_code']}",
             "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
         }
-        self.__cooldown = response["interval"]
+        self.__cooldown = float(response.get("interval", 5))
         self.__expiry = int(response["expires_in"]) + time.time()
 
         return self
@@ -133,7 +148,8 @@ def authentication_complete(self) -> bool:
             if response["error"] == "authorization_pending":
                 return False
             if response["error"] == "slow_down":
-                self.__cooldown = int(response["interval"])
+                # Fall back safely if interval is omitted.
+                self.__cooldown = float(response.get("interval", self.__cooldown + 5))
                 return False
             raise RuntimeError(f"Unknown error from Github: {response}")
         if "access_token" in response:
@@ -158,7 +174,7 @@ def __init__(self, auth_file: str | None, options: dict[str, Any]) -> None:
         self.repository_owner = options["repository_owner"]
         self.repository_name = options["repository_name"]
         self.app_id = options["github_app_id"]
-        self.labels = set(options["labels"])
+        self.labels = set(options.get("labels", []))
         self.issue_url = None
         self.github: Github | None = None
 
@@ -197,7 +213,7 @@ def upload(
             raise RuntimeError("Failed to login to Github")
 
         data = self._format_report(report)
-        if not (self.repository_name is None and self.repository_owner is None):
+        if self.repository_owner is not None and self.repository_name is not None:
             response = self.github.api_open_issue(
                 self.repository_owner, self.repository_name, data
             )

tests/integration/test_github.py

@@ -47,11 +47,57 @@ def test__format_report(self) -> None:
         self.assertIn("title", result)
         self.assertEqual(self.crashdb.labels, set(result["labels"]))
 
+    @patch("apport.crashdb_impl.github.Github.api_authentication")
+    def test_enter_without_interval(self, mock_api: MagicMock) -> None:
+        """Default cooldown is used when interval is omitted."""
+        response = self.api_auth_return_value.copy()
+        response.pop("interval")
+        mock_api.return_value = response
+
+        with self.github:
+            self.assertEqual(self.github._Github__cooldown, 5.0)
+
+    @patch("apport.crashdb_impl.github.Github.api_open_issue")
+    def test_authentication_complete_slow_down_without_interval(
+        self, mock_api: MagicMock
+    ) -> None:
+        """Fallback cooldown is used if slow_down omits interval."""
+        mock_api.side_effect = [
+            self.api_auth_return_value,
+            {"error": "slow_down"},
+        ]
+
+        with self.github as github:
+            self.assertEqual(github.authentication_complete())
+            self.assertEqual(github._Github__cooldown, 6.0)
+
+    @patch("apport.crashdb_impl.github.Github.authentication_complete")
+    def test_post_adds_authorization_header(self, mock_urlopen: MagicMock) -> None:
+        """Authenticated requests include Authorization header."""
+        response = MagicMock()
+        response.read.return_value = b"{}"
+        mock_urlopen.return_value.__enter__.return_value = response
+
+        self.github._Github__access_token = "token"
+
+        self.github._post(
+            "https://example.com",
+            "{}",
+            "application/json",
+            "application/json",
+        )
+
+        request = mock_urlopen.call_args.args[0]
+        self.assertEqual(request.headers["Authorization"], "token token")
+
     @patch("apport.crashdb_impl.github.Github.api_authentication")
     @patch("apport.crashdb_impl.github.Github.api_open_issue")
     @patch("apport.crashdb_impl.github.Github.authentication_complete")
     def test_upload(
-        self, mock_auth: MagicMock, mock_api: MagicMock, mock_api_auth: MagicMock
+        self,
+        mock_auth: MagicMock,
+        mock_api: MagicMock,
+        mock_api_auth: MagicMock,
     ) -> None:
         mock_api.return_value = {"html_url": "doesntmatterhere"}
         mock_auth.return_value = True