diff --git a/.github/scripts/expire_sponsor.py b/.github/scripts/expire_sponsor.py new file mode 100644 index 00000000..8c3e6f34 --- /dev/null +++ b/.github/scripts/expire_sponsor.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import os +from datetime import date, datetime +from pathlib import Path +from zoneinfo import ZoneInfo + + +README = Path("README.md") +START_MARKER = "" +END_MARKER = "" +EXPIRE_ON = date.fromisoformat(os.environ.get("SPONSOR_EXPIRE_ON", "2026-08-28")) +TIMEZONE = ZoneInfo(os.environ.get("SPONSOR_TIMEZONE", "Asia/Shanghai")) + + +def current_date() -> date: + override = os.environ.get("SPONSOR_TODAY") + if override: + return date.fromisoformat(override) + return datetime.now(TIMEZONE).date() + + +def main() -> None: + today = current_date() + if today < EXPIRE_ON: + print(f"Sponsor block is still active until {EXPIRE_ON}; today is {today}.") + return + + text = README.read_text(encoding="utf-8") + start = text.find(START_MARKER) + end = text.find(END_MARKER) + + if start == -1 and end == -1: + print("Sponsor block markers are absent; nothing to remove.") + return + if start == -1 or end == -1 or end < start: + raise SystemExit("Sponsor block markers are incomplete or out of order.") + + end += len(END_MARKER) + updated = text[:start].rstrip() + "\n\n" + text[end:].lstrip() + + if updated == text: + print("Sponsor block is already removed.") + return + + README.write_text(updated, encoding="utf-8") + print(f"Removed sponsor block on {today}.") + + +if __name__ == "__main__": + main() diff --git a/.github/workflows/expire-sponsor.yml b/.github/workflows/expire-sponsor.yml new file mode 100644 index 00000000..5e88f5c1 --- /dev/null +++ b/.github/workflows/expire-sponsor.yml @@ -0,0 +1,33 @@ +name: Expire sponsor block + +on: + schedule: + # 00:05 in Asia/Shanghai. + - cron: "5 16 * * *" + workflow_dispatch: + +permissions: + contents: write + +jobs: + remove-sponsor: + runs-on: ubuntu-latest + steps: + - name: Check out repository + uses: actions/checkout@v4 + + - name: Remove expired sponsor block + run: python3 .github/scripts/expire_sponsor.py + + - name: Commit README update + run: | + if git diff --quiet; then + echo "No README changes to commit." + exit 0 + fi + + git config user.name "github-actions[bot]" + git config user.email "41898282+github-actions[bot]@users.noreply.github.com" + git add README.md + git commit -m "Remove expired sponsor block" + git push diff --git a/.gitignore b/.gitignore index fedead23..721f067a 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ files/* +docs/superpowers/ .DS_Store .idea/* */.idea/* @@ -15,4 +16,4 @@ dist/ media/ .MWebMetaData/ push.sh -assets/ \ No newline at end of file +assets/ diff --git a/README.md b/README.md index c7713c34..210733b2 100644 --- a/README.md +++ b/README.md @@ -110,25 +110,19 @@ FirstSpider|2021-02-09 14:55:14,620|air_spider.py|run|line:80|INFO| 无任务， ## 感谢以下代理赞助商 -### Rapidproxy代理 + - +### LokiProxy - +> 支持免费测试，住宅代理低至 $0.48/GB，移动代理低至 $0.91/GB，不限量住宅代理仅 $12/小时即可体验 ⚡， 支持ASN精准定位及无限并发会话，高稳定、高性价比 + - +![](https://markdown-media.oss-cn-beijing.aliyuncs.com/2026/07/21/17846416079896.jpg) +链接： https://www.lokiproxy.com/?utm_t=1&utm_i=158 -### SWIFTPROXY - - - - - - - - + ## 参与贡献 diff --git a/feapder/db/redisdb.py b/feapder/db/redisdb.py index d882e687..ee4e0cd9 100644 --- a/feapder/db/redisdb.py +++ b/feapder/db/redisdb.py @@ -6,11 +6,11 @@ --------- @author: Boris """ -import os import time from typing import Union, List import redis +from redis.cluster import ClusterNode, RedisCluster from redis.connection import Encoder as _Encoder from redis.exceptions import ConnectionError, TimeoutError from redis.exceptions import DataError @@ -140,29 +140,24 @@ def get_connect(self): startup_nodes = [] for ip_port in ip_ports: ip, port = ip_port.split(":") - startup_nodes.append({"host": ip, "port": port}) + startup_nodes.append(ClusterNode(ip, int(port))) if self._service_name: # log.debug("使用redis哨兵模式") - hosts = [(node["host"], node["port"]) for node in startup_nodes] + hosts = [(node.host, node.port) for node in startup_nodes] sentinel = Sentinel(hosts, socket_timeout=3, **self._kwargs) self._redis = sentinel.master_for( self._service_name, password=self._user_pass, db=self._db, - redis_class=redis.StrictRedis, + redis_class=redis.Redis, decode_responses=self._decode_responses, max_connections=self._max_connections, **self._kwargs, ) + self._is_redis_cluster = False else: - try: - from rediscluster import RedisCluster - except ModuleNotFoundError as e: - log.error('请安装 pip install "feapder[all]"') - os._exit(0) - # log.debug("使用redis集群模式") self._redis = RedisCluster( startup_nodes=startup_nodes, @@ -171,11 +166,10 @@ def get_connect(self): max_connections=self._max_connections, **self._kwargs, ) - - self._is_redis_cluster = True + self._is_redis_cluster = True else: ip, port = ip_ports[0].split(":") - self._redis = redis.StrictRedis( + self._redis = redis.Redis( host=ip, port=port, db=self._db, @@ -186,7 +180,7 @@ def get_connect(self): ) self._is_redis_cluster = False else: - self._redis = redis.StrictRedis.from_url( + self._redis = redis.Redis.from_url( self._url, decode_responses=self._decode_responses, **self._kwargs ) self._is_redis_cluster = False @@ -573,7 +567,8 @@ def zexists(self, table, values): if isinstance(values, list): pipe = self._redis.pipeline() - pipe.multi() + if not self._is_redis_cluster: + pipe.multi() for value in values: pipe.zscore(table, value) is_exists_temp = pipe.execute() @@ -773,7 +768,8 @@ def setbit( else: assert len(offsets) == len(values), "offsets值要与values值一一对应" pipe = self._redis.pipeline() - pipe.multi() + if not self._is_redis_cluster: + pipe.multi() for offset, value in zip(offsets, values): pipe.setbit(table, offset, value) @@ -792,7 +788,8 @@ def getbit(self, table, offsets): """ if isinstance(offsets, list): pipe = self._redis.pipeline() - pipe.multi() + if not self._is_redis_cluster: + pipe.multi() for offset in offsets: pipe.getbit(table, offset) diff --git a/feapder/network/proxy_pool_old.py b/feapder/network/proxy_pool_old.py index 2e3bb6c1..fddaabc1 100644 --- a/feapder/network/proxy_pool_old.py +++ b/feapder/network/proxy_pool_old.py @@ -147,7 +147,7 @@ def get_proxy_from_redis(proxy_source_url, **kwargs): @return: [{'http':'http://xxx.xxx.xxx:xxx', 'https':'http://xxx.xxx.xxx.xxx:xxx'}] """ - redis_conn = redis.StrictRedis.from_url(proxy_source_url) + redis_conn = redis.Redis.from_url(proxy_source_url) key = kwargs.get("redis_proxies_key") assert key, "从redis中获取代理 需要指定 redis_proxies_key" proxies = redis_conn.zrange(key, 0, -1) diff --git a/feapder/requirements.txt b/feapder/requirements.txt index 21717674..888d27f2 100644 --- a/feapder/requirements.txt +++ b/feapder/requirements.txt @@ -1,21 +1,20 @@ better-exceptions>=0.2.2 -DBUtils>=2.0 -parsel>=1.5.2 +DBUtils>=3.0 +parsel>=1.8.1 PyExecJS>=1.5.1 -pymongo>=3.10.1 -PyMySQL>=0.9.3 -redis>=2.10.6,<4.0.0 -requests>=2.22.0 -selenium>=3.141.0 -bs4>=0.0.1 -ipython>=7.14.0 -bitarray>=1.5.3 -redis-py-cluster>=2.1.0 -cryptography>=3.3.2 -urllib3>=1.25.8 +pymongo>=4.0.0 +PyMySQL>=1.1.0 +redis>=5.0.0,<9.0.0 +requests>=2.31.0 +selenium>=4.10.0 +beautifulsoup4>=4.12.0 +ipython>=8.0.0 +bitarray>=2.8.0 +cryptography>=41.0.0 +urllib3>=2.0.0,<3.0.0 loguru>=0.5.3 influxdb>=5.3.1 pyperclip>=1.8.2 webdriver-manager>=4.0.0 terminal-layout>=2.1.3 -playwright \ No newline at end of file +playwright>=1.40.0 diff --git a/feapder/utils/webdriver/selenium_driver.py b/feapder/utils/webdriver/selenium_driver.py index 9f46d54b..06eb5493 100644 --- a/feapder/utils/webdriver/selenium_driver.py +++ b/feapder/utils/webdriver/selenium_driver.py @@ -8,13 +8,13 @@ @email: boris_liu@foxmail.com """ +import inspect import json import logging import os from typing import Optional, Union, List from selenium import webdriver -from selenium.webdriver.common.desired_capabilities import DesiredCapabilities from selenium.webdriver.remote.webdriver import WebDriver as RemoteWebDriver from webdriver_manager.chrome import ChromeDriverManager from webdriver_manager.firefox import GeckoDriverManager @@ -33,41 +33,7 @@ class SeleniumDriver(WebDriver, RemoteWebDriver): PHANTOMJS = "PHANTOMJS" FIREFOX = "FIREFOX" - __CHROME_ATTRS__ = { - "executable_path", - "port", - "options", - "service_args", - "desired_capabilities", - "service_log_path", - "chrome_options", - "keep_alive", - } - - __EDGE_ATTRS__ = __CHROME_ATTRS__ - - __FIREFOX_ATTRS__ = { - "firefox_profile", - "firefox_binary", - "timeout", - "capabilities", - "proxy", - "executable_path", - "options", - "service_log_path", - "firefox_options", - "service_args", - "desired_capabilities", - "log_path", - "keep_alive", - } - __PHANTOMJS_ATTRS__ = { - "executable_path", - "port", - "desired_capabilities", - "service_args", - "service_log_path", - } + __DRIVER_ATTRS__ = {"keep_alive"} def __init__(self, xhr_url_regexes: list = None, **kwargs): """ @@ -131,41 +97,117 @@ def filter_kwargs(self, kwargs: dict, driver_attrs: set): return data + def get_options(self, default_options, *option_keys): + for option_key in option_keys: + options = self._kwargs.get(option_key) + if options is not None: + return options + + return default_options + + def get_driver_kwargs(self): + return self.filter_kwargs(self._kwargs, self.__DRIVER_ATTRS__) + + def apply_capabilities(self, options, *capability_keys): + for capability_key in capability_keys: + capabilities = self._kwargs.get(capability_key) + if not capabilities: + continue + + for key, value in capabilities.items(): + options.set_capability(key, value) + + return options + + def build_service(self, service_cls, driver_manager_cls=None): + service = self._kwargs.get("service") + if service is not None: + return service + + service_kwargs = {} + service_args = self._kwargs.get("service_args") + if service_args is not None: + service_kwargs["service_args"] = service_args + + port = self._kwargs.get("port") + if port is not None: + service_kwargs["port"] = port + + log_path = self._kwargs.get("service_log_path") + if log_path is None: + log_path = self._kwargs.get("log_path") + if log_path is not None: + log_param = ( + "log_output" + if "log_output" in inspect.signature(service_cls).parameters + else "log_path" + ) + service_kwargs[log_param] = log_path + + if self._executable_path: + return service_cls(self._executable_path, **service_kwargs) + + if self._auto_install_driver and driver_manager_cls is not None: + return service_cls(driver_manager_cls().install(), **service_kwargs) + + if service_kwargs: + return service_cls(**service_kwargs) + + return None + + def create_driver(self, driver_cls, options, service): + kwargs = self.get_driver_kwargs() + if service is not None: + kwargs["service"] = service + + return driver_cls(options=options, **kwargs) + + def get_proxy(self): + return self._proxy() if callable(self._proxy) else self._proxy + + def get_user_agent(self): + return self._user_agent() if callable(self._user_agent) else self._user_agent + def get_driver(self): return self.driver def firefox_driver(self): - if webdriver.__version__ >= "4.0.0": - raise Exception( - f"暂未适配selenium=={webdriver.__version__}版本的firefox API，建议安装selenium==3.141.0版本或使用CHROME浏览器" - ) + from selenium.webdriver.firefox.service import Service - firefox_profile = webdriver.FirefoxProfile() - firefox_options = webdriver.FirefoxOptions() - firefox_capabilities = webdriver.DesiredCapabilities.FIREFOX - try: - from selenium.webdriver.firefox.service import Service - except (ImportError, ModuleNotFoundError): - Service = None + firefox_options = self.get_options( + webdriver.FirefoxOptions(), "options", "firefox_options" + ) + firefox_profile = self._kwargs.get("firefox_profile") + if firefox_profile is not None: + firefox_options.profile = firefox_profile + firefox_binary = self._kwargs.get("firefox_binary") + if firefox_binary is not None: + firefox_options.binary_location = ( + getattr(firefox_binary, "path", None) + or getattr(firefox_binary, "_start_cmd", None) + or firefox_binary + ) + self.apply_capabilities(firefox_options, "desired_capabilities", "capabilities") if self._proxy: - proxy = self._proxy() if callable(self._proxy) else self._proxy - firefox_capabilities["marionette"] = True - firefox_capabilities["proxy"] = { - "proxyType": "MANUAL", - "httpProxy": proxy, - "ftpProxy": proxy, - "sslProxy": proxy, - } + proxy = self.get_proxy() + firefox_options.set_capability( + "proxy", + { + "proxyType": "MANUAL", + "httpProxy": proxy, + "ftpProxy": proxy, + "sslProxy": proxy, + }, + ) if self._user_agent: - firefox_profile.set_preference( - "general.useragent.override", - self._user_agent() if callable(self._user_agent) else self._user_agent, + firefox_options.set_preference( + "general.useragent.override", self.get_user_agent() ) if not self._load_images: - firefox_profile.set_preference("permissions.default.image", 2) + firefox_options.set_preference("permissions.default.image", 2) if self._headless: firefox_options.add_argument("--headless") @@ -176,25 +218,8 @@ def firefox_driver(self): for arg in self._custom_argument: firefox_options.add_argument(arg) - kwargs = self.filter_kwargs(self._kwargs, self.__FIREFOX_ATTRS__) - - if Service is None: - if self._executable_path: - kwargs.update(executable_path=self._executable_path) - elif self._auto_install_driver: - kwargs.update(executable_path=GeckoDriverManager().install()) - else: - if self._executable_path: - kwargs.update(service=Service(self._executable_path)) - elif self._auto_install_driver: - kwargs.update(service=Service(GeckoDriverManager().install())) - - driver = webdriver.Firefox( - capabilities=firefox_capabilities, - options=firefox_options, - firefox_profile=firefox_profile, - **kwargs, - ) + service = self.build_service(Service, GeckoDriverManager) + driver = self.create_driver(webdriver.Firefox, firefox_options, service) if self._window_size: driver.set_window_size(*self._window_size) @@ -202,31 +227,22 @@ def firefox_driver(self): return driver def chrome_driver(self): - chrome_options = webdriver.ChromeOptions() + chrome_options = self.get_options( + webdriver.ChromeOptions(), "options", "chrome_options" + ) # 此步骤很重要，设置为开发者模式，防止被各大网站识别出来使用了Selenium chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"]) chrome_options.add_experimental_option("useAutomationExtension", False) # docker 里运行需要 chrome_options.add_argument("--no-sandbox") - try: - from selenium.webdriver.chrome.service import Service - except (ImportError, ModuleNotFoundError): - Service = None + from selenium.webdriver.chrome.service import Service + + self.apply_capabilities(chrome_options, "desired_capabilities") if self._proxy: - chrome_options.add_argument( - "--proxy-server={}".format( - self._proxy() if callable(self._proxy) else self._proxy - ) - ) + chrome_options.add_argument("--proxy-server={}".format(self.get_proxy())) if self._user_agent: - chrome_options.add_argument( - "user-agent={}".format( - self._user_agent() - if callable(self._user_agent) - else self._user_agent - ) - ) + chrome_options.add_argument("user-agent={}".format(self.get_user_agent())) if not self._load_images: chrome_options.add_experimental_option( "prefs", {"profile.managed_default_content_settings.images": 2} @@ -254,19 +270,8 @@ def chrome_driver(self): for arg in self._custom_argument: chrome_options.add_argument(arg) - kwargs = self.filter_kwargs(self._kwargs, self.__CHROME_ATTRS__) - if Service is None: - if self._executable_path: - kwargs.update(executable_path=self._executable_path) - elif self._auto_install_driver: - kwargs.update(executable_path=ChromeDriverManager().install()) - else: - if self._executable_path: - kwargs.update(service=Service(self._executable_path)) - elif self._auto_install_driver: - kwargs.update(service=Service(ChromeDriverManager().install())) - - driver = webdriver.Chrome(options=chrome_options, **kwargs) + service = self.build_service(Service, ChromeDriverManager) + driver = self.create_driver(webdriver.Chrome, chrome_options, service) # 隐藏浏览器特征 if self._use_stealth_js: @@ -293,44 +298,30 @@ def chrome_driver(self): ) if self._download_path: - driver.command_executor._commands["send_command"] = ( - "POST", - "/session/$sessionId/chromium/send_command", + driver.execute_cdp_cmd( + "Page.setDownloadBehavior", + {"behavior": "allow", "downloadPath": self._download_path}, ) - params = { - "cmd": "Page.setDownloadBehavior", - "params": {"behavior": "allow", "downloadPath": self._download_path}, - } - driver.execute("send_command", params) return driver def edge_driver(self): - edge_options = webdriver.EdgeOptions() + edge_options = self.get_options( + webdriver.EdgeOptions(), "options", "edge_options" + ) # 此步骤很重要，设置为开发者模式，防止被各大网站识别出来使用了Selenium edge_options.add_experimental_option("excludeSwitches", ["enable-automation"]) edge_options.add_experimental_option("useAutomationExtension", False) # docker 里运行需要 edge_options.add_argument("--no-sandbox") - try: - from selenium.webdriver.edge.service import Service - except (ImportError, ModuleNotFoundError): - Service = None + from selenium.webdriver.edge.service import Service + + self.apply_capabilities(edge_options, "desired_capabilities") if self._proxy: - edge_options.add_argument( - "--proxy-server={}".format( - self._proxy() if callable(self._proxy) else self._proxy - ) - ) + edge_options.add_argument("--proxy-server={}".format(self.get_proxy())) if self._user_agent: - edge_options.add_argument( - "user-agent={}".format( - self._user_agent() - if callable(self._user_agent) - else self._user_agent - ) - ) + edge_options.add_argument("user-agent={}".format(self.get_user_agent())) if not self._load_images: edge_options.add_experimental_option( "prefs", {"profile.managed_default_content_settings.images": 2} @@ -358,19 +349,8 @@ def edge_driver(self): for arg in self._custom_argument: edge_options.add_argument(arg) - kwargs = self.filter_kwargs(self._kwargs, self.__CHROME_ATTRS__) - if Service is None: - if self._executable_path: - kwargs.update(executable_path=self._executable_path) - elif self._auto_install_driver: - raise NotImplementedError("edge not support auto install driver") - else: - if self._executable_path: - kwargs.update(service=Service(self._executable_path)) - elif self._auto_install_driver: - raise NotImplementedError("edge not support auto install driver") - - driver = webdriver.Edge(options=edge_options, **kwargs) + service = self.build_service(Service) + driver = self.create_driver(webdriver.Edge, edge_options, service) # 隐藏浏览器特征 if self._use_stealth_js: @@ -397,58 +377,19 @@ def edge_driver(self): ) if self._download_path: - driver.command_executor._commands["send_command"] = ( - "POST", - "/session/$sessionId/chromium/send_command", + driver.execute_cdp_cmd( + "Page.setDownloadBehavior", + {"behavior": "allow", "downloadPath": self._download_path}, ) - params = { - "cmd": "Page.setDownloadBehavior", - "params": {"behavior": "allow", "downloadPath": self._download_path}, - } - driver.execute("send_command", params) return driver def phantomjs_driver(self): - import warnings - - warnings.filterwarnings("ignore") - - service_args = [] - dcap = DesiredCapabilities.PHANTOMJS - - if self._proxy: - service_args.append( - "--proxy=%s" % self._proxy() if callable(self._proxy) else self._proxy - ) - if self._user_agent: - dcap["phantomjs.page.settings.userAgent"] = ( - self._user_agent() if callable(self._user_agent) else self._user_agent - ) - if not self._load_images: - service_args.append("--load-images=no") - - # 添加自定义的配置参数 - if self._custom_argument: - for arg in self._custom_argument: - service_args.append(arg) - - kwargs = self.filter_kwargs(self._kwargs, self.__PHANTOMJS_ATTRS__) - - if self._executable_path: - kwargs.update(executable_path=self._executable_path) - - driver = webdriver.PhantomJS( - service_args=service_args, desired_capabilities=dcap, **kwargs + raise NotImplementedError( + "PhantomJS is not supported by Selenium 4. " + "Please use CHROME, EDGE, or FIREFOX." ) - if self._window_size: - driver.set_window_size(self._window_size[0], self._window_size[1]) - - del warnings - - return driver - @property def domain(self): return tools.get_domain(self.url or self.driver.current_url) diff --git a/setup.py b/setup.py index cf4fe542..14542b97 100644 --- a/setup.py +++ b/setup.py @@ -13,8 +13,8 @@ import setuptools -if version_info < (3, 6, 0): - raise SystemExit("Sorry! feapder requires python 3.6.0 or later.") +if version_info < (3, 9, 0): + raise SystemExit("Sorry! feapder requires python 3.9.0 or later.") with open(join(dirname(__file__), "feapder/VERSION"), "rb") as fh: version = fh.read().decode("ascii").strip() @@ -35,15 +35,15 @@ requires = [ "better-exceptions>=0.2.2", - "DBUtils>=2.0", - "parsel>=1.5.2", - "PyMySQL>=0.9.3", - "redis>=2.10.6,<4.0.0", - "requests>=2.22.0", - "bs4>=0.0.1", - "ipython>=7.14.0", - "cryptography>=3.3.2", - "urllib3>=1.25.8", + "DBUtils>=3.0", + "parsel>=1.8.1", + "PyMySQL>=1.1.0", + "redis>=5.0.0,<9.0.0", + "requests>=2.31.0", + "beautifulsoup4>=4.12.0", + "ipython>=8.0.0", + "cryptography>=41.0.0", + "urllib3>=2.0.0,<3.0.0", "loguru>=0.5.3", "influxdb>=5.3.1", "pyperclip>=1.8.2", @@ -52,15 +52,14 @@ render_requires = [ "webdriver-manager>=4.0.0", - "playwright", - "selenium>=3.141.0", + "playwright>=1.40.0", + "selenium>=4.10.0", ] all_requires = [ - "bitarray>=1.5.3", + "bitarray>=2.8.0", "PyExecJS>=1.5.1", - "pymongo>=3.10.1", - "redis-py-cluster>=2.1.0", + "pymongo>=4.0.0", ] + render_requires setuptools.setup( @@ -69,7 +68,7 @@ author="Boris", license="MIT", author_email="feapder@qq.com", - python_requires=">=3.6", + python_requires=">=3.9", description="feapder是一款支持分布式、批次采集、数据防丢、报警丰富的python爬虫框架", long_description=long_description, long_description_content_type="text/markdown", diff --git a/tests/test_dependency_modernization.py b/tests/test_dependency_modernization.py new file mode 100644 index 00000000..e9a73b2a --- /dev/null +++ b/tests/test_dependency_modernization.py @@ -0,0 +1,115 @@ +import inspect + +from feapder.db.redisdb import RedisDB +from feapder.utils.webdriver.selenium_driver import SeleniumDriver + + +class OldStyleService: + def __init__( + self, executable_path=None, port=0, service_args=None, log_path=None, **kwargs + ): + self.executable_path = executable_path + self.port = port + self.service_args = service_args + self.log_path = log_path + + +class NewStyleService: + def __init__( + self, executable_path=None, port=0, service_args=None, log_output=None, **kwargs + ): + self.executable_path = executable_path + self.port = port + self.service_args = service_args + self.log_output = log_output + + +class FakeBrowser: + def set_window_size(self, *args): + self.window_size = args + + +def make_selenium_driver(**kwargs): + driver = object.__new__(SeleniumDriver) + driver._kwargs = kwargs + driver._executable_path = kwargs.pop("executable_path", "/tmp/driver") + driver._auto_install_driver = False + driver._proxy = None + driver._user_agent = None + driver._load_images = True + driver._headless = False + driver._custom_argument = None + driver._window_size = None + return driver + + +def test_selenium_service_log_path_supports_old_and_new_service_api(): + driver = make_selenium_driver( + service_log_path="/tmp/webdriver.log", service_args=["--verbose"], port=1234 + ) + + old_service = driver.build_service(OldStyleService) + new_service = driver.build_service(NewStyleService) + + assert old_service.executable_path == "/tmp/driver" + assert old_service.service_args == ["--verbose"] + assert old_service.port == 1234 + assert old_service.log_path == "/tmp/webdriver.log" + assert new_service.executable_path == "/tmp/driver" + assert new_service.service_args == ["--verbose"] + assert new_service.port == 1234 + assert new_service.log_output == "/tmp/webdriver.log" + + +def test_selenium_firefox_binary_maps_to_options_binary_location(): + driver = make_selenium_driver(firefox_binary="/tmp/firefox") + captured = {} + + def create_driver(driver_cls, options, service): + captured["options"] = options + captured["service"] = service + return FakeBrowser() + + driver.create_driver = create_driver + driver.build_service = lambda *args, **kwargs: None + + assert driver.firefox_driver() is not None + assert captured["options"].binary_location == "/tmp/firefox" + assert captured["service"] is None + + +def test_selenium_driver_kwargs_keep_public_constructor_args_internal(): + driver = make_selenium_driver( + keep_alive=False, + executable_path="/tmp/driver", + desired_capabilities={"acceptInsecureCerts": True}, + service_args=["--verbose"], + ) + + assert driver.get_driver_kwargs() == {"keep_alive": False} + + +def test_redisdb_public_api_signatures_are_preserved(): + expected = { + "__init__": [ + "self", + "ip_ports", + "db", + "user_pass", + "url", + "decode_responses", + "service_name", + "max_connections", + "kwargs", + ], + "from_url": ["url"], + "zadd": ["self", "table", "values", "prioritys"], + "zget": ["self", "table", "count", "is_pop"], + "zexists": ["self", "table", "values"], + "setbit": ["self", "table", "offsets", "values"], + "getbit": ["self", "table", "offsets"], + } + + for method_name, parameters in expected.items(): + signature = inspect.signature(getattr(RedisDB, method_name)) + assert list(signature.parameters) == parameters