Python (pytest, Selenium, Playwright)

英語から自動翻訳されています。

ヘルパーを追加する

このヘルパーをテストフォルダーにコピーします。標準ライブラリだけを使うのでインストールは不要で、API キーを PROXYMAIL_QA_API_KEY から読み取ります：

"""Disposable inboxes for Python tests (pytest with Selenium or Playwright):
create a test address, sign up with it in your app, then wait for the code or
link from the email it gets. Guide: https://proxymail.qa/docs/python

The API key comes from the test run's environment (PROXYMAIL_QA_API_KEY).
When PROXYMAIL_QA_RUN is set (start_run, or the GitHub Action), every call
adds what it saw to that run. Standard library only; Python 3.9 or later.
"""

import contextlib
import json
import math
import os
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone

BASE_URL = os.environ.get("PROXYMAIL_QA_API_URL", "https://api.proxymail.qa/v1")
# Sent with every call. Some networks and firewalls turn away Python's
# default user agent (Python-urllib), so the helper names itself.
USER_AGENT = "proxymail-qa-python/1"


class ProxyMailQAError(Exception):
    """An API error. When a wait finds nothing, hint says why (an email
    provider in sandbox mode, say), and the message includes it."""

    def __init__(self, status, message, hint=None):
        super().__init__(f"{message} {hint}" if hint else message)
        self.status = status
        self.hint = hint


def _request(method, path, body=None, retry=False):
    key = os.environ.get("PROXYMAIL_QA_API_KEY")
    if not key:
        raise RuntimeError("Set PROXYMAIL_QA_API_KEY for the test run.")
    headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json", "User-Agent": USER_AGENT}
    run = os.environ.get("PROXYMAIL_QA_RUN")
    if run:
        headers["Proxy-Mail-QA-Run"] = run
        # A wait that will be asked again: its "nothing arrived" isn't evidence yet.
        if retry:
            headers["Proxy-Mail-QA-Run-Retry"] = "true"
        # The test this call belongs to (set_test, or the pytest hooks below).
        if _test["name"]:
            headers["Proxy-Mail-QA-Test"] = urllib.parse.quote(_test["name"], safe="")
    request = urllib.request.Request(
        BASE_URL + path,
        method=method,
        data=None if body is None else json.dumps(body).encode(),
        headers=headers,
    )
    try:
        with urllib.request.urlopen(request, timeout=40) as response:
            raw = response.read()
            if response.headers.get("Content-Type", "").startswith("text/markdown"):
                return raw.decode()
            return json.loads(raw) if raw else None
    except urllib.error.HTTPError as error:
        try:
            payload = json.loads(error.read() or b"{}")
        except ValueError:
            # Not the API's answer: a wrong address, or a proxy or firewall on the way.
            payload = {"message": f"HTTP {error.code} from {BASE_URL}, but not from the Proxy Mail QA API. Check PROXYMAIL_QA_API_URL, and whether a proxy or firewall turned the call away."}
        raise ProxyMailQAError(error.code, payload.get("message", str(error)), payload.get("hint")) from None


def create_address(**options):
    """A new test address with a persona and a password for the sign-up form:
    {"address": ..., "password": ..., "persona": {...}}. Options include
    label, lifetime_days, domain, age ("adult", "under13", ...), country, and
    where it's used: site (a host or page URL) and environment ("staging")."""
    return _request("POST", "/addresses", {"persona": True, "password": True, **options})


def form_values(fields, address=None, **options):
    """What to type or pick in each field of a sign-up form. fields lists what
    the page says about each one: {"id", "type", "name", "label",
    "placeholder", "autocomplete", "pattern", "min_length", "max_length",
    "options": [{"value", "label"}]}. Returns {"address", "password",
    "values": [{"id", "kind", "value", "note"}, ...]}, from a new address made
    for the form (options as for create_address, such as age) or the one
    given. You fill in the form."""
    body = {"fields": fields, **options}
    if address:
        body["address"] = address
    return _request("POST", "/form-values", body)


def find_addresses(tags=None, country=None, min_age=None, max_age=None, has_password=None, search=None, limit=None, site=None, environment=None):
    """Test accounts made earlier, newest first, where every filter given
    matches: find_addresses(tags=["age-gate"], country="GB", max_age=17)
    finds under-18s in GB. Ages are whole years today, from the persona's
    date of birth; search is part of the address, label, notes, name or
    username; site and environment are where it's used."""
    query = [("tag", tag) for tag in tags or []]
    filters = (("country", country), ("min_age", min_age), ("max_age", max_age), ("search", search), ("limit", limit), ("site", site), ("environment", environment))
    for name, value in filters:
        if value is not None:
            query.append((name, value))
    if has_password is not None:
        query.append(("has_password", "true" if has_password else "false"))
    return _request("GET", "/addresses" + ("?" + urllib.parse.urlencode(query) if query else ""))


def check_out_account(tags=None, country=None, min_age=None, max_age=None, has_password=None, search=None, minutes=None, site=None, environment=None):
    """An account from the pool: one matching every filter given that nobody
    else has checked out (the least recently used first), held for `minutes`
    (default 10). Raises ProxyMailQAError 409 when every match is checked
    out, 404 when none match. Tag accounts with their state (verified,
    premium, ...) to make pools."""
    filters = {"tags": tags, "country": country, "min_age": min_age, "max_age": max_age, "has_password": has_password, "search": search, "minutes": minutes}
    filters.update(site=site, environment=environment)
    return _request("POST", "/addresses/checkout", {name: value for name, value in filters.items() if value is not None})


def sign_in_account(site=None, environment=None, tags=None):
    """The saved test account to sign in with, and its password, in one call:
    one with a saved password and every tag given, used on this site first,
    then the one that got mail most recently. None when none fits. Needs a
    key created with "Can read saved passwords"; every reveal is logged."""
    body = {name: value for name, value in (("site", site), ("environment", environment), ("tags", tags)) if value is not None}
    try:
        return _request("POST", "/addresses/sign-in", body)
    except ProxyMailQAError as error:
        if error.status == 404:
            return None
        raise


def mark_dev_host(host):
    """Marks a host ("localhost:3000") as a dev environment of the project:
    links to it then pass check_email's local-links check."""
    return _request("POST", "/project/dev-hosts", {"host": host})["dev_hosts"]


def update_project(name=None, website=None, environments=None, email_prefix=None, mail_domain=None):
    """Changes the project, as in Project settings: its name, website,
    environments, email prefix or default mail domain. Only the arguments given
    change; website="" or email_prefix="" removes one. Needs the project's lead
    (or an organisation owner or admin), and is in the project's activity log."""
    fields = {"name": name, "website": website, "environments": environments, "email_prefix": email_prefix, "mail_domain": mail_domain}
    return _request("PATCH", "/project", {k: v for k, v in fields.items() if v is not None})


def extend_checkout(account, minutes=None):
    """Holds a checked-out account longer: `minutes` from now (default 10,
    at most 60)."""
    body = {"token": account["checkout"]["token"]}
    if minutes is not None:
        body["minutes"] = minutes
    return _request("POST", f"/addresses/{urllib.parse.quote(account['address'])}/extend", body)


def return_account(account, add_tags=None, remove_tags=None):
    """Frees a checked-out account for the next test, recording the state
    it was left in (tags to add and remove)."""
    body = {"token": account["checkout"]["token"], "add_tags": add_tags, "remove_tags": remove_tags}
    return _request("POST", f"/addresses/{urllib.parse.quote(account['address'])}/return", body)


@contextlib.contextmanager
def checked_out_account(create=None, check=None, reset=None, new_account=None, attempts=3, **filters):
    """with checked_out_account(tags=["premium"]) as account: ... checks an
    account out and always returns it.

    Hooks make it an account in the state the test needs. Your test code runs
    them, with your own access to the app; Proxy Mail QA never signs in to it.
    create(account) makes a new one when none is free: sign up with
    account["address"] and account["password"], verify it and set its state
    (new_account holds create_address options such as age or label).
    check(account) confirms the state: return False or raise when it isn't,
    and it's reset and checked again, or another account is tried; one that
    still fails loses the filter's tags and is tagged check-failed.
    reset(account) restores the state afterwards; if it fails, the account
    loses the filter's tags and is tagged needs-reset."""
    account = _prepare_account(create, check, reset, new_account or {}, attempts, filters)
    try:
        yield account
    except BaseException:
        with contextlib.suppress(Exception):
            _restore_account(account, reset, filters)
        raise
    _restore_account(account, reset, filters)


def _prepare_account(create, check, reset, new_account, attempts, filters):
    tried, last = [], ""
    for _ in range(attempts):
        try:
            account = check_out_account(**filters)
        except ProxyMailQAError as error:
            if error.status not in (404, 409):
                raise
            # None matches, or every match is busy: make one, if the test can.
            if create:
                return _new_account(create, check, new_account, filters)
            if tried:  # accounts that failed their check have left the pool
                break
            raise
        if account["address"] in tried:  # every match has been tried
            with contextlib.suppress(Exception):
                return_account(account)
            break
        tried.append(account["address"])
        problem = _failed_check(check, account)
        if problem is None:
            return account
        if reset:
            try:
                reset(account)
                problem = _failed_check(check, account)
            except Exception as error:
                problem = f"reset failed: {error}"
            if problem is None:
                return account
        last = problem
        _note(f"Test account failed its check ({problem}); it loses its tags and is tagged check-failed.", account["address"])
        with contextlib.suppress(Exception):
            return_account(account, add_tags=["check-failed"], remove_tags=filters.get("tags"))
    if create:
        return _new_account(create, check, new_account, filters)
    tried_text = f": tried {', '.join(tried)}" + (f" (last: {last})" if last else "") if tried else ""
    raise RuntimeError(f"No test account passed the check{tried_text}. Pass create to make one.")


def _new_account(create, check, new_account, filters):
    """Makes an account with the create hook, held from the start so no
    parallel test takes it half made, then tags and checks it."""
    tags = [tag.lower() for tag in filters.get("tags") or []]
    min_age, max_age = filters.get("min_age"), filters.get("max_age")
    if (min_age is not None or max_age is not None) and not (new_account.get("age") or new_account.get("date_of_birth")):
        raise ValueError("Give new_account an age or date_of_birth, so new accounts match min_age and max_age.")
    options = {**new_account}
    if filters.get("country") and "country" not in options:
        options["country"] = filters["country"]
    created = create_address(**options)
    held = check_out_account(search=created["address"], minutes=filters.get("minutes"))
    if held["address"] != created["address"]:
        with contextlib.suppress(Exception):
            return_account(held)
        raise RuntimeError(f"Couldn't hold the new account {created['address']}.")
    try:
        create({**held, "password": created["password"]})
        tagged = _set_tags(created["address"], sorted(set(held["tags"]) | set(tags)))
        account = {**tagged, "checkout": held["checkout"]}
        age = _age((account.get("persona") or {}).get("date_of_birth"))
        if age is not None and ((min_age is not None and age < min_age) or (max_age is not None and age > max_age)):
            raise ValueError(f"The new account is {age}, outside min_age and max_age: give new_account a matching age.")
        problem = _failed_check(check, account)
        if problem is not None:
            raise RuntimeError(f"The new account {created['address']} failed its check after create: {problem}")
        _note("Made a new test account" + (f" tagged {', '.join(tags)}." if tags else "."), created["address"])
        return account
    except BaseException:
        # Untagged, it stays out of the pool; it expires with its address.
        with contextlib.suppress(Exception):
            _set_tags(created["address"], held["tags"])
        with contextlib.suppress(Exception):
            return_account(held)
        raise


def _restore_account(account, reset, filters):
    """Runs reset, then returns the account. If reset fails, the account
    leaves the pool's state (needs-reset) and the error is raised."""
    if reset:
        try:
            reset(account)
        except Exception as error:
            _note(f"Couldn't reset the test account ({error}); it loses its tags and is tagged needs-reset.", account["address"])
            with contextlib.suppress(Exception):
                return_account(account, add_tags=["needs-reset"], remove_tags=filters.get("tags"))
            raise
    with contextlib.suppress(Exception):
        return_account(account)


def _failed_check(check, account):
    """None when the account passes (or there's no check), or why it doesn't."""
    if check is None:
        return None
    try:
        return "check returned False" if check(account) is False else None
    except Exception as error:  # such as an AssertionError
        return str(error) or type(error).__name__


def _set_tags(address, tags):
    return _request("PATCH", f"/addresses/{urllib.parse.quote(address)}", {"tags": tags})


def _age(date_of_birth):
    """Whole years from a YYYY-MM-DD date of birth, or None without one."""
    if not date_of_birth:
        return None
    born, today = datetime.strptime(date_of_birth, "%Y-%m-%d").date(), datetime.now(timezone.utc).date()
    return today.year - born.year - ((today.month, today.day) < (born.month, born.day))


def _note(text, address):
    """Adds a line to the current run, if there is one, without failing the test."""
    if os.environ.get("PROXYMAIL_QA_RUN"):
        with contextlib.suppress(Exception):
            add_run_note(text, address=address)


def _wait(kind, address, since, timeout, extra):
    deadline = time.monotonic() + timeout
    since = since.astimezone(timezone.utc).isoformat()
    while True:
        left = max(0, min(30, int(deadline - time.monotonic())))
        query = urllib.parse.urlencode({"since": since, "timeout": left, **extra})
        more = time.monotonic() + left < deadline - 0.5
        try:
            return _request("GET", f"/addresses/{urllib.parse.quote(address)}/{kind}?{query}", retry=more)
        except ProxyMailQAError as error:
            # Only a wait that found nothing is repeated, until the timeout.
            if error.status != 404 or not str(error).startswith(f"No {kind} arrived") or time.monotonic() >= deadline - 0.5:
                raise


def wait_for_code(address, since, timeout=30, unused=False, type=None):
    """The code from the newest email received since `since` (a datetime),
    waiting up to `timeout` seconds. unused=True skips codes already used
    (returned by an earlier wait, or copied or filled in the app or Proxy
    Mail QA for Chrome). type="coupon" waits for a coupon or promo code
    instead, "any" for either."""
    extra = {"unused": "true"} if unused else {}
    if type and type != "otp":
        extra["type"] = type
    return _wait("code", address, since, timeout, extra)["value"]


def wait_for_coupon(address, since, timeout=300, unused=False):
    """The coupon or promo code from the newest offer email received since
    `since`, waiting up to `timeout` seconds (offers often come minutes
    later): {"code", "discount", "valid_until", "after_seconds"}, where
    discount is our "20% off" and valid_until a date, when the email states
    them."""
    match = _wait("code", address, since, timeout, {"type": "coupon", **({"unused": "true"} if unused else {})})
    return {"code": match["value"], "discount": match.get("discount"), "valid_until": match.get("valid_until"), "after_seconds": match.get("after_seconds")}


def wait_for_link(address, since, type="verify", app=None, timeout=30, unused=False):
    """The link from the newest email received since `since`: verify,
    magic_signin, invite, reset or any; or unsubscribe (the email's own
    unsubscribe link) or list_unsubscribe (the List-Unsubscribe header's).
    app=True asks for the link that opens a mobile app (myapp://...) when the
    email also has a web one; unused=True skips links already used."""
    extra = {"type": type, **({} if app is None else {"app": "true" if app else "false"}), **({"unused": "true"} if unused else {})}
    return _wait("link", address, since, timeout, extra)["value"]


def _count_query(since, type, subtype, service, category=None, **extra):
    query = {key: value for key, value in extra.items()}
    if category:
        query["category"] = category
    if since is not None:
        query["since"] = since if isinstance(since, str) else since.astimezone(timezone.utc).isoformat()
    if service:
        query["service"] = service
    if type:
        query["type"] = type
    if subtype:
        query["subtype"] = subtype if isinstance(subtype, str) else ",".join(subtype)
    return "?" + urllib.parse.urlencode(query) if query else ""


def count_emails(address, since=None, type=None, subtype=None, service=None, category=None):
    """How many emails reached the address, each summarised without its
    content: {"count": ..., "emails": [{"received_at", "service", "type",
    "subtype", "category", "gap_seconds", "attachments"}, ...]}. type is
    "code" or "link"; category is "auth", "transactional" or "marketing";
    without since, every email the address has received. Count after waiting
    for the first email, so a duplicate has had time to arrive."""
    query = _count_query(since, type, subtype, service, category)
    return _request("GET", f"/addresses/{urllib.parse.quote(address)}/count{query}")


def check_email(
    address, since=None, type=None, subtype=None, service=None, email_id=None, expect_host=None, expect_text=None,
    expect_language=None, expect_within=None, category=None, expect_attachment=None,
):
    """Checks one email the address received (the newest that matches, or
    email_id): SPF, DKIM and DMARC, local and plain http links, where its link
    goes (expect_host), a stated expiry, a plain-text part, size, image-only
    emails, phrases it should contain (expect_text, a list) and its language
    (expect_language, such as "de"), and how soon after since it arrived
    (expect_within, such as "5m"), and the attachments it should have
    (expect_attachment, a list: "application/pdf", ".ics", "invoice-*.pdf").
    Returns {"passed", "failed", "warnings",
    "checks": [{"check", "status", "detail"}, ...]}, never the email's content.
    Wait for the email first."""
    query = _count_query(since, type, subtype, service, category)
    extra = [(name, value) for name, value in (("email_id", email_id), ("expect_host", expect_host), ("expect_language", expect_language), ("expect_within", expect_within)) if value]
    extra += [("expect_text", phrase) for phrase in expect_text or []]
    extra += [("expect_attachment", pattern) for pattern in expect_attachment or []]
    if extra:
        query += ("&" if query else "?") + urllib.parse.urlencode(extra)
    return _request("GET", f"/addresses/{urllib.parse.quote(address)}/checks{query}")


def expect_no_email(address, since=None, timeout=30, type=None, subtype=None, service=None, category=None):
    """Returns when no matching email arrives within `timeout` seconds, for
    checks such as "an unsubscribed user gets nothing" (category="marketing",
    so receipts can still arrive); raises
    ProxyMailQAError (status 409) as soon as one does. Pass since as the time
    of the action, so an email that came before the call still counts;
    without it, from now."""
    deadline = time.monotonic() + timeout
    while True:
        left = max(0, min(30, math.ceil(deadline - time.monotonic())))
        query = _count_query(since, type, subtype, service, category, timeout=left)
        more = time.monotonic() + left < deadline - 0.5
        result = _request("GET", f"/addresses/{urllib.parse.quote(address)}/none{query}", retry=more)
        # The server's since, so later requests watch from the same moment.
        since = result["since"]
        if time.monotonic() >= deadline - 0.5:
            return result


def mark_unsubscribed(address, email_id=None, service=None, at=None, grace=None):
    """Records that the address unsubscribed from a sender, after the test did
    it in its own browser (Proxy Mail QA never opens unsubscribe links). Any
    later marketing email from that sender fails after_unsubscribe, however
    long after. Name the sender with email_id or service (a name or domain);
    with neither, the newest marketing email decides. at backdates it, and
    grace ("2h", "1d") lets marketing already on its way through."""
    if at is not None and not isinstance(at, str):
        at = at.astimezone(timezone.utc).isoformat()
    body = {"email_id": email_id, "service": service, "at": at, "grace": grace}
    return _request("POST", f"/addresses/{urllib.parse.quote(address)}/unsubscribed", {k: v for k, v in body.items() if v is not None})


def mark_resubscribed(address, service=None, email_id=None):
    """The address subscribed again: ends the watch for that sender (the only
    one, when there's one)."""
    query = urllib.parse.urlencode({k: v for k, v in {"service": service, "email_id": email_id}.items() if v is not None})
    return _request("DELETE", f"/addresses/{urllib.parse.quote(address)}/unsubscribed" + (f"?{query}" if query else ""))


def check_unsubscribe(address):
    """Each unsubscribe and the marketing emails from that sender that came
    after it, by id and how long after: {"passed", "watches": [...]}."""
    return _request("GET", f"/addresses/{urllib.parse.quote(address)}/unsubscribed")


def offers(address):
    """Each sender's coupons and which version of the address each went to
    (plain, capitals or +tag). Mail to all of them reaches one inbox, so
    passed is False when an offer reached two or more: one person got it
    twice."""
    return _request("GET", f"/addresses/{urllib.parse.quote(address)}/offers")


def check_sequence(address, steps, since=None):
    """Whether a welcome series or drip campaign came in order and on time:
    each step is a dict with any of subject (a phrase), category, type ("code"
    or "link"), after ("1d") and within ("2h"). Each answers pass, early,
    late, missing or waiting; "done" is False while one is still waiting."""
    if since is not None and not isinstance(since, str):
        since = since.astimezone(timezone.utc).isoformat()
    body = {"steps": steps, **({"since": since} if since else {})}
    return _request("POST", f"/addresses/{urllib.parse.quote(address)}/sequence", body)


def record_coupon_result(address, result, code=None, email_id=None, message=None, one_time=False, discount_applied=None):
    """After applying a coupon at checkout (on staging with test payments,
    never real card details), records what the app did: "accepted",
    "rejected" or "error", with its message. The answer's "check" is the
    coupon_result verdict: it fails a coupon accepted after its end date, for
    an offer already given to another version of the address, or twice when
    one_time."""
    body = {"result": result, "code": code, "email_id": email_id, "message": message, "one_time": one_time, "discount_applied": discount_applied}
    return _request("POST", f"/addresses/{urllib.parse.quote(address)}/coupon-result", {k: v for k, v in body.items() if v is not None})


_KEEP = object()


def record_consent(address, marketing=_KEEP, prechecked=_KEEP, double_opt_in=_KEEP):
    """Records what the address agreed to at sign-up: marketing ("opted_in"
    or "declined"), prechecked (the box was ticked when the form loaded) and
    double_opt_in ("confirmed" or "pending"). Only what's given changes.
    check_email then judges marketing email by the rules where the persona
    lives (a test aid, not legal advice)."""
    given = {"marketing": marketing, "prechecked": prechecked, "double_opt_in": double_opt_in}
    return _request("PUT", f"/addresses/{urllib.parse.quote(address)}/consent", {k: v for k, v in given.items() if v is not _KEEP})


def consent(address):
    """What the address agreed to, as recorded."""
    return _request("GET", f"/addresses/{urllib.parse.quote(address)}/consent")


def start_run(name, source=None, meta=None):
    """Starts a run: later calls (from this process and any it starts) add
    what they saw, for a report on the pull request. Returns {"id", "url",
    ...}. In GitHub Actions, the Proxy Mail QA action does this for you."""
    run = _request("POST", "/runs", {"name": name, "meta": meta or {}, **({"source": source} if source else {})})
    os.environ["PROXYMAIL_QA_RUN"] = run["id"]
    return run


def add_run_note(text, failed=False, address=None):
    """A line of your own in the run; failed=True fails it."""
    _request("POST", f"/runs/{os.environ['PROXYMAIL_QA_RUN']}/notes", {"text": text, "failed": failed, "address": address})


def finish_run(failed=False):
    """Finishes the run: passed unless an email was missing or came when none
    should have, an email check, a test, or an unsubscribe, offer or coupon
    check failed, a note failed it, or failed=True says the tests failed."""
    return _request("POST", f"/runs/{os.environ.pop('PROXYMAIL_QA_RUN')}/finish", {"failed": failed})


def set_test(name):
    """Names the test the following calls belong to (None for none), so the
    run's report says which test each email came from. The pytest hooks at
    the end of this file do it for you."""
    _test["name"] = " ".join(str(name).split())[:200] if name else None


def record_test(name, status, duration_ms=None, error=None):
    """A test's result in the run: status is "passed", "failed", "flaky" or
    "skipped", and a failed test fails the run."""
    body = {"name": name, "status": status, "duration_ms": duration_ms, "error": error}
    _request("POST", f"/runs/{os.environ['PROXYMAIL_QA_RUN']}/tests", {key: value for key, value in body.items() if value is not None})


def record_test_totals(passed, failed, skipped=0, flaky=0):
    """How the suite went, for the top of the run's report."""
    totals = {"passed": passed, "failed": failed, "skipped": skipped, "flaky": flaky}
    _request("POST", f"/runs/{os.environ['PROXYMAIL_QA_RUN']}/tests", {"totals": totals})


def run_report(run_id=None):
    """The run's report in Markdown, for a pull request or job summary."""
    return _request("GET", f"/runs/{run_id or os.environ['PROXYMAIL_QA_RUN']}?format=markdown")


def update_address(address, **changes):
    """Changes a test account's details, such as after the app changed them:
    first_name, last_name, username, date_of_birth, country, label, notes or
    tags. None clears one."""
    return _request("PATCH", f"/addresses/{urllib.parse.quote(address)}", changes)


def set_password(address, password=None, label="password"):
    """Saves the password a test account uses now, after the app changed or
    reset it. Without one, a strong password is generated and returned once,
    as "password"."""
    body = {"label": label}
    if password is not None:
        body["password"] = password
    return _request("PUT", f"/addresses/{urllib.parse.quote(address)}/password", body)


def variants(address):
    """Versions of the address for testing a sign-up form: "delivered" ones
    (capitals, plus tags with special characters) whose mail reaches its
    inbox, "undeliverable" ones, and "invalid" ones it should reject."""
    return _request("GET", f"/addresses/{urllib.parse.quote(address)}/variants")


def name_variants(field=None, max_length=None):
    """Awkward names for a form's name field ("first_name", "last_name",
    "full_name" or "display_name"): "accept" (apostrophes, accents, other
    scripts, one letter, exactly max_length), "reject" (over max_length,
    empty, only spaces), "store_as_text" (HTML or SQL the app must store as
    typed, never run) and "decide" (digits, symbols, emoji, spaces around,
    very long ones: note what the app did)."""
    query = {k: v for k, v in {"field": field, "max_length": max_length}.items() if v is not None}
    return _request("GET", "/variants/names" + ("?" + urllib.parse.urlencode(query) if query else ""))


def password_variants(min_length=None, max_length=None, symbols=None):
    """Awkward passwords within the field's rules: "accept" (the minimum and
    maximum, each symbol class, spaces, non-ASCII), "reject" (one under or
    over, a symbol where none are allowed) and "decide" (only digits, only
    lower case, a space at the start or end, emoji, composed and decomposed
    accents, bcrypt's 72 bytes). Fresh each call."""
    query = {k: v for k, v in {"min_length": min_length, "max_length": max_length}.items() if v is not None}
    if symbols is not None:
        query["symbols"] = "true" if symbols else "false"
    return _request("GET", "/variants/passwords" + ("?" + urllib.parse.urlencode(query) if query else ""))


def delete_address(address):
    _request("DELETE", f"/addresses/{urllib.parse.quote(address)}")


def now():
    """The time to pass as `since`: call it just before submitting the form."""
    return datetime.now(timezone.utc)


# pytest: in a run (PROXYMAIL_QA_RUN), the report says which test each email
# came from, records each failed test with the start of its error, and shows
# the totals. Turn it on by importing the hooks into your conftest.py:
#
#   from proxymail_qa import pytest_runtest_setup, pytest_runtest_teardown, pytest_runtest_logreport, pytest_sessionfinish

_test = {"name": None}
_totals = {"passed": 0, "failed": 0, "skipped": 0, "flaky": 0}


def _test_name(nodeid):
    """tests/test_signup.py::TestSignUp::test_code[firefox] reads as
    TestSignUp › test_code[firefox]."""
    return " › ".join(nodeid.split("::")[1:]) or nodeid


def pytest_runtest_setup(item):
    set_test(_test_name(item.nodeid))


def pytest_runtest_teardown(item):
    set_test(None)


def pytest_runtest_logreport(report):
    if not os.environ.get("PROXYMAIL_QA_RUN"):
        return
    if report.skipped and report.when in ("setup", "call"):
        _totals["skipped"] += 1
    elif report.failed and report.when in ("setup", "call"):
        _totals["failed"] += 1
        # pytest's one-line summary of the failure, such as "AssertionError: …".
        crash = getattr(getattr(report.longrepr, "reprcrash", None), "message", None)
        error = " ".join((crash or str(report.longrepr).strip().splitlines()[-1]).split())[:300]
        with contextlib.suppress(Exception):
            record_test(_test_name(report.nodeid), "failed", round(report.duration * 1000), error)
    elif report.passed and report.when == "call":
        _totals["passed"] += 1


def pytest_sessionfinish(session, exitstatus):
    if os.environ.get("PROXYMAIL_QA_RUN") and any(_totals.values()):
        with contextlib.suppress(Exception):
            record_test_totals(**_totals)

Selenium

Selenium を使い、コードを入力するサインアップテストです。

from selenium.webdriver.common.by import By

from proxymail_qa import create_address, delete_address, now, wait_for_code


def test_sign_up_sends_a_working_code(driver):  # your WebDriver fixture
    account = create_address()
    driver.get("https://staging.example.com/signup")
    driver.find_element(By.NAME, "email").send_keys(account["address"])
    driver.find_element(By.NAME, "password").send_keys(account["password"])

    submitted = now()
    driver.find_element(By.CSS_SELECTOR, "button[type=submit]").click()

    code = wait_for_code(account["address"], since=submitted)
    driver.find_element(By.NAME, "code").send_keys(code)
    delete_address(account["address"])

Python 向けの Playwright
Section titled “Python 向けの Playwright”
または、Python 向けの Playwright を使い、確認リンクを開きます。
since を使うと、フォーム送信前に送られたメールは待機の対象外になります。待機は最大 30 秒です。もっと長くするには timeout を渡します。何も届かない場合、ProxyMailQAError は Proxy Mail QA がわかる範囲で理由を示します。たとえば、アプリのメールプロバイダーが確認済みアドレスにしか送信しない、などです。
リンクではなくコードの場合は wait_for_code を、モバイルアプリを開くリンクの場合は wait_for_link(address, since, app=True) を使います。
test_signup.py
```
import re

from playwright.sync_api import Page, expect

from proxymail_qa import create_address, delete_address, now, wait_for_link


def test_sign_up_verifies_the_email(page: Page):
    account = create_address()
    page.goto("/signup")
    page.get_by_label("Email").fill(account["address"])
    page.get_by_label("Password").fill(account["password"])

    submitted = now()
    page.get_by_role("button", name="Sign up").click()

    page.goto(wait_for_link(account["address"], since=submitted))
    expect(page).to_have_url(re.compile("verified"))
    delete_address(account["address"])
```

サインアップフォーム用の値

form_values は、フォーム用に作った新しいアドレスをもとに、各フィールドに入力または選択する値を返します。フィールドのルールに合うユーザー名とパスワード、フィールドの形式に合った生年月日、ドロップダウンにある国などです。各フィールドをページの表記どおりに記述し、入力します：

result = form_values([
    {"id": "email", "type": "email", "label": "Email"},
    {"id": "pw", "type": "password", "label": "Password", "min_length": 12},
    {"id": "dob", "type": "text", "label": "Date of birth", "placeholder": "DD/MM/YYYY"},
])
for field in result["values"]:
    if field["value"] is not None:
        page.locator(f"#{field['id']}").fill(field["value"])

age= と country= で新しいアドレスの内容を決めます。address= は以前作ったアドレスを使います。詳しくは REST API ガイドを参照してください。

メールが届かなかったこと、または 1 回だけ届いたことの確認

何も起きないことで合格するテストもあります。配信停止したユーザーにニュースレターが届かない、1 分以内の 2 回目のリセット要求が拒否される、などです。expect_no_email はアドレスを監視し、一致するメールが届いた時点で ProxyMailQAError（ステータス 409）を発生させます。count_emails は届いたメールを数えるので、「確認メールがちょうど 1 通」を確認できます：

import time

from proxymail_qa import count_emails, expect_no_email, now, wait_for_link


def test_unsubscribed_users_get_no_newsletter(app, unsubscribed):
    sent = now()
    app.send_newsletter()
    expect_no_email(unsubscribed["address"], since=sent, timeout=20)


def test_sign_up_sends_one_verification_email(page, account):
    submitted = now()
    page.get_by_role("button", name="Sign up").click()
    wait_for_link(account["address"], since=submitted)
    time.sleep(10)  # give a duplicate time to arrive
    assert count_emails(account["address"], since=submitted, type="link")["count"] == 1

どちらもメールの内容は返しません。各メールは、届いた時刻、送信者、コードやリンクの有無で要約されます。フィルター：type（"code" または "link"）、subtype（"verify" や "reset" など）、service、category（"marketing"。領収書などは引き続き届いてかまいません）。

配信停止そのものをテストするには、テストのブラウザーで配信停止してから記録します：mark_unsubscribed(address, email_id=...)（または service="acme.com"。すでに送信中のメールには grace="1d"）。その後、その送信者からマーケティングメールが届くと after_unsubscribe は不合格になり、届いた場合 check_unsubscribe(address)["passed"] は false になります。mark_resubscribed(address) で監視を終了します。そのほかはすべて REST API ガイドにあり、キー、並列実行、後片付けについては CI での実行にあります。

メールの送信と構成の確認

check_email は、届いたメールにチェックを実行します：SPF、DKIM、DMARC、localhost や通常の http へのリンク、リンクの行き先、サインインコードやリンクの有効期限が書かれているか、プレーンテキストのパート、サイズ、画像だけのメール、そして指定した場合は、含まれるべきフレーズ（expect_text=[...]）と言語（expect_language="de"）です。返すのは判定結果だけで、メールの内容は返しません：

from proxymail_qa import check_email, now, wait_for_link


def test_verification_email_is_well_built(page, account):
    submitted = now()
    page.get_by_role("button", name="Sign up").click()
    wait_for_link(account["address"], since=submitted)
    result = check_email(account["address"], since=submitted, type="link", expect_host="staging.example.com")
    assert result["passed"], [c for c in result["checks"] if c["status"] == "fail"]

各チェックの意味は REST API ガイドにあります。自分のマシンで動くアプリをテストしていますか？mark_dev_host("localhost:3000") を 1 回実行すると、そこへのリンクは local_links で合格します。

並列テスト用のアカウントプール

アカウントに状態のタグを付け、各テストにほかのテストが使っていないものを 1 つずつ借りさせます。with checked_out_account(tags=["verified", "premium"]) as account: はアカウントを 1 つ借り出し、必ず返却します。テストで変更した内容を記録するには、check_out_account(...) の後に自分で return_account(account, add_tags=[...], remove_tags=[...]) を呼び出します。詳しくは REST API ガイドを参照してください。

アプリ内で特定の状態（確認済み、プレミアムプラン、注文なしなど）のアカウントがテストに必要なときは、3 つのフックを渡します。フックはテストコードが自身のアプリへのアクセス権で実行します。Proxy Mail QA がアプリにサインインすることはありません。ここで app は、アプリとやり取りする独自のコードを表します。

def create(account):  # when none is free: sign up, verify, upgrade
    submitted = now()
    app.sign_up(account["address"], account["password"])
    app.open(wait_for_link(account["address"], since=submitted))
    app.admin.set_plan(account["address"], "premium")


def check(account):  # still in that state? Return False or raise if not
    return app.admin.user(account["address"])["plan"] == "premium"


def reset(account):  # after the test: put it back
    app.admin.reset_user(account["address"], plan="premium", orders=[])


with checked_out_account(tags=["verified", "premium"], create=create, check=check, reset=reset) as account:
    ...

check に失敗したアカウントはリセットされ、もう一度チェックされます。それでも失敗するとプールのタグを外されて check-failed のタグが付き、次のアカウントが試されます（attempts、3 回）。create は、一致するものがないとき、またはすべて借り出し中のときに、最初からこのテスト用に確保された新しいアドレスで実行されます（new_account は age などの create_address オプションを受け取ります）。テスト後に reset が失敗すると、アカウントに needs-reset のタグが付き、プールから外れます。Node.js クライアントも同じように動作します。

実行のレポート

PROXYMAIL_QA_RUN が設定されていると、すべての呼び出しが確認した内容をその実行に追加します。アドレス、届いたメールと届くまでの速さ、コード、各リンクの行き先、チェック、届かなかったメールです。GitHub Actions では、Proxy Mail QA アクションがこれを設定し、レポートをプルリクエストに投稿します。それ以外の環境では、テストの前に start_run("Sign-up tests") を（たとえばセッションスコープのフィクスチャで）、後に finish_run(failed=...) を呼び出します。run_report() はレポートを Markdown で返します。

各メールがどのテストから来たか

pytest では、ヘルパーのフックを conftest.py にインポートします。これで実行中は各呼び出しがテスト名を伝えるため、レポートに各メールがどのテストから来たかが表示され、実行には失敗した各テストとそのエラーの冒頭、合計が記録されます：

from proxymail_qa import pytest_runtest_setup, pytest_runtest_teardown, pytest_runtest_logreport, pytest_sessionfinish

実行外では何もしません。ほかのランナーは set_test(name)、record_test(name, status, duration_ms, error)、record_test_totals(passed, failed, skipped, flaky) を自分で呼び出せます。

テストアカウントをもう一度見つける

find_addresses は、指定したフィルターがすべて一致する、以前作ったテストアカウントを新しい順に返します：tags、country、min_age、max_age（ペルソナの生年月日から計算した今日時点の満年齢）、has_password、search（アドレス、ラベル、メモ、名前、ユーザー名の一部）、site と environment（使用先。create_address(site=..., environment=...) で設定）。

from proxymail_qa import find_addresses

teen = find_addresses(tags=["age-gate"], country="GB", max_age=17, has_password=True)[0]

sign_in_account(site=..., tags=[...]) はさらに一歩進んで、サインインに使う保存済みテストアカウントをパスワードとともに返します。このサイトで使ったことのあるアカウントが優先され、合うものがなければ None を返します。「Can read saved passwords」が付いたキーが必要です。

from proxymail_qa import sign_in_account

account = sign_in_account(site="staging.example.com", tags=["verified"])
page.get_by_label("Email").fill(account["address"])
page.get_by_label("Password").fill(account["password"])

テストアカウントを最新の状態に保つ

テストでアプリ内のアカウントを変更したときは、保存済みの情報で引き続きサインインできるよう記録します。パスワードの変更やリセットの後には set_password(address, "the-new-password")（パスワードを省くと生成され、1 回だけ返されます）、新しい名前、ユーザー名、生年月日、国、またはラベル、メモ、タグには update_address(address, username="new_name") を使います。

特殊でも有効なアドレスでサインアップフォームをテストするには、variants(address) が、メールが元のアドレスの受信トレイに届くバリエーションと、フォームが拒否すべき無効なものを一覧にします（詳細）。ほかのフィールドには、name_variants(field="first_name", max_length=50) と password_variants(min_length=8, max_length=64, symbols=True) が入力用の名前とパスワードを返します。アプリがそれぞれをどう扱うべきかでグループ分けされています：accept、reject、store_as_text（決して実行してはならない HTML や SQL）、decide（詳細）。プールのアカウントを予定より長く使うテストは、extend_checkout(account, minutes=20) を呼び出せます。

クーポンとオファー

offers、record_coupon_result、check_sequence、record_consent、consent、mark_unsubscribed、mark_resubscribed、check_unsubscribe は「マーケティングメールとオファーメール」の一部です。これは Labs の機能で、キーの作成者がアプリの Account settings → Labs で有効にします。有効にするまでは、その旨を示す 403 を返します。

オファーに登録した後、wait_for_coupon はクーポンを待ち（オファーは時間がかかることが多いため、デフォルトは 5 分）、メールに記載された割引と有効期限、そして since からどれだけ後に届いたかとともにコードを返します。

from proxymail_qa import check_email, now, wait_for_coupon

signed_up = now()
page.get_by_role("button", name="Get 20% off").click()
offer = wait_for_coupon(account["address"], since=signed_up)
assert offer["discount"] == "20% off"
assert check_email(account["address"], since=signed_up, type="code", subtype="coupon", expect_within="5m")["passed"]

ウェルカムオファーが 1 人に 1 回だけ届くことを確認するには、同じアドレスの別バージョン（variants(address)["delivered"] が name+promo1@… などを一覧にします）で再登録します。オファーが複数のバージョンに届いた場合、offers(address)["passed"] は False になります。

チェックアウトでクーポンを適用した後（ステージング環境でテスト決済を使用）、record_coupon_result(address, "accepted", message=...) がアプリの動作を記録します。その ["check"] は coupon_result の判定で、終了日以降に受け付けられたクーポン、再利用されたオファー、one_time=True で 2 回受け付けられたクーポンは失敗になります。

ウェルカムシリーズやドリップキャンペーンでは、check_sequence(address, [{"subject": "Welcome"}, {"subject": "tips", "after": "1d"}], since=signed_up) が各メールが順番どおり、時間どおりに届いたかを示します。

国ごとのマーケティングルールについては、country を付けてアドレスを作成し、登録フォームの後に record_consent(address, marketing="declined", prechecked=False) を実行します。送るべきでなかったマーケティングメールがあると、check_email(address, category="marketing") は marketing_consent を失敗にします。consent(address) で読み戻せます。

wait_for_code(address, since, type="coupon") はコードだけを返します。

このページについて質問がありますか？

質問はどの言語でもかまいません。ドキュメントからすぐに回答を得ることも、Proxy Mail QA チームに質問を送ることもできます。チームからの返信は英語です。

ドキュメントに質問