diff --git a/README.rst b/README.rst index 731883a..efb5a0b 100644 --- a/README.rst +++ b/README.rst @@ -19,7 +19,7 @@ python-anticaptcha .. introduction-start Client library for solve captchas with `Anticaptcha.com support`_. -The library supports both Python 2.7 and Python 3. +The library requires Python >= 3.9. The library is cyclically and automatically tested for proper operation. We are constantly making the best efforts for its effective operation. @@ -67,9 +67,9 @@ Example snippet for Recaptcha: client = AnticaptchaClient(api_key) task = NoCaptchaTaskProxylessTask(url, site_key) - job = client.createTask(task) + job = client.create_task(task) job.join() - print job.get_solution_response() + print(job.get_solution_response()) The full integration example is available in file ``examples/recaptcha.py``. @@ -89,9 +89,9 @@ measures for automated training and analysis. For provide that pass client = AnticaptchaClient(api_key) task = NoCaptchaTaskProxylessTask(url, site_key, is_invisible=True) - job = client.createTask(task) + job = client.create_task(task) job.join() - print job.get_solution_response() + print(job.get_solution_response()) Solve text captcha @@ -107,9 +107,9 @@ Example snippet for text captcha: captcha_fp = open('examples/captcha_ms.jpeg', 'rb') client = AnticaptchaClient(api_key) task = ImageToTextTask(captcha_fp) - job = client.createTask(task) + job = client.create_task(task) job.join() - print job.get_captcha_text() + print(job.get_captcha_text()) Solve funcaptcha ################ @@ -125,13 +125,13 @@ Example snippet for funcaptcha: api_key = '174faff8fbc769e94a5862391ecfd010' site_key = 'DE0B0BB7-1EE4-4D70-1853-31B835D4506B' # grab from site url = 'https://www.google.com/recaptcha/api2/demo' - proxy = Proxy.parse_url("socks5://login:password@123.123.123.123") + proxy = Proxy.parse_url("socks5://login:password@123.123.123.123:1080") client = AnticaptchaClient(api_key) - task = FunCaptchaTask(url, site_key, proxy=proxy, user_agent=user_agent) - job = client.createTask(task) + task = FunCaptchaTask(url, site_key, user_agent=UA, **proxy.to_kwargs()) + job = client.create_task(task) job.join() - print job.get_token_response() + print(job.get_token_response()) Report incorrect image ###################### @@ -146,10 +146,10 @@ Example snippet for reporting an incorrect image task: captcha_fp = open('examples/captcha_ms.jpeg', 'rb') client = AnticaptchaClient(api_key) task = ImageToTextTask(captcha_fp) - job = client.createTask(task) + job = client.create_task(task) job.join() - print job.get_captcha_text() - job.report_incorrect() + print(job.get_captcha_text()) + job.report_incorrect_image() Setup proxy ########### @@ -181,20 +181,24 @@ We recommend entering IP-based access control for incoming addresses to proxy. I Error handling ############## -In the event of an application error, the AnticaptchaException exception is thrown. To handle the exception, do the following: +In the event of an application error, the ``AnticaptchaException`` exception is thrown. To handle the exception, do the following: .. code:: python - from python_anticaptcha import AnticatpchaException, ImageToTextTask + from python_anticaptcha import AnticaptchaException, ImageToTextTask try: # any actions - except AnticatpchaException as e: + except AnticaptchaException as e: if e.error_code == 'ERROR_ZERO_BALANCE': notify_about_no_funds(e.error_id, e.error_code, e.error_description) else: raise +.. note:: + + The legacy misspelled ``AnticatpchaException`` alias is still available for backward compatibility. + .. usage-end Versioning diff --git a/pyproject.toml b/pyproject.toml index f550004..7d3fc6b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,3 +1,8 @@ +[project] +name = "python-anticaptcha" +requires-python = ">=3.9" +dynamic = ["version", "description", "readme", "dependencies", "optional-dependencies"] + [tool.pytest.ini_options] testpaths = ["tests"] markers = [ diff --git a/python_anticaptcha/__init__.py b/python_anticaptcha/__init__.py index 80213de..1cb621f 100644 --- a/python_anticaptcha/__init__.py +++ b/python_anticaptcha/__init__.py @@ -1,6 +1,7 @@ from importlib.metadata import version, PackageNotFoundError -from .base import AnticaptchaClient +from .base import AnticaptchaClient, Job +from .proxy import Proxy from .tasks import ( NoCaptchaTaskProxylessTask, RecaptchaV2TaskProxyless, @@ -28,3 +29,27 @@ except PackageNotFoundError: # package is not installed pass + +__all__ = [ + "AnticaptchaClient", + "Job", + "Proxy", + "NoCaptchaTaskProxylessTask", + "RecaptchaV2TaskProxyless", + "NoCaptchaTask", + "RecaptchaV2Task", + "FunCaptchaProxylessTask", + "FunCaptchaTask", + "ImageToTextTask", + "RecaptchaV3TaskProxyless", + "HCaptchaTaskProxyless", + "HCaptchaTask", + "RecaptchaV2EnterpriseTaskProxyless", + "RecaptchaV2EnterpriseTask", + "GeeTestTaskProxyless", + "GeeTestTask", + "AntiGateTaskProxyless", + "AntiGateTask", + "AnticaptchaException", + "AnticatpchaException", +] diff --git a/python_anticaptcha/base.py b/python_anticaptcha/base.py index 5753deb..a5188f7 100644 --- a/python_anticaptcha/base.py +++ b/python_anticaptcha/base.py @@ -1,7 +1,10 @@ +from __future__ import annotations + import requests import time import json import warnings +from typing import Any from urllib.parse import urljoin from .exceptions import AnticaptchaException @@ -15,49 +18,49 @@ class Job: task_id = None _last_result = None - def __init__(self, client, task_id): + def __init__(self, client: AnticaptchaClient, task_id: int) -> None: self.client = client self.task_id = task_id - def _update(self): + def _update(self) -> None: self._last_result = self.client.getTaskResult(self.task_id) - def check_is_ready(self): + def check_is_ready(self) -> bool: self._update() return self._last_result["status"] == "ready" - def get_solution_response(self): # Recaptcha + def get_solution_response(self) -> str: # Recaptcha return self._last_result["solution"]["gRecaptchaResponse"] - def get_solution(self): + def get_solution(self) -> dict[str, Any]: return self._last_result["solution"] - def get_token_response(self): # Funcaptcha + def get_token_response(self) -> str: # Funcaptcha return self._last_result["solution"]["token"] - def get_answers(self): + def get_answers(self) -> dict[str, str]: return self._last_result["solution"]["answers"] - def get_captcha_text(self): # Image + def get_captcha_text(self) -> str: # Image return self._last_result["solution"]["text"] - def get_cells_numbers(self): + def get_cells_numbers(self) -> list[int]: return self._last_result["solution"]["cellNumbers"] - def report_incorrect(self): + def report_incorrect(self) -> bool: warnings.warn( "report_incorrect is deprecated, use report_incorrect_image instead", DeprecationWarning, ) return self.client.reportIncorrectImage() - def report_incorrect_image(self): + def report_incorrect_image(self) -> bool: return self.client.reportIncorrectImage(self.task_id) - def report_incorrect_recaptcha(self): + def report_incorrect_recaptcha(self) -> bool: return self.client.reportIncorrectRecaptcha(self.task_id) - def join(self, maximum_time=None): + def join(self, maximum_time: int | None = None) -> None: elapsed_time = 0 maximum_time = maximum_time or MAXIMUM_JOIN_TIME while not self.check_is_ready(): @@ -86,8 +89,8 @@ class AnticaptchaClient: response_timeout = 5 def __init__( - self, client_key, language_pool="en", host="api.anti-captcha.com", use_ssl=True - ): + self, client_key: str, language_pool: str = "en", host: str = "api.anti-captcha.com", use_ssl: bool = True, + ) -> None: self.client_key = client_key self.language_pool = language_pool self.base_url = "{proto}://{host}/".format( @@ -96,14 +99,14 @@ def __init__( self.session = requests.Session() @property - def client_ip(self): + def client_ip(self) -> str: if not hasattr(self, "_client_ip"): self._client_ip = self.session.get( "https://api.myip.com", timeout=self.response_timeout ).json()["ip"] return self._client_ip - def _check_response(self, response): + def _check_response(self, response: dict[str, Any]) -> None: if response.get("errorId", False) == 11: response[ "errorDescription" @@ -115,7 +118,7 @@ def _check_response(self, response): response["errorId"], response["errorCode"], response["errorDescription"] ) - def createTask(self, task): + def createTask(self, task: Any) -> Job: request = { "clientKey": self.client_key, "task": task.serialize(), @@ -130,7 +133,7 @@ def createTask(self, task): self._check_response(response) return Job(self, response["taskId"]) - def createTaskSmee(self, task, timeout=MAXIMUM_JOIN_TIME): + def createTaskSmee(self, task: Any, timeout: int = MAXIMUM_JOIN_TIME) -> Job: """ Beta method to stream response from smee.io """ @@ -172,7 +175,7 @@ def createTaskSmee(self, task, timeout=MAXIMUM_JOIN_TIME): job._last_result = payload["body"] return job - def getTaskResult(self, task_id): + def getTaskResult(self, task_id: int) -> dict[str, Any]: request = {"clientKey": self.client_key, "taskId": task_id} response = self.session.post( urljoin(self.base_url, self.TASK_RESULT_URL), json=request @@ -180,7 +183,7 @@ def getTaskResult(self, task_id): self._check_response(response) return response - def getBalance(self): + def getBalance(self) -> float: request = { "clientKey": self.client_key, "softId": self.SOFT_ID, @@ -191,7 +194,7 @@ def getBalance(self): self._check_response(response) return response["balance"] - def getAppStats(self, soft_id, mode): + def getAppStats(self, soft_id: int, mode: str) -> dict[str, Any]: request = {"clientKey": self.client_key, "softId": soft_id, "mode": mode} response = self.session.post( urljoin(self.base_url, self.APP_STAT_URL), json=request @@ -199,7 +202,7 @@ def getAppStats(self, soft_id, mode): self._check_response(response) return response - def reportIncorrectImage(self, task_id): + def reportIncorrectImage(self, task_id: int) -> bool: request = {"clientKey": self.client_key, "taskId": task_id} response = self.session.post( urljoin(self.base_url, self.REPORT_IMAGE_URL), json=request @@ -207,10 +210,19 @@ def reportIncorrectImage(self, task_id): self._check_response(response) return response.get("status", False) != False - def reportIncorrectRecaptcha(self, task_id): + def reportIncorrectRecaptcha(self, task_id: int) -> bool: request = {"clientKey": self.client_key, "taskId": task_id} response = self.session.post( urljoin(self.base_url, self.REPORT_RECAPTCHA_URL), json=request ).json() self._check_response(response) return response["status"] == "success" + + # Snake_case aliases + create_task = createTask + create_task_smee = createTaskSmee + get_task_result = getTaskResult + get_balance = getBalance + get_app_stats = getAppStats + report_incorrect_image = reportIncorrectImage + report_incorrect_recaptcha = reportIncorrectRecaptcha diff --git a/python_anticaptcha/exceptions.py b/python_anticaptcha/exceptions.py index 108febd..ffeb085 100644 --- a/python_anticaptcha/exceptions.py +++ b/python_anticaptcha/exceptions.py @@ -1,5 +1,8 @@ +from __future__ import annotations + + class AnticaptchaException(Exception): - def __init__(self, error_id, error_code, error_description, *args): + def __init__(self, error_id: int | str | None, error_code: int | str, error_description: str, *args: object) -> None: super().__init__( "[{}:{}]{}".format(error_code, error_id, error_description) ) diff --git a/python_anticaptcha/proxy.py b/python_anticaptcha/proxy.py new file mode 100644 index 0000000..c2c5493 --- /dev/null +++ b/python_anticaptcha/proxy.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +from dataclasses import dataclass +from urllib.parse import urlparse + + +@dataclass(frozen=True) +class Proxy: + proxy_type: str + proxy_address: str + proxy_port: int + proxy_login: str = "" + proxy_password: str = "" + + @classmethod + def parse_url(cls, url: str) -> Proxy: + parsed = urlparse(url) + if not parsed.hostname or not parsed.port: + raise ValueError(f"Invalid proxy URL: {url}") + return cls( + proxy_type=parsed.scheme, + proxy_address=parsed.hostname, + proxy_port=parsed.port, + proxy_login=parsed.username or "", + proxy_password=parsed.password or "", + ) + + def to_kwargs(self) -> dict[str, str | int]: + return { + "proxy_type": self.proxy_type, + "proxy_address": self.proxy_address, + "proxy_port": self.proxy_port, + "proxy_login": self.proxy_login, + "proxy_password": self.proxy_password, + } diff --git a/python_anticaptcha/py.typed b/python_anticaptcha/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/python_anticaptcha/tasks.py b/python_anticaptcha/tasks.py index 3ef6890..0464233 100644 --- a/python_anticaptcha/tasks.py +++ b/python_anticaptcha/tasks.py @@ -1,31 +1,34 @@ +from __future__ import annotations + import base64 +from typing import Any, BinaryIO class BaseTask: type = None - def serialize(self, **result): + def serialize(self, **result: Any) -> dict[str, Any]: result["type"] = self.type return result class UserAgentMixin(BaseTask): - def __init__(self, *args, **kwargs): - self.userAgent = kwargs.pop("user_agent") + def __init__(self, *args: Any, **kwargs: Any) -> None: + self.userAgent: str = kwargs.pop("user_agent") super().__init__(*args, **kwargs) - def serialize(self, **result): + def serialize(self, **result: Any) -> dict[str, Any]: data = super().serialize(**result) data["userAgent"] = self.userAgent return data class CookieMixin(BaseTask): - def __init__(self, *args, **kwargs): - self.cookies = kwargs.pop("cookies", "") + def __init__(self, *args: Any, **kwargs: Any) -> None: + self.cookies: str = kwargs.pop("cookies", "") super().__init__(*args, **kwargs) - def serialize(self, **result): + def serialize(self, **result: Any) -> dict[str, Any]: data = super().serialize(**result) if self.cookies: data["cookies"] = self.cookies @@ -33,15 +36,15 @@ def serialize(self, **result): class ProxyMixin(BaseTask): - def __init__(self, *args, **kwargs): - self.proxyType = kwargs.pop("proxy_type") - self.proxyAddress = kwargs.pop("proxy_address") - self.proxyPort = kwargs.pop("proxy_port") - self.proxyLogin = kwargs.pop("proxy_login") - self.proxyPassword = kwargs.pop("proxy_password") + def __init__(self, *args: Any, **kwargs: Any) -> None: + self.proxyType: str = kwargs.pop("proxy_type") + self.proxyAddress: str = kwargs.pop("proxy_address") + self.proxyPort: int = kwargs.pop("proxy_port") + self.proxyLogin: str = kwargs.pop("proxy_login") + self.proxyPassword: str = kwargs.pop("proxy_password") super().__init__(*args, **kwargs) - def serialize(self, **result): + def serialize(self, **result: Any) -> dict[str, Any]: data = super().serialize(**result) data["proxyType"] = self.proxyType data["proxyAddress"] = self.proxyAddress @@ -61,14 +64,14 @@ class NoCaptchaTaskProxylessTask(BaseTask): def __init__( self, - website_url, - website_key, - website_s_token=None, - is_invisible=None, - recaptcha_data_s_value=None, - *args, - **kwargs - ): + website_url: str, + website_key: str, + website_s_token: str | None = None, + is_invisible: bool | None = None, + recaptcha_data_s_value: str | None = None, + *args: Any, + **kwargs: Any, + ) -> None: self.websiteURL = website_url self.websiteKey = website_key self.websiteSToken = website_s_token @@ -76,7 +79,7 @@ def __init__( self.isInvisible = is_invisible super().__init__(*args, **kwargs) - def serialize(self, **result): + def serialize(self, **result: Any) -> dict[str, Any]: data = super().serialize(**result) data["websiteURL"] = self.websiteURL data["websiteKey"] = self.websiteKey @@ -111,15 +114,15 @@ class FunCaptchaProxylessTask(BaseTask): data = None def __init__( - self, website_url, website_key, subdomain=None, data=None, *args, **kwargs - ): + self, website_url: str, website_key: str, subdomain: str | None = None, data: str | None = None, *args: Any, **kwargs: Any, + ) -> None: self.websiteURL = website_url self.websiteKey = website_key self.funcaptchaApiJSSubdomain = subdomain self.data = data super().__init__(*args, **kwargs) - def serialize(self, **result): + def serialize(self, **result: Any) -> dict[str, Any]: data = super().serialize(**result) data["websiteURL"] = self.websiteURL data["websitePublicKey"] = self.websiteKey @@ -148,18 +151,18 @@ class ImageToTextTask(BaseTask): def __init__( self, - fp, - phrase=None, - case=None, - numeric=None, - math=None, - min_length=None, - max_length=None, - comment=None, - website_url=None, - *args, - **kwargs - ): + fp: BinaryIO, + phrase: bool | None = None, + case: bool | None = None, + numeric: int | None = None, + math: bool | None = None, + min_length: int | None = None, + max_length: int | None = None, + comment: str | None = None, + website_url: str | None = None, + *args: Any, + **kwargs: Any, + ) -> None: self.fp = fp self.phrase = phrase self.case = case @@ -171,7 +174,7 @@ def __init__( self.websiteUrl = website_url super().__init__(*args, **kwargs) - def serialize(self, **result): + def serialize(self, **result: Any) -> dict[str, Any]: data = super().serialize(**result) data["body"] = base64.b64encode(self.fp.read()).decode("utf-8") if self.phrase is not None: @@ -202,8 +205,8 @@ class RecaptchaV3TaskProxyless(BaseTask): isEnterprise = False def __init__( - self, website_url, website_key, min_score, page_action, is_enterprise=False, *args, **kwargs - ): + self, website_url: str, website_key: str, min_score: float, page_action: str, is_enterprise: bool = False, *args: Any, **kwargs: Any, + ) -> None: self.websiteURL = website_url self.websiteKey = website_key self.minScore = min_score @@ -211,7 +214,7 @@ def __init__( self.isEnterprise = is_enterprise super().__init__(*args, **kwargs) - def serialize(self, **result): + def serialize(self, **result: Any) -> dict[str, Any]: data = super().serialize(**result) data["websiteURL"] = self.websiteURL data["websiteKey"] = self.websiteKey @@ -226,12 +229,12 @@ class HCaptchaTaskProxyless(BaseTask): websiteURL = None websiteKey = None - def __init__(self, website_url, website_key, *args, **kwargs): + def __init__(self, website_url: str, website_key: str, *args: Any, **kwargs: Any) -> None: self.websiteURL = website_url self.websiteKey = website_key super().__init__(*args, **kwargs) - def serialize(self, **result): + def serialize(self, **result: Any) -> dict[str, Any]: data = super().serialize(**result) data["websiteURL"] = self.websiteURL data["websiteKey"] = self.websiteKey @@ -249,14 +252,14 @@ class RecaptchaV2EnterpriseTaskProxyless(BaseTask): enterprisePayload = None apiDomain = None - def __init__(self, website_url, website_key, enterprise_payload, api_domain, *args, **kwargs): + def __init__(self, website_url: str, website_key: str, enterprise_payload: dict[str, Any] | None, api_domain: str | None, *args: Any, **kwargs: Any) -> None: self.websiteURL = website_url self.websiteKey = website_key self.enterprisePayload = enterprise_payload self.apiDomain = api_domain super().__init__(*args, **kwargs) - def serialize(self, **result): + def serialize(self, **result: Any) -> dict[str, Any]: data = super().serialize(**result) data["websiteURL"] = self.websiteURL data["websiteKey"] = self.websiteKey @@ -282,8 +285,8 @@ class GeeTestTaskProxyless(BaseTask): geetestGetLib = None def __init__( - self, website_url, gt, challenge, subdomain=None, lib=None, *args, **kwargs - ): + self, website_url: str, gt: str, challenge: str, subdomain: str | None = None, lib: str | None = None, *args: Any, **kwargs: Any, + ) -> None: self.websiteURL = website_url self.gt = gt self.challenge = challenge @@ -291,7 +294,7 @@ def __init__( self.geetestGetLib = lib super().__init__(*args, **kwargs) - def serialize(self, **result): + def serialize(self, **result: Any) -> dict[str, Any]: data = super().serialize(**result) data["websiteURL"] = self.websiteURL data["gt"] = self.gt @@ -304,7 +307,7 @@ def serialize(self, **result): class GeeTestTask(ProxyMixin, UserAgentMixin, GeeTestTaskProxyless): - pass + type = "GeeTestTask" class AntiGateTaskProxyless(BaseTask): @@ -313,13 +316,13 @@ class AntiGateTaskProxyless(BaseTask): templateName = None variables = None - def __init__(self, website_url, template_name, variables, *args, **kwargs): + def __init__(self, website_url: str, template_name: str, variables: dict[str, Any], *args: Any, **kwargs: Any) -> None: self.websiteURL = website_url self.templateName = template_name self.variables = variables super().__init__(*args, **kwargs) - def serialize(self, **result): + def serialize(self, **result: Any) -> dict[str, Any]: data = super().serialize(**result) data["websiteURL"] = self.websiteURL data["templateName"] = self.templateName diff --git a/tests/test_base.py b/tests/test_base.py index e18611d..0582aad 100644 --- a/tests/test_base.py +++ b/tests/test_base.py @@ -147,3 +147,34 @@ def test_timeout_raises(self, mock_sleep): with pytest.raises(AnticaptchaException) as exc_info: job.join(maximum_time=SLEEP_EVERY_CHECK_FINISHED) assert "exceeded" in str(exc_info.value).lower() + + +class TestSnakeCaseAliases: + def test_create_task_alias(self): + assert AnticaptchaClient.create_task is AnticaptchaClient.createTask + + def test_create_task_smee_alias(self): + assert AnticaptchaClient.create_task_smee is AnticaptchaClient.createTaskSmee + + def test_get_task_result_alias(self): + assert AnticaptchaClient.get_task_result is AnticaptchaClient.getTaskResult + + def test_get_balance_alias(self): + assert AnticaptchaClient.get_balance is AnticaptchaClient.getBalance + + def test_get_app_stats_alias(self): + assert AnticaptchaClient.get_app_stats is AnticaptchaClient.getAppStats + + def test_report_incorrect_image_alias(self): + assert AnticaptchaClient.report_incorrect_image is AnticaptchaClient.reportIncorrectImage + + def test_report_incorrect_recaptcha_alias(self): + assert AnticaptchaClient.report_incorrect_recaptcha is AnticaptchaClient.reportIncorrectRecaptcha + + def test_alias_works_on_instance(self): + client = AnticaptchaClient("key123") + mock_response = MagicMock() + mock_response.json.return_value = {"errorId": 0, "balance": 5.0} + with patch.object(client.session, "post", return_value=mock_response): + balance = client.get_balance() + assert balance == 5.0 diff --git a/tests/test_proxy.py b/tests/test_proxy.py new file mode 100644 index 0000000..cfec1bb --- /dev/null +++ b/tests/test_proxy.py @@ -0,0 +1,69 @@ +import pytest + +from python_anticaptcha.proxy import Proxy + + +class TestProxyParseUrl: + def test_full_url(self): + proxy = Proxy.parse_url("socks5://user:pass@host.example.com:8080") + assert proxy.proxy_type == "socks5" + assert proxy.proxy_address == "host.example.com" + assert proxy.proxy_port == 8080 + assert proxy.proxy_login == "user" + assert proxy.proxy_password == "pass" + + def test_without_credentials(self): + proxy = Proxy.parse_url("http://123.123.123.123:9190") + assert proxy.proxy_type == "http" + assert proxy.proxy_address == "123.123.123.123" + assert proxy.proxy_port == 9190 + assert proxy.proxy_login == "" + assert proxy.proxy_password == "" + + def test_socks4(self): + proxy = Proxy.parse_url("socks4://10.0.0.1:1080") + assert proxy.proxy_type == "socks4" + assert proxy.proxy_address == "10.0.0.1" + assert proxy.proxy_port == 1080 + + def test_invalid_url_no_port(self): + with pytest.raises(ValueError, match="Invalid proxy URL"): + Proxy.parse_url("socks5://host.example.com") + + def test_invalid_url_no_host(self): + with pytest.raises(ValueError, match="Invalid proxy URL"): + Proxy.parse_url("socks5://:8080") + + +class TestProxyToKwargs: + def test_output(self): + proxy = Proxy( + proxy_type="http", + proxy_address="1.2.3.4", + proxy_port=8080, + proxy_login="user", + proxy_password="pass", + ) + kwargs = proxy.to_kwargs() + assert kwargs == { + "proxy_type": "http", + "proxy_address": "1.2.3.4", + "proxy_port": 8080, + "proxy_login": "user", + "proxy_password": "pass", + } + + def test_unpack_into_task(self): + """Verify to_kwargs() can be unpacked into proxy-requiring task constructors.""" + proxy = Proxy.parse_url("socks5://u:p@h:8080") + kwargs = proxy.to_kwargs() + assert "proxy_type" in kwargs + assert "proxy_address" in kwargs + assert "proxy_port" in kwargs + + +class TestProxyFrozen: + def test_immutable(self): + proxy = Proxy.parse_url("http://1.2.3.4:80") + with pytest.raises(AttributeError): + proxy.proxy_type = "socks5" diff --git a/tests/test_tasks.py b/tests/test_tasks.py index f6994c2..9ee0e96 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -292,8 +292,7 @@ def test_type(self): **USER_AGENT_KWARGS, **PROXY_KWARGS, ) - # GeeTestTask doesn't set type explicitly, inherits from GeeTestTaskProxyless - assert task.serialize()["type"] == "GeeTestTaskProxyless" + assert task.serialize()["type"] == "GeeTestTask" class TestAntiGateTaskProxyless: