chore: minor cleanup for coverage (#115)

Author: mxschmitt

playwright/async_api.py

@@ -4664,13 +4664,15 @@ def expect_file_chooser(
 
     def expect_request(
         self,
+        url: typing.Union[str, typing.Pattern, typing.Callable[[str], bool]] = None,
         predicate: typing.Union[typing.Callable[["Request"], bool]] = None,
         timeout: int = None,
     ) -> AsyncEventContextManager["Request"]:
         return AsyncEventContextManager(self, "request", predicate, timeout)
 
     def expect_response(
         self,
+        url: typing.Union[str, typing.Pattern, typing.Callable[[str], bool]] = None,
         predicate: typing.Union[typing.Callable[["Response"], bool]] = None,
         timeout: int = None,
     ) -> AsyncEventContextManager["Response"]:

playwright/event_context_manager.py

@@ -58,5 +58,5 @@ def __init__(
     async def __aenter__(self) -> EventInfoImpl[T]:
         return self._event
 
-    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+    async def __aexit__(self, *args: Any) -> None:
         await self._event.value

playwright/network.py

@@ -53,7 +53,7 @@ def postData(self) -> Optional[str]:
         b64_content = self._initializer.get("postData")
         if not b64_content:
             return None
-        return base64.b64decode(bytes(b64_content, "utf-8")).decode()
+        return base64.b64decode(b64_content).decode()
 
     @property
     def headers(self) -> Dict[str, str]:
@@ -130,7 +130,7 @@ async def continue_(
         if headers:
             overrides["headers"] = serialize_headers(headers)
         if isinstance(postData, str):
-            overrides["postData"] = base64.b64encode(bytes(postData, "utf-8")).decode()
+            overrides["postData"] = base64.b64encode(postData.encode()).decode()
         elif isinstance(postData, bytes):
             overrides["postData"] = base64.b64encode(postData).decode()
         await self._channel.send("continue", cast(Any, overrides))
@@ -173,7 +173,7 @@ async def body(self) -> bytes:
 
     async def text(self) -> str:
         content = await self.body()
-        return content.decode("utf-8")
+        return content.decode()
 
     async def json(self) -> Union[Dict, List]:
         return json.loads(await self.text())

playwright/page.py

@@ -395,12 +395,9 @@ async def waitForNavigation(
     ) -> Optional[Response]:
         return await self._main_frame.waitForNavigation(**locals_to_params(locals()))
 
-    async def waitForRequest(
-        self,
-        url: URLMatch = None,
-        predicate: Callable[[Request], bool] = None,
-        timeout: int = None,
-    ) -> Request:
+    def _get_request_predicate(
+        self, url: Optional[URLMatch], predicate: Optional[Callable[[Request], bool]]
+    ) -> Callable[[Any], bool]:
         matcher = URLMatcher(url) if url else None
 
         def my_predicate(request: Request) -> bool:
@@ -410,10 +407,34 @@ def my_predicate(request: Request) -> bool:
                 return predicate(request)
             return True
 
+        return my_predicate
+
+    def _get_response_predicate(
+        self, url: Optional[URLMatch], predicate: Optional[Callable[[Response], bool]]
+    ) -> Callable[[Any], bool]:
+        matcher = URLMatcher(url) if url else None
+
+        def my_predicate(response: Response) -> bool:
+            if matcher:
+                return matcher.matches(response.url)
+            if predicate:
+                return predicate(response)
+            return True
+
+        return my_predicate
+
+    async def waitForRequest(
+        self,
+        url: URLMatch = None,
+        predicate: Callable[[Request], bool] = None,
+        timeout: int = None,
+    ) -> Request:
+        predicate = self._get_request_predicate(url, predicate)
+
         return cast(
             Request,
             await self.waitForEvent(
-                Page.Events.Request, predicate=my_predicate, timeout=timeout
+                Page.Events.Request, predicate=predicate, timeout=timeout
             ),
         )
 
@@ -423,19 +444,12 @@ async def waitForResponse(
         predicate: Callable[[Response], bool] = None,
         timeout: int = None,
     ) -> Response:
-        matcher = URLMatcher(url) if url else None
-
-        def my_predicate(response: Response) -> bool:
-            if matcher:
-                return matcher.matches(response.url)
-            if predicate:
-                return predicate(response)
-            return True
+        predicate = self._get_response_predicate(url, predicate)
 
         return cast(
             Response,
             await self.waitForEvent(
-                Page.Events.Response, predicate=my_predicate, timeout=timeout
+                Page.Events.Response, predicate=predicate, timeout=timeout
             ),
         )
 
@@ -741,13 +755,21 @@ def expect_file_chooser(
         return EventContextManagerImpl(self, "filechooser", predicate, timeout)
 
     def expect_request(
-        self, predicate: Callable[[Request], bool] = None, timeout: int = None,
+        self,
+        url: URLMatch = None,
+        predicate: Callable[[Request], bool] = None,
+        timeout: int = None,
     ) -> EventContextManagerImpl[Request]:
+        predicate = self._get_request_predicate(url, predicate)
         return EventContextManagerImpl(self, "request", predicate, timeout)
 
     def expect_response(
-        self, predicate: Callable[[Response], bool] = None, timeout: int = None,
+        self,
+        url: URLMatch = None,
+        predicate: Callable[[Response], bool] = None,
+        timeout: int = None,
     ) -> EventContextManagerImpl[Response]:
+        predicate = self._get_response_predicate(url, predicate)
         return EventContextManagerImpl(self, "response", predicate, timeout)
 
     def expect_popup(

playwright/sync_api.py

@@ -4866,13 +4866,15 @@ def expect_file_chooser(
 
     def expect_request(
         self,
+        url: typing.Union[str, typing.Pattern, typing.Callable[[str], bool]] = None,
         predicate: typing.Union[typing.Callable[["Request"], bool]] = None,
         timeout: int = None,
     ) -> EventContextManager["Request"]:
         return EventContextManager(self, "request", predicate, timeout)
 
     def expect_response(
         self,
+        url: typing.Union[str, typing.Pattern, typing.Callable[[str], bool]] = None,
         predicate: typing.Union[typing.Callable[["Response"], bool]] = None,
         timeout: int = None,
     ) -> EventContextManager["Response"]:

playwright/transport.py

@@ -53,11 +53,10 @@ async def _run(self) -> None:
                     data = await self._input.readexactly(to_read)
                     length -= to_read
                     if len(buffer):
-                        buffer = b"".join([buffer, data])
+                        buffer = buffer + data
                     else:
                         buffer = data
-                msg = buffer.decode("utf-8")
-                obj = json.loads(msg)
+                obj = json.loads(buffer)
 
                 if "DEBUGP" in os.environ:  # pragma: no cover
                     print("\x1b[33mRECV>\x1b[0m", json.dumps(obj, indent=2))
@@ -70,7 +69,7 @@ def send(self, message: Dict) -> None:
         msg = json.dumps(message)
         if "DEBUGP" in os.environ:  # pragma: no cover
             print("\x1b[32mSEND>\x1b[0m", json.dumps(message, indent=2))
-        data = bytes(msg, "utf-8")
+        data = msg.encode()
         self._output.write(
             len(data).to_bytes(4, byteorder="little", signed=False) + data
         )

scripts/generate_api.py

@@ -18,7 +18,9 @@
 from typing import (  # type: ignore
     Any,
     List,
+    Match,
     Union,
+    cast,
     get_args,
     get_origin,
     get_type_hints,
@@ -81,10 +83,6 @@ def signature(func: FunctionType, indent: int) -> str:
             else:
                 raise ValueError(f"value {default_value} not recognized")
             tokens.append(f"{name}: {processed} = {default_value}")
-        elif name == "contentScript":
-            tokens.append(f"{name}: {processed} = False")
-        elif name == "arg":
-            tokens.append(f"{name}: typing.Any")
         else:
             tokens.append(f"{name}: {processed}")
     return split.join(tokens)
@@ -113,10 +111,8 @@ def return_type(func: FunctionType) -> str:
 
 
 def short_name(t: Any) -> str:
-    match = re.compile(r"playwright\.[^.]+\.([^']+)").search(str(t))
-    if match:
-        return match.group(1)
-    return str(t)
+    match = cast(Match[str], re.compile(r"playwright\.[^.]+\.([^']+)").search(str(t)))
+    return match.group(1)
 
 
 def return_value(value: Any) -> List[str]:

scripts/generate_async_api.py

@@ -135,5 +135,5 @@ def main() -> None:
     documentation_provider.print_remainder()
 
 
-if __name__ == "__main__":
+if __name__ == "__main__":  # pragma: no cover
     main()

scripts/generate_sync_api.py

@@ -118,5 +118,5 @@ def main() -> None:
     documentation_provider.print_remainder()
 
 
-if __name__ == "__main__":
+if __name__ == "__main__":  # pragma: no cover
     main()

tests/server.py

@@ -106,9 +106,9 @@ def process(self):
                     return
                 file_content = None
                 try:
-                    file_content = open(
-                        static_path / request.path.decode()[1:], "rb"
-                    ).read()
+                    file_content = (
+                        static_path / request.path.decode()[1:]
+                    ).read_bytes()
                 except (FileNotFoundError, IsADirectoryError):
                     request.setResponseCode(HTTPStatus.NOT_FOUND)
                 if file_content: