feat: add Response assertions (#1125)

* feat: add Response assertions

* roll

Author: mxschmitt

README.md

@@ -4,9 +4,9 @@ Playwright is a Python library to automate [Chromium](https://www.chromium.org/H
 
 |          | Linux | macOS | Windows |
 |   :---   | :---: | :---: | :---:   |
-| Chromium <!-- GEN:chromium-version -->99.0.4812.0<!-- GEN:stop --> | ✅ | ✅ | ✅ |
+| Chromium <!-- GEN:chromium-version -->99.0.4837.0<!-- GEN:stop --> | ✅ | ✅ | ✅ |
 | WebKit <!-- GEN:webkit-version -->15.4<!-- GEN:stop --> | ✅ | ✅ | ✅ |
-| Firefox <!-- GEN:firefox-version -->95.0<!-- GEN:stop --> | ✅ | ✅ | ✅ |
+| Firefox <!-- GEN:firefox-version -->96.0.1<!-- GEN:stop --> | ✅ | ✅ | ✅ |
 
 ## Documentation
 

playwright/_impl/_assertions.py

@@ -16,6 +16,7 @@
 from urllib.parse import urljoin
 
 from playwright._impl._api_structures import ExpectedTextValue, FrameExpectOptions
+from playwright._impl._fetch import APIResponse
 from playwright._impl._locator import Locator
 from playwright._impl._page import Page
 from playwright._impl._str_utils import escape_regex_flags
@@ -532,6 +533,37 @@ async def not_to_be_focused(
         await self._not.to_be_focused(timeout)
 
 
+class APIResponseAssertions:
+    def __init__(self, response: APIResponse, is_not: bool = False) -> None:
+        self._loop = response._loop
+        self._dispatcher_fiber = response._dispatcher_fiber
+        self._is_not = is_not
+        self._actual = response
+
+    @property
+    def _not(self) -> "APIResponseAssertions":
+        return APIResponseAssertions(self._actual, not self._is_not)
+
+    async def to_be_ok(
+        self,
+    ) -> None:
+        __tracebackhide__ = True
+        if self._is_not is not self._actual.ok:
+            return
+        message = f"Response status expected to be within [200..299] range, was '{self._actual.status}'"
+        if self._is_not:
+            message = message.replace("expected to", "expected not to")
+        log_list = await self._actual._fetch_log()
+        log = "\n".join(log_list).strip()
+        if log:
+            message += f"\n Call log:\n{log}"
+        raise AssertionError(message)
+
+    async def not_to_be_ok(self) -> None:
+        __tracebackhide__ = True
+        await self._not.to_be_ok()
+
+
 def expected_regex(
     pattern: Pattern, match_substring: bool, normalize_white_space: bool
 ) -> ExpectedTextValue:

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, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Union, cast
 
 from playwright._impl._api_structures import (
     Geolocation,
@@ -115,10 +115,11 @@ async def new_context(
         await normalize_context_params(self._connection._is_sync, params)
 
         channel = await self._channel.send("newContext", params)
-        context = from_channel(channel)
+        context = cast(BrowserContext, from_channel(channel))
         self._contexts.append(context)
         context._browser = self
         context._options = params
+        context._tracing._local_utils = self._local_utils
         return context
 
     async def new_page(

playwright/_impl/_browser_context.py

@@ -47,7 +47,6 @@
     locals_to_params,
     to_impl,
 )
-from playwright._impl._local_utils import LocalUtils
 from playwright._impl._network import Request, Response, Route, serialize_headers
 from playwright._impl._page import BindingCall, Page, Worker
 from playwright._impl._tracing import Tracing
@@ -83,11 +82,10 @@ def __init__(
         self._options: Dict[str, Any] = {}
         self._background_pages: Set[Page] = set()
         self._service_workers: Set[Worker] = set()
-        self._tracing = Tracing(self)
+        self._tracing = cast(Tracing, from_channel(initializer["tracing"]))
         self._request: APIRequestContext = from_channel(
             initializer["APIRequestContext"]
         )
-        _local_utils: LocalUtils
         self._channel.on(
             "bindingCall",
             lambda params: self._on_binding(from_channel(params["binding"])),

playwright/_impl/_browser_type.py

@@ -89,7 +89,7 @@ async def launch(
         browser = cast(
             Browser, from_channel(await self._channel.send("launch", params))
         )
-        browser._utils = self._playwright._utils
+        browser._local_utils = self._playwright._utils
         return browser
 
     async def launch_persistent_context(
@@ -143,11 +143,12 @@ async def launch_persistent_context(
         params = locals_to_params(locals())
         await normalize_context_params(self._connection._is_sync, params)
         normalize_launch_params(params)
-        context = from_channel(
-            await self._channel.send("launchPersistentContext", params)
+        context = cast(
+            BrowserContext,
+            from_channel(await self._channel.send("launchPersistentContext", params)),
         )
         context._options = params
-        context._local_utils = self._playwright._utils
+        context.tracing._local_utils = self._playwright._utils
         return context
 
     async def connect_over_cdp(
@@ -160,7 +161,7 @@ async def connect_over_cdp(
         params = locals_to_params(locals())
         response = await self._channel.send_return_as_dict("connectOverCDP", params)
         browser = cast(Browser, from_channel(response["browser"]))
-        browser._utils = self._playwright._utils
+        browser._local_utils = self._playwright._utils
 
         default_context = cast(
             Optional[BrowserContext],
@@ -210,7 +211,7 @@ async def connect(
         assert pre_launched_browser
         browser = cast(Browser, from_channel(pre_launched_browser))
         browser._should_close_connection_on_close = True
-        browser._utils = self._playwright._utils
+        browser._local_utils = self._playwright._utils
 
         def handle_transport_close() -> None:
             for context in browser.contexts:

playwright/_impl/_fetch.py

@@ -41,6 +41,7 @@
     object_to_array,
 )
 from playwright._impl._network import serialize_headers
+from playwright._impl._tracing import Tracing
 
 if typing.TYPE_CHECKING:
     from playwright._impl._playwright import Playwright
@@ -72,14 +73,20 @@ async def new_context(
                 )
         if "extraHTTPHeaders" in params:
             params["extraHTTPHeaders"] = serialize_headers(params["extraHTTPHeaders"])
-        return from_channel(await self.playwright._channel.send("newRequest", params))
+        context = cast(
+            APIRequestContext,
+            from_channel(await self.playwright._channel.send("newRequest", params)),
+        )
+        context._tracing._local_utils = self.playwright._utils
+        return context
 
 
 class APIRequestContext(ChannelOwner):
     def __init__(
         self, parent: ChannelOwner, type: str, guid: str, initializer: Dict
     ) -> None:
         super().__init__(parent, type, guid, initializer)
+        self._tracing: Tracing = from_channel(initializer["tracing"])
 
     async def dispose(self) -> None:
         await self._channel.send("dispose")
@@ -400,6 +407,14 @@ async def dispose(self) -> None:
     def _fetch_uid(self) -> str:
         return self._initializer["fetchUid"]
 
+    async def _fetch_log(self) -> List[str]:
+        return await self._request._channel.send(
+            "fetchLog",
+            {
+                "fetchUid": self._fetch_uid(),
+            },
+        )
+
 
 def is_json_content_type(headers: network.HeadersArray = None) -> bool:
     if not headers:

playwright/_impl/_object_factory.py

@@ -32,6 +32,7 @@
 from playwright._impl._playwright import Playwright
 from playwright._impl._selectors import Selectors
 from playwright._impl._stream import Stream
+from playwright._impl._tracing import Tracing
 
 
 class DummyObject(ChannelOwner):
@@ -82,6 +83,8 @@ def create_remote_object(
         return Route(parent, type, guid, initializer)
     if type == "Stream":
         return Stream(parent, type, guid, initializer)
+    if type == "Tracing":
+        return Tracing(parent, type, guid, initializer)
     if type == "WebSocket":
         return WebSocket(parent, type, guid, initializer)
     if type == "Worker":

playwright/_impl/_tracing.py

@@ -13,22 +13,20 @@
 # limitations under the License.
 
 import pathlib
-from typing import TYPE_CHECKING, Optional, Union, cast
+from typing import Dict, Optional, Union, cast
 
 from playwright._impl._artifact import Artifact
-from playwright._impl._connection import from_nullable_channel
+from playwright._impl._connection import ChannelOwner, from_nullable_channel
 from playwright._impl._helper import locals_to_params
+from playwright._impl._local_utils import LocalUtils
 
-if TYPE_CHECKING:  # pragma: no cover
-    from playwright._impl._browser_context import BrowserContext
 
-
-class Tracing:
-    def __init__(self, context: "BrowserContext") -> None:
-        self._context = context
-        self._channel = context._channel
-        self._loop = context._loop
-        self._dispatcher_fiber = context._channel._connection._dispatcher_fiber
+class Tracing(ChannelOwner):
+    def __init__(
+        self, parent: ChannelOwner, type: str, guid: str, initializer: Dict
+    ) -> None:
+        super().__init__(parent, type, guid, initializer)
+        _local_utils: LocalUtils
 
     async def start(
         self,
@@ -54,7 +52,7 @@ async def stop(self, path: Union[pathlib.Path, str] = None) -> None:
         await self._channel.send("tracingStop")
 
     async def _do_stop_chunk(self, file_path: Union[pathlib.Path, str] = None) -> None:
-        is_local = not self._channel._connection.is_remote
+        is_local = not self._connection.is_remote
 
         mode = "doNotSave"
         if file_path:
@@ -88,4 +86,4 @@ async def _do_stop_chunk(self, file_path: Union[pathlib.Path, str] = None) -> No
 
         # Add local sources to the remote trace if necessary.
         if result.get("sourceEntries", []):
-            await self._context._local_utils.zip(file_path, result["sourceEntries"])
+            await self._local_utils.zip(file_path, result["sourceEntries"])

playwright/async_api/__init__.py

@@ -23,6 +23,9 @@
 import playwright._impl._api_structures
 import playwright._impl._api_types
 import playwright.async_api._generated
+from playwright._impl._assertions import (
+    APIResponseAssertions as APIResponseAssertionsImpl,
+)
 from playwright._impl._assertions import LocatorAssertions as LocatorAssertionsImpl
 from playwright._impl._assertions import PageAssertions as PageAssertionsImpl
 from playwright.async_api._context_manager import PlaywrightContextManager
@@ -31,6 +34,7 @@
     APIRequest,
     APIRequestContext,
     APIResponse,
+    APIResponseAssertions,
     Browser,
     BrowserContext,
     BrowserType,
@@ -83,23 +87,30 @@ def async_playwright() -> PlaywrightContextManager:
 
 
 @overload
-def expect(page_or_locator: Page) -> PageAssertions:
+def expect(actual: Page) -> PageAssertions:
+    ...
+
+
+@overload
+def expect(actual: Locator) -> LocatorAssertions:
     ...
 
 
 @overload
-def expect(page_or_locator: Locator) -> LocatorAssertions:
+def expect(actual: APIResponse) -> APIResponseAssertions:
     ...
 
 
 def expect(
-    page_or_locator: Union[Page, Locator]
-) -> Union[PageAssertions, LocatorAssertions]:
-    if isinstance(page_or_locator, Page):
-        return PageAssertions(PageAssertionsImpl(page_or_locator._impl_obj))
-    elif isinstance(page_or_locator, Locator):
-        return LocatorAssertions(LocatorAssertionsImpl(page_or_locator._impl_obj))
-    raise ValueError(f"Unsupported type: {type(page_or_locator)}")
+    actual: Union[Page, Locator, APIResponse]
+) -> Union[PageAssertions, LocatorAssertions, APIResponseAssertions]:
+    if isinstance(actual, Page):
+        return PageAssertions(PageAssertionsImpl(actual._impl_obj))
+    elif isinstance(actual, Locator):
+        return LocatorAssertions(LocatorAssertionsImpl(actual._impl_obj))
+    elif isinstance(actual, APIResponse):
+        return APIResponseAssertions(APIResponseAssertionsImpl(actual._impl_obj))
+    raise ValueError(f"Unsupported type: {type(actual)}")
 
 
 __all__ = [

playwright/async_api/_generated.py

@@ -43,6 +43,9 @@
     ViewportSize,
 )
 from playwright._impl._api_types import Error
+from playwright._impl._assertions import (
+    APIResponseAssertions as APIResponseAssertionsImpl,
+)
 from playwright._impl._assertions import LocatorAssertions as LocatorAssertionsImpl
 from playwright._impl._assertions import PageAssertions as PageAssertionsImpl
 from playwright._impl._async_base import (
@@ -15597,3 +15600,41 @@ async def not_to_be_focused(self, *, timeout: float = None) -> NoneType:
 
 
 mapping.register(LocatorAssertionsImpl, LocatorAssertions)
+
+
+class APIResponseAssertions(AsyncBase):
+    async def to_be_ok(self) -> NoneType:
+        """APIResponseAssertions.to_be_ok
+
+        Ensures the response status code is within [200..299] range.
+
+        ```py
+        from playwright.async_api import expect
+
+        # ...
+        await expect(response).to_be_ok()
+        ```
+        """
+        __tracebackhide__ = True
+
+        return mapping.from_maybe_impl(
+            await self._async(
+                "api_response_assertions.to_be_ok", self._impl_obj.to_be_ok()
+            )
+        )
+
+    async def not_to_be_ok(self) -> NoneType:
+        """APIResponseAssertions.not_to_be_ok
+
+        The opposite of `a_pi_response_assertions.to_be_ok()`.
+        """
+        __tracebackhide__ = True
+
+        return mapping.from_maybe_impl(
+            await self._async(
+                "api_response_assertions.not_to_be_ok", self._impl_obj.not_to_be_ok()
+            )
+        )
+
+
+mapping.register(APIResponseAssertionsImpl, APIResponseAssertions)