Python selenium.webdriver 模块，Remote() 实例源码

def _webdriver(self):
        className = self._driverClassName()
        self._remoteUsername = self._remoteUsername()
        remoteURL = self._remoteURL()
        capabilities = self._capabilities()
        if remoteURL:
            self.log("RemoteURL")
            # Remote driver requires a executor (typically a Remote URL)
            browserProfile = FirefoxProfile()
            return webdriver.Remote(command_executor = remoteURL, keep_alive = True, \
                browser_profile = browserProfile, desired_capabilities = capabilities)
        clazz = self._driverClass(className)
        try:
            self.log("" + className + "(capabilities)")
            return clazz(desired_capabilities = capabilities)
        except Exception:
            self.log("Setting up " + className)
            if className == 'PhantomJS':
                # PhantomJS cannot handle capabilities
                self.log("Creating " + className)
                return clazz()
            else:
                # Firefox and Ie drivers have different name for desired capabilities parameter
                #return clazz(capabilities = capabilities)
                return clazz()

def getDriver(self):
        count = 0
        while(True):
            try:
                ce = "http://%s:4444/wd/hub" % os.environ["HUB"]
                driver = webdriver.Remote(
                    command_executor=ce,
                    desired_capabilities={
                        "browserName": os.environ.get("browser", "firefox"),
                        "platform": "Linux"
                    }
                )
                return driver
            except WebDriverException as e:
                s = "%s" % e
                print("Got exception %s" % s)
                print("%s" % dir(s))
                if "Empty pool of VM for setup Capabilities" not in s:
                    raise
                time.sleep(5)
            if count == 60:
                raise Exception("Time out trying to get a browser")
            count += 1

def set_selenium_remote_session(self, selenium_url=''):
        """Starts remote session for a selenium server.
         Useful for docker setup."""
        if self.aborting:
            return self

        if self.use_firefox:
            self.browser = webdriver.Remote(
                command_executor=selenium_url,
                desired_capabilities=DesiredCapabilities.FIREFOX)
        else:
            self.browser = webdriver.Remote(
                command_executor=selenium_url,
                desired_capabilities=DesiredCapabilities.CHROME)

        self.logger.info('Session started - %s'
                         % (datetime.now().strftime('%Y-%m-%d %H:%M:%S')))

        return self

def __init__(self):
        try:
            self.driver = webdriver.Remote(command_executor='http://firefox:4444/wd/hub',
                                           desired_capabilities=DesiredCapabilities.FIREFOX)
        except URLError as e:
            print("Failed to connect!")
            raise e

def __init__(self):
        try:
            self.driver = webdriver.Remote(command_executor='http://chrome:4444/wd/hub',
                                           desired_capabilities=DesiredCapabilities.CHROME)
        except URLError as e:
            print("Failed to connect!")
            raise e

def start(self):
        self.driver = webdriver.Remote(
           command_executor=self.remote,
           desired_capabilities=self.browser_type)

def _create_remote_web_driver(self, capabilities_type, webdriver_remote_url,
                                  desired_capabilities=None, profile=None):
        """parses the string based desired_capabilities if neccessary and
        creates the associated remote web driver"""

        desired_capabilities_object = capabilities_type.copy()

        if isinstance(desired_capabilities, (str, unicode)):
            desired_capabilities = self._parse_capabilities_string(desired_capabilities)

        desired_capabilities_object.update(desired_capabilities or {})

        return webdriver.Remote(desired_capabilities=desired_capabilities_object,
                                command_executor=str(webdriver_remote_url),
                                browser_profile=profile)

def test_bot_using_webdriver_phantomjs(self):
        """Create a bot."""
        config.ready = True
        config.extend({
            'webdriver_type': u'phantomjs',
            'webdriver_url': pytest.config.getvalue("webdriver").decode("ascii")
        })
        from dallinger.bots import BotBase
        bot = BotBase("http://dallinger.io")
        assert isinstance(bot.driver, webdriver.Remote)
        assert bot.driver.capabilities['browserName'] == 'phantomjs'

def test_bot_using_webdriver_firefox(self):
        """Create a bot."""
        config.ready = True
        config.extend({
            'webdriver_type': u'firefox',
            'webdriver_url': pytest.config.getvalue("webdriver").decode("ascii")
        })
        from dallinger.bots import BotBase
        bot = BotBase("http://dallinger.io")
        assert isinstance(bot.driver, webdriver.Remote)
        assert bot.driver.capabilities['browserName'] == 'firefox'

def test_bot_using_webdriver_chrome(self):
        """Create a bot."""
        config.ready = True
        config.extend({
            'webdriver_type': u'chrome',
            'webdriver_url': pytest.config.getvalue("webdriver").decode("ascii")
        })
        from dallinger.bots import BotBase
        bot = BotBase("http://dallinger.io")
        assert isinstance(bot.driver, webdriver.Remote)
        assert bot.driver.capabilities['browserName'] == 'chrome'

def driver(self):
        """Returns a Selenium WebDriver instance of the type requested in the
        configuration."""
        from dallinger.config import get_config
        config = get_config()
        if not config.ready:
            config.load()
        driver_url = config.get('webdriver_url', None)
        driver_type = config.get('webdriver_type', 'phantomjs').lower()

        if driver_url:
            capabilities = {}
            if driver_type == 'firefox':
                capabilities = webdriver.DesiredCapabilities.FIREFOX
            elif driver_type == 'chrome':
                capabilities = webdriver.DesiredCapabilities.CHROME
            elif driver_type == 'phantomjs':
                capabilities = webdriver.DesiredCapabilities.PHANTOMJS
            else:
                raise ValueError(
                    'Unsupported remote webdriver_type: {}'.format(driver_type))
            driver = webdriver.Remote(
                desired_capabilities=capabilities,
                command_executor=driver_url
            )
        elif driver_type == 'phantomjs':
            driver = webdriver.PhantomJS()
        elif driver_type == 'firefox':
            driver = webdriver.Firefox()
        elif driver_type == 'chrome':
            driver = webdriver.Chrome()
        else:
            raise ValueError(
                'Unsupported webdriver_type: {}'.format(driver_type))
        driver.set_window_size(1024, 768)
        logger.info("Created {} webdriver.".format(driver_type))
        return driver

def __init__(self):
        self.driver = webdriver.Remote(
        command_executor='http://127.0.0.1:4444/wd/hub',
        desired_capabilities=DesiredCapabilities.FIREFOX)
        pass

def spawnBrowser(self):
        """Spawns the browser to use when internetting."""
        self.driver = webdriver.Remote(
            command_executor='http://127.0.0.1:4444/wd/hub',
            desired_capabilities=DesiredCapabilities.FIREFOX)

def driver(self, package=None, attach=True, activity=None, process=None):
        """
        Args:
            - package(string): default current running app
            - attach(bool): default true, Attach to an already-running app instead of launching the app with a clear data directory
            - activity(string): Name of the Activity hosting the WebView.
            - process(string): Process name of the Activity hosting the WebView (as given by ps). 
                If not given, the process name is assumed to be the same as androidPackage.

        Returns:
            selenium driver
        """
        app = self._d.current_app()
        capabilities = {
            'chromeOptions': {
                'androidDeviceSerial': self._d.serial,
                'androidPackage': package or app.package,
                'androidUseRunningApp': attach,
                'androidProcess': process or app.package,
                'androidActivity': activity or app.activity,
            }
        }

        try:
            dr = webdriver.Remote('http://localhost:%d' % self._port, capabilities)
        except URLError:
            self._launch_webdriver()
            dr = webdriver.Remote('http://localhost:%d' % self._port, capabilities)

        # always quit driver when done
        atexit.register(dr.quit)
        return dr

def initialize_driver(self, driver=None):
        if self.command_executor:
            chrome_options = Options()
            chrome_options.add_argument("--disable-notifications")
            if self.proxy:
                chrome_options.add_argument('--proxy-server=%s' % self.proxy)
            self.driver = webdriver.Remote(
                command_executor=self.command_executor,
                desired_capabilities=chrome_options.to_capabilities()
            )
        else:
            if self.which_driver == 'phantomjs':
                dcap = dict(DesiredCapabilities.PHANTOMJS)
                dcap["phantomjs.page.settings.userAgent"] = (
                    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/53 "
                    "(KHTML, like Gecko) Chrome/15.0.87"
                )
                driver = webdriver.PhantomJS(desired_capabilities=dcap)
                driver.set_window_size(1400, 1000)
                self.driver = driver
            elif self.which_driver == 'chrome':
                chrome_options = Options()
                chrome_options.add_argument("--disable-notifications")
                if self.proxy:
                    chrome_options.add_argument('--proxy-server=%s' % self.proxy)
                self.driver = webdriver.Chrome(chrome_options=chrome_options)
            # otherwise use the driver passed in
            else:
                self.driver = driver
        # set page load timeout
        self.driver.set_page_load_timeout(time_to_wait=240)

def selenium(request: FixtureRequest) -> webdriver.Remote:
    if request.config.getoption('--local-chrome'):
        driver = initialize_local_chrome()
        request.addfinalizer(lambda: deinitialize_selenium(driver))
        return driver
    driver = request.getfixturevalue('selenium')  # type: webdriver.Remote
    yield driver
    # https://github.com/SeleniumHQ/docker-selenium/issues/87#issuecomment-286231268
    driver.close()

def setup(self, base_url: str, console_ip: str, pages: PageObjectContainer, selenium: webdriver.Remote):
        self.base_url = base_url
        self.console_ip = console_ip
        self.console_port = 5673
        self.ConnectPage = pages.connect_page
        self.OverviewPage = pages.overview_page
        self.EntitiesPage = pages.entities_page
        self.selenium = selenium
        self.test_name = None
        return self

def setup(self, base_url: str, console_ip: str, pages: PageObjectContainer, selenium: webdriver.Remote):
        self.base_url = base_url
        self.console_ip = console_ip
        self.LogsPage = pages.logs_page
        self.selenium = selenium
        self.test_name = None
        return self

def __init__(self, selenium: webdriver.Remote):
        self.selenium = selenium

def wait(cls, selenium: webdriver.Remote):
        locator = (By.CSS_SELECTOR, '.active a[ng-href="#/logs"]')
        WebDriverWait(selenium, 30).until(EC.visibility_of_element_located(locator))

def wait(cls, selenium: webdriver.Remote):
        # wait for Connect link in the top bar to be active
        locator = (By.CSS_SELECTOR, '.active a[ng-href="#/{}/connect"]'.format(PLUGIN_NAME))
        WebDriverWait(selenium, 30).until(EC.presence_of_element_located(locator))

def __init__(self, selenium: webdriver.Remote):
        super().__init__(selenium)
        self.node_count = None  # type: int

def __init__(self, selenium: webdriver.Remote):
        super().__init__(selenium)
        self.node_count = 5
        # Wait for at least one node expander to appear
        _ = self.wait_locate_visible_element(self.expander_locator)

def wait(cls, selenium: webdriver.Remote):
        # wait for Overview link in the top bar to be active
        locator = (By.CSS_SELECTOR, '.active a[ng-href="#/{}/overview"]'.format(PLUGIN_NAME))
        WebDriverWait(selenium, 30).until(EC.presence_of_element_located(locator))

def __init__(self, selenium: webdriver.Remote):
        super().__init__(selenium)
        self.node_count = 17
        # Wait for at least one node expander to appear
        _ = self.wait_locate_visible_element(self.expander_locator)

def __init__(self, selenium: webdriver.Remote):
        super().__init__(selenium)
        # Wait for at least one node expander to appear
        _ = self.wait_locate_visible_element(self.expander_locator)

def wait(cls, selenium: webdriver.Remote):
        # wait for Overview link in the top bar to be active
        locator = (By.CSS_SELECTOR, 'li.active > a[ng-href="#!/overview"]')
        WebDriverWait(selenium, 30).until(EC.presence_of_element_located(locator))

def __init__(self, selenium: webdriver.Remote):
        super().__init__(selenium)
        # Wait for at least one node expander to appear
        _ = self.wait_locate_visible_element(self.expander_locator)

def wait(cls, selenium: webdriver.Remote):
        # wait for Overview link in the top bar to be active
        locator = (By.CSS_SELECTOR, 'li.active > a[ng-href="#!/list"]')
        WebDriverWait(selenium, 30).until(EC.presence_of_element_located(locator))

def __init__(self, selenium: webdriver.Remote):
        super().__init__(selenium)

def setup(self, base_url: str, console_ip: str, pages: PageObjectContainer, selenium: webdriver.Remote):
        self.base_url = base_url
        self.console_ip = console_ip
        self.ConnectPage = pages.connect_page
        self.OverviewPage = pages.overview_page
        self.selenium = selenium
        self.test_name = None
        return self

def setup(self, base_url: str, console_ip: str, pages: PageObjectContainer, selenium: webdriver.Remote):
        self.base_url = base_url
        self.console_ip = console_ip
        self.console_port = 5673
        self.ConnectPage = pages.connect_page
        self.OverviewPage = pages.overview_page
        self.selenium = selenium
        self.test_name = None
        return self

def initialize_remote_selenium():

    selenium_url = 'http://192.168.33.10:4444/wd/hub'
    capabilities = {'browserName': 'chrome'}

    # selenium_url = 'http://127.0.0.1:4444/wd/hub'
    # capabilities = {
    #     'browserName': 'jbrowserdriver',
    #     'version': '1',
    #     'platform': 'ANY',
    # }

    driver = webdriver.Remote(selenium_url, desired_capabilities=capabilities)
    driver.implicitly_wait(10)
    return driver

def deinitialize_selenium(driver: webdriver.Remote):
    driver.close()

def start(self):
        try:

            if self.browser == "Firefox":
                driver = webdriver.Firefox()
                driver.get(self.url)
                return driver

            elif self.browser == "IE":
                driver = webdriver.Ie()
                driver.get(self.url)

                return driver
            elif self.browser == "HtmlUnit":
                # ????? selenium-server
                driver = webdriver.Remote("http://127.0.0.1:4444/wd/hub/",
                                          desired_capabilities=webdriver.DesiredCapabilities.HTMLUNITWITHJS)
                return driver

            elif self.browser =="Chrome":
                driver = webdriver.Chrome()
                driver.get(self.url)
                return driver
            else: # ??
                #

                driver = webdriver.Remote(command_executor='http://192.168.50.162:1111/wd/hub',
                                          desired_capabilities=webdriver.DesiredCapabilities.CHROME)
                driver.get(self.url)
                return driver

        except:

            print(str(sys.exc_info()))

def start_headless(self):
        """Headless Chrome initiator."""
        print('Start headless browser')
        option_set = options.Options()
        option_set.add_arguments("test-type")
        option_set.add_arguments("start-maximized")
        option_set.add_arguments("--js-flags=--expose-gc")
        option_set.add_arguments("--enable-precise-memory-info")
        option_set.add_argument('headless')
        option_set.add_argument('disable-notifications')
        option_set.add_argument('disable-gpu')
        option_set.add_argument('disable-infobars')
        option_set.add_arguments("--disable-default-apps")
        option_set.add_arguments("test-type=browser")
        option_set.add_experimental_option(
                'prefs', {
                    'credentials_enable_service': False,
                    'profile': {
                        'password_manager_enabled': False
                    }
                }
            )
        option_set.binary_location = os.getenv(
            'CHROME_CANARY',
            '/Applications/Google Chrome Canary.app' +
            '/Contents/MacOS/Google Chrome Canary'
        )
        webdriver_service = service.Service(
            os.getenv(
                'CHROMEDRIVER',
                '/Applications/chromedriver'
            )
        )
        webdriver_service.start()
        print('Service started; returning Remote webdriver')
        return webdriver.Remote(
            webdriver_service.service_url,
            option_set.to_capabilities()
        )

def start_opera(self, location):
        """Opera initiator."""
        webdriver_service = service.Service(location)
        webdriver_service.start()
        return webdriver.Remote(
            webdriver_service.service_url,
            DesiredCapabilities.OPERA.copy()
        )

def start_htmlunit(self, implicity_wait_timeout=60):
        browser = webdriver.Remote(desired_capabilities=DesiredCapabilities.HTMLUNIT)
        browser.implicitly_wait(implicity_wait_timeout)
        browser.get(self.url)

def init_driver_chrome():
    '''
    Initialize chrome browser. it needs a webdriver service
    '''
    import selenium.webdriver.chrome.service as service
    global service  # global variable because its needed in quit_driver()
    service = service.Service('chromedriver')
    service.start()
    print "service initialized for chrome browser"

    capabilities = {'chrome.loadAsync': 'true'}
    driver = webdriver.Remote(service.service_url, capabilities)
    driver.wait = WebDriverWait(driver, 5)
    driver.implicitly_wait(10)
    return driver

def start(self):

        if self.is_started is False:
            self.service.start()

            capabilities = {
                'chrome.binary':
                '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome'
            }
            self.browser = (
                webdriver.Remote(self.service.service_url, capabilities))
            self.is_started = True

        return

def connect_to_selenium(port, t=30, c=1):
  global Browser

  if c >= t:
    return False

  try:
    if verbose:
      print ('Connecting to docker selenium server '
      'http://127.0.0.1:%s/wd/hub [%s/%s]...') % (str(port), str(c), str(t))

    if driver == 'chrome':
      desired_caps = DesiredCapabilities.CHROME
    elif driver == 'firefox':
      desired_caps = DesiredCapabilities.FIREFOX

    Browser = webdriver.Remote(
      command_executor='http://127.0.0.1:%s/wd/hub' % str(port),
      desired_capabilities=desired_caps
    )
  except:
    c += 1
    sleep(1)
    connect_to_selenium(port, t, c)

  return True

def selenium_setup(self):
        capabilities = {'browserName': 'chrome'}
        try:
            self.pydriver = webdriver.Remote(self.config_dict[TABLEAU_SELENIUM_URL], capabilities)
        except (WebDriverException, Exception), we:
            ACLogger().get_logger().error('unable to start webdriver.Remote: %s ' % we)
            return False
        return True

def driver(base_url, osinfo, browserinfo):
    ostype, os_version = osinfo
    browser, browser_version = browserinfo
    buildid = '{0}_{1}_{2}_{3}'.format(ostype.lower().replace(' ', '_'),
                                       os_version.lower().replace(' ', '_'), browser, browser_version)
    # skip some combinations
    if os_version == 'Sierra' and browser == 'safari' and browser_version == '9.1':
        pytest.skip('cannot have Sierra running safari 9.1')
    elif os_version == 'El Capitan' and browser == 'safari' and browser_version == '10':
        pytest.skip('cannot have El Capitan running safari 10')

    # set driver
    if browserstack:
        username = os.environ.get('BROWSERSTACK_USER', None)
        access = os.environ.get('BROWSERSTACK_ACCESS_KEY', None)
        desired_cap = {'os': ostype, 'os_version': os_version, 'browser': browser, 'build': buildid,
                       'browser_version': browser_version, 'project': 'marvin', 'resolution': '1920x1080'}
        desired_cap['browserstack.local'] = True
        desired_cap['browserstack.debug'] = True
        desired_cap['browserstack.localIdentifier'] = os.environ['BROWSERSTACK_LOCAL_IDENTIFIER']
        driver = webdriver.Remote(
            command_executor='http://{0}:{1}@hub.browserstack.com:80/wd/hub'.format(username, access),
            desired_capabilities=desired_cap)
    else:
        if browser == 'chrome':
            driver = webdriver.Chrome()
        elif browser == 'firefox':
            driver = webdriver.Firefox()
        elif browser == 'safari':
            driver = webdriver.Safari()

    driver.get(base_url)
    yield driver
    # teardown
    driver.quit()

def start_driver_session(context: Context, session_name: str):
    desired_capabilities = context.desired_capabilities
    desired_capabilities["name"] = session_name
    if CONFIG["hub_url"]:
        context.driver = webdriver.Remote(
            desired_capabilities=desired_capabilities,
            command_executor=CONFIG["hub_url"])
    else:
        browser_name = CONFIG["environments"][0]["browser"]
        drivers = {
            "chrome": webdriver.Chrome,
            "edge": webdriver.Edge,
            "firefox": webdriver.Firefox,
            "ie": webdriver.Ie,
            "phantomjs": webdriver.PhantomJS,
            "safari": webdriver.Safari,
        }
        # start the browser
        context.driver = drivers[browser_name.lower()]()
    context.driver.set_page_load_timeout(time_to_wait=27)
    try:
        context.driver.maximize_window()
        logging.debug("Maximized the window.")
    except WebDriverException:
        logging.debug("Failed to maximize the window.")
        try:
            context.driver.set_window_size(1600, 1200)
            logging.warning("Set window size to 1600x1200")
        except WebDriverException:
            logging.warning("Failed to set window size, will continue as is")
    logging.debug("Browser Capabilities: %s", context.driver.capabilities)

def get_appium_driver(url, desired_capabilities) -> Remote:
    """
    Return the same instance to the Appium driver.

    :param url: the URL (address and port) where the service runs.
    :param desired_capabilities: session configuration data.
    :return: returns the SAME instance of the driver
    """
    return SingletonFactory.build(Remote,
                                  command_executor=url,
                                  desired_capabilities=desired_capabilities)