Verify files status in parallel while using bulk upload

[inbarw]

Your question

I'm trying to check status of files while using bulk upload.
for single upload I'm using this verification:
def verify_entry_process_status(self, entry_name):
file_status = self.file_status_selector.replace("ENTRY_NAME", entry_name)
expect(self.page.locator(file_status)).not_to_have_text("Processing", timeout=100000)
expect(self.page.locator(file_status)).to_have_text("Ready")

I'm waiting for the status text not to be processing and then that it was changed to ready.
So for bulk upload I want to use this function for all the files that I uploaded at the same time.
Tried async, multiprocessing, but I'm pretty stuck of how I should to it (new with playwright/async/multiprocessing)

[rwoll]

Thanks for your report! I'm not fully following. Can you include your previous attempts with some comments in the code so I can get a better sense of what you're trying to do? Thanks!

[inbarw]

Ok, so here are two examples:

1. ThreadPoolExecutor:
   def verify_upload(page, entry_name):
   file_status = '//span[text()="ENTRY_NAME"]/following-sibling::p'.replace("ENTRY_NAME", entry_name)
   expect(page.locator(file_status)).not_to_have_text("Processing", timeout=100000)
   expect(page.locator(file_status)).to_have_text("Ready")
   print(entry_name + " verified")
   print(page.title)

if name == "main":
with sync_playwright() as playwright:
browser = playwright.chromium.launch(channel="chrome", headless=False)
context = browser.new_context()
page = context.new_page()
page.goto("url")

    # login
    page.locator('#Login-username').fill("username")
    page.locator('#Login-password').fill("pass")
    page.locator('#Login-login').click()
    # Navigate to upload page
    page.locator('text=create').click()
    page.locator('//a[@href="https://github.com/uploadmedia" and text()="Upload"]').click()

    # upload files
    page.locator('input[type=file]').set_input_files(file_list)
    #for entry_name in entries_list:
        # p = multiprocessing.Process(target=verify_upload, args=(entry_name,))
        # p.start()
        # processes.append(p)
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        #executor.map(verify_upload, [page, entries_list])
        #for entry_name in entries_list:
        future = executor.submit(verify_upload, page, "video")

        print(future)

    # Joins all the processes
    # for p in processes:
    #     p.join()

3. Async:
   import asyncio
   from playwright.async_api import async_playwright

async def verify_upload(page, entry_name):
file_status = '//span[text()="ENTRY_NAME"]/following-sibling::p'.replace("ENTRY_NAME", entry_name)
await expect(page.locator(file_status)).not_to_have_text("Processing", timeout=100000)
await expect(page.locator(file_status)).to_have_text("Ready")
#await page.locator(file_status).inner_text()
print(entry_name + " verified")
print(page.title)

async def main():
async with async_playwright() as playwright:
browser = await playwright.chromium.launch(channel="chrome", headless=False)
page = await browser.new_page()
await page.goto("url")
# login
await page.locator('#Login-username').fill("user")
await page.locator('#Login-password').fill("pass")
await page.locator('#Login-login').click()
# Navigate to upload page
await page.locator('text=create').click()
await page.locator('//a[@href="/uploadmedia" and text()="Upload"]').click()

    # upload files
    await page.locator('input[type=file]').set_input_files(file_list)
    await asyncio.gather(verify_upload(page, "video"), verify_upload(page, "image"))

asyncio.run(main())

I'm not sure how to pass the page so that it will work, I'm getting error when I'm getting to the expect in verify_upload, also not sure that it's even the correct way, I got a little bit confused how to do it.

[rwoll]

Thanks—I recommend keeping it as simple as possible. Avoid using multiprocessing and ThreadPoolExecutor.

Your line:

await page.locator('input[type=file]').set_input_files(file_list)

looks good. If you run the script to that point and manually looks at the page, is it doing the correct thing, or is it stuck before there?

Once you manually confirm the files are uploading, I recommend verifying the uploads serially. With async, you'll be able to parallelize to a degree, but it'll be helpful to first get it working inline.

[rwoll]

Here's a standalone example for some inspiration:

import asyncio
from playwright.async_api import async_playwright, expect

async def main():
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)
        page = await browser.new_page()
        await page.goto("https://pw.rwoll.dev/file-uploads")
        
        await page.locator("input[type=file]").set_input_files(["a.txt", "b.txt", "c.txt"])
        await page.wait_for_timeout(10000)

        # option 1: simple multi text match
        await expect(page.locator("li")).to_contain_text([
          "Name: a.txt",
          "Name: b.txt",
          "Name: c.txt",
        ])

        # alternatives…

        # define some verification function with multiple verification assertions
        async def verify(name):
          await expect(page.locator("li", has_text=name)).to_be_visible()
          # …
          # any other work you need to do to verify things

        # OPTION 2: Run them serially:
        for name in ["a.txt", "b.txt", "c.txt"]:
          await verify(name) 


        # OPTION 3: Run them concurrently
        tasks = []
        for name in ["a.txt", "b.txt", "c.txt"]:
          tasks.append(verify(name)) # notice no await!

        # c: await all of them to complete
        await asyncio.gather(*tasks)

asyncio.run(main())

If you save files a.txt, b.txt, c.txt next to the script, you can run it and play around.

[inbarw]

Thanks for the detailed answer! I tried with option 3 and it works good, but I have another question.
I'm setting the browser and page in conftest. The fixture function return a page, and then in my test I'm doing all the steps/verifications:
My conftest:
@pytest.fixture(scope="session")
def set_up_browser(playwright):
if local_settings.LOCAL_RUNNING_BROWSER == "chromium":
browser = playwright.chromium.launch(headless=headless_value)
elif local_settings.LOCAL_RUNNING_BROWSER == "firefox":
browser = playwright.firefox.launch(headless=headless_value)
elif local_settings.LOCAL_RUNNING_BROWSER == "chrome":
browser = playwright.chromium.launch(channel="chrome",headless=headless_value)
return browser

@pytest.fixture(scope="session")
def set_up_page_and_test(set_up_browser, request):
browser = set_up_browser
context = browser.new_context()
# Open new page
page = context.new_page()
# Get test number from file name
#test_num = request.keywords.node.items[0].parent.name.replace(".py","").split("")[-1] #When running function
test_num = request.keywords.node.config.args[0].split(".py")[0].split("")[-1] # When running class
# Set up test url/cred/folders
set_up_test(test_num)
# Navigate to page
page.goto(local_settings.LOCAL_SETTINGS_TEST_BASE_URL)

yield page

@pytest.fixture(scope='session')
def set_up_page_and_login(set_up_page_and_test, request):
page = set_up_page_and_test
login_page = Login(page)
user_name = local_settings.LOCAL_SETTINGS_LOGIN_USERNAME
password = local_settings.LOCAL_SETTINGS_LOGIN_PASSWORD
login_page.login_to_site(user_name, password)

yield page

Example test:
class Test:
def test_upload(self, set_up_page_and_login):
# Entries details
file_path_list = [file1, file2]
entries_name = [name1, name2]

    # Pages
    page = set_up_page_and_login
    header = Header(page)
    upload_page = Upload(page)
   
   # tests steps: upload files, verify file etc

So my question is how to combine the set up in conftest with the async function and the test, also how to pass all the variables

[rwoll]

Are you using pytest-playwright already? If so, you'll want to stick sync version of option 2 (i.e. drop the awaits, and don't do concurrent) since pytest-playwright uses sync API.

also how to pass all the variables

Which variables and where?

Also, please format your code in future comments so it's easier to read/follow.

[inbarw]

Yes, I'm using pytest-playwright, but If I'm using option 2 I won't really be able to see what happens to all media that I'm uploading in realtime, until one will finish, other also can, so I won't know if they got the real status.
The variable are media name and paths, it asked for how to combine between the conftest/test and a async option.
So there is no option to use async using pytest-playwright?

[rwoll]

So there is no option to use async using pytest-playwright?

Unfortunately not today—please watch for support: microsoft/playwright-pytest#74. Until then, we recommend using the sync API in the testing.

The variable are media name and paths, it asked for how to combine between the conftest/test and a async option.

Option 2 of #1480 (comment) can be used. You can either make a helper function that takes in your required variables and calls your check, or create a helper function as a fixture that already has access to page fixture. Neither are Playwright-specific, so the the pytest docs, Community Slack, or Stack Overflow would be alternative resources for generic pytest/python pointers.