I'm trying to run two(or more) selenium webdrivers with Python at the same time

I have so far tried using Python's Multiprocessing module, I have used it this way:

def _():
    sets = list()
    pool = Pool()
    for i in range(len(master)):
        driver = setProxy(proxy,f'Tread#{i 1}')
        sets.append(
            [f'Thread#{i 1}',
            driver,
            master[i]]
        )
    for i in range(len(sets)):
        pool.apply_async(enterPoint, args=(sets[i][0],sets[i][1],sets[i][2]))
    pool.close()
    pool.join()

The function above calls setProxy() to get a driver instance with a proxy set to it, which works perfectly and opens a chromedriver len(master) amount of times and accesses a link to check the IP. The sets list is a list of lists that consist of 3 objects that are the Thread number, the driver which will run and a list with the data that the driver will use. Pool's apply_async() should run enterPoint() len(sets) of times, and the args are Thread number, driver and the data Here's enterPoint code:

def enterPoint(thread,driver,accounts):
    print('I exist!')
    for account in accounts:
        cEEPE(thread,driver,account)

But the 'I exist' statement never gets printed out in the CLI I'm running the application at. cEEPE() is where the magic happens. I've tested my code without applying multiprocessing to it and it works as it should. I suspect there's a problem in Pool's apply_async() method, which I might have used it the wrong way.

CodePudding user response：

The code provided in the question is in isolation, so its harder to comment on, but I would set about using this process given the problem described:

1. import multiprocessing & selenium
2. use start & join methods.

This would produce the two (or more) processes that you ask for.

import multiprocessing
from selenium import webdriver

def open_browser(name):
    driver = webdriver.Firefox()
    driver.get("http://www.google.com")
    print(name, driver.title)
    driver.quit()

if __name__ == '__main__':
    process1 = multiprocessing.Process(target=open_browser, args=("Process-1",))
    process2 = multiprocessing.Process(target=open_browser, args=("Process-2",))

    process1.start()
    process2.start()

    process1.join()
    process2.join()

CodePudding user response：

So, I got the code above to work, here's how I fixed it: instead of writing the apply_async() method like this:

pool.apply_async(enterPoint, args=(sets[i][0],sets[i][1],sets[i][2]))

here's how I wrote it:

pool.apply_async(enterPoint(sets[i][0],sets[i][1],sets[i][2]))

But still, this does not fix my issue since I would like enterPoint to run twice at the same time..

CodePudding user response：

It can be done easily with SeleniumBase, which can multi-thread tests (Eg: -n=3 for 3 threads), or even set a proxy server (--proxy=USER:PASS@SERVER:PORT)

pip install seleniumbase, then run with python:

from parameterized import parameterized
from seleniumbase import BaseCase
BaseCase.main(__name__, __file__, "-n=3")

class GoogleTests(BaseCase):
    @parameterized.expand(
        [
            ["Download Python", "Download Python", "img.python-logo"],
            ["Wikipedia", "www.wikipedia.org", "img.central-featured-logo"],
            ["SeleniumBase.io Docs", "SeleniumBase", 'img[alt*="SeleniumB"]'],
        ]
    )
    def test_parameterized_google_search(self, search_key, expected_text, img):
        self.open("https://google.com/ncr")
        self.hide_elements("iframe")
        self.type('input[title="Search"]', search_key   "\n")
        self.assert_text(expected_text, "#search")
        self.click('a:contains("%s")' % expected_text)
        self.assert_element(img)

(This example uses parameterized to turn one test into three different ones.) You can also apply the multi-threading to multiple files, etc.