fix(chrome): add robust CfT JSON parsing fallback for compressed responses (#701) (#724)

Author: aleksandr-kotlyar

tests/test_chromium_driver.py

@@ -1,4 +1,7 @@
 import os
+import gzip
+import json
+import zlib
 
 import pytest
 
@@ -386,3 +389,39 @@ def test_chrome_115_plus_uses_chrome_for_testing_macos_download_url_after_milest
         CHROME_FOR_TESTING_LATEST_VERSIONS_PER_MILESTONE_URL,
         CHROME_FOR_TESTING_KNOWN_GOOD_VERSIONS_URL,
     ]
+
+
+def test_chromium_cft_json_parser_handles_gzip_compressed_response():
+    class Response:
+        headers = {"Content-Encoding": "gzip"}
+        text = ""
+        content = gzip.compress(
+            json.dumps({"builds": {"120.0.6099": {"version": "120.0.6099.109"}}}).encode("utf-8")
+        )
+
+        @staticmethod
+        def json():
+            raise ValueError("broken json parser")
+
+    driver, _ = chrome_driver_for(browser_version="120.0.6099.71", responses={})
+    parsed = driver._parse_json_response(Response())
+
+    assert parsed["builds"]["120.0.6099"]["version"] == "120.0.6099.109"
+
+
+def test_chromium_cft_json_parser_handles_deflate_compressed_response():
+    class Response:
+        headers = {"Content-Encoding": "deflate"}
+        text = ""
+        content = zlib.compress(
+            json.dumps({"versions": [{"version": "120.0.6099.109"}]}).encode("utf-8")
+        )
+
+        @staticmethod
+        def json():
+            raise ValueError("broken json parser")
+
+    driver, _ = chrome_driver_for(browser_version="120.0.6099.71", responses={})
+    parsed = driver._parse_json_response(Response())
+
+    assert parsed["versions"][0]["version"] == "120.0.6099.109"

webdriver_manager/drivers/chrome.py

@@ -4,7 +4,9 @@
 from webdriver_manager.core.logger import log
 from webdriver_manager.core.os_manager import ChromeType
 
+import gzip
 import json
+import zlib
 
 CHROME_FOR_TESTING_LATEST_PATCH_VERSIONS_PER_BUILD_URL = (
     "https://googlechromelabs.github.io/chrome-for-testing/latest-patch-versions-per-build.json"
@@ -142,7 +144,7 @@ def _get_cft_json(self, url):
         response = self._http_client.get(url)
 
         try:
-            return json.loads(response.text)
+            return self._parse_json_response(response)
         except ValueError as error:
             raise ValueError(
                 f"Could not parse Chrome for Testing metadata from {url}."
@@ -163,7 +165,12 @@ def _raise_missing_chromedriver_version(self, browser_version, checked_urls, det
     def get_url_for_version_and_platform(self, browser_version, platform):
         url = "https://googlechromelabs.github.io/chrome-for-testing/known-good-versions-with-downloads.json"
         response = self._http_client.get(url)
-        data = response.json()
+        try:
+            data = self._parse_json_response(response)
+        except ValueError as error:
+            raise ValueError(
+                f"Could not parse Chrome for Testing metadata from {url}."
+            ) from error
         versions = data["versions"]
 
         if version.parse(browser_version) >= version.parse("115"):
@@ -189,3 +196,38 @@ def get_url_for_version_and_platform(self, browser_version, platform):
                             return d["url"]
 
         raise Exception(f"No such driver version {browser_version} for {platform}")
+
+    def _parse_json_response(self, response):
+        try:
+            return response.json()
+        except Exception:
+            pass
+
+        text = getattr(response, "text", None)
+        if text:
+            try:
+                return json.loads(text)
+            except ValueError:
+                pass
+
+        raw = getattr(response, "content", None)
+        if not raw:
+            raise ValueError("Response body is empty")
+
+        headers = getattr(response, "headers", {}) or {}
+        encoding = (headers.get("Content-Encoding") or "").lower().strip()
+
+        if encoding == "gzip":
+            raw = gzip.decompress(raw)
+        elif encoding == "br":
+            try:
+                import brotli
+            except ImportError as error:
+                raise ValueError(
+                    "Response is brotli-compressed but brotli package is not installed"
+                ) from error
+            raw = brotli.decompress(raw)
+        elif encoding == "deflate":
+            raw = zlib.decompress(raw)
+
+        return json.loads(raw.decode("utf-8"))