如何在 pytest 中识别验证码，又不让整个运行卡住

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

# 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)

# 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

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

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"
    )

# 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,
    })

@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

# 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

# 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

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)),
    )

# 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"

# 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)

# 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

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)

# 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

# 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 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

# 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)