pytest में बिना रन अटकाए कैप्चा कैसे हल करें

pytest में कैप्चा हल करने के लिए conftest.py में हर सेशन के लिए एक CapSkip क्लाइंट बनाएँ और हल को हर टेस्ट के अंदर करें, क्योंकि token सिर्फ़ एक बार काम करता है। जब CapSkip चल नहीं रहा हो, तब कैप्चा टेस्ट स्किप कर दें। फिर SDK को उसके अपने timeouts दें। Windows पर pytest-timeout सिर्फ़ अपना thread मेथड इस्तेमाल कर सकता है, और जब वह चलता है तो पूरा रन खत्म कर देता है। Python SDK कोई HTTP timeout सेट नहीं करता, इसलिए चुप पड़ा CapSkip बिना किसी error के किसी टेस्ट को अटका देता है। CapSkip आपके अपने Windows PC पर चलता है, इसलिए जो suite हर रात सैकड़ों बार हल करता है, उस पर प्रति-हल कोई खर्च नहीं आता। यह pytest कैप्चा सेटअप इन सभी जालों को कवर करता है, साथ में xdist के साथ पैरेलल रन और async टेस्ट को भी।

आपको क्या चाहिए

• किसी Windows PC पर चल रहा CapSkip, जिसमें reCAPTCHA सॉल्वर चालू हो। पहली बार शुरू करने के चरण सेटअप गाइड में समझाए गए हैं।
• Python 3.10 या उसके बाद का वर्ज़न। capskip 1.3.0 पैकेज और pytest 9, दोनों को यही चाहिए। नीचे का सब कुछ Windows 11 पर Python 3.14.0 के साथ चलाया गया, pytest 9.1.1, pytest-timeout 2.4.0, pytest-xdist 3.8.0, pytest-asyncio 1.4.0 और requests 2.34.2 के साथ। कनेक्शन ठुकराए जाने के समय 3.13.7 पर भी वही निकले।
• एक PowerShell prompt। कमांड PowerShell syntax में लिखी गई हैं।
• आपकी अपनी एक staging साइट, जिसके लॉगिन फॉर्म में reCAPTCHA v2 checkbox हो। उदाहरण Django जैसे एक फॉर्म में लॉग इन करता है जिसमें CSRF token होता है। फ़ील्ड के नाम अपनी साइट के हिसाब से बदलें।
• सॉल्वर के लिए एक पता। Local मोड में CapSkip केवल उसी डिवाइस के लिए 127.0.0.1 पर जवाब देता है। Server मोड में वह आपके नेटवर्क पते या पब्लिक IP पर सुनता है, ताकि किसी दूसरी मशीन पर चल रहे टेस्ट API के ज़रिए उसे कॉल कर सकें। दोनों यहाँ सेट किए जाते हैं: कनेक्शन सेटिंग्स.

नीचे की माप एक ही PC पर चल रहे mock CapSkip और mock लॉगिन पेज से ली गई हैं (यानी टेस्ट के लिए बनाए गए नकली CapSkip और नकली पेज), और इनके साथ असली pytest और असली SDK इस्तेमाल हुआ। pytest और SDK का व्यवहार आपकी अपनी साइट पर भी वैसा ही रहता है। हल करने का समय, thread की सीमाएँ और लॉगिन पेज के स्वीकार व अस्वीकार करने के नियम mock के अपने हैं।

चरण 1: प्रोजेक्ट बनाएँ

सब कुछ एक ही प्रोजेक्ट फ़ोल्डर में रखें: conftest.py, tokens.py और टेस्ट फ़ाइलें। virtual environment इस्तेमाल करें। नीचे की हर कमांड virtual environment के अपने Python को उसके path से चलाती है, इसलिए आपको उसे activate करने की ज़रूरत नहीं। pytest को python -m pytest के रूप में शुरू करें। यह रूप मौजूदा फ़ोल्डर को import path में जोड़ देता है। अगर आप conftest.py को किसी tests फ़ोल्डर में ले जाएँ, तो tokens.py को प्रोजेक्ट फ़ोल्डर में ही रखें। तब pytest.exe ModuleNotFoundError: No module named ‘tokens’ के साथ फेल होता है, और python -m pytest काम करता है।

# Windows PowerShell, in your project folder
python -m venv .venv
.venv\Scripts\python -m pip install pytest "capskip==1.3.0" requests

चरण 2: पूरे रन के लिए एक CapSkip क्लाइंट

इसे conftest.py के रूप में सेव करें। यह CapSkip की सेटिंग्स एनवायरनमेंट से पढ़ता है, क्योंकि SDK उन्हें खुद नहीं पढ़ता, और डिफ़ॉल्ट वैल्यू की वजह से यह बिना किसी सेटअप के चल जाता है। CAPSKIP_API_KEY तभी मायने रखता है जब आपने CapSkip में API key validation चालू किया हो। fixture पूरे सेशन के लिए एक क्लाइंट बनाता है। साथ ही वह यह भी पता लगाता है कि CapSkip चल रहा है या नहीं।

# conftest.py (next to your tests): pip install pytest capskip requests
import os
import socket

import pytest
from capskip import CapSkip

# The SDK reads no environment variables, so the fixture reads them.
HOST = os.environ.get("CAPSKIP_HOST", "127.0.0.1")
PORT = int(os.environ.get("CAPSKIP_PORT", "8080"))
API_KEY = os.environ.get("CAPSKIP_API_KEY", "capskip")


@pytest.fixture(scope="session")
def capskip_client():
    # CapSkip has no ping endpoint, so try a plain TCP connect. Use the IP,
    # not "localhost": a refused connect takes 2 s, or 4 s for localhost.
    try:
        socket.create_connection((HOST, PORT), timeout=3).close()
    except OSError as exc:
        pytest.skip(f"CapSkip is not running on {HOST}:{PORT}: {exc}")
    # One client for the whole run. A solve gives up after 90 s of polling.
    return CapSkip(apiKey=API_KEY, host=HOST, port=PORT, recaptchaTimeout=90)

CapSkip में कोई ping endpoint नहीं है, इसलिए एक सादा TCP connect पूछता है कि कोई सुन रहा है या नहीं, और वह कोई solve job नहीं भेजता। अगर कोई नहीं सुन रहा, तो fixture फेल होने के बजाय स्किप कर देता है। CapSkip बंद होने पर 127.0.0.1 से ठुकराए गए connect में Windows पर 2.05 सेकंड लगे, और localhost नाम पर करीब 4.1 सेकंड। स्किप cache हो जाता है, इसलिए पूरे सेशन में यह कीमत सिर्फ़ एक बार चुकानी पड़ती है: दो फ़ाइलों के चार कैप्चा टेस्ट में कुल 2.05 सेकंड लगे।

IP इस्तेमाल करें। localhost नाम की एक दूसरी कीमत तब भी है जब CapSkip चल रहा हो। सिर्फ़ IPv4 पर सुनने वाले listener के सामने, दो लॉगिन टेस्ट localhost के साथ 10.3 सेकंड में और 127.0.0.1 के साथ 3.75 सेकंड में पूरे हुए। यानी कनेक्शन जाँच के लिए एक बार करीब 2 सेकंड, और हर हल पर 1 से 2.5 सेकंड और।

चरण 3: एक helper जो एक बार हल करता है, और एक बार रीट्राई करता है

इसे tokens.py के रूप में सेव करें। कुछ हल फेल हो जाते हैं, और नई कोशिश काम कर सकती है। SDK एक ApiException उठाता है जिसका संदेश ERROR_CAPTCHA_UNSOLVABLE होता है, जब तक कि आपने CapSkip में कोई Error Replacement टेक्स्ट सेट न किया हो। तब संदेश में वही टेक्स्ट होता है, और नीचे की जाँच कभी मेल नहीं खाती। helper सिर्फ़ उसी नाकामी पर रीट्राई करता है। CapSkip जिस submit को ठुकरा दे, जैसे ERROR_GOOGLEKEY या ERROR_PAGEURL, वह तुरंत exception उठाता है। जिस sitekey को Google ठुकरा दे, वह भी UNSOLVABLE जैसा ही दिखता है, इसलिए फेल होने से पहले उसे दो बार आज़माया जाता है। इस error का मतलब जानने के लिए ERROR_CAPTCHA_UNSOLVABLE देखें।

# tokens.py: the helper your tests (and your app) call
from capskip import ApiException


def fresh_token(solver, sitekey, url, attempts=2):
    """Solve a reCAPTCHA v2 and return the token string."""
    for attempt in range(attempts):
        try:
            return solver.recaptcha(sitekey=sitekey, url=url)["code"]
        except ApiException as exc:
            # UNSOLVABLE is how CapSkip reports a solve that failed this
            # time. A refused submit (ERROR_GOOGLEKEY, ERROR_PAGEURL) will
            # fail again, so it raises at once.
            last = attempt == attempts - 1
            if last or "ERROR_CAPTCHA_UNSOLVABLE" not in str(exc):
                raise

चरण 4: हर टेस्ट के लिए एक fixture, और एक marker

अगला ब्लॉक conftest.py के आख़िर में जोड़ें। यह एक factory है, token नहीं। इसे हर बार कॉल करना एक नया हल है, और हर टेस्ट अपने लिए अलग माँगता है, क्योंकि reCAPTCHA token सिर्फ़ एक बार स्वीकार होता है। Python फ़ाइल के बीच में भी imports की अनुमति देता है, इसलिए import लाइन वहीं रह सकती है जहाँ ब्लॉक उसे रखता है।

from tokens import fresh_token


@pytest.fixture
def solve_recaptcha(capskip_client):
    # A factory, not a token: every call is a new solve, because a
    # reCAPTCHA token is accepted once. Function scope, so each test
    # asks for its own.
    def solve(sitekey, url):
        return fresh_token(capskip_client, sitekey, url)

    return solve

फिर नीचे का marker hook conftest.py में जोड़ें। यह captcha नाम का एक marker register करता है, ताकि आप तेज़ टेस्ट CapSkip के बिना और live वाले उसके साथ चला सकें। strict markers ऑप्शन चालू होने पर, जिसे चरण 6 चालू करता है, कोई भी ऐसा marker जो आपने register नहीं किया, error बन जाता है। captcha को register करने से असली वाला चलता रहता है और गलत वर्तनी error बन जाती है।

def pytest_configure(config):
    # Tell pytest the marker exists, so --strict-markers accepts it and a
    # misspelling such as "captcah" is an error instead of a silent mark.
    config.addinivalue_line(
        "markers", "captcha: needs a running CapSkip and solves a real CAPTCHA"
    )

चरण 5: लॉगिन टेस्ट लिखें

अगले दो ब्लॉक को एक ही फ़ाइल test_login.py के रूप में सेव करें। यह पहला ब्लॉक आपका अपना है, CapSkip का नहीं। यह लॉगिन पेज खोलता है, CSRF token और sitekey पढ़ता है, और फॉर्म पोस्ट करता है। requests फॉर्म पोस्ट पर कोई Referer नहीं भेजता। HTTPS पर Django जैसी सख़्त साइट बिना Referer वाली पोस्ट को ठुकरा देती है, और mock ने भी 403 के साथ यही किया। इसलिए helper उसे सेट करता है। दोनों patterns और फ़ील्ड के नाम अपने markup के हिसाब से बदलें, और post_form में किसी टेस्ट अकाउंट का email और password रखें।

# test_login.py: set LOGIN_URL to your own login page
import os
import re

import pytest
import requests

LOGIN_URL = os.environ.get("LOGIN_URL", "https://example.com/login")


def open_form(session):
    html = session.get(LOGIN_URL, timeout=15).text
    csrf = re.search(r'name="csrfmiddlewaretoken" value="([^"]+)"', html)
    sitekey = re.search(r'data-sitekey="([^"]+)"', html)
    return csrf.group(1), sitekey.group(1)


def post_form(session, csrf, token, password="secret"):
    # requests sends no Referer, and a Django site refuses an HTTPS post
    # that has none, so set it by hand.
    return session.post(LOGIN_URL, timeout=15, headers={"Referer": LOGIN_URL}, data={
        "csrfmiddlewaretoken": csrf,
        "email": "[email protected]",
        "password": password,
        "g-recaptcha-response": token,
    })

टेस्ट सबसे आख़िर में हल करते हैं और तुरंत पोस्ट करते हैं, क्योंकि token करीब दो मिनट बाद एक्सपायर हो जाता है। इस पहलू को reCAPTCHA token expiration की गाइड समझाती है। Welcome, 403 और error टेक्स्ट को वैसा बदलें जैसा आपकी अपनी साइट दिखाती है। एक तयशुदा username दूसरी रात के रन में टकरा जाता है, इसलिए अगर आपका फॉर्म अकाउंट बनाता है तो हर बार नया इस्तेमाल करें।

@pytest.mark.captcha
def test_login(solve_recaptcha):
    session = requests.Session()
    csrf, sitekey = open_form(session)
    token = solve_recaptcha(sitekey, LOGIN_URL)  # solve last, post at once
    resp = post_form(session, csrf, token)
    assert resp.status_code == 200 and "Welcome" in resp.text


@pytest.mark.captcha
def test_wrong_password_is_refused(solve_recaptcha):
    session = requests.Session()
    csrf, sitekey = open_form(session)
    token = solve_recaptcha(sitekey, LOGIN_URL)  # a new solve, not test 1's
    resp = post_form(session, csrf, token, password="wrong")
    assert resp.status_code == 403
    assert "Invalid email or password" in resp.text

चरण 6: पहले तेज़ टेस्ट चलाएँ, फिर live वाले

पहली लाइन टेस्ट को आपके लॉगिन पेज की ओर पॉइंट करती है। दूसरी लाइन गलत वर्तनी वाले marker को error बना देती है। यह एक एनवायरनमेंट वेरिएबल सेट करती है जिसे pytest अतिरिक्त ऑप्शन के लिए पढ़ता है, और वह उस PowerShell सेशन के बाकी समय तक बना रहता है। तीसरी लाइन कैप्चा टेस्ट को छोड़कर बाकी सब चलाती है। चौथी सिर्फ़ live वाले चलाती है। skip वाला switch बताता है कि टेस्ट क्यों स्किप हुआ, और durations वाला switch सबसे धीमे टेस्ट की सूची देता है।

# Windows PowerShell: your login page, then make a misspelled marker an error
$env:LOGIN_URL = "https://staging.example.com/login"
$env:PYTEST_ADDOPTS = "--strict-markers"
.venv\Scripts\python -m pytest -m "not captcha"      # fast suite, no CapSkip
.venv\Scripts\python -m pytest -m captcha -rs -v --durations=5   # live tests

इस मोड़ पर तेज़ रन के लिए चलाने को कुछ नहीं मिलता, और pytest exit code 5 के साथ बंद होता है। नीचे के unit tests वाले सेक्शन से test_unit.py सेव करने के बाद उसमें टेस्ट आ जाते हैं। तब उसे सॉल्वर से कुछ नहीं चाहिए: CapSkip बंद होने पर भी उसके चारों टेस्ट 0.02 सेकंड में पास हो गए। live रन इस तरह की एक लाइन के साथ स्किप हुआ, जो conftest.py की ओर नहीं, टेस्ट की ओर इशारा करती है:

• SKIPPED [1] test_login.py:29: CapSkip is not running on 127.0.0.1:8080: [WinError 10061] No connection could be made because the target machine actively refused it

स्किप हुआ suite exit code 0 के साथ बंद होता है, और pytest 9.1.1 में कोई switch नहीं है जो स्किप को फेल में बदल दे। अगर आपके build को CapSkip चाहिए ही, तो junitxml ऑप्शन से एक junit फ़ाइल (XML टेस्ट रिपोर्ट) लिखें और उसकी skipped गिनती शून्य से ऊपर होने पर build फेल कर दें। strict ऑप्शन चालू होने पर, किसी टेस्ट में captcha की जगह captcah टाइप करने से collection exit code 2 और संदेश ‘captcah’ not found in `markers` configuration option के साथ रुक जाता है। strict ऑप्शन के बिना यह typo सिर्फ़ एक warning होती, और वह टेस्ट तेज़ suite में चल जाता।

हर टेस्ट के अंदर हल करें

एक fixture में एक बार हल करके token साझा न करें। टेस्टिंग में, module-scoped fixture ने एक बार हल किया और दो टेस्ट ने वही token पोस्ट किया। पहला स्वीकार हुआ। दूसरा mock के reCAPTCHA failed: token already used के साथ फेल हुआ। असली साइट इसे अलग शब्दों में कहती है। कौन-सा टेस्ट हारता है, यह क्रम पर निर्भर है, इसलिए नाकामी इधर-उधर होती रहती है और flaky लगती है। xdist के साथ यह और बुरा है, जैसा पैरेलल सेक्शन दिखाता है। यह रही पूरी गलती, ताकि आप उसे पहचान सकें। उसे फेल होते देखने के बाद फ़ाइल डिलीट कर दें, वरना वह आगे के रन को अनियमित रूप से फेल या पास कराएगी।

# test_shared_token.py: the mistake, so you can see it fail
import pytest
import requests

from test_login import LOGIN_URL, open_form, post_form


@pytest.fixture(scope="module")
def shared_token(capskip_client):
    csrf, sitekey = open_form(requests.Session())
    return capskip_client.recaptcha(sitekey=sitekey, url=LOGIN_URL)["code"]


@pytest.mark.captcha
@pytest.mark.parametrize("n", [1, 2])
def test_each_login_reuses_one_token(shared_token, n):
    session = requests.Session()
    csrf, sitekey = open_form(session)
    resp = post_form(session, csrf, shared_token)
    assert resp.status_code == 200, resp.text  # n=2 fails: used already

हर टेस्ट के लिए एक हल में कुछ सेकंड लगते हैं। जिस टेस्ट को खुद कैप्चा की ज़रूरत नहीं, उसे fixture माँगना ही नहीं चाहिए। नीचे का fake solver उसके आसपास के लॉजिक को कवर करता है।

अटके हुए हल को फेल करवाएँ, रन को खत्म न होने दें

यह वह हिस्सा है जो Windows पर अलग है। pytest-timeout प्लगइन के दो मेथड हैं। signal मेथड को SIGALRM चाहिए, जो Windows में नहीं है, इसलिए वह वहाँ उपलब्ध नहीं है। उसे माँगने पर INTERNALERROR और AttributeError: module ‘signal’ has no attribute ‘SIGALRM’ के साथ खत्म हुआ। thread मेथड डिफ़ॉल्ट है, और वही अकेला है जो काम करता है। 10 सेकंड की सीमा और रन में चार टेस्ट के साथ, किसी टेस्ट के अटकने पर उसने यह किया:

• उसने main thread का stack प्रिंट किया, करीब 100 लाइनें, और प्रोसेस को 12.5 सेकंड पर exit code 1 के साथ खत्म कर दिया।
• अटके हुए टेस्ट को कोई PASSED या FAILED लाइन नहीं मिली।
• कोई summary प्रिंट नहीं हुई और कोई junit फ़ाइल नहीं लिखी गई।
• उसके बाद के दो टेस्ट कभी चले ही नहीं।

xdist के साथ नुकसान छोटा होता है। सिर्फ़ वही worker मरता है जो वह टेस्ट चला रहा था। टेस्ट को FAILED बताया जाता है, और साथ में worker gw1 crashed while running लिखा आता है, एक बदले का worker शुरू हो जाता है, और रन summary और exit code 1 के साथ पूरा होता है।

इसलिए thread timeout आख़िरी उपाय है, टेस्ट timeout नहीं। आप चाहते हैं कि SDK पहले फेल हो, एक सामान्य exception के साथ, ताकि टेस्ट फेल हो और रन आगे चलता रहे। Python SDK 1.3.0 इसे मुश्किल बना देता है। वह requests को बिना timeout के कॉल करता है, इसलिए जो CapSkip कॉल स्वीकार करके कभी जवाब नहीं देता, वह उसे अटका देता है। किसी बदलाव के बिना, चुप पड़ा submit तब भी अटका हुआ था जब हमने उसे 45 सेकंड पर रोका। उसका recaptchaTimeout सिर्फ़ दो poll के बीच जाँचा जाता है, और अटके हुए poll को अभी तैयार नहीं माना जाता।

फ़िक्स यह है कि SDK जिन दो request functions का इस्तेमाल करता है, उन्हें timeout वाले functions से बदल दें। अगला ब्लॉक conftest.py में जोड़ें। यह SDK के अंदर हाथ डालता है, इसलिए capskip को वर्ज़न 1.3.0 पर पिन करें। hasattr वाली लाइन यह तय करती है कि अगर कोई बाद का वर्ज़न requests इस्तेमाल करना बंद कर दे तो यह कुछ न करे। URL के रूप में दिए गए image कैप्चा के डाउनलोड को यह कवर नहीं करता, क्योंकि SDK वह अपनी अलग कॉल से करता है।

import functools
import types

import capskip.api
import requests

# capskip 1.3.0 sets no HTTP timeout, so a CapSkip that accepts a call
# and never answers would hang the test. Give every SDK call 5 s to
# connect and 30 s to answer. This swaps a name inside the SDK, so pin
# capskip==1.3.0 and do nothing if the SDK stops using requests.
if hasattr(capskip.api, "requests"):
    capskip.api.requests = types.SimpleNamespace(
        RequestException=requests.RequestException,
        post=functools.partial(requests.post, timeout=(5, 30)),
        get=functools.partial(requests.get, timeout=(5, 30)),
    )

patch के साथ, जिस CapSkip ने submit का कभी जवाब नहीं दिया, उसने 30.0 सेकंड पर Read timed out. (read timeout=30) संदेश वाला NetworkException उठाया। pytest टेस्ट 30.8 सेकंड पर फेल हुआ, और रन सही-सलामत रहा। चुप पड़ा poll अलग है। SDK उसे अभी तैयार नहीं मानकर निगल लेता है, और 300 सेकंड वाला डिफ़ॉल्ट recaptchaTimeout TimeoutException के साथ 333.0 सेकंड पर खत्म हुआ। इसीलिए चरण 2 recaptchaTimeout=90 पास करता है। 90 के साथ टेस्ट 92.6 सेकंड पर फेल हुआ।

समय-सीमा करीब 35 सेकंड तक लाँघी जा सकती है। वह poll के बीच जाँची जाती है, एक poll पूरे 30 सेकंड ले सकता है, और SDK जाँचने से पहले 5 सेकंड तक रुकता है। 45 की सीमा के साथ नाकामी 60.8 सेकंड पर आई, और 20 के साथ 30.3 सेकंड पर। इसलिए इन संख्याओं के साथ एक कोशिश की सबसे बुरी स्थिति करीब 125 सेकंड है: polling के 90, और एक अटका हुआ poll। चरण 3 का helper ज़्यादा से ज़्यादा दो कोशिशें करता है, इसलिए सबसे बुरी स्थिति करीब 250 सेकंड है। pytest-timeout की सीमा उससे ऊपर रखें। अगला ब्लॉक प्लगइन इंस्टॉल करता है और 300 सेकंड सेट करता है। एनवायरनमेंट वेरिएबल चरण 6 वाले को बदल देता है, इसलिए वह strict ऑप्शन को दोहराता है। यही दो ऑप्शन आप pytest ini फ़ाइल में timeout और timeout_method के रूप में भी रख सकते हैं। वेरिएबल बस चिपकाने में आसान है।

# Windows PowerShell: pytest-timeout has only the thread method here.
# 300 s is above two attempts at the slowest SDK failure (about 125 s each),
# so the SDK fails first.
.venv\Scripts\python -m pip install pytest-timeout
$env:PYTEST_ADDOPTS = "--strict-markers --timeout=300 --timeout-method=thread"

xdist के साथ टेस्ट पैरेलल चलाएँ

हल करना धीमा है और ज़्यादातर इंतज़ार है, इसलिए xdist मदद करता है। हर xdist worker एक अलग प्रोसेस है। इसलिए session fixture हर worker के लिए एक बार चलता है, और हर worker अपना क्लाइंट बनाता है। चार worker और आठ टेस्ट के साथ, रन में चार प्रोसेस ID और चार क्लाइंट objects दिखे। xdist के बिना हर एक का एक ही होता है। टेस्ट workers-0.log से workers-7.log तक लिखता है, और Get-Content workers-*.log उन्हें दिखाता है। हर worker के लिए एक क्लाइंट ठीक है, क्योंकि क्लाइंट ऐसा कुछ नहीं रखता जिसे worker को साझा करना पड़े।

# test_workers.py: shows one client per worker process
import os
from pathlib import Path

import pytest


@pytest.mark.captcha
@pytest.mark.parametrize("n", range(8))
def test_which_client(capskip_client, worker_id, n):
    line = f"{worker_id} pid={os.getpid()} client={id(capskip_client)}"
    Path(f"workers-{n}.log").write_text(line)

worker की संख्या खुद चुनें। auto सेटिंग का मतलब CPU की गिनती है, यह नहीं कि CapSkip एक साथ कितना हल कर सकता है। टेस्ट PC पर इसने 24 worker दिए। चार unit tests में इसके साथ करीब 4 सेकंड लगे और चार worker के साथ करीब 1.1 सेकंड, worker शुरू करने की लागत की वजह से। CapSkip की अपनी सीमा reCAPTCHA Max. Threads सेटिंग है, जो डिफ़ॉल्ट रूप से 10 है। यह टेबल उसी सीमा और 5 सेकंड के हल वाले mock के सामने 24 लॉगिन की है।

10 worker के बाद फ़ायदा छोटा है: 10 worker के साथ 26.1 सेकंड और 24 के साथ 23.1। अतिरिक्त जॉब CapSkip के अंदर खाली thread का इंतज़ार करती हैं, और जो जॉब Wait Timeout से ज़्यादा इंतज़ार करती है वह फेल हो जाती है। एक छोटे पैमाने के टेस्ट में, आठ टेस्ट एक साथ चले, जबकि सीमा 2 थी, प्रतीक्षा-सीमा 6 सेकंड और हल 4 सेकंड का। चार जॉब ERROR_CAPTCHA_UNSOLVABLE के साथ फेल हुईं। tokens.py की रीट्राई ने उन्हें फिर कतार में डाल दिया, और आठों टेस्ट 22.3 सेकंड में पास हो गए। रीट्राई की यह दूसरी वजह है।

5 सेकंड के हल के लिए हर टेस्ट में करीब 7.9 सेकंड लगे। SDK 0, 0.25, 0.75, 1.75, 3.75 और 7.75 सेकंड पर poll करता है, इसलिए वह पूरे हो चुके हल को अगले poll पर पहचानता है। xdist का इंस्टॉल और कमांड नीचे हैं।

# Each xdist worker is its own process, so each builds its own client.
# Pick the number yourself: -n auto means CPU count, not solver capacity.
.venv\Scripts\python -m pip install pytest-xdist
.venv\Scripts\python -m pytest -m captcha -n 4 -v

एक और जाल। xdist के तहत, पहले बताई गई साझा-token वाली गलती पास भी हो सकती है। module-scoped token हर worker पर एक बार हल होता है, इसलिए दो टेस्ट तभी टकराते हैं जब वे एक ही worker पर आ जाएँ। जो टेस्ट कुछ रन में पास हो और कुछ में फेल, वह साझा token ढूँढने का संकेत है।

AsyncCapSkip के साथ async टेस्ट

Python SDK में एक असली async क्लाइंट है, AsyncCapSkip। यह httpx पर बना है। यह fixture conftest.py में जोड़ें। यह capskip_client माँगता है, इसलिए स्किप की जाँच अब भी लागू रहती है। async क्लाइंट कोई event loop नहीं रखता, इसलिए एक सादा fixture काफ़ी है और टेस्ट डिफ़ॉल्ट function-scoped loops पर चलते हैं।

from capskip import AsyncCapSkip


@pytest.fixture(scope="session")
def async_capskip_client(capskip_client):
    # Asking for capskip_client keeps the skip-when-closed check. The
    # async client holds no event loop, so a plain fixture is enough.
    return AsyncCapSkip(apiKey=API_KEY, host=HOST, port=PORT, recaptchaTimeout=90)

pytest-asyncio 1.4.0 strict मोड में चलता है, और pytest 9 marker के बिना async टेस्ट को Failed: async def functions are not natively supported के साथ फेल कर देता है। अगले ब्लॉक को test_async.py के रूप में सेव करें। YOUR_SITEKEY को पेज के sitekey से बदलें।

# test_async.py: .venv\Scripts\python -m pip install pytest-asyncio
import asyncio

import pytest

from test_login import LOGIN_URL


@pytest.mark.captcha
@pytest.mark.asyncio  # needed: pytest-asyncio runs in strict mode
async def test_five_solves_at_once(async_capskip_client):
    solves = [
        async_capskip_client.recaptcha(sitekey="YOUR_SITEKEY", url=LOGIN_URL)
        for _ in range(5)
    ]
    results = await asyncio.gather(*solves)
    assert len({r["code"] for r in results}) == 5  # five different tokens

यह उम्मीद न रखें कि यह सीधी रेखा में बढ़ेगा। जिस mock ने 1.5 सेकंड में हल किया, उसके सामने 1, 5, 10 और 25 हल को gather करने में 2.6, 2.5, 4.6 और 9.8 सेकंड लगे। क्लाइंट हर कॉल के लिए एक नया httpx कनेक्शन खोलता है, जिससे event loop हर बार थोड़ी देर व्यस्त रहता है। timeout वाले सेक्शन का patch इस क्लाइंट पर लागू नहीं होता। इसकी सीमा httpx का डिफ़ॉल्ट है, यानी हर रिक्वेस्ट के लिए 5 सेकंड, हर हल के लिए नहीं, इसलिए 12 सेकंड का हल फिर भी काम कर गया और 14.1 सेकंड पर पूरा हुआ। चुप पड़े submit पर 5.2 सेकंड में खाली संदेश वाला NetworkException आया। चुप पड़ा poll, 20 की सीमा के साथ, 24.7 सेकंड पर TimeoutException के साथ खत्म हुआ। Python में concurrency के बारे में और जानने के लिए देखें: asyncio के साथ पैरेलल में कैप्चा हल करना.

ऐसे unit tests जिन्हें सॉल्वर की ज़रूरत नहीं

आपके ज़्यादातर टेस्ट को CapSkip की ज़रूरत नहीं होनी चाहिए। आपके अपने कोड के जो टेस्ट सॉल्वर को कॉल करते हैं, वे fake इस्तेमाल कर सकते हैं। उसका नतीजा असली वाले से बिल्कुल मेल खाना चाहिए। SDK captchaId और code वाला एक सादा dict लौटाता है, इसलिए fake भी वही करता है। आख़िरी टेस्ट दूसरा तरीका दिखाता है: असली क्लाइंट बनाएँ, जो हल करने तक कोई कनेक्शन नहीं खोलता, और monkeypatch से एक मेथड बदल दें। चारों CapSkip बंद होने पर भी पास होते हैं। ब्लॉक को test_unit.py के रूप में सेव करें।

# test_unit.py: runs with CapSkip closed
import pytest
from capskip import ApiException, CapSkip

from tokens import fresh_token

URL = "https://example.com/login"


class FakeSolver:
    """Same call and same result shape as CapSkip.recaptcha()."""

    def __init__(self, *outcomes):
        self.outcomes = list(outcomes)
        self.calls = 0

    def recaptcha(self, sitekey, url, **kwargs):
        self.calls += 1
        outcome = self.outcomes.pop(0)
        if isinstance(outcome, Exception):
            raise outcome
        return {"captchaId": "1001", "code": outcome}


def test_returns_the_code():
    assert fresh_token(FakeSolver("tok-1"), "KEY", URL) == "tok-1"


def test_retries_once_when_unsolvable():
    fake = FakeSolver(ApiException("ERROR_CAPTCHA_UNSOLVABLE"), "tok-2")
    assert fresh_token(fake, "KEY", URL) == "tok-2"
    assert fake.calls == 2


def test_does_not_retry_a_bad_sitekey():
    fake = FakeSolver(ApiException("ERROR_GOOGLEKEY"), "never-used")
    with pytest.raises(ApiException):
        fresh_token(fake, "KEY", URL)
    assert fake.calls == 1


def test_real_client_with_monkeypatch(monkeypatch):
    solver = CapSkip()  # opens no connection until a solve is called
    fake = {"captchaId": "1", "code": "tok-3"}
    monkeypatch.setattr(solver, "recaptcha", lambda **kw: fake)
    assert fresh_token(solver, "KEY", URL) == "tok-3"

टेस्ट को दूसरी मशीन पर चल रहे CapSkip के सामने चलाएँ

टेस्ट और सॉल्वर का एक ही PC पर होना ज़रूरी नहीं। कनेक्शन सेटिंग्स में CapSkip को Server मोड पर कर दें, तो वह आपके नेटवर्क पते या पब्लिक IP पर सुनता है। भरोसेमंद रिमोट एक्सेस के लिए स्टैटिक पब्लिक IP की सलाह दी जाती है। फिर तीन वेरिएबल सेट करें और वही suite बिना बदलाव के चलता है। सॉल्वर अब भी आपकी अपनी Windows मशीन है, और हल अब भी बिना मीटर वाले हैं।

# CapSkip on another Windows machine in Server mode: set these first
$env:CAPSKIP_HOST = "192.168.1.50"
$env:CAPSKIP_PORT = "8080"
$env:CAPSKIP_API_KEY = "your-key"   # only if API key validation is on
.venv\Scripts\python -m pytest -m captcha -rs

दो बातों का ध्यान रखें। SDK, CapSkip से सादे HTTP पर बात करता है, इसलिए API key clear text में जाती है। उस लिंक को निजी नेटवर्क या VPN पर रखें, और API key validation चालू करें। और Server मोड शुरू होते ही CapSkip अपना पोर्ट हर पते के लिए खोल देता है। पहुँच को सिर्फ़ अपेक्षित मशीनों तक सीमित रखने के लिए, बाकी पतों के लिए Windows Firewall में एक block नियम जोड़ें। इस गाइड के लिए रिमोट सेटअप चलाया नहीं गया। सिर्फ़ वेरिएबल पढ़े गए।

पूरी conftest.py

यह रही conftest.py, चरण 2 और 4 से, साथ में timeout patch और async fixture, एक ही फ़ाइल में। यह उन ब्लॉकों की जगह लेती है जो आपने एक-एक करके जोड़े थे, इसलिए पहले उन्हें हटा दें।

# conftest.py, complete: pip install pytest capskip==1.3.0 requests
import functools
import os
import socket
import types

import capskip.api
import pytest
import requests
from capskip import AsyncCapSkip, CapSkip

from tokens import fresh_token

# The SDK reads no environment variables, so this file reads them.
HOST = os.environ.get("CAPSKIP_HOST", "127.0.0.1")
PORT = int(os.environ.get("CAPSKIP_PORT", "8080"))
API_KEY = os.environ.get("CAPSKIP_API_KEY", "capskip")

# capskip 1.3.0 sets no HTTP timeout. Give every SDK call 5 s to connect
# and 30 s to answer. Pin capskip==1.3.0: this swaps a name inside it.
if hasattr(capskip.api, "requests"):
    capskip.api.requests = types.SimpleNamespace(
        RequestException=requests.RequestException,
        post=functools.partial(requests.post, timeout=(5, 30)),
        get=functools.partial(requests.get, timeout=(5, 30)),
    )


def pytest_configure(config):
    # Register the marker so --strict-markers accepts it.
    config.addinivalue_line(
        "markers", "captcha: needs a running CapSkip and solves a real CAPTCHA"
    )


@pytest.fixture(scope="session")
def capskip_client():
    # No ping endpoint, so a plain TCP connect. Use the IP, not "localhost".
    try:
        socket.create_connection((HOST, PORT), timeout=3).close()
    except OSError as exc:
        pytest.skip(f"CapSkip is not running on {HOST}:{PORT}: {exc}")
    return CapSkip(apiKey=API_KEY, host=HOST, port=PORT, recaptchaTimeout=90)


@pytest.fixture
def solve_recaptcha(capskip_client):
    # A factory: every call is a new solve, and a token works once.
    def solve(sitekey, url):
        return fresh_token(capskip_client, sitekey, url)

    return solve


@pytest.fixture(scope="session")
def async_capskip_client(capskip_client):
    return AsyncCapSkip(apiKey=API_KEY, host=HOST, port=PORT, recaptchaTimeout=90)

pytest कैप्चा टेस्ट की आम errors

FAQ

संक्षेप में

pytest कैप्चा suite को हर सेशन के लिए एक CapSkip क्लाइंट और हर टेस्ट के अंदर एक नया हल चाहिए। CapSkip बंद हो तो स्किप करें, live टेस्ट पर marker लगाएँ, और CapSkip के 10 threads के नीचे worker की स्पष्ट संख्या रखें। Windows पर SDK की अपनी सीमाएँ pytest-timeout की सीमा से काफ़ी नीचे रखें, ताकि अटका हुआ हल रन को खत्म करने के बजाय एक टेस्ट को फेल करे।

• कैप्चा के वे सभी टाइप जिन्हें Python पैकेज संभालता है: Python कैप्चा सॉल्वर पेज.
• सादी Python script में checkbox वाला reCAPTCHA: Python में reCAPTCHA v2 हल करना.
• checkbox वाला सॉल्वर कैसे काम करता है: reCAPTCHA v2 सॉल्वर पेज.

हर रात चलने वाला suite कुछ सौ बार हल करता है, और यह गिनती जल्दी ही बहुत बढ़ जाती है। अपनी खुद की Windows मशीन पर चल रहे कैप्चा सॉल्वर से इसे चलाने का मतलब है कि ये हल कभी बिल में नहीं बदलते।