diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index df94488..974b25c 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -27,7 +27,7 @@ jobs: - uses: nanasess/setup-chromedriver@v2 - name: Run integration tests - run: make test + run: make test_e2e env: KEY: ${{ secrets.anticaptcha_key }} PROXY_URL: "${{ secrets.proxy_url }}" diff --git a/Makefile b/Makefile index 2723b0e..d85d900 100644 --- a/Makefile +++ b/Makefile @@ -1,7 +1,7 @@ CHROMEDRIVER_VERSION=99.0.4844.17 CHROMEDRIVER_DIR=${PWD}/geckodriver -.PHONY: lint fmt build docs install test gecko +.PHONY: lint fmt build docs install test test_e2e gecko build: python setup.py sdist bdist_wheel @@ -25,7 +25,10 @@ gecko: rm ${CHROMEDRIVER_DIR}/chromedriver_linux64.zip test: - PATH=$$PWD/geckodriver:$$PATH nose2 --verbose + pytest + +test_e2e: + PATH=$$PWD/geckodriver:$$PATH pytest -m e2e --override-ini="addopts=" clean: rm -r build geckodriver diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f550004 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,6 @@ +[tool.pytest.ini_options] +testpaths = ["tests"] +markers = [ + "e2e: end-to-end tests requiring API keys and network access", +] +addopts = "-m 'not e2e'" diff --git a/setup.cfg b/setup.cfg index 12db776..b30d3fa 100644 --- a/setup.cfg +++ b/setup.cfg @@ -3,6 +3,3 @@ current_version = 0.4.2 commit = True tag = True tag_name = {new_version} - -[nosetests] -process-timeout = 600 diff --git a/setup.py b/setup.py index d8f87d7..6522cf8 100644 --- a/setup.py +++ b/setup.py @@ -7,12 +7,11 @@ long_description = f.read() -tests_deps = ["retry", "nose2", "selenium"] +tests_deps = ["retry", "pytest", "selenium"] extras = {"tests": tests_deps, "docs": "sphinx"} setup( - test_suite="nose2.collector.collector", name="python-anticaptcha", description="Client library for solve captchas with Anticaptcha.com support.", long_description=long_description, @@ -39,6 +38,5 @@ keywords="recaptcha captcha development", packages=["python_anticaptcha"], install_requires=["requests"], - tests_require=tests_deps, extras_require=extras, ) diff --git a/tests/test_base.py b/tests/test_base.py new file mode 100644 index 0000000..e18611d --- /dev/null +++ b/tests/test_base.py @@ -0,0 +1,149 @@ +from unittest.mock import patch, MagicMock +import pytest + +from python_anticaptcha.base import AnticaptchaClient, Job, SLEEP_EVERY_CHECK_FINISHED +from python_anticaptcha.exceptions import AnticaptchaException + + +class TestAnticaptchaClientInit: + def test_https_url(self): + client = AnticaptchaClient("key123") + assert client.base_url == "https://api.anti-captcha.com/" + assert client.client_key == "key123" + + def test_http_url(self): + client = AnticaptchaClient("key123", use_ssl=False) + assert client.base_url == "http://api.anti-captcha.com/" + + def test_custom_host(self): + client = AnticaptchaClient("key123", host="custom.host.com") + assert client.base_url == "https://custom.host.com/" + + def test_language_pool(self): + client = AnticaptchaClient("key123", language_pool="rn") + assert client.language_pool == "rn" + + +class TestCheckResponse: + def setup_method(self): + self.client = AnticaptchaClient("key123") + + def test_success_passthrough(self): + response = {"errorId": 0, "taskId": 42} + self.client._check_response(response) + + def test_error_raises(self): + response = { + "errorId": 1, + "errorCode": "ERROR_KEY_DOES_NOT_EXIST", + "errorDescription": "Account authorization key not found", + } + with pytest.raises(AnticaptchaException) as exc_info: + self.client._check_response(response) + assert exc_info.value.error_id == 1 + assert exc_info.value.error_code == "ERROR_KEY_DOES_NOT_EXIST" + + def test_error_id_11_appends_ip(self): + response = { + "errorId": 11, + "errorCode": "ERROR_IP_NOT_ALLOWED", + "errorDescription": "IP not allowed", + } + with patch.object( + type(self.client), "client_ip", new_callable=lambda: property(lambda self: "5.6.7.8") + ): + with pytest.raises(AnticaptchaException) as exc_info: + self.client._check_response(response) + assert "5.6.7.8" in exc_info.value.error_description + + +class TestCreateTask: + def test_payload_structure(self): + client = AnticaptchaClient("key123") + mock_task = MagicMock() + mock_task.serialize.return_value = {"type": "NoCaptchaTaskProxyless", "websiteURL": "https://example.com"} + + mock_response = MagicMock() + mock_response.json.return_value = {"errorId": 0, "taskId": 99} + + with patch.object(client.session, "post", return_value=mock_response) as mock_post: + job = client.createTask(mock_task) + + call_kwargs = mock_post.call_args + payload = call_kwargs[1]["json"] if "json" in call_kwargs[1] else call_kwargs.kwargs["json"] + assert payload["clientKey"] == "key123" + assert payload["task"] == {"type": "NoCaptchaTaskProxyless", "websiteURL": "https://example.com"} + assert payload["softId"] == 847 + assert payload["languagePool"] == "en" + assert isinstance(job, Job) + assert job.task_id == 99 + + +class TestGetBalance: + def test_returns_balance(self): + client = AnticaptchaClient("key123") + mock_response = MagicMock() + mock_response.json.return_value = {"errorId": 0, "balance": 3.21} + + with patch.object(client.session, "post", return_value=mock_response): + balance = client.getBalance() + assert balance == 3.21 + + +class TestJobCheckIsReady: + def test_ready(self): + client = MagicMock() + client.getTaskResult.return_value = {"status": "ready", "solution": {}} + job = Job(client, task_id=1) + assert job.check_is_ready() is True + + def test_processing(self): + client = MagicMock() + client.getTaskResult.return_value = {"status": "processing"} + job = Job(client, task_id=1) + assert job.check_is_ready() is False + + +class TestJobSolutionGetters: + def setup_method(self): + self.client = MagicMock() + self.job = Job(self.client, task_id=1) + self.job._last_result = { + "status": "ready", + "solution": { + "gRecaptchaResponse": "recaptcha-token", + "token": "funcaptcha-token", + "text": "captcha text", + "cellNumbers": [1, 3, 5], + "answers": {"q1": "a1"}, + }, + } + + def test_get_solution_response(self): + assert self.job.get_solution_response() == "recaptcha-token" + + def test_get_token_response(self): + assert self.job.get_token_response() == "funcaptcha-token" + + def test_get_captcha_text(self): + assert self.job.get_captcha_text() == "captcha text" + + def test_get_cells_numbers(self): + assert self.job.get_cells_numbers() == [1, 3, 5] + + def test_get_answers(self): + assert self.job.get_answers() == {"q1": "a1"} + + def test_get_solution(self): + assert self.job.get_solution() == self.job._last_result["solution"] + + +class TestJobJoinTimeout: + @patch("python_anticaptcha.base.time.sleep") + def test_timeout_raises(self, mock_sleep): + client = MagicMock() + client.getTaskResult.return_value = {"status": "processing"} + job = Job(client, task_id=1) + with pytest.raises(AnticaptchaException) as exc_info: + job.join(maximum_time=SLEEP_EVERY_CHECK_FINISHED) + assert "exceeded" in str(exc_info.value).lower() diff --git a/tests/test_examples.py b/tests/test_examples.py index 1a7de99..d3573bf 100644 --- a/tests/test_examples.py +++ b/tests/test_examples.py @@ -3,12 +3,11 @@ from retry import retry import os +import pytest from python_anticaptcha import AnticatpchaException from contextlib import contextmanager -_multiprocess_can_split_ = True - def missing_key(*args, **kwargs): return skipIf( @@ -23,6 +22,7 @@ def missing_proxy(*args, **kwargs): )(*args, **kwargs) +@pytest.mark.e2e @missing_key class AntiGateTestCase(TestCase): @retry(tries=3) @@ -34,6 +34,7 @@ def test_process_antigate(self): self.assertIn(key, solution) +@pytest.mark.e2e @missing_key @missing_proxy class FuncaptchaTestCase(TestCase): @@ -46,6 +47,7 @@ def test_funcaptcha(self): self.assertIn("Solved!", funcaptcha_request.process()) +@pytest.mark.e2e @missing_key class RecaptchaRequestTestCase(TestCase): # Anticaptcha responds is not fully reliable. @@ -56,6 +58,7 @@ def test_process(self): self.assertIn(recaptcha_request.EXPECTED_RESULT, recaptcha_request.process()) +@pytest.mark.e2e @missing_key @skipIf(True, "Anti-captcha unable to provide required score, but we tests via proxy") class RecaptchaV3ProxylessTestCase(TestCase): @@ -78,6 +81,7 @@ def open_driver(*args, **kwargs): driver.quit() +@pytest.mark.e2e @missing_key class RecaptchaSeleniumtTestCase(TestCase): # Anticaptcha responds is not fully reliable. @@ -98,6 +102,7 @@ def test_process(self): ) +@pytest.mark.e2e @missing_key class TextTestCase(TestCase): def test_process(self): @@ -106,6 +111,7 @@ def test_process(self): self.assertEqual(text.process(text.IMAGE).lower(), text.EXPECTED_RESULT.lower()) +@pytest.mark.e2e @missing_key @skipIf(True, "We testing via proxy for performance reason.") class HCaptchaTaskProxylessTestCase(TestCase): @@ -116,6 +122,7 @@ def test_process(self): self.assertIn(hcaptcha_request.EXPECTED_RESULT, hcaptcha_request.process()) +@pytest.mark.e2e @missing_key @missing_proxy class HCaptchaTaskTestCase(TestCase): diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py new file mode 100644 index 0000000..ccac6ce --- /dev/null +++ b/tests/test_exceptions.py @@ -0,0 +1,42 @@ +from python_anticaptcha.exceptions import ( + AnticaptchaException, + AnticatpchaException, + InvalidWidthException, + MissingNameException, +) + + +class TestAnticaptchaException: + def test_attributes(self): + exc = AnticaptchaException(1, "ERROR_KEY", "Some description") + assert exc.error_id == 1 + assert exc.error_code == "ERROR_KEY" + assert exc.error_description == "Some description" + + def test_str_format(self): + exc = AnticaptchaException(1, "ERROR_KEY", "Some description") + assert str(exc) == "[ERROR_KEY:1]Some description" + + def test_typo_alias(self): + assert AnticatpchaException is AnticaptchaException + + +class TestInvalidWidthException: + def test_message(self): + exc = InvalidWidthException(75) + assert exc.width == 75 + assert "75" in str(exc) + assert "100, 50, 33, 25" in str(exc) + assert exc.error_id == "AC-1" + assert exc.error_code == 1 + + +class TestMissingNameException: + def test_message(self): + exc = MissingNameException("MyClass") + assert exc.cls == "MyClass" + assert "MyClass" in str(exc) + assert '__init__(name="X")' in str(exc) + assert 'serialize(name="X")' in str(exc) + assert exc.error_id == "AC-2" + assert exc.error_code == 2 diff --git a/tests/test_tasks.py b/tests/test_tasks.py new file mode 100644 index 0000000..f6994c2 --- /dev/null +++ b/tests/test_tasks.py @@ -0,0 +1,365 @@ +import io + +from python_anticaptcha.tasks import ( + NoCaptchaTaskProxylessTask, + RecaptchaV2TaskProxyless, + NoCaptchaTask, + RecaptchaV2Task, + FunCaptchaProxylessTask, + FunCaptchaTask, + ImageToTextTask, + RecaptchaV3TaskProxyless, + HCaptchaTaskProxyless, + HCaptchaTask, + RecaptchaV2EnterpriseTaskProxyless, + RecaptchaV2EnterpriseTask, + GeeTestTaskProxyless, + GeeTestTask, + AntiGateTaskProxyless, + AntiGateTask, +) + +PROXY_KWARGS = dict( + proxy_type="http", + proxy_address="1.2.3.4", + proxy_port=8080, + proxy_login="user", + proxy_password="pass", +) + +USER_AGENT_KWARGS = dict(user_agent="Mozilla/5.0") + + +class TestNoCaptchaTaskProxylessTask: + def test_type(self): + task = NoCaptchaTaskProxylessTask( + website_url="https://example.com", website_key="key123" + ) + data = task.serialize() + assert data["type"] == "NoCaptchaTaskProxyless" + assert data["websiteURL"] == "https://example.com" + assert data["websiteKey"] == "key123" + + def test_optional_fields_omitted(self): + task = NoCaptchaTaskProxylessTask( + website_url="https://example.com", website_key="key123" + ) + data = task.serialize() + assert "websiteSToken" not in data + assert "isInvisible" not in data + assert "recaptchaDataSValue" not in data + + def test_optional_fields_included(self): + task = NoCaptchaTaskProxylessTask( + website_url="https://example.com", + website_key="key123", + website_s_token="stoken", + is_invisible=True, + recaptcha_data_s_value="svalue", + ) + data = task.serialize() + assert data["websiteSToken"] == "stoken" + assert data["isInvisible"] is True + assert data["recaptchaDataSValue"] == "svalue" + + +class TestRecaptchaV2TaskProxyless: + def test_type(self): + task = RecaptchaV2TaskProxyless( + website_url="https://example.com", website_key="key123" + ) + assert task.serialize()["type"] == "RecaptchaV2TaskProxyless" + + +class TestNoCaptchaTask: + def test_type_and_proxy(self): + task = NoCaptchaTask( + website_url="https://example.com", + website_key="key123", + **USER_AGENT_KWARGS, + **PROXY_KWARGS, + ) + data = task.serialize() + assert data["type"] == "NoCaptchaTask" + assert data["proxyType"] == "http" + assert data["proxyAddress"] == "1.2.3.4" + assert data["proxyPort"] == 8080 + assert data["proxyLogin"] == "user" + assert data["proxyPassword"] == "pass" + assert data["userAgent"] == "Mozilla/5.0" + + +class TestRecaptchaV2Task: + def test_type(self): + task = RecaptchaV2Task( + website_url="https://example.com", + website_key="key123", + **USER_AGENT_KWARGS, + **PROXY_KWARGS, + ) + assert task.serialize()["type"] == "RecaptchaV2Task" + + +class TestFunCaptchaProxylessTask: + def test_type_and_required_fields(self): + task = FunCaptchaProxylessTask( + website_url="https://example.com", website_key="pubkey" + ) + data = task.serialize() + assert data["type"] == "FunCaptchaTaskProxyless" + assert data["websiteURL"] == "https://example.com" + assert data["websitePublicKey"] == "pubkey" + assert "funcaptchaApiJSSubdomain" not in data + assert "data" not in data + + def test_optional_fields(self): + task = FunCaptchaProxylessTask( + website_url="https://example.com", + website_key="pubkey", + subdomain="sub.example.com", + data="extra", + ) + data = task.serialize() + assert data["funcaptchaApiJSSubdomain"] == "sub.example.com" + assert data["data"] == "extra" + + +class TestFunCaptchaTask: + def test_type(self): + task = FunCaptchaTask( + website_url="https://example.com", + website_key="pubkey", + **USER_AGENT_KWARGS, + **PROXY_KWARGS, + ) + assert task.serialize()["type"] == "FunCaptchaTask" + + +class TestImageToTextTask: + def test_serialize_base64(self): + fp = io.BytesIO(b"fake image data") + task = ImageToTextTask(fp=fp) + data = task.serialize() + assert data["type"] == "ImageToTextTask" + assert data["body"] == "ZmFrZSBpbWFnZSBkYXRh" + + def test_optional_fields_omitted(self): + fp = io.BytesIO(b"data") + task = ImageToTextTask(fp=fp) + data = task.serialize() + for key in ["phrase", "case", "numeric", "math", "minLength", "maxLength", "comment", "websiteUrl"]: + assert key not in data + + def test_optional_fields_included(self): + fp = io.BytesIO(b"data") + task = ImageToTextTask( + fp=fp, phrase=True, case=True, numeric=1, + math=False, min_length=3, max_length=10, + comment="solve this", website_url="https://example.com", + ) + data = task.serialize() + assert data["phrase"] is True + assert data["case"] is True + assert data["numeric"] == 1 + assert data["math"] is False + assert data["minLength"] == 3 + assert data["maxLength"] == 10 + assert data["comment"] == "solve this" + assert data["websiteUrl"] == "https://example.com" + + +class TestRecaptchaV3TaskProxyless: + def test_serialize(self): + task = RecaptchaV3TaskProxyless( + website_url="https://example.com", + website_key="key123", + min_score=0.9, + page_action="verify", + ) + data = task.serialize() + assert data["type"] == "RecaptchaV3TaskProxyless" + assert data["websiteURL"] == "https://example.com" + assert data["websiteKey"] == "key123" + assert data["minScore"] == 0.9 + assert data["pageAction"] == "verify" + assert data["isEnterprise"] is False + + def test_enterprise_flag(self): + task = RecaptchaV3TaskProxyless( + website_url="https://example.com", + website_key="key123", + min_score=0.3, + page_action="login", + is_enterprise=True, + ) + assert task.serialize()["isEnterprise"] is True + + +class TestHCaptchaTaskProxyless: + def test_serialize(self): + task = HCaptchaTaskProxyless( + website_url="https://example.com", website_key="hkey" + ) + data = task.serialize() + assert data["type"] == "HCaptchaTaskProxyless" + assert data["websiteURL"] == "https://example.com" + assert data["websiteKey"] == "hkey" + + +class TestHCaptchaTask: + def test_type(self): + task = HCaptchaTask( + website_url="https://example.com", + website_key="hkey", + **USER_AGENT_KWARGS, + **PROXY_KWARGS, + ) + assert task.serialize()["type"] == "HCaptchaTask" + + +class TestRecaptchaV2EnterpriseTaskProxyless: + def test_required_fields(self): + task = RecaptchaV2EnterpriseTaskProxyless( + website_url="https://example.com", + website_key="ekey", + enterprise_payload=None, + api_domain=None, + ) + data = task.serialize() + assert data["type"] == "RecaptchaV2EnterpriseTaskProxyless" + assert data["websiteURL"] == "https://example.com" + assert data["websiteKey"] == "ekey" + assert "enterprisePayload" not in data + assert "apiDomain" not in data + + def test_optional_fields(self): + task = RecaptchaV2EnterpriseTaskProxyless( + website_url="https://example.com", + website_key="ekey", + enterprise_payload={"s": "value"}, + api_domain="recaptcha.net", + ) + data = task.serialize() + assert data["enterprisePayload"] == {"s": "value"} + assert data["apiDomain"] == "recaptcha.net" + + +class TestRecaptchaV2EnterpriseTask: + def test_type(self): + task = RecaptchaV2EnterpriseTask( + website_url="https://example.com", + website_key="ekey", + enterprise_payload=None, + api_domain=None, + **USER_AGENT_KWARGS, + **PROXY_KWARGS, + ) + assert task.serialize()["type"] == "RecaptchaV2EnterpriseTask" + + +class TestGeeTestTaskProxyless: + def test_required_fields(self): + task = GeeTestTaskProxyless( + website_url="https://example.com", gt="gt123", challenge="ch456" + ) + data = task.serialize() + assert data["type"] == "GeeTestTaskProxyless" + assert data["websiteURL"] == "https://example.com" + assert data["gt"] == "gt123" + assert data["challenge"] == "ch456" + assert "geetestApiServerSubdomain" not in data + assert "geetestGetLib" not in data + + def test_optional_fields(self): + task = GeeTestTaskProxyless( + website_url="https://example.com", + gt="gt123", + challenge="ch456", + subdomain="api.geetest.com", + lib="getlib.js", + ) + data = task.serialize() + assert data["geetestApiServerSubdomain"] == "api.geetest.com" + assert data["geetestGetLib"] == "getlib.js" + + +class TestGeeTestTask: + def test_type(self): + task = GeeTestTask( + website_url="https://example.com", + gt="gt123", + challenge="ch456", + **USER_AGENT_KWARGS, + **PROXY_KWARGS, + ) + # GeeTestTask doesn't set type explicitly, inherits from GeeTestTaskProxyless + assert task.serialize()["type"] == "GeeTestTaskProxyless" + + +class TestAntiGateTaskProxyless: + def test_serialize(self): + task = AntiGateTaskProxyless( + website_url="https://example.com", + template_name="my_template", + variables={"var1": "val1", "var2": "val2"}, + ) + data = task.serialize() + assert data["type"] == "AntiGateTask" + assert data["websiteURL"] == "https://example.com" + assert data["templateName"] == "my_template" + assert data["variables"] == {"var1": "val1", "var2": "val2"} + + +class TestAntiGateTask: + def test_type(self): + task = AntiGateTask( + website_url="https://example.com", + template_name="tmpl", + variables={}, + **PROXY_KWARGS, + ) + # AntiGateTask doesn't set type explicitly, inherits AntiGateTask type + assert task.serialize()["type"] == "AntiGateTask" + + +class TestProxyMixin: + def test_proxy_login_omitted_when_falsy(self): + task = NoCaptchaTask( + website_url="https://example.com", + website_key="key123", + proxy_type="http", + proxy_address="1.2.3.4", + proxy_port=8080, + proxy_login="", + proxy_password="", + user_agent="Mozilla/5.0", + ) + data = task.serialize() + assert "proxyLogin" not in data + assert "proxyPassword" not in data + assert data["proxyType"] == "http" + assert data["proxyAddress"] == "1.2.3.4" + assert data["proxyPort"] == 8080 + + +class TestCookieMixin: + def test_cookies_omitted_when_empty(self): + task = NoCaptchaTask( + website_url="https://example.com", + website_key="key123", + **USER_AGENT_KWARGS, + **PROXY_KWARGS, + ) + data = task.serialize() + assert "cookies" not in data + + def test_cookies_included_when_set(self): + task = NoCaptchaTask( + website_url="https://example.com", + website_key="key123", + cookies="session=abc", + **USER_AGENT_KWARGS, + **PROXY_KWARGS, + ) + data = task.serialize() + assert data["cookies"] == "session=abc" diff --git a/tox.ini b/tox.ini index 802ff7d..6e64824 100644 --- a/tox.ini +++ b/tox.ini @@ -2,7 +2,10 @@ envlist = py39,py310,py311,py312,py313,py314 [testenv] -passenv=KEY PROXY_URL -deps=.[tests] -commands= - nose2 --verbose +deps = .[tests] +commands = pytest {posargs} + +[testenv:e2e] +passenv = KEY, PROXY_URL +deps = .[tests] +commands = pytest -m e2e --override-ini="addopts=" {posargs} diff --git a/unittest.cfg b/unittest.cfg deleted file mode 100644 index 23b5951..0000000 --- a/unittest.cfg +++ /dev/null @@ -1,7 +0,0 @@ -[unittest] -plugins = nose2.plugins.mp - -[multiprocess] -always-on = true -processes = 8 -test-run-timeout = 5 \ No newline at end of file