feat: added Download.saveAs(path) (#147)

Author: mxschmitt

playwright/async_api.py

@@ -2939,6 +2939,18 @@ async def path(self) -> typing.Union[str, NoneType]:
         """
         return mapping.from_maybe_impl(await self._impl_obj.path())
 
+    async def saveAs(self, path: typing.Union[pathlib.Path, str]) -> NoneType:
+        """Download.saveAs
+
+        Saves the download to a user-specified path.
+
+        Parameters
+        ----------
+        path : Union[pathlib.Path, str]
+            Path where the download should be saved.
+        """
+        return mapping.from_maybe_impl(await self._impl_obj.saveAs(path=path))
+
 
 mapping.register(DownloadImpl, Download)
 

playwright/download.py

@@ -12,7 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Dict, Optional
+from pathlib import Path
+from typing import Dict, Optional, Union
 
 from playwright.connection import ChannelOwner
 
@@ -39,3 +40,7 @@ async def failure(self) -> Optional[str]:
 
     async def path(self) -> Optional[str]:
         return await self._channel.send("path")
+
+    async def saveAs(self, path: Union[Path, str]) -> None:
+        path = str(Path(path))
+        return await self._channel.send("saveAs", dict(path=path))

playwright/sync_api.py

@@ -3065,6 +3065,18 @@ def path(self) -> typing.Union[str, NoneType]:
         """
         return mapping.from_maybe_impl(self._sync(self._impl_obj.path()))
 
+    def saveAs(self, path: typing.Union[pathlib.Path, str]) -> NoneType:
+        """Download.saveAs
+
+        Saves the download to a user-specified path.
+
+        Parameters
+        ----------
+        path : Union[pathlib.Path, str]
+            Path where the download should be saved.
+        """
+        return mapping.from_maybe_impl(self._sync(self._impl_obj.saveAs(path=path)))
+
 
 mapping.register(DownloadImpl, Download)
 

tests/async/test_download.py

@@ -14,11 +14,12 @@
 import asyncio
 import os
 from asyncio.futures import Future
+from pathlib import Path
 from typing import Optional
 
 import pytest
 
-from playwright import Error as PlaywrightError
+from playwright import Error
 from playwright.async_api import Browser, Page
 
 
@@ -53,10 +54,10 @@ async def test_should_report_downloads_with_acceptDownloads_false(page: Page, se
     download = (await asyncio.gather(page.waitForEvent("download"), page.click("a")))[0]
     assert download.url == f"{server.PREFIX}/downloadWithFilename"
     assert download.suggestedFilename == "file.txt"
-    error: Optional[PlaywrightError] = None
+    error: Optional[Error] = None
     try:
         await download.path()
-    except PlaywrightError as exc:
+    except Error as exc:
         error = exc
     assert "acceptDownloads" in await download.failure()
     assert error
@@ -73,6 +74,107 @@ async def test_should_report_downloads_with_acceptDownloads_true(browser, server
     await page.close()
 
 
+async def test_should_save_to_user_specified_path(tmpdir: Path, browser, server):
+    page = await browser.newPage(acceptDownloads=True)
+    await page.setContent(f'<a href="{server.PREFIX}/download">download</a>')
+    [download, _] = await asyncio.gather(page.waitForEvent("download"), page.click("a"))
+    user_path = tmpdir / "download.txt"
+    await download.saveAs(user_path)
+    assert user_path.exists()
+    assert user_path.read_text("utf-8") == "Hello world"
+    await page.close()
+
+
+async def test_should_save_to_user_specified_path_without_updating_original_path(
+    tmpdir, browser, server
+):
+    page = await browser.newPage(acceptDownloads=True)
+    await page.setContent(f'<a href="{server.PREFIX}/download">download</a>')
+    [download, _] = await asyncio.gather(page.waitForEvent("download"), page.click("a"))
+    user_path = tmpdir / "download.txt"
+    await download.saveAs(user_path)
+    assert user_path.exists()
+    assert user_path.read_text("utf-8") == "Hello world"
+
+    originalPath = Path(await download.path())
+    assert originalPath.exists()
+    assert originalPath.read_text("utf-8") == "Hello world"
+    await page.close()
+
+
+async def test_should_save_to_two_different_paths_with_multiple_saveAs_calls(
+    tmpdir, browser, server
+):
+    page = await browser.newPage(acceptDownloads=True)
+    await page.setContent(f'<a href="{server.PREFIX}/download">download</a>')
+    [download, _] = await asyncio.gather(page.waitForEvent("download"), page.click("a"))
+    user_path = tmpdir / "download.txt"
+    await download.saveAs(user_path)
+    assert user_path.exists()
+    assert user_path.read_text("utf-8") == "Hello world"
+
+    anotheruser_path = tmpdir / "download (2).txt"
+    await download.saveAs(anotheruser_path)
+    assert anotheruser_path.exists()
+    assert anotheruser_path.read_text("utf-8") == "Hello world"
+    await page.close()
+
+
+async def test_should_save_to_overwritten_filepath(tmpdir: Path, browser, server):
+    page = await browser.newPage(acceptDownloads=True)
+    await page.setContent(f'<a href="{server.PREFIX}/download">download</a>')
+    [download, _] = await asyncio.gather(page.waitForEvent("download"), page.click("a"))
+    user_path = tmpdir / "download.txt"
+    await download.saveAs(user_path)
+    assert len(list(Path(tmpdir).glob("*.*"))) == 1
+    await download.saveAs(user_path)
+    assert len(list(Path(tmpdir).glob("*.*"))) == 1
+    assert user_path.exists()
+    assert user_path.read_text("utf-8") == "Hello world"
+    await page.close()
+
+
+async def test_should_create_subdirectories_when_saving_to_non_existent_user_specified_path(
+    tmpdir, browser, server
+):
+    page = await browser.newPage(acceptDownloads=True)
+    await page.setContent(f'<a href="{server.PREFIX}/download">download</a>')
+    [download, _] = await asyncio.gather(page.waitForEvent("download"), page.click("a"))
+    nested_path = tmpdir / "these" / "are" / "directories" / "download.txt"
+    await download.saveAs(nested_path)
+    assert nested_path.exists()
+    assert nested_path.read_text("utf-8") == "Hello world"
+    await page.close()
+
+
+async def test_should_error_when_saving_with_downloads_disabled(
+    tmpdir, browser, server
+):
+    page = await browser.newPage(acceptDownloads=False)
+    await page.setContent(f'<a href="{server.PREFIX}/download">download</a>')
+    [download, _] = await asyncio.gather(page.waitForEvent("download"), page.click("a"))
+    user_path = tmpdir / "download.txt"
+    with pytest.raises(Error) as exc:
+        await download.saveAs(user_path)
+    assert (
+        "Pass { acceptDownloads: true } when you are creating your browser context"
+        in exc.value.message
+    )
+    await page.close()
+
+
+async def test_should_error_when_saving_after_deletion(tmpdir, browser, server):
+    page = await browser.newPage(acceptDownloads=True)
+    await page.setContent(f'<a href="{server.PREFIX}/download">download</a>')
+    [download, _] = await asyncio.gather(page.waitForEvent("download"), page.click("a"))
+    user_path = tmpdir / "download.txt"
+    await download.delete()
+    with pytest.raises(Error) as exc:
+        await download.saveAs(user_path)
+    assert "Download already deleted. Save before deleting." in exc.value.message
+    await page.close()
+
+
 async def test_should_report_non_navigation_downloads(browser, server):
     # Mac WebKit embedder does not download in this case, although Safari does.
     def handle_download(request):