chore: port update option for routeFromHar (#1392)

Fixes #1389.

Ports:

  - [x] microsoft/playwright@6a8d835 (chore: allow updating har while routing (#15197))

playwright/_impl/_browser.py

@@ -16,7 +16,7 @@
 import json
 from pathlib import Path
 from types import SimpleNamespace
-from typing import TYPE_CHECKING, Any, Dict, List, Pattern, Union, cast
+from typing import TYPE_CHECKING, Dict, List, Pattern, Union, cast
 
 from playwright._impl._api_structures import (
     Geolocation,
@@ -38,11 +38,10 @@
     async_readfile,
     is_safe_close_error,
     locals_to_params,
+    prepare_record_har_options,
 )
-from playwright._impl._local_utils import LocalUtils
 from playwright._impl._network import serialize_headers
 from playwright._impl._page import Page
-from playwright._impl._str_utils import escape_regex_flags
 
 if TYPE_CHECKING:  # pragma: no cover
     from playwright._impl._browser_type import BrowserType
@@ -64,7 +63,6 @@ def __init__(
         self._should_close_connection_on_close = False
 
         self._contexts: List[BrowserContext] = []
-        _utils: LocalUtils
         self._channel.on("close", lambda _: self._on_close())
 
     def __repr__(self) -> str:
@@ -131,6 +129,7 @@ async def new_context(
         self._contexts.append(context)
         context._browser = self
         context._options = params
+        context._set_browser_type(self._browser_type)
         return context
 
     async def new_page(
@@ -177,6 +176,11 @@ async def new_page(
         context._owner_page = page
         return page
 
+    def _set_browser_type(self, browser_type: "BrowserType") -> None:
+        self._browser_type = browser_type
+        for context in self._contexts:
+            context._set_browser_type(browser_type)
+
     async def close(self) -> None:
         if self._is_closed_or_closing:
             return
@@ -224,32 +228,7 @@ async def normalize_context_params(is_sync: bool, params: Dict) -> None:
     if "extraHTTPHeaders" in params:
         params["extraHTTPHeaders"] = serialize_headers(params["extraHTTPHeaders"])
     if "recordHarPath" in params:
-        recordHar: Dict[str, Any] = {"path": str(params["recordHarPath"])}
-        params["recordHar"] = recordHar
-        if "recordHarUrlFilter" in params:
-            opt = params["recordHarUrlFilter"]
-            if isinstance(opt, str):
-                params["recordHar"]["urlGlob"] = opt
-            if isinstance(opt, Pattern):
-                params["recordHar"]["urlRegexSource"] = opt.pattern
-                params["recordHar"]["urlRegexFlags"] = escape_regex_flags(opt)
-            del params["recordHarUrlFilter"]
-        if "recordHarMode" in params:
-            params["recordHar"]["mode"] = params["recordHarMode"]
-            del params["recordHarMode"]
-
-        new_content_api = None
-        old_content_api = None
-        if "recordHarContent" in params:
-            new_content_api = params["recordHarContent"]
-            del params["recordHarContent"]
-        if "recordHarOmitContent" in params:
-            old_content_api = params["recordHarOmitContent"]
-            del params["recordHarOmitContent"]
-        content = new_content_api or ("omit" if old_content_api else None)
-        if content:
-            params["recordHar"]["content"] = content
-
+        params["recordHar"] = prepare_record_har_options(params)
         del params["recordHarPath"]
     if "recordVideoDir" in params:
         params["recordVideo"] = {"dir": str(params["recordVideoDir"])}

playwright/_impl/_browser_context.py

@@ -16,7 +16,18 @@
 import json
 from pathlib import Path
 from types import SimpleNamespace
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Set, Union, cast
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Pattern,
+    Set,
+    Union,
+    cast,
+)
 
 from playwright._impl._api_structures import (
     Cookie,
@@ -37,6 +48,7 @@
 from playwright._impl._frame import Frame
 from playwright._impl._har_router import HarRouter
 from playwright._impl._helper import (
+    HarRecordingMetadata,
     RouteFromHarNotFoundPolicy,
     RouteHandler,
     RouteHandlerCallback,
@@ -47,6 +59,7 @@
     async_writefile,
     is_safe_close_error,
     locals_to_params,
+    prepare_record_har_options,
     to_impl,
 )
 from playwright._impl._network import Request, Response, Route, serialize_headers
@@ -56,6 +69,7 @@
 
 if TYPE_CHECKING:  # pragma: no cover
     from playwright._impl._browser import Browser
+    from playwright._impl._browser_type import BrowserType
 
 
 class BrowserContext(ChannelOwner):
@@ -85,6 +99,7 @@ def __init__(
         self._background_pages: Set[Page] = set()
         self._service_workers: Set[Worker] = set()
         self._tracing = cast(Tracing, from_channel(initializer["tracing"]))
+        self._har_recorders: Dict[str, HarRecordingMetadata] = {}
         self._request: APIRequestContext = from_channel(
             initializer["APIRequestContext"]
         )
@@ -201,6 +216,14 @@ def pages(self) -> List[Page]:
     def browser(self) -> Optional["Browser"]:
         return self._browser
 
+    def _set_browser_type(self, browser_type: "BrowserType") -> None:
+        self._browser_type = browser_type
+        if self._options.get("recordHar"):
+            self._har_recorders[""] = {
+                "path": self._options["recordHar"]["path"],
+                "content": self._options["recordHar"].get("content"),
+            }
+
     async def new_page(self) -> Page:
         if self._owner_page:
             raise Error("Please use browser.new_context()")
@@ -294,12 +317,37 @@ async def unroute(
         if len(self._routes) == 0:
             await self._disable_interception()
 
+    async def _record_into_har(
+        self,
+        har: Union[Path, str],
+        page: Optional[Page] = None,
+        url: Union[Pattern, str] = None,
+    ) -> None:
+        params = {
+            "options": prepare_record_har_options(
+                {
+                    "recordHarPath": har,
+                    "recordHarContent": "attach",
+                    "recordHarMode": "minimal",
+                    "recordHarUrlFilter": url,
+                }
+            )
+        }
+        if page:
+            params["page"] = page._channel
+        har_id = await self._channel.send("harStart", params)
+        self._har_recorders[har_id] = {"path": str(har), "content": "attach"}
+
     async def route_from_har(
         self,
         har: Union[Path, str],
-        url: URLMatch = None,
+        url: Union[Pattern, str] = None,
         not_found: RouteFromHarNotFoundPolicy = None,
+        update: bool = None,
     ) -> None:
+        if update:
+            await self._record_into_har(har=har, page=None, url=url)
+            return
         router = await HarRouter.create(
             local_utils=self._connection.local_utils,
             file=str(har),
@@ -338,13 +386,26 @@ def _on_close(self) -> None:
 
     async def close(self) -> None:
         try:
-            if self._options.get("recordHar"):
+            for har_id, params in self._har_recorders.items():
                 har = cast(
-                    Artifact, from_channel(await self._channel.send("harExport"))
-                )
-                await har.save_as(
-                    cast(Dict[str, str], self._options["recordHar"])["path"]
+                    Artifact,
+                    from_channel(
+                        await self._channel.send("harExport", {"harId": har_id})
+                    ),
                 )
+                # Server side will compress artifact if content is attach or if file is .zip.
+                is_compressed = params.get("content") == "attach" or params[
+                    "path"
+                ].endswith(".zip")
+                need_compressed = params["path"].endswith(".zip")
+                if is_compressed and not need_compressed:
+                    tmp_path = params["path"] + ".tmp"
+                    await har.save_as(tmp_path)
+                    await self._connection.local_utils.har_unzip(
+                        zipFile=tmp_path, harFile=params["path"]
+                    )
+                else:
+                    await har.save_as(params["path"])
                 await har.delete()
             await self._channel.send("close")
             await self._closed_future

playwright/_impl/_browser_type.py

@@ -92,6 +92,7 @@ async def launch(
         browser = cast(
             Browser, from_channel(await self._channel.send("launch", params))
         )
+        browser._set_browser_type(self)
         return browser
 
     async def launch_persistent_context(
@@ -154,6 +155,7 @@ async def launch_persistent_context(
             from_channel(await self._channel.send("launchPersistentContext", params)),
         )
         context._options = params
+        context._set_browser_type(self)
         return context
 
     async def connect_over_cdp(
@@ -174,6 +176,7 @@ async def connect_over_cdp(
         if default_context:
             browser._contexts.append(default_context)
             default_context._browser = browser
+        browser._set_browser_type(self)
         return browser
 
     async def connect(
@@ -230,6 +233,7 @@ def handle_transport_close() -> None:
 
         transport.once("close", handle_transport_close)
 
+        browser._set_browser_type(self)
         return browser
 
 

playwright/_impl/_helper.py

@@ -41,6 +41,7 @@
 
 from playwright._impl._api_structures import NameValue
 from playwright._impl._api_types import Error, TimeoutError
+from playwright._impl._str_utils import escape_regex_flags
 
 if sys.version_info >= (3, 8):  # pragma: no cover
     from typing import Literal, TypedDict
@@ -85,6 +86,40 @@ class FallbackOverrideParameters(TypedDict, total=False):
     postData: Optional[Union[str, bytes]]
 
 
+class HarRecordingMetadata(TypedDict, total=False):
+    path: str
+    content: Optional[HarContentPolicy]
+
+
+def prepare_record_har_options(params: Dict) -> Dict[str, Any]:
+    out_params: Dict[str, Any] = {"path": str(params["recordHarPath"])}
+    if "recordHarUrlFilter" in params:
+        opt = params["recordHarUrlFilter"]
+        if isinstance(opt, str):
+            out_params["urlGlob"] = opt
+        if isinstance(opt, Pattern):
+            out_params["urlRegexSource"] = opt.pattern
+            out_params["urlRegexFlags"] = escape_regex_flags(opt)
+        del params["recordHarUrlFilter"]
+    if "recordHarMode" in params:
+        out_params["mode"] = params["recordHarMode"]
+        del params["recordHarMode"]
+
+    new_content_api = None
+    old_content_api = None
+    if "recordHarContent" in params:
+        new_content_api = params["recordHarContent"]
+        del params["recordHarContent"]
+    if "recordHarOmitContent" in params:
+        old_content_api = params["recordHarOmitContent"]
+        del params["recordHarOmitContent"]
+    content = new_content_api or ("omit" if old_content_api else None)
+    if content:
+        out_params["content"] = content
+
+    return out_params
+
+
 class ParsedMessageParams(TypedDict):
     type: str
     guid: str

playwright/_impl/_local_utils.py

@@ -53,3 +53,7 @@ async def har_lookup(
     async def har_close(self, harId: str) -> None:
         params = locals_to_params(locals())
         await self._channel.send("harClose", params)
+
+    async def har_unzip(self, zipFile: str, harFile: str) -> None:
+        params = locals_to_params(locals())
+        await self._channel.send("harUnzip", params)

playwright/_impl/_page.py

@@ -605,9 +605,13 @@ async def unroute(
     async def route_from_har(
         self,
         har: Union[Path, str],
-        url: URLMatch = None,
+        url: Union[Pattern, str] = None,
         not_found: RouteFromHarNotFoundPolicy = None,
+        update: bool = None,
     ) -> None:
+        if update:
+            await self._browser_context._record_into_har(har=har, page=self, url=url)
+            return
         router = await HarRouter.create(
             local_utils=self._connection.local_utils,
             file=str(har),

playwright/async_api/_generated.py

@@ -7744,8 +7744,9 @@ async def route_from_har(
         self,
         har: typing.Union[pathlib.Path, str],
         *,
-        url: typing.Union[str, typing.Pattern, typing.Callable[[str], bool]] = None,
-        not_found: Literal["abort", "fallback"] = None
+        url: typing.Union[str, typing.Pattern] = None,
+        not_found: Literal["abort", "fallback"] = None,
+        update: bool = None
     ) -> NoneType:
         """Page.route_from_har
 
@@ -7761,19 +7762,21 @@ async def route_from_har(
         har : Union[pathlib.Path, str]
             Path to a [HAR](http://www.softwareishard.com/blog/har-12-spec) file with prerecorded network data. If `path` is a
             relative path, then it is resolved relative to the current working directory.
-        url : Union[Callable[[str], bool], Pattern, str, NoneType]
+        url : Union[Pattern, str, NoneType]
             A glob pattern, regular expression or predicate to match the request URL. Only requests with URL matching the pattern
             will be surved from the HAR file. If not specified, all requests are served from the HAR file.
         not_found : Union["abort", "fallback", NoneType]
             - If set to 'abort' any request not found in the HAR file will be aborted.
             - If set to 'fallback' missing requests will be sent to the network.
 
             Defaults to abort.
+        update : Union[bool, NoneType]
+            If specified, updates the given HAR with the actual network information instead of serving from file.
         """
 
         return mapping.from_maybe_impl(
             await self._impl_obj.route_from_har(
-                har=har, url=self._wrap_handler(url), not_found=not_found
+                har=har, url=url, not_found=not_found, update=update
             )
         )
 
@@ -10477,8 +10480,9 @@ async def route_from_har(
         self,
         har: typing.Union[pathlib.Path, str],
         *,
-        url: typing.Union[str, typing.Pattern, typing.Callable[[str], bool]] = None,
-        not_found: Literal["abort", "fallback"] = None
+        url: typing.Union[str, typing.Pattern] = None,
+        not_found: Literal["abort", "fallback"] = None,
+        update: bool = None
     ) -> NoneType:
         """BrowserContext.route_from_har
 
@@ -10494,19 +10498,21 @@ async def route_from_har(
         har : Union[pathlib.Path, str]
             Path to a [HAR](http://www.softwareishard.com/blog/har-12-spec) file with prerecorded network data. If `path` is a
             relative path, then it is resolved relative to the current working directory.
-        url : Union[Callable[[str], bool], Pattern, str, NoneType]
+        url : Union[Pattern, str, NoneType]
             A glob pattern, regular expression or predicate to match the request URL. Only requests with URL matching the pattern
             will be surved from the HAR file. If not specified, all requests are served from the HAR file.
         not_found : Union["abort", "fallback", NoneType]
             - If set to 'abort' any request not found in the HAR file will be aborted.
             - If set to 'fallback' falls through to the next route handler in the handler chain.
 
             Defaults to abort.
+        update : Union[bool, NoneType]
+            If specified, updates the given HAR with the actual network information instead of serving from file.
         """
 
         return mapping.from_maybe_impl(
             await self._impl_obj.route_from_har(
-                har=har, url=self._wrap_handler(url), not_found=not_found
+                har=har, url=url, not_found=not_found, update=update
             )
         )