tests: added missing interception tests (microsoft#122)

Author: mxschmitt

local-requirements.txt

@@ -3,6 +3,8 @@ pytest-asyncio==0.14.0
 pytest-cov==2.10.0
 pytest-sugar==0.9.4
 pytest-xdist==1.33.0
+pixelmatch==0.2.1
+Pillow==7.2.0
 mypy==0.782
 setuptools==49.1.0
 twisted==20.3.0

playwright/async_api.py

@@ -376,6 +376,7 @@ async def fulfill(
         status: int = None,
         headers: typing.Union[typing.Dict[str, str]] = None,
         body: typing.Union[str, bytes] = None,
+        path: typing.Union[str, pathlib.Path] = None,
         contentType: str = None,
     ) -> NoneType:
         """Route.fulfill
@@ -392,12 +393,18 @@ async def fulfill(
             Optional response headers. Header values will be converted to a string.
         body : Optional[str, bytes]
             Optional response body.
+        path : Optional[str, pathlib.Path]
+            Optional file path to respond with. The content type will be inferred from file extension. If `path` is a relative path, then it is resolved relative to current working directory.
         contentType : Optional[str]
             If set, equals to setting `Content-Type` response header.
         """
         return mapping.from_maybe_impl(
             await self._impl_obj.fulfill(
-                status=status, headers=headers, body=body, contentType=contentType
+                status=status,
+                headers=headers,
+                body=body,
+                path=path,
+                contentType=contentType,
             )
         )
 

playwright/impl_to_api_mapping.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import inspect
 from typing import Any, Callable, Dict, List, Optional
 from weakref import WeakKeyDictionary
 
@@ -80,7 +81,10 @@ def wrap_handler(self, handler: Callable[..., None]) -> Callable[..., None]:
             return self._instances[handler]
 
         def wrapper(*args: Any) -> Any:
-            return handler(*list(map(lambda a: self.from_maybe_impl(a), args)))
+            arg_count = len(inspect.signature(handler).parameters)
+            return handler(
+                *list(map(lambda a: self.from_maybe_impl(a), args))[:arg_count]
+            )
 
         self._instances[handler] = wrapper
         return wrapper

playwright/network.py

@@ -14,6 +14,8 @@
 
 import base64
 import json
+import mimetypes
+from pathlib import Path
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast
 
 from playwright.connection import ChannelOwner, from_channel, from_nullable_channel
@@ -100,22 +102,36 @@ async def fulfill(
         status: int = None,
         headers: Dict[str, str] = None,
         body: Union[str, bytes] = None,
+        path: Union[str, Path] = None,
         contentType: str = None,
     ) -> None:
         params = locals_to_params(locals())
-        if contentType:
-            if headers is None:
-                headers = {}
-            headers["Content-Type"] = contentType
-            del params["contentType"]
-        if headers:
-            params["headers"] = serialize_headers(headers)
+        length = 0
         if isinstance(body, str):
             params["body"] = body
             params["isBase64"] = False
+            length = len(body.encode())
         elif isinstance(body, bytes):
             params["body"] = base64.b64encode(body).decode()
             params["isBase64"] = True
+            length = len(body)
+        elif path:
+            del params["path"]
+            file_content = Path(path).read_bytes()
+            params["body"] = base64.b64encode(file_content).decode()
+            params["isBase64"] = True
+            length = len(file_content)
+
+        headers = {k.lower(): str(v) for k, v in params.get("headers", {}).items()}
+        if params.get("contentType"):
+            headers["content-type"] = params["contentType"]
+        elif path:
+            headers["content-type"] = (
+                mimetypes.guess_type(str(Path(path)))[0] or "application/octet-stream"
+            )
+        if length and "content-length" not in headers:
+            headers["content-length"] = str(length)
+        params["headers"] = serialize_headers(headers)
         await self._channel.send("fulfill", params)
 
     async def continue_(

playwright/sync_api.py

@@ -378,6 +378,7 @@ def fulfill(
         status: int = None,
         headers: typing.Union[typing.Dict[str, str]] = None,
         body: typing.Union[str, bytes] = None,
+        path: typing.Union[str, pathlib.Path] = None,
         contentType: str = None,
     ) -> NoneType:
         """Route.fulfill
@@ -394,13 +395,19 @@ def fulfill(
             Optional response headers. Header values will be converted to a string.
         body : Optional[str, bytes]
             Optional response body.
+        path : Optional[str, pathlib.Path]
+            Optional file path to respond with. The content type will be inferred from file extension. If `path` is a relative path, then it is resolved relative to current working directory.
         contentType : Optional[str]
             If set, equals to setting `Content-Type` response header.
         """
         return mapping.from_maybe_impl(
             self._sync(
                 self._impl_obj.fulfill(
-                    status=status, headers=headers, body=body, contentType=contentType
+                    status=status,
+                    headers=headers,
+                    body=body,
+                    path=path,
+                    contentType=contentType,
                 )
             )
         )

tests/conftest.py

@@ -13,15 +13,21 @@
 # limitations under the License.
 
 import asyncio
+import io
 import sys
 
 import pytest
+from PIL import Image
+from pixelmatch import pixelmatch
+from pixelmatch.contrib.PIL import from_PIL_to_raw_data
 
 from playwright import async_playwright
+from playwright.path_utils import get_file_dirname
 
 from .server import test_server
 from .utils import utils as utils_object
 
+_dirname = get_file_dirname()
 
 # Will mark all the tests as async
 def pytest_collection_modifyitems(items):
@@ -213,3 +219,25 @@ def pytest_addoption(parser):
         default=False,
         help="Run tests in headful mode.",
     )
+
+
+@pytest.fixture(scope="session")
+def assert_to_be_golden(browser_name: str):
+    def compare(received_raw: bytes, golden_name: str):
+        golden_file = (_dirname / f"golden-{browser_name}" / golden_name).read_bytes()
+        received_image = Image.open(io.BytesIO(received_raw))
+        golden_image = Image.open(io.BytesIO(golden_file))
+
+        if golden_image.size != received_image.size:
+            pytest.fail("Image size differs to golden image")
+            return
+        diff_pixels = pixelmatch(
+            from_PIL_to_raw_data(received_image),
+            from_PIL_to_raw_data(golden_image),
+            golden_image.size[0],
+            golden_image.size[1],
+            threshold=0.2,
+        )
+        assert diff_pixels == 0
+
+    return compare