From 543652d0545c229e7872e51caa29861fe0d96f69 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Nov 2026 14:02:41 +0000 Subject: [PATCH 1/4] feat(journeys): journey file format and prescale journeys verify --- CHANGELOG.md | 13 + README.md | 39 +++ cli/src/prescale_cli/commands/journeys.py | 164 +++++++++++ cli/src/prescale_cli/journeys/__init__.py | 6 + cli/src/prescale_cli/journeys/model.py | 332 ++++++++++++++++++++++ cli/src/prescale_cli/journeys/replay.py | 133 +++++++++ cli/src/prescale_cli/main.py | 3 + cli/tests/test_journeys.py | 260 +++++++++++++++++ examples/README.md | 9 + examples/fragile-shop.py | 71 ++++- examples/journeys/browse.yaml | 12 + examples/journeys/checkout.yaml | 25 ++ 12 files changed, 1064 insertions(+), 3 deletions(-) create mode 100644 cli/src/prescale_cli/commands/journeys.py create mode 100644 cli/src/prescale_cli/journeys/__init__.py create mode 100644 cli/src/prescale_cli/journeys/model.py create mode 100644 cli/src/prescale_cli/journeys/replay.py create mode 100644 cli/tests/test_journeys.py create mode 100644 examples/journeys/browse.yaml create mode 100644 examples/journeys/checkout.yaml diff --git a/CHANGELOG.md b/CHANGELOG.md index 0240859..e2909bb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,19 @@ ## Unreleased +### Added +- **Journeys: your users' real paths, saved in your repo.** A journey file + (`prescale/journeys/.yaml`) lists the requests a user's path makes (log in, add to + cart, place an order), carrying tokens and IDs from one step to the next. + `prescale journeys` lists them; `prescale journeys verify ` replays each once, as one + user, and names the step that no longer answers as expected (the journey is stale, or the + app is broken). It exits non-zero, so it works as a CI check. + - Secrets come from environment variables (`{{env.NAME}}`); a literal password, token or + API key in a journey file is refused. + - Journeys that change data only run against local hosts or hosts you allow with + `--allow-writes`; journeys that send messages or delete data are never replayed. + - The example shop has a JSON API and two example journeys in `examples/journeys/`. + ### Fixed - **`prescale mcp` works on a fresh install again.** `pip install 'prescale[mcp]'` picked up the new mcp 2.x, which renamed the server class PreScale uses, so the MCP server wouldn't start and told diff --git a/README.md b/README.md index 178406b..552c8a1 100644 --- a/README.md +++ b/README.md @@ -94,6 +94,7 @@ From source: `git clone https://github.com/pyjeebz/PreScale.git && pip install . | [`run `](#run--what-breaks-first) | Ramp traffic; report what breaks first and at what load | | [`investigate `](#investigate--and-why) | …and *why* it breaks, plus how to fix it (`--app` to start your app first: [replica mode](#replica-mode--test-your-app-not-your-host)) | | [`audit `](#audit--hygiene-no-load) | Load-free HTTP hygiene check (compression, caching, CDN, HTTP/2) | +| [`journeys`](#journeys--your-users-real-paths) | List your journey files; `journeys verify ` checks they still replay | | [`compare`](#ci--gate-on-capacity) | Capacity diff of two runs; regression gating for CI | | [`profiles`](#frame-it-as-a-launch) | Launch scenarios for `--profile` | | [`history` / `show`](#saved-runs-and-the-json-contract) | List and re-open saved runs | @@ -166,6 +167,44 @@ Try this → Add a pooler (e.g. pgbouncer) and check the pool checkout timeout. ``` +### `journeys` — your users' real paths + +A journey is a path a user takes through your app (log in, add to cart, place an order), saved as the requests it makes. Journeys live in your repo as `prescale/journeys/.yaml`: + +```yaml +version: 1 +name: checkout +type: checkout +steps: + - name: log in + request: + method: POST + path: /api/login + json: {email: "{{unique.email}}", password: "{{env.SHOP_PASSWORD}}"} + expect: {status: 200} + extract: {token: body.token} + - name: place order + request: + method: POST + path: /api/orders + headers: {Authorization: "Bearer {{token}}"} + expect: {status: 201} +``` + +```bash +prescale journeys # list them +prescale journeys verify http://localhost:3000 # replay each once, as one user +``` + +`verify` stops at the first step that doesn't answer as expected and calls the journey **stale**: the app changed, or it's broken. It exits non-zero, so it works as a CI check. + +- `{{name}}` is a value an earlier step pulled out with `extract` (`body.` or `header.`); `{{unique.email}}` and `{{unique.id}}` are fresh each run; `{{env.NAME}}` reads an environment variable. +- **Secrets never go in the file.** A literal password, token, API key or `Authorization` value is refused; use `{{env.NAME}}`. +- Paths must start with `/`: a journey only calls the app it's run against. +- Journeys that change data (POST, PUT, PATCH, DELETE) run against local hosts, or a host you name with `--allow-writes staging.myapp.com`. Types `side_effect` (emails, SMS, invites) and `destructive` are never replayed. + +Try it on the example shop: `python examples/fragile-shop.py 8400`, then `SHOP_PASSWORD=demo prescale journeys verify http://localhost:8400 --dir examples/journeys`. + ### `audit` — hygiene, no load A fast, load-free check of the HTTP-level footguns that decide how you scale — compression, static-asset caching, CDN, HTTP version, cookies on assets. Cheap enough to run on every commit. diff --git a/cli/src/prescale_cli/commands/journeys.py b/cli/src/prescale_cli/commands/journeys.py new file mode 100644 index 0000000..626b39a --- /dev/null +++ b/cli/src/prescale_cli/commands/journeys.py @@ -0,0 +1,164 @@ +"""`prescale journeys` - list the app's journeys, and check they still replay. + +Journeys live in the app's repo as `prescale/journeys/*.yaml`. `verify` replays +each one once, as a single user: a step that fails there means the journey is +stale (the app changed) or the app is broken, never that it broke under load. +""" + +from __future__ import annotations + +import asyncio +import os +from pathlib import Path +from urllib.parse import urlparse + +import click +from rich.console import Console +from rich.markup import escape +from rich.table import Table + +from prescale_cli.commands.upload import header_option +from prescale_cli.journeys.model import DEFAULT_DIR, Journey, load_dir +from prescale_cli.journeys.replay import JourneyCheck, check + +console = Console() + +_LOCAL_HOSTS = {"localhost", "127.0.0.1", "0.0.0.0", "::1"} + +dir_option = click.option( + "--dir", "directory", type=click.Path(file_okay=False, path_type=Path), + default=DEFAULT_DIR, show_default=True, help="Where the journey files are.") + + +@click.group(invoke_without_command=True) +@dir_option +@click.pass_context +def journeys(ctx: click.Context, directory: Path) -> None: + """List the app's journeys (prescale/journeys/*.yaml). + + \b + Check they still replay against your app: + prescale journeys verify http://localhost:3000 + """ + if ctx.invoked_subcommand is not None: + return + found = _load_or_exit(directory) + table = Table(title=f"Journeys in {directory}") + for col in ("Name", "Type", "Weight", "Steps", "Changes data", "File"): + table.add_column(col) + for j in found: + table.add_row(j.name, j.type, f"{j.weight:g}", str(len(j.steps)), + "yes" if j.writes else "no", str(j.source)) + console.print(table) + + +@journeys.command() +@click.argument("url") +@dir_option +@click.option("--journey", "only", multiple=True, + help="Check only this journey (repeatable).") +@click.option("--allow-writes", "allow_writes", multiple=True, metavar="HOST", + help="Let journeys that change data (POST, PUT, PATCH, DELETE) run against " + "this non-local host, e.g. staging.myapp.com (repeatable).") +@click.option("--timeout", default=10.0, type=float, help="Per-request timeout in seconds.") +@header_option +def verify(url: str, directory: Path, only: tuple[str, ...], allow_writes: tuple[str, ...], + timeout: float, headers: dict[str, str]) -> None: + """Replay each journey once, as one user, and report any step that fails. + + Exits non-zero if a journey is stale or the app couldn't be reached. + + \b + prescale journeys verify http://localhost:3000 + prescale journeys verify https://staging.myapp.com --allow-writes staging.myapp.com + """ + parsed = urlparse(url) + if parsed.scheme not in ("http", "https") or not parsed.netloc: + console.print(f"[red]Error:[/red] '{url}' is not a URL " + "(expected e.g. http://localhost:3000).") + raise SystemExit(2) + found = _load_or_exit(directory) + if only: + missing = sorted(set(only) - {j.name for j in found}) + if missing: + console.print(f"[red]Error:[/red] no journey named {', '.join(missing)} " + f"in {directory}.") + raise SystemExit(2) + found = [j for j in found if j.name in only] + + host = (parsed.hostname or "").lower() + _refuse_unallowed_writes(found, host, {h.lower() for h in allow_writes}) + unset = sorted({name for j in found for name in j.env_names()} - set(os.environ)) + if unset: + console.print(f"[red]Error:[/red] the journeys read environment variable(s) that " + f"aren't set: {', '.join(unset)}.") + raise SystemExit(2) + + base = url if url.endswith("/") else url + "/" + results = asyncio.run(_check_all(found, base, timeout=timeout, headers=headers)) + for result in results: + _print_check(result) + + broken = [r for r in results if not r.ok] + if broken: + stale = [r.journey.name for r in broken if r.stale] + if stale: + console.print(f"\n[yellow]{len(stale)} stale:[/yellow] {', '.join(stale)}. The app " + "no longer answers the way these journeys expect. If the change was " + "on purpose, update the journey files; if not, it's a bug.") + unreached = [r.journey.name for r in broken if not r.stale] + if unreached: + console.print(f"\n[red]{len(unreached)} couldn't run:[/red] {', '.join(unreached)}.") + raise SystemExit(1) + console.print(f"\n[green]✓ All {len(results)} journeys replay.[/green]") + + +def _load_or_exit(directory: Path) -> list[Journey]: + found, errors = load_dir(directory) + for error in errors: + console.print(f"[red]Error:[/red] {escape(error)}", highlight=False) + if errors: + raise SystemExit(2) + if not found: + console.print(f"No journeys in {directory}. Add one as {directory}/.yaml.") + raise SystemExit(1) + return found + + +def _refuse_unallowed_writes(found: list[Journey], host: str, allowed: set[str]) -> None: + """Journeys that change data run on local hosts, or on hosts named with + --allow-writes. Anywhere else they could change real data.""" + if host in _LOCAL_HOSTS or host in allowed: + return + writers = [j.name for j in found if j.writes] + if writers: + console.print(f"[red]Error:[/red] {', '.join(writers)} change data (POST, PUT, PATCH " + f"or DELETE). Run them against {host} only if it's a staging or test " + f"environment, with --allow-writes {host}.") + raise SystemExit(2) + + +async def _check_all(found: list[Journey], base: str, *, timeout: float, + headers: dict[str, str]) -> list[JourneyCheck]: + return [await check(j, base, timeout=timeout, headers=headers) for j in found] + + +def _print_check(result: JourneyCheck) -> None: + journey = result.journey + mark = "[green]✓[/green]" if result.ok else ("[yellow]stale[/yellow]" if result.stale + else "[red]✗[/red]") + console.print(f"\n{mark} [bold]{journey.name}[/bold] [dim]({journey.type}, " + f"{len(journey.steps)} steps)[/dim]") + for i, step in enumerate(journey.steps, 1): + done = result.steps[i - 1] if i <= len(result.steps) else None + if done is None: + console.print(f" [dim]{i}. {step.name} {step.label} (not run)[/dim]") + elif done.ok: + console.print(f" {i}. {step.name} [dim]{step.label}[/dim] " + f"{done.status} {done.latency * 1000:.0f}ms") + else: + console.print(f" {i}. {step.name} [dim]{step.label}[/dim] " + f"[red]{escape(done.problem or '')}[/red]", highlight=False) + if done.status is not None and done.status >= 500: + console.print(" [dim]A 5xx with no load is usually a bug in the app, " + "not a stale journey.[/dim]") diff --git a/cli/src/prescale_cli/journeys/__init__.py b/cli/src/prescale_cli/journeys/__init__.py new file mode 100644 index 0000000..3dab79d --- /dev/null +++ b/cli/src/prescale_cli/journeys/__init__.py @@ -0,0 +1,6 @@ +"""Journeys: a user's path through the app (log in, search, add to cart, order), +saved as the HTTP requests it makes, so PreScale can replay it. + +A journey lives in the app's own repo as `prescale/journeys/.yaml`. +`model` loads and checks those files; `replay` runs a journey's steps. +""" diff --git a/cli/src/prescale_cli/journeys/model.py b/cli/src/prescale_cli/journeys/model.py new file mode 100644 index 0000000..1ca708b --- /dev/null +++ b/cli/src/prescale_cli/journeys/model.py @@ -0,0 +1,332 @@ +"""The journey file format (version 1), and the templating its steps use. + +A journey file is YAML, one journey per file, committed to the app's repo: + + version: 1 + name: checkout + type: checkout + steps: + - name: log in + request: {method: POST, path: /api/login, + json: {email: "{{unique.email}}", password: "{{env.SHOP_PASSWORD}}"}} + expect: {status: 200} + extract: {token: body.token} + - name: place order + request: {method: POST, path: /api/orders, + headers: {Authorization: "Bearer {{token}}"}} + expect: {status: [200, 201]} + +Placeholders: `{{name}}` is a value an earlier step extracted; `{{unique.email}}` +and `{{unique.id}}` are fresh for each run of the journey; `{{env.NAME}}` reads an +environment variable, which is how secrets stay out of the file. Everything here +is pure: loading never touches the network. +""" + +from __future__ import annotations + +import re +import secrets +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import yaml + +FORMAT_VERSION = 1 +DEFAULT_DIR = Path("prescale") / "journeys" + +# Journey types PreScale replays, and each one's default share of the traffic mix. +TYPES: dict[str, int] = { + "browse": 60, "search": 25, "account": 8, "checkout": 5, "write": 5, "auth": 2, +} +# Types PreScale never replays: they message real people or destroy data. +REFUSED_TYPES = { + "side_effect": "it sends email, SMS or invites to real people", + "destructive": "it deletes data or changes admin settings", +} +METHODS = {"GET", "HEAD", "OPTIONS", "POST", "PUT", "PATCH", "DELETE"} +READ_METHODS = {"GET", "HEAD", "OPTIONS"} + +_PLACEHOLDER = re.compile(r"\{\{\s*([A-Za-z_][\w.]*)\s*\}\}") +_SENSITIVE_KEY = re.compile(r"pass(word|wd)?|secret|token|api[-_]?key|authorization|cookie", + re.IGNORECASE) +_UNIQUE = {"email", "id"} +_SOURCES = ("body", "header") + + +class JourneyError(Exception): + """A journey file that can't be used, with a message that says how to fix it.""" + + +@dataclass +class Step: + name: str + method: str + path: str + headers: dict[str, str] = field(default_factory=dict) + json: Any = None + body: str | None = None + expect: list[int] | None = None # None: any status below 400 + extract: dict[str, str] = field(default_factory=dict) + think: float = 0.0 + + @property + def label(self) -> str: + """The route as reports show it: method plus path, placeholders kept.""" + return f"{self.method} {self.path.split('?')[0]}" + + def accepts(self, status: int) -> bool: + if self.expect is None: + return status < 400 + return status in self.expect + + +@dataclass +class Journey: + name: str + type: str + weight: float + steps: list[Step] + source: Path | None = None + + @property + def writes(self) -> bool: + """Whether replaying it can change data on the target.""" + return any(step.method not in READ_METHODS for step in self.steps) + + def env_names(self) -> set[str]: + """Environment variables its steps read.""" + return {name[4:] for step in self.steps for name in _names(_step_values(step)) + if name.startswith("env.")} + + +# --- loading -------------------------------------------------------------- + +def load_dir(directory: Path = DEFAULT_DIR) -> tuple[list[Journey], list[str]]: + """Every journey in DIRECTORY, sorted by name, plus one error message per file + that couldn't be loaded. A missing directory is no journeys, not an error.""" + if not directory.is_dir(): + return [], [] + journeys: list[Journey] = [] + errors: list[str] = [] + for path in sorted([*directory.glob("*.yaml"), *directory.glob("*.yml")]): + try: + journeys.append(load_file(path)) + except JourneyError as exc: + errors.append(f"{path}: {exc}") + seen: dict[str, Path] = {} + for journey in journeys: + if journey.name in seen: + errors.append(f"{journey.source}: the name '{journey.name}' is already used by " + f"{seen[journey.name]}") + seen.setdefault(journey.name, journey.source) + return sorted(journeys, key=lambda j: j.name), errors + + +def load_file(path: Path) -> Journey: + try: + data = yaml.safe_load(path.read_text()) + except yaml.YAMLError as exc: + raise JourneyError(f"not valid YAML ({exc})") from exc + journey = parse(data) + journey.source = path + return journey + + +def parse(data: Any) -> Journey: + """Check a decoded journey document and build a `Journey`.""" + if not isinstance(data, dict): + raise JourneyError("a journey file must be a mapping with version, name and steps") + _only(data, {"version", "name", "type", "weight", "steps"}, "the journey") + if data.get("version") != FORMAT_VERSION: + raise JourneyError(f"version must be {FORMAT_VERSION} (got {data.get('version')!r})") + name = data.get("name") + if not isinstance(name, str) or not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", name): + raise JourneyError("name must be lowercase letters, digits, - or _ (e.g. checkout)") + jtype = data.get("type", "browse") + if jtype in REFUSED_TYPES: + raise JourneyError(f"type '{jtype}' is never replayed: {REFUSED_TYPES[jtype]}") + if jtype not in TYPES: + raise JourneyError(f"type must be one of {', '.join(TYPES)} (got {jtype!r})") + weight = data.get("weight", TYPES[jtype]) + if isinstance(weight, bool) or not isinstance(weight, (int, float)) or weight <= 0: + raise JourneyError("weight must be a positive number") + raw_steps = data.get("steps") + if not isinstance(raw_steps, list) or not raw_steps: + raise JourneyError("steps must be a non-empty list") + + steps = [_parse_step(raw, i) for i, raw in enumerate(raw_steps, 1)] + _check_placeholders(steps) + return Journey(name=name, type=jtype, weight=float(weight), steps=steps) + + +def _parse_step(raw: Any, index: int) -> Step: + where = f"step {index}" + if not isinstance(raw, dict): + raise JourneyError(f"{where} must be a mapping with name and request") + _only(raw, {"name", "request", "expect", "extract", "think"}, where) + name = raw.get("name") or f"step {index}" + if not isinstance(name, str): + raise JourneyError(f"{where}: name must be text") + where = f"step {index} ({name})" + + request = raw.get("request") + if not isinstance(request, dict): + raise JourneyError(f"{where}: request must be a mapping with method and path") + _only(request, {"method", "path", "headers", "json", "body"}, f"{where} request") + method = str(request.get("method", "GET")).upper() + if method not in METHODS: + raise JourneyError(f"{where}: method {method} isn't supported") + path = request.get("path") + if not isinstance(path, str) or not path.startswith("/"): + raise JourneyError(f"{where}: path must start with / (journeys only call the app " + "they're run against, never another site)") + headers = request.get("headers") or {} + if not isinstance(headers, dict) or not all(isinstance(v, str) for v in headers.values()): + raise JourneyError(f"{where}: headers must map names to text") + if "json" in request and "body" in request: + raise JourneyError(f"{where}: give either json or body, not both") + body = request.get("body") + if body is not None and not isinstance(body, str): + raise JourneyError(f"{where}: body must be text (use json for structured data)") + _check_secrets({"headers": headers, "json": request.get("json")}, where) + + return Step( + name=name, method=method, path=path, + headers={str(k): v for k, v in headers.items()}, + json=request.get("json"), body=body, + expect=_parse_expect(raw.get("expect"), where), + extract=_parse_extract(raw.get("extract"), where), + think=_parse_seconds(raw.get("think", 0), where), + ) + + +def _parse_expect(raw: Any, where: str) -> list[int] | None: + if raw is None: + return None + if not isinstance(raw, dict) or set(raw) != {"status"}: + raise JourneyError(f"{where}: expect must be {{status: 200}} or {{status: [200, 201]}}") + status = raw["status"] + codes = status if isinstance(status, list) else [status] + if not codes or not all(isinstance(c, int) and 100 <= c <= 599 for c in codes): + raise JourneyError(f"{where}: expect status must be HTTP status codes") + return codes + + +def _parse_extract(raw: Any, where: str) -> dict[str, str]: + if raw is None: + return {} + if not isinstance(raw, dict): + raise JourneyError(f"{where}: extract must map names to body. or header.") + for name, source in raw.items(): + if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z_]\w*", name) \ + or name in {"env", "unique"}: + raise JourneyError(f"{where}: can't extract into {name!r}") + if not isinstance(source, str) or source.split(".", 1)[0] not in _SOURCES \ + or "." not in source: + raise JourneyError(f"{where}: extract {name} must come from body. " + "or header.") + return dict(raw) + + +def _parse_seconds(raw: Any, where: str) -> float: + if isinstance(raw, bool): + raise JourneyError(f"{where}: think must be seconds, e.g. 2 or 2s or 500ms") + if isinstance(raw, (int, float)) and raw >= 0: + return float(raw) + match = re.fullmatch(r"(\d+(?:\.\d+)?)(ms|s)", str(raw).strip()) + if not match: + raise JourneyError(f"{where}: think must be seconds, e.g. 2 or 2s or 500ms") + value = float(match.group(1)) + return value / 1000 if match.group(2) == "ms" else value + + +def _only(data: dict, allowed: set[str], where: str) -> None: + unknown = sorted(set(data) - allowed) + if unknown: + raise JourneyError(f"{where} has unknown field(s): {', '.join(map(str, unknown))}") + + +def _check_secrets(value: Any, where: str, key: str = "") -> None: + """Refuse a literal value under a secret-looking key: secrets come from {{env.*}} + (or an earlier step's extract), so they never end up committed in the file.""" + if isinstance(value, dict): + for k, v in value.items(): + _check_secrets(v, where, str(k)) + elif isinstance(value, list): + for v in value: + _check_secrets(v, where, key) + elif isinstance(value, str) and key and _SENSITIVE_KEY.search(key) \ + and not _PLACEHOLDER.search(value): + raise JourneyError(f"{where}: {key} has a literal value. Put secrets in an " + f"environment variable and write \"{{{{env.NAME}}}}\" instead") + + +def _check_placeholders(steps: list[Step]) -> None: + """Every {{name}} must be extracted by an earlier step, or be env./unique.""" + known: set[str] = set() + for i, step in enumerate(steps, 1): + for name in _names(_step_values(step)): + if name.startswith("env."): + continue + if name.startswith("unique."): + if name[7:] not in _UNIQUE: + raise JourneyError(f"step {i} ({step.name}): {{{{{name}}}}} isn't known; " + "use unique.email or unique.id") + continue + if name not in known: + raise JourneyError(f"step {i} ({step.name}) uses {{{{{name}}}}}, but no " + "earlier step extracts it") + known.update(step.extract) + + +def _step_values(step: Step) -> list[Any]: + return [step.path, step.headers, step.json, step.body] + + +def _names(value: Any) -> list[str]: + if isinstance(value, str): + return _PLACEHOLDER.findall(value) + if isinstance(value, dict): + return [n for k, v in value.items() for n in (*_names(k), *_names(v))] + if isinstance(value, list): + return [n for v in value for n in _names(v)] + return [] + + +# --- templating ----------------------------------------------------------- + +class TemplateError(Exception): + """A placeholder with no value at replay time (e.g. an unset env variable).""" + + +def fresh_uniques() -> dict[str, str]: + """Values for {{unique.*}}, new for each run of a journey.""" + token = secrets.token_hex(6) + # example.com is reserved for examples (RFC 2606): mail to it goes nowhere. + return {"unique.email": f"prescale+{token}@example.com", "unique.id": token} + + +def render(value: Any, variables: dict[str, Any], env: dict[str, str]) -> Any: + """Fill placeholders in a step value. A string that is exactly one placeholder + keeps the value's type (an extracted id stays a number in JSON).""" + if isinstance(value, str): + whole = _PLACEHOLDER.fullmatch(value.strip()) + if whole: + return _lookup(whole.group(1), variables, env) + return _PLACEHOLDER.sub(lambda m: str(_lookup(m.group(1), variables, env)), value) + if isinstance(value, dict): + return {render(k, variables, env): render(v, variables, env) for k, v in value.items()} + if isinstance(value, list): + return [render(v, variables, env) for v in value] + return value + + +def _lookup(name: str, variables: dict[str, Any], env: dict[str, str]) -> Any: + if name.startswith("env."): + if name[4:] not in env: + raise TemplateError(f"environment variable {name[4:]} isn't set") + return env[name[4:]] + if name not in variables: + raise TemplateError(f"{{{{{name}}}}} has no value") + return variables[name] diff --git a/cli/src/prescale_cli/journeys/replay.py b/cli/src/prescale_cli/journeys/replay.py new file mode 100644 index 0000000..b6e67ba --- /dev/null +++ b/cli/src/prescale_cli/journeys/replay.py @@ -0,0 +1,133 @@ +"""Run a journey's steps against an app. + +`check` replays a journey once, as a single user, to see whether it still works +against the app as it is now. A step that fails its expectation here, with no +load at all, means the journey is stale (the app changed), not that the app +broke under load. Each run gets its own client, so cookies and extracted values +never leak between users. +""" + +from __future__ import annotations + +import os +import time +from dataclasses import dataclass, field +from typing import Any +from urllib.parse import urljoin + +import httpx + +from prescale_cli.journeys.model import Journey, Step, TemplateError, fresh_uniques, render +from prescale_cli.loadtest import client_headers + +_SNIPPET = 160 + + +@dataclass +class StepCheck: + step: Step + status: int | None = None + latency: float = 0.0 + ok: bool = False + problem: str | None = None # why it failed, in plain words + + +@dataclass +class JourneyCheck: + journey: Journey + steps: list[StepCheck] = field(default_factory=list) + + @property + def ok(self) -> bool: + return len(self.steps) == len(self.journey.steps) and all(s.ok for s in self.steps) + + @property + def failed(self) -> StepCheck | None: + return next((s for s in self.steps if not s.ok), None) + + @property + def stale(self) -> bool: + """The app answered, but not the way the journey expects.""" + failed = self.failed + return failed is not None and failed.status is not None + + +async def check(journey: Journey, base_url: str, *, timeout: float = 10.0, + headers: dict[str, str] | None = None, env: dict[str, str] | None = None, + transport: httpx.AsyncBaseTransport | None = None) -> JourneyCheck: + """Replay JOURNEY once against BASE_URL and stop at the first step that fails.""" + env = dict(os.environ) if env is None else env + result = JourneyCheck(journey) + variables: dict[str, Any] = fresh_uniques() + async with httpx.AsyncClient(timeout=timeout, follow_redirects=True, + headers=client_headers(headers), transport=transport) as client: + for step in journey.steps: + outcome = await run_step(client, step, base_url, variables, env) + result.steps.append(outcome) + if not outcome.ok: + break + return result + + +async def run_step(client: httpx.AsyncClient, step: Step, base_url: str, + variables: dict[str, Any], env: dict[str, str]) -> StepCheck: + """Send one step, judge it against `expect`, and pull its `extract` values + into VARIABLES for the steps after it.""" + outcome = StepCheck(step) + try: + path = render(step.path, variables, env) + req_headers = render(step.headers, variables, env) + payload = render(step.json, variables, env) if step.json is not None else None + body = render(step.body, variables, env) if step.body is not None else None + except TemplateError as exc: + outcome.problem = str(exc) + return outcome + + start = time.perf_counter() + try: + resp = await client.request(step.method, urljoin(base_url, path), + headers=req_headers, json=payload, content=body) + except httpx.TimeoutException: + outcome.problem = "timed out" + return outcome + except httpx.HTTPError as exc: + outcome.problem = f"couldn't reach the app ({exc.__class__.__name__})" + return outcome + outcome.latency = time.perf_counter() - start + outcome.status = resp.status_code + + if not step.accepts(resp.status_code): + wanted = (" or ".join(map(str, step.expect)) if step.expect else "a status below 400") + outcome.problem = f"returned {resp.status_code}, expected {wanted}" + snippet = resp.text.strip().replace("\n", " ")[:_SNIPPET] + if snippet: + outcome.problem += f": {snippet}" + return outcome + + for name, source in step.extract.items(): + value = extract(resp, source) + if value is None: + outcome.problem = f"couldn't find {source} in the response to extract {name}" + return outcome + variables[name] = value + outcome.ok = True + return outcome + + +def extract(resp: httpx.Response, source: str) -> Any: + """Read `body.a.b.0` (JSON) or `header.Name` from a response; None if missing.""" + kind, _, where = source.partition(".") + if kind == "header": + return resp.headers.get(where) + try: + value: Any = resp.json() + except ValueError: + return None + for part in where.split("."): + if isinstance(value, dict) and part in value: + value = value[part] + elif isinstance(value, list) and part.isdigit() and int(part) < len(value): + value = value[int(part)] + else: + return None + return value diff --git a/cli/src/prescale_cli/main.py b/cli/src/prescale_cli/main.py index d87c4c5..67f66a8 100644 --- a/cli/src/prescale_cli/main.py +++ b/cli/src/prescale_cli/main.py @@ -9,6 +9,7 @@ compare, history, investigate, + journeys, login, mcp, profiles, @@ -33,6 +34,7 @@ def cli() -> None: prescale run http://localhost:8000 # load test: what breaks first prescale investigate http://localhost:8000 # ...and why, plus how to fix it prescale audit https://myapp.com # static scaling-hygiene check + prescale journeys verify http://localhost:8000 # check your journeys still replay prescale history # list saved runs prescale compare # capacity diff of the last two runs prescale profiles # launch scenarios for --profile @@ -45,6 +47,7 @@ def cli() -> None: cli.add_command(run.run) cli.add_command(investigate.investigate) +cli.add_command(journeys.journeys) cli.add_command(audit.audit) cli.add_command(history.history) cli.add_command(compare.compare) diff --git a/cli/tests/test_journeys.py b/cli/tests/test_journeys.py new file mode 100644 index 0000000..68c8142 --- /dev/null +++ b/cli/tests/test_journeys.py @@ -0,0 +1,260 @@ +"""Journey files: the format, its safety rules, and replaying a journey once.""" + +import asyncio +import json + +import httpx +import pytest +from click.testing import CliRunner + +from prescale_cli.journeys.model import ( + JourneyError, + TemplateError, + load_dir, + parse, + render, +) +from prescale_cli.journeys.replay import check +from prescale_cli.main import cli + +CHECKOUT = { + "version": 1, + "name": "checkout", + "type": "checkout", + "steps": [ + {"name": "log in", + "request": {"method": "POST", "path": "/api/login", + "json": {"email": "{{unique.email}}", "password": "{{env.PW}}"}}, + "expect": {"status": 200}, + "extract": {"token": "body.token"}}, + {"name": "add to cart", + "request": {"method": "POST", "path": "/api/cart", + "headers": {"Authorization": "Bearer {{token}}"}}, + "expect": {"status": [200, 201]}, + "extract": {"cart_id": "body.id"}}, + {"name": "place order", + "request": {"method": "POST", "path": "/api/carts/{{cart_id}}/order", + "headers": {"Authorization": "Bearer {{token}}"}}, + "think": "500ms"}, + ], +} + + +def _with(**changes): + doc = json.loads(json.dumps(CHECKOUT)) + doc.update(changes) + return doc + + +def _step(i, **changes): + doc = json.loads(json.dumps(CHECKOUT)) + doc["steps"][i].update(changes) + return doc + + +# --- the format ------------------------------------------------------------- + +def test_parse_valid_journey(): + j = parse(CHECKOUT) + assert j.name == "checkout" and j.type == "checkout" + assert j.weight == 5 # the type's default share + assert [s.label for s in j.steps] == [ + "POST /api/login", "POST /api/cart", "POST /api/carts/{{cart_id}}/order"] + assert j.steps[1].expect == [200, 201] + assert j.steps[2].expect is None and j.steps[2].think == 0.5 + assert j.writes + assert j.env_names() == {"PW"} + + +def test_read_only_journey_does_not_write(): + j = parse({"version": 1, "name": "browse", "steps": [{"request": {"path": "/"}}]}) + assert j.type == "browse" and not j.writes + assert j.steps[0].method == "GET" and j.steps[0].name == "step 1" + + +@pytest.mark.parametrize("doc, message", [ + (_with(version=2), "version must be 1"), + (_with(name="Check Out"), "name must be lowercase"), + (_with(type="spam"), "type must be one of"), + (_with(weight=0), "weight must be a positive number"), + (_with(steps=[]), "steps must be a non-empty list"), + (_with(extra=True), "unknown field(s): extra"), + (_step(0, request={"method": "POST", "path": "https://evil.example/x"}), "path must start with /"), + (_step(0, request={"method": "BREW", "path": "/"}), "method BREW isn't supported"), + (_step(0, expect={"status": "ok"}), "HTTP status codes"), + (_step(0, extract={"token": "cookie.sid"}), "body. or header."), + (_step(0, think="soon"), "think must be seconds"), +]) +def test_parse_rejects_bad_journeys(doc, message): + with pytest.raises(JourneyError, match=message.replace("(", r"\(").replace(")", r"\)")): + parse(doc) + + +@pytest.mark.parametrize("jtype", ["side_effect", "destructive"]) +def test_refused_types_are_never_loaded(jtype): + with pytest.raises(JourneyError, match="never replayed"): + parse(_with(type=jtype)) + + +def test_placeholder_must_be_extracted_by_an_earlier_step(): + doc = json.loads(json.dumps(CHECKOUT)) + del doc["steps"][0]["extract"] + with pytest.raises(JourneyError, match=r"uses \{\{token\}\}, but no earlier step"): + parse(doc) + + +def test_unknown_unique_placeholder_is_rejected(): + doc = _step(0, request={"method": "GET", "path": "/u/{{unique.phone}}"}) + with pytest.raises(JourneyError, match="unique.email or unique.id"): + parse(doc) + + +@pytest.mark.parametrize("request_part", [ + {"json": {"email": "a@b.co", "password": "hunter2"}}, + {"headers": {"Authorization": "Bearer abc123"}}, + {"json": {"user": {"api_key": "sk-live"}}}, +]) +def test_literal_secrets_are_refused(request_part): + doc = _step(0, request={"method": "POST", "path": "/api/login", **request_part}) + with pytest.raises(JourneyError, match="literal value"): + parse(doc) + + +def test_load_dir_reports_bad_files_and_duplicate_names(tmp_path): + (tmp_path / "a.yaml").write_text(json.dumps(CHECKOUT)) + (tmp_path / "b.yml").write_text(json.dumps(CHECKOUT)) # same name + (tmp_path / "c.yaml").write_text("version: 1\nname: [oops\n") # not YAML + found, errors = load_dir(tmp_path) + assert [j.name for j in found] == ["checkout", "checkout"] + assert any("not valid YAML" in e for e in errors) + assert any("already used" in e for e in errors) + + +def test_load_dir_missing_directory_is_empty(tmp_path): + assert load_dir(tmp_path / "nope") == ([], []) + + +# --- templating ------------------------------------------------------------- + +def test_render_keeps_type_of_whole_placeholder(): + variables = {"id": 7, "token": "t1"} + assert render({"cart": "{{id}}"}, variables, {}) == {"cart": 7} + assert render("/carts/{{id}}/order", variables, {}) == "/carts/7/order" + assert render("Bearer {{token}}", variables, {}) == "Bearer t1" + assert render("{{env.PW}}", {}, {"PW": "s3cret"}) == "s3cret" + + +def test_render_missing_env_is_an_error(): + with pytest.raises(TemplateError, match="PW isn't set"): + render("{{env.PW}}", {}, {}) + + +# --- replaying once ---------------------------------------------------------- + +def _shop(order_status=201, cart_path="/api/cart"): + seen = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + if request.url.path == "/api/login": + body = json.loads(request.content) + assert body["password"] == "pw" and body["email"].endswith("@example.com") + return httpx.Response(200, json={"token": "t1"}) + assert request.headers["authorization"] == "Bearer t1" + if request.url.path == cart_path: + return httpx.Response(201, json={"id": 42}) + if request.url.path == "/api/carts/42/order": + return httpx.Response(order_status, text="pool exhausted") + return httpx.Response(404, json={"error": "not found"}) + + return httpx.MockTransport(handler), seen + + +def _check(transport): + return asyncio.run(check(parse(CHECKOUT), "http://shop.test/", env={"PW": "pw"}, + transport=transport)) + + +def test_check_replays_and_carries_extracted_values(): + transport, seen = _shop() + result = _check(transport) + assert result.ok and not result.stale + assert [s.status for s in result.steps] == [200, 201, 201] + assert str(seen[2].url) == "http://shop.test/api/carts/42/order" + + +def test_check_marks_a_changed_app_as_stale(): + transport, _ = _shop(cart_path="/api/basket") + result = _check(transport) + assert not result.ok and result.stale + assert len(result.steps) == 2 # stops at the failing step + assert result.failed.problem.startswith("returned 404, expected 200 or 201") + + +def test_check_without_expect_fails_on_4xx_and_5xx(): + transport, _ = _shop(order_status=500) + result = _check(transport) + assert result.stale + assert "expected a status below 400: pool exhausted" in result.failed.problem + + +def test_check_unreachable_app_is_not_stale(): + def boom(request): + raise httpx.ConnectError("refused") + result = _check(httpx.MockTransport(boom)) + assert not result.ok and not result.stale + assert result.failed.problem.startswith("couldn't reach the app") + + +def test_check_missing_extract_fails_the_step(): + def handler(request): + return httpx.Response(200, json={"no_token": True}) + result = _check(httpx.MockTransport(handler)) + assert "couldn't find body.token" in result.failed.problem + + +# --- the command ------------------------------------------------------------- + +def _out(res): + return " ".join(res.output.split()) # rich wraps long lines + + +def _write(tmp_path, doc=CHECKOUT, name="checkout.yaml"): + (tmp_path / name).write_text(json.dumps(doc)) + return tmp_path + + +def test_journeys_lists_files(tmp_path): + res = CliRunner().invoke(cli, ["journeys", "--dir", str(_write(tmp_path))]) + assert res.exit_code == 0, res.output + assert "checkout" in res.output and "yes" in res.output + + +def test_verify_refuses_writes_on_a_non_local_host(tmp_path): + res = CliRunner().invoke(cli, ["journeys", "verify", "https://shop.example.com", + "--dir", str(_write(tmp_path))], env={"PW": "pw"}) + assert res.exit_code == 2 + assert "--allow-writes shop.example.com" in _out(res) + + +def test_verify_refuses_when_env_is_missing(tmp_path): + res = CliRunner().invoke(cli, ["journeys", "verify", "http://localhost:1", + "--dir", str(_write(tmp_path))], env={"PW": None}) + assert res.exit_code == 2 + assert "aren't set: PW" in _out(res) + + +def test_verify_reports_invalid_files(tmp_path): + bad = _with(name="Bad Name") + res = CliRunner().invoke(cli, ["journeys", "verify", "http://localhost:1", + "--dir", str(_write(tmp_path, bad))]) + assert res.exit_code == 2 + assert "name must be lowercase" in _out(res) + + +def test_verify_unknown_journey_name(tmp_path): + res = CliRunner().invoke(cli, ["journeys", "verify", "http://localhost:1", + "--dir", str(_write(tmp_path)), "--journey", "nope"], + env={"PW": "pw"}) + assert res.exit_code == 2 + assert "no journey named nope" in _out(res) diff --git a/examples/README.md b/examples/README.md index 175dd94..bdd75ef 100644 --- a/examples/README.md +++ b/examples/README.md @@ -17,3 +17,12 @@ PreScale reports that `/checkout` breaks first, then diagnoses it as a connection-pool ceiling and suggests raising the pool / adding pgbouncer. It's the app behind the demo in the README. Tune the `Semaphore(25)` to move the breaking point. + +### Journeys + +The shop also has a small JSON API (log in, add to cart, place an order; the order +draws from the same pool) and two journeys for it in `journeys/`: + +```bash +SHOP_PASSWORD=demo prescale journeys verify http://localhost:8400 --dir examples/journeys +``` diff --git a/examples/fragile-shop.py b/examples/fragile-shop.py index 0edb2af..2a6bd2d 100644 --- a/examples/fragile-shop.py +++ b/examples/fragile-shop.py @@ -7,10 +7,18 @@ python examples/fragile-shop.py 8400 prescale investigate http://localhost:8400 --path /search --path /product --path /checkout + +It also has a small JSON API for journeys (log in, add to cart, place an order; +the order draws from the same pool), with journey files in examples/journeys/: + + SHOP_PASSWORD=demo prescale journeys verify http://localhost:8400 --dir examples/journeys """ import asyncio +import itertools +import json import random +import secrets import sys PAGE = (b"" @@ -18,7 +26,13 @@ b'' b"acme.shop") -_REASON = {200: "OK", 404: "Not Found", 500: "Internal Server Error"} +_REASON = {200: "OK", 201: "Created", 400: "Bad Request", 401: "Unauthorized", + 404: "Not Found", 500: "Internal Server Error"} + +PRODUCTS = [{"id": 1, "name": "mug"}, {"id": 2, "name": "tee"}, {"id": 3, "name": "cap"}] +_tokens: set[str] = set() +_carts: dict[int, str] = {} # cart id -> owner token +_cart_ids = itertools.count(1) def _resp(code, body=b"ok", ctype="text/plain"): @@ -36,13 +50,21 @@ async def _handle(reader, writer, pool): if not line: return parts = line.decode(errors="replace").split() + method = parts[0] if parts else "GET" path = parts[1].split("?")[0] if len(parts) >= 2 else "/" - while True: # drain request headers + headers = {} + while True: # read request headers header = await reader.readline() if header in (b"\r\n", b"\n", b""): break + name, _, value = header.decode(errors="replace").partition(":") + headers[name.strip().lower()] = value.strip() + length = int(headers.get("content-length") or 0) + body = await reader.readexactly(length) if length else b"" - if path == "/": + if path.startswith("/api/"): + out = await _api(method, path, headers, body, pool) + elif path == "/": await asyncio.sleep(0.005) out = _resp(200, PAGE, "text/html") elif path in ("/app.css", "/app.js"): @@ -75,6 +97,49 @@ async def _handle(reader, writer, pool): pass +def _json(code, data): + return _resp(code, json.dumps(data).encode(), "application/json") + + +async def _api(method, path, headers, body, pool): + """The shop's JSON API, used by the example journeys.""" + token = headers.get("authorization", "").removeprefix("Bearer ").strip() + if method == "GET" and path == "/api/products": + await asyncio.sleep(random.uniform(0.005, 0.02)) + return _json(200, PRODUCTS) + if method == "POST" and path == "/api/login": + try: + creds = json.loads(body or b"{}") + except ValueError: + return _json(400, {"error": "send JSON"}) + if not creds.get("email") or not creds.get("password"): + return _json(401, {"error": "email and password required"}) + await asyncio.sleep(0.02) # password hashing + new = secrets.token_hex(8) + _tokens.add(new) + return _json(200, {"token": new}) + if token not in _tokens: + return _json(401, {"error": "log in first"}) + if method == "POST" and path == "/api/cart": + cart = next(_cart_ids) + _carts[cart] = token + return _json(201, {"id": cart}) + if method == "POST" and path.startswith("/api/carts/") and path.endswith("/order"): + cart = path.split("/")[3] + if not cart.isdigit() or _carts.get(int(cart)) != token: + return _json(404, {"error": "no such cart"}) + try: # grab a DB connection + await asyncio.wait_for(pool.acquire(), timeout=0.001) + except asyncio.TimeoutError: + return _json(500, {"error": "could not acquire DB connection (pool exhausted)"}) + try: + await asyncio.sleep(random.uniform(0.08, 0.12)) + return _json(201, {"order": f"o-{cart}"}) + finally: + pool.release() + return _json(404, {"error": "not found"}) + + async def main(port): pool = asyncio.Semaphore(25) # 25 DB connections server = await asyncio.start_server( diff --git a/examples/journeys/browse.yaml b/examples/journeys/browse.yaml new file mode 100644 index 0000000..b10e145 --- /dev/null +++ b/examples/journeys/browse.yaml @@ -0,0 +1,12 @@ +version: 1 +name: browse +type: browse +steps: + - name: home + request: {method: GET, path: /} + - name: list products + request: {method: GET, path: /api/products} + expect: {status: 200} + extract: {product_id: body.0.id} + - name: product page + request: {method: GET, path: /product} diff --git a/examples/journeys/checkout.yaml b/examples/journeys/checkout.yaml new file mode 100644 index 0000000..6725e2b --- /dev/null +++ b/examples/journeys/checkout.yaml @@ -0,0 +1,25 @@ +version: 1 +name: checkout +type: checkout +steps: + - name: log in + request: + method: POST + path: /api/login + json: {email: "{{unique.email}}", password: "{{env.SHOP_PASSWORD}}"} + expect: {status: 200} + extract: {token: body.token} + - name: add to cart + request: + method: POST + path: /api/cart + headers: {Authorization: "Bearer {{token}}"} + expect: {status: 201} + extract: {cart_id: body.id} + - name: place order + request: + method: POST + path: "/api/carts/{{cart_id}}/order" + headers: {Authorization: "Bearer {{token}}"} + expect: {status: 201} + think: 1s From c3b3c62be8b1d30624225a4cee772f4d87cb50f3 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Nov 2026 14:07:32 +0000 Subject: [PATCH 2/4] fix(journeys): keep every request on the app, tighten the secrets rule --- README.md | 2 +- cli/src/prescale_cli/commands/journeys.py | 19 ++-- cli/src/prescale_cli/journeys/model.py | 43 +++++++--- cli/src/prescale_cli/journeys/replay.py | 48 +++++++++-- cli/tests/test_journeys.py | 100 +++++++++++++++++++++- examples/fragile-shop.py | 12 ++- 6 files changed, 193 insertions(+), 31 deletions(-) diff --git a/README.md b/README.md index 552c8a1..7be10e9 100644 --- a/README.md +++ b/README.md @@ -200,7 +200,7 @@ prescale journeys verify http://localhost:3000 # replay each once, as one us - `{{name}}` is a value an earlier step pulled out with `extract` (`body.` or `header.`); `{{unique.email}}` and `{{unique.id}}` are fresh each run; `{{env.NAME}}` reads an environment variable. - **Secrets never go in the file.** A literal password, token, API key or `Authorization` value is refused; use `{{env.NAME}}`. -- Paths must start with `/`: a journey only calls the app it's run against. +- Paths start with `/` and are relative to the URL you give (`http://host/app` + `/api/cart` is `http://host/app/api/cart`). A journey only calls that app: a request or redirect to another host fails the step. - Journeys that change data (POST, PUT, PATCH, DELETE) run against local hosts, or a host you name with `--allow-writes staging.myapp.com`. Types `side_effect` (emails, SMS, invites) and `destructive` are never replayed. Try it on the example shop: `python examples/fragile-shop.py 8400`, then `SHOP_PASSWORD=demo prescale journeys verify http://localhost:8400 --dir examples/journeys`. diff --git a/cli/src/prescale_cli/commands/journeys.py b/cli/src/prescale_cli/commands/journeys.py index 626b39a..23f112a 100644 --- a/cli/src/prescale_cli/commands/journeys.py +++ b/cli/src/prescale_cli/commands/journeys.py @@ -48,7 +48,7 @@ def journeys(ctx: click.Context, directory: Path) -> None: table.add_column(col) for j in found: table.add_row(j.name, j.type, f"{j.weight:g}", str(len(j.steps)), - "yes" if j.writes else "no", str(j.source)) + "yes" if j.writes else "no", escape(str(j.source))) console.print(table) @@ -94,8 +94,7 @@ def verify(url: str, directory: Path, only: tuple[str, ...], allow_writes: tuple f"aren't set: {', '.join(unset)}.") raise SystemExit(2) - base = url if url.endswith("/") else url + "/" - results = asyncio.run(_check_all(found, base, timeout=timeout, headers=headers)) + results = asyncio.run(_check_all(found, url, timeout=timeout, headers=headers)) for result in results: _print_check(result) @@ -120,7 +119,8 @@ def _load_or_exit(directory: Path) -> list[Journey]: if errors: raise SystemExit(2) if not found: - console.print(f"No journeys in {directory}. Add one as {directory}/.yaml.") + console.print(f"No journeys in {escape(str(directory))}. Add one as " + f"{escape(str(directory))}/.yaml.", highlight=False) raise SystemExit(1) return found @@ -147,17 +147,18 @@ def _print_check(result: JourneyCheck) -> None: journey = result.journey mark = "[green]✓[/green]" if result.ok else ("[yellow]stale[/yellow]" if result.stale else "[red]✗[/red]") - console.print(f"\n{mark} [bold]{journey.name}[/bold] [dim]({journey.type}, " + console.print(f"\n{mark} [bold]{escape(journey.name)}[/bold] [dim]({journey.type}, " f"{len(journey.steps)} steps)[/dim]") for i, step in enumerate(journey.steps, 1): done = result.steps[i - 1] if i <= len(result.steps) else None + name, label = escape(step.name), escape(step.label) if done is None: - console.print(f" [dim]{i}. {step.name} {step.label} (not run)[/dim]") + console.print(f" [dim]{i}. {name} {label} (not run)[/dim]", highlight=False) elif done.ok: - console.print(f" {i}. {step.name} [dim]{step.label}[/dim] " - f"{done.status} {done.latency * 1000:.0f}ms") + console.print(f" {i}. {name} [dim]{label}[/dim] " + f"{done.status} {done.latency * 1000:.0f}ms", highlight=False) else: - console.print(f" {i}. {step.name} [dim]{step.label}[/dim] " + console.print(f" {i}. {name} [dim]{label}[/dim] " f"[red]{escape(done.problem or '')}[/red]", highlight=False) if done.status is not None and done.status >= 500: console.print(" [dim]A 5xx with no load is usually a bug in the app, " diff --git a/cli/src/prescale_cli/journeys/model.py b/cli/src/prescale_cli/journeys/model.py index 1ca708b..699133f 100644 --- a/cli/src/prescale_cli/journeys/model.py +++ b/cli/src/prescale_cli/journeys/model.py @@ -24,11 +24,14 @@ from __future__ import annotations +import json +import math import re import secrets from dataclasses import dataclass, field from pathlib import Path from typing import Any +from urllib.parse import parse_qsl, urlsplit import yaml @@ -48,8 +51,14 @@ READ_METHODS = {"GET", "HEAD", "OPTIONS"} _PLACEHOLDER = re.compile(r"\{\{\s*([A-Za-z_][\w.]*)\s*\}\}") -_SENSITIVE_KEY = re.compile(r"pass(word|wd)?|secret|token|api[-_]?key|authorization|cookie", - re.IGNORECASE) +# A key is secret-looking when it is, or ends with, one of these words: +# password, client_secret, x-api-key, access_token, Authorization, Cookie. +_SENSITIVE_KEY = re.compile( + r"(^|[-_.])(password|passwd|pwd|secret|token|api[-_]?key|authorization|cookie)$", + re.IGNORECASE) +# The secret itself must be a placeholder, optionally after an auth scheme. +_SECRET_OK = re.compile(r"((Bearer|Basic|Token)\s+)?\{\{\s*[A-Za-z_][\w.]*\s*\}\}", + re.IGNORECASE) _UNIQUE = {"email", "id"} _SOURCES = ("body", "header") @@ -149,7 +158,8 @@ def parse(data: Any) -> Journey: if jtype not in TYPES: raise JourneyError(f"type must be one of {', '.join(TYPES)} (got {jtype!r})") weight = data.get("weight", TYPES[jtype]) - if isinstance(weight, bool) or not isinstance(weight, (int, float)) or weight <= 0: + if isinstance(weight, bool) or not isinstance(weight, (int, float)) \ + or not math.isfinite(weight) or weight <= 0: raise JourneyError("weight must be a positive number") raw_steps = data.get("steps") if not isinstance(raw_steps, list) or not raw_steps: @@ -168,7 +178,8 @@ def _parse_step(raw: Any, index: int) -> Step: name = raw.get("name") or f"step {index}" if not isinstance(name, str): raise JourneyError(f"{where}: name must be text") - where = f"step {index} ({name})" + if raw.get("name"): + where = f"step {index} ({name})" request = raw.get("request") if not isinstance(request, dict): @@ -178,9 +189,9 @@ def _parse_step(raw: Any, index: int) -> Step: if method not in METHODS: raise JourneyError(f"{where}: method {method} isn't supported") path = request.get("path") - if not isinstance(path, str) or not path.startswith("/"): - raise JourneyError(f"{where}: path must start with / (journeys only call the app " - "they're run against, never another site)") + if not isinstance(path, str) or not path.startswith("/") or path.startswith(("//", "/\\")): + raise JourneyError(f"{where}: path must start with a single / (it's relative to the " + "URL the journey runs against, and never reaches another site)") headers = request.get("headers") or {} if not isinstance(headers, dict) or not all(isinstance(v, str) for v in headers.values()): raise JourneyError(f"{where}: headers must map names to text") @@ -189,7 +200,9 @@ def _parse_step(raw: Any, index: int) -> Step: body = request.get("body") if body is not None and not isinstance(body, str): raise JourneyError(f"{where}: body must be text (use json for structured data)") - _check_secrets({"headers": headers, "json": request.get("json")}, where) + _check_secrets({"headers": headers, "json": request.get("json"), + "query": dict(parse_qsl(urlsplit(path).query)), + "body": _decode_body(body)}, where) return Step( name=name, method=method, path=path, @@ -247,6 +260,16 @@ def _only(data: dict, allowed: set[str], where: str) -> None: raise JourneyError(f"{where} has unknown field(s): {', '.join(map(str, unknown))}") +def _decode_body(body: str | None) -> Any: + """A string body as data, so secrets in it are checked too (JSON or a form).""" + if not body: + return None + try: + return json.loads(body) + except ValueError: + return dict(parse_qsl(body)) + + def _check_secrets(value: Any, where: str, key: str = "") -> None: """Refuse a literal value under a secret-looking key: secrets come from {{env.*}} (or an earlier step's extract), so they never end up committed in the file.""" @@ -256,8 +279,8 @@ def _check_secrets(value: Any, where: str, key: str = "") -> None: elif isinstance(value, list): for v in value: _check_secrets(v, where, key) - elif isinstance(value, str) and key and _SENSITIVE_KEY.search(key) \ - and not _PLACEHOLDER.search(value): + elif value is not None and key and _SENSITIVE_KEY.search(key) \ + and not (isinstance(value, str) and _SECRET_OK.fullmatch(value.strip())): raise JourneyError(f"{where}: {key} has a literal value. Put secrets in an " f"environment variable and write \"{{{{env.NAME}}}}\" instead") diff --git a/cli/src/prescale_cli/journeys/replay.py b/cli/src/prescale_cli/journeys/replay.py index b6e67ba..a5222e0 100644 --- a/cli/src/prescale_cli/journeys/replay.py +++ b/cli/src/prescale_cli/journeys/replay.py @@ -1,5 +1,9 @@ """Run a journey's steps against an app. +A step's path is relative to the URL the journey runs against: with +`http://host/app`, the path `/api/cart` is `http://host/app/api/cart`. Every +request, redirects included, must stay on that URL's scheme, host and port. + `check` replays a journey once, as a single user, to see whether it still works against the app as it is now. A step that fails its expectation here, with no load at all, means the journey is stale (the app changed), not that the app @@ -13,7 +17,6 @@ import time from dataclasses import dataclass, field from typing import Any -from urllib.parse import urljoin import httpx @@ -23,6 +26,26 @@ _SNIPPET = 160 +class OffOriginError(Exception): + """A request that would leave the app the journey runs against.""" + + +def step_url(base_url: str, path: str) -> str: + """The step's URL: PATH appended to BASE_URL, refused if it leaves its origin. + Appending (not urljoin) keeps a base path like /app and never lets a rendered + path such as //other-host/x change the host.""" + if not path.startswith("/") or path.startswith(("//", "/\\")): + raise OffOriginError(f"path {path!r} must start with a single /") + url = base_url.rstrip("/") + path + if _origin(httpx.URL(url)) != _origin(httpx.URL(base_url)): + raise OffOriginError(f"{url} isn't on {base_url}") + return url + + +def _origin(url: httpx.URL) -> tuple[str, str, int | None]: + return url.scheme, url.host, url.port + + @dataclass class StepCheck: step: Step @@ -59,8 +82,16 @@ async def check(journey: Journey, base_url: str, *, timeout: float = 10.0, env = dict(os.environ) if env is None else env result = JourneyCheck(journey) variables: dict[str, Any] = fresh_uniques() + origin = _origin(httpx.URL(base_url)) + + async def same_origin(request: httpx.Request) -> None: + # Runs for redirect hops too, so an app can't bounce a write elsewhere. + if _origin(request.url) != origin: + raise OffOriginError(f"redirected to {request.url.host}, outside the app") + async with httpx.AsyncClient(timeout=timeout, follow_redirects=True, - headers=client_headers(headers), transport=transport) as client: + headers=client_headers(headers), transport=transport, + event_hooks={"request": [same_origin]}) as client: for step in journey.steps: outcome = await run_step(client, step, base_url, variables, env) result.steps.append(outcome) @@ -75,18 +106,21 @@ async def run_step(client: httpx.AsyncClient, step: Step, base_url: str, into VARIABLES for the steps after it.""" outcome = StepCheck(step) try: - path = render(step.path, variables, env) - req_headers = render(step.headers, variables, env) + url = step_url(base_url, str(render(step.path, variables, env))) + req_headers = {k: str(v) for k, v in render(step.headers, variables, env).items()} payload = render(step.json, variables, env) if step.json is not None else None body = render(step.body, variables, env) if step.body is not None else None - except TemplateError as exc: + except (TemplateError, OffOriginError) as exc: outcome.problem = str(exc) return outcome start = time.perf_counter() try: - resp = await client.request(step.method, urljoin(base_url, path), - headers=req_headers, json=payload, content=body) + resp = await client.request(step.method, url, headers=req_headers, json=payload, + content=body) + except OffOriginError as exc: + outcome.problem = str(exc) + return outcome except httpx.TimeoutException: outcome.problem = "timed out" return outcome diff --git a/cli/tests/test_journeys.py b/cli/tests/test_journeys.py index 68c8142..889c3e0 100644 --- a/cli/tests/test_journeys.py +++ b/cli/tests/test_journeys.py @@ -79,7 +79,10 @@ def test_read_only_journey_does_not_write(): (_with(weight=0), "weight must be a positive number"), (_with(steps=[]), "steps must be a non-empty list"), (_with(extra=True), "unknown field(s): extra"), - (_step(0, request={"method": "POST", "path": "https://evil.example/x"}), "path must start with /"), + (_step(0, request={"method": "POST", "path": "https://evil.example/x"}), "single /"), + (_step(0, request={"method": "DELETE", "path": "//evil.example/x"}), "single /"), + (_with(weight=float("nan")), "weight must be a positive number"), + (_with(weight=float("inf")), "weight must be a positive number"), (_step(0, request={"method": "BREW", "path": "/"}), "method BREW isn't supported"), (_step(0, expect={"status": "ok"}), "HTTP status codes"), (_step(0, extract={"token": "cookie.sid"}), "body. or header."), @@ -120,6 +123,30 @@ def test_literal_secrets_are_refused(request_part): parse(doc) +@pytest.mark.parametrize("request_part", [ + {"body": '{"email": "a@b.co", "password": "hunter2"}'}, + {"body": "user=a&password=hunter2"}, + {"path": "/a?api_key=sk_live_123"}, + {"json": {"password": 123456}}, + {"headers": {"Authorization": "Basic dXNlcjpwdw== {{unique.id}}"}}, +]) +def test_literal_secrets_are_refused_anywhere(request_part): + request_part.setdefault("path", "/api/login") + doc = _step(0, request={"method": "POST", "path": "/api/login", **request_part}) + with pytest.raises(JourneyError, match="literal value"): + parse(doc) + + +@pytest.mark.parametrize("request_part", [ + {"json": {"passenger_name": "Ada", "passport": "X1", "token_type": "bearer", + "cookie_consent": "yes", "bypass": "no"}}, + {"headers": {"Authorization": "Bearer {{env.TOKEN}}", "X-Api-Key": "{{env.KEY}}"}}, + {"headers": {"Authorization": "Basic {{env.BASIC}}"}}, +]) +def test_ordinary_fields_and_placeholder_secrets_are_allowed(request_part): + parse(_step(0, request={"method": "POST", "path": "/api/login", **request_part})) + + def test_load_dir_reports_bad_files_and_duplicate_names(tmp_path): (tmp_path / "a.yaml").write_text(json.dumps(CHECKOUT)) (tmp_path / "b.yml").write_text(json.dumps(CHECKOUT)) # same name @@ -213,6 +240,68 @@ def handler(request): assert "couldn't find body.token" in result.failed.problem +def _one_step(path="/next", **request): + return parse({"version": 1, "name": "j", "steps": [ + {"request": {"method": "GET", "path": "/start"}, "extract": {"next": "body.next"}}, + {"request": {"method": "POST", "path": path, **request}}]}) + + +def test_rendered_path_cannot_change_host(): + seen = [] + + def handler(request): + seen.append(str(request.url)) + if request.url.path == "/start": + return httpx.Response(200, json={"next": "/evil.example/x", "pid": 1}) + return httpx.Response(200) + + j = _one_step(path="/{{next}}") + result = asyncio.run(check(j, "http://shop.test", env={}, transport=httpx.MockTransport(handler))) + assert seen == ["http://shop.test/start"] # nothing sent off the app + assert not result.ok and "single /" in result.failed.problem + + +def test_base_url_path_prefix_is_kept(): + seen = [] + + def handler(request): + seen.append(request.url.path) + return httpx.Response(200, json={"next": "x"}) + + asyncio.run(check(_one_step(), "http://shop.test/app/", env={}, + transport=httpx.MockTransport(handler))) + assert seen == ["/app/start", "/app/next"] + + +def test_redirect_to_another_host_fails_the_step(): + seen = [] + + def handler(request): + seen.append(request.url.host) + if request.url.path == "/start": + return httpx.Response(200, json={"next": "x"}) + return httpx.Response(307, headers={"Location": "https://evil.example/steal"}) + + result = asyncio.run(check(_one_step(), "http://shop.test", env={}, + transport=httpx.MockTransport(handler))) + assert "evil.example" not in seen + assert not result.ok and "outside the app" in result.failed.problem + + +def test_extracted_number_in_a_header_is_sent_as_text(): + sent = {} + + def handler(request): + if request.url.path == "/start": + return httpx.Response(200, json={"next": 7}) + sent["x-id"] = request.headers["x-id"] + return httpx.Response(200) + + result = asyncio.run(check(_one_step(headers={"X-Id": "{{next}}"}), "http://shop.test", + env={}, transport=httpx.MockTransport(handler))) + assert result.ok and sent["x-id"] == "7" + + # --- the command ------------------------------------------------------------- def _out(res): @@ -252,6 +341,15 @@ def test_verify_reports_invalid_files(tmp_path): assert "name must be lowercase" in _out(res) +def test_verify_prints_markup_like_names_safely(tmp_path): + doc = {"version": 1, "name": "j", "steps": [ + {"name": "[/bold] oops", "request": {"method": "GET", "path": "/"}}]} + res = CliRunner().invoke(cli, ["journeys", "verify", "http://127.0.0.1:9", + "--dir", str(_write(tmp_path, doc))]) + assert res.exit_code == 1, res.output + assert "[/bold] oops" in res.output + + def test_verify_unknown_journey_name(tmp_path): res = CliRunner().invoke(cli, ["journeys", "verify", "http://localhost:1", "--dir", str(_write(tmp_path)), "--journey", "nope"], diff --git a/examples/fragile-shop.py b/examples/fragile-shop.py index 2a6bd2d..c176011 100644 --- a/examples/fragile-shop.py +++ b/examples/fragile-shop.py @@ -59,8 +59,12 @@ async def _handle(reader, writer, pool): break name, _, value = header.decode(errors="replace").partition(":") headers[name.strip().lower()] = value.strip() - length = int(headers.get("content-length") or 0) - body = await reader.readexactly(length) if length else b"" + length = headers.get("content-length") or "0" + if not length.isdigit(): + writer.write(_resp(400, b"bad Content-Length")) + await writer.drain() + return + body = await reader.readexactly(int(length)) if int(length) else b"" if path.startswith("/api/"): out = await _api(method, path, headers, body, pool) @@ -111,7 +115,9 @@ async def _api(method, path, headers, body, pool): try: creds = json.loads(body or b"{}") except ValueError: - return _json(400, {"error": "send JSON"}) + creds = None + if not isinstance(creds, dict): + return _json(400, {"error": "send a JSON object"}) if not creds.get("email") or not creds.get("password"): return _json(401, {"error": "email and password required"}) await asyncio.sleep(0.02) # password hashing From 6ac5abed647ed156528ee63097eafc2362c109e5 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Nov 2026 14:27:06 +0000 Subject: [PATCH 3/4] feat(journeys): prescale run replays journeys at load --- CHANGELOG.md | 6 + README.md | 17 +- cli/src/prescale_cli/commands/journeys.py | 20 ++- cli/src/prescale_cli/commands/run.py | 164 ++++++++++++++++-- cli/src/prescale_cli/journeys/load.py | 122 +++++++++++++ cli/src/prescale_cli/journeys/replay.py | 112 ++++++++++-- cli/src/prescale_cli/loadtest.py | 39 +++-- cli/src/prescale_cli/render.py | 38 ++++ .../prescale_cli/schema/result.schema.json | 63 +++++++ cli/tests/test_journeys_load.py | 152 ++++++++++++++++ 10 files changed, 675 insertions(+), 58 deletions(-) create mode 100644 cli/src/prescale_cli/journeys/load.py create mode 100644 cli/tests/test_journeys_load.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e2909bb..6204b14 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,12 @@ - Journeys that change data only run against local hosts or hosts you allow with `--allow-writes`; journeys that send messages or delete data are never replayed. - The example shop has a JSON API and two example journeys in `examples/journeys/`. +- **`prescale run` replays journeys at load.** When `prescale/journeys/` has journeys, virtual + users walk whole journeys (with each step's think time and their own cookies) instead of hitting + single URLs. Each journey is ramped on its own; the report names the journey and the step that + break first, with per-step timings, and saved runs carry a `journeys` field (plus + `verdict.culprit_journey`). Stale journeys are skipped with the reason. `--journey` picks + journeys; `--urls` tests URLs as before. ### Fixed - **`prescale mcp` works on a fresh install again.** `pip install 'prescale[mcp]'` picked up the new diff --git a/README.md b/README.md index 7be10e9..16371a8 100644 --- a/README.md +++ b/README.md @@ -94,7 +94,7 @@ From source: `git clone https://github.com/pyjeebz/PreScale.git && pip install . | [`run `](#run--what-breaks-first) | Ramp traffic; report what breaks first and at what load | | [`investigate `](#investigate--and-why) | …and *why* it breaks, plus how to fix it (`--app` to start your app first: [replica mode](#replica-mode--test-your-app-not-your-host)) | | [`audit `](#audit--hygiene-no-load) | Load-free HTTP hygiene check (compression, caching, CDN, HTTP/2) | -| [`journeys`](#journeys--your-users-real-paths) | List your journey files; `journeys verify ` checks they still replay | +| [`journeys`](#journeys--your-users-real-paths) | List your journey files; `journeys verify ` checks they still replay. `run` replays them at load | | [`compare`](#ci--gate-on-capacity) | Capacity diff of two runs; regression gating for CI | | [`profiles`](#frame-it-as-a-launch) | Launch scenarios for `--profile` | | [`history` / `show`](#saved-runs-and-the-json-contract) | List and re-open saved runs | @@ -198,10 +198,23 @@ prescale journeys verify http://localhost:3000 # replay each once, as one us `verify` stops at the first step that doesn't answer as expected and calls the journey **stale**: the app changed, or it's broken. It exits non-zero, so it works as a CI check. +When `prescale/journeys/` has journeys, **`prescale run` replays them at load**: each virtual user walks a whole journey (logs in, adds to cart, orders), pauses for each step's `think` time, then starts again as a new user. Each journey is ramped on its own, so its verdict doesn't depend on any traffic mix, and the report names the journey and the step that break first: + +``` + Journeys (each ramped on its own) + Journey Survives Breaks at step Why + browse 400+ (held) — + checkout ~20 (20–30) 3. place order POST /api/carts/{{cart_id}}/order + Server returned 5xx under load… +``` + +A stale journey isn't load-tested (a failure with no load says nothing about capacity); it's listed with the reason. `--journey NAME` picks journeys, and `--urls` (or `--path`, `--from-sitemap`) tests URLs instead. + - `{{name}}` is a value an earlier step pulled out with `extract` (`body.` or `header.`); `{{unique.email}}` and `{{unique.id}}` are fresh each run; `{{env.NAME}}` reads an environment variable. - **Secrets never go in the file.** A literal password, token, API key or `Authorization` value is refused; use `{{env.NAME}}`. - Paths start with `/` and are relative to the URL you give (`http://host/app` + `/api/cart` is `http://host/app/api/cart`). A journey only calls that app: a request or redirect to another host fails the step. -- Journeys that change data (POST, PUT, PATCH, DELETE) run against local hosts, or a host you name with `--allow-writes staging.myapp.com`. Types `side_effect` (emails, SMS, invites) and `destructive` are never replayed. +- Journeys that change data (POST, PUT, PATCH, DELETE) run against local hosts, or a host you name with `--allow-writes staging.myapp.com` (on `verify` and `run`). Types `side_effect` (emails, SMS, invites) and `destructive` are never replayed. +- Each run of a journey has its own cookies and extracted values, so one user's login never leaks to another. Try it on the example shop: `python examples/fragile-shop.py 8400`, then `SHOP_PASSWORD=demo prescale journeys verify http://localhost:8400 --dir examples/journeys`. diff --git a/cli/src/prescale_cli/commands/journeys.py b/cli/src/prescale_cli/commands/journeys.py index 23f112a..5918f59 100644 --- a/cli/src/prescale_cli/commands/journeys.py +++ b/cli/src/prescale_cli/commands/journeys.py @@ -87,12 +87,8 @@ def verify(url: str, directory: Path, only: tuple[str, ...], allow_writes: tuple found = [j for j in found if j.name in only] host = (parsed.hostname or "").lower() - _refuse_unallowed_writes(found, host, {h.lower() for h in allow_writes}) - unset = sorted({name for j in found for name in j.env_names()} - set(os.environ)) - if unset: - console.print(f"[red]Error:[/red] the journeys read environment variable(s) that " - f"aren't set: {', '.join(unset)}.") - raise SystemExit(2) + refuse_unallowed_writes(found, host, {h.lower() for h in allow_writes}) + refuse_missing_env(found) results = asyncio.run(_check_all(found, url, timeout=timeout, headers=headers)) for result in results: @@ -125,7 +121,17 @@ def _load_or_exit(directory: Path) -> list[Journey]: return found -def _refuse_unallowed_writes(found: list[Journey], host: str, allowed: set[str]) -> None: +def refuse_missing_env(found: list[Journey]) -> None: + """Stop before any request if a journey reads an environment variable that + isn't set, instead of failing a step halfway through.""" + unset = sorted({name for j in found for name in j.env_names()} - set(os.environ)) + if unset: + console.print(f"[red]Error:[/red] the journeys read environment variable(s) that " + f"aren't set: {', '.join(unset)}.") + raise SystemExit(2) + + +def refuse_unallowed_writes(found: list[Journey], host: str, allowed: set[str]) -> None: """Journeys that change data run on local hosts, or on hosts named with --allow-writes. Anywhere else they could change real data.""" if host in _LOCAL_HOSTS or host in allowed: diff --git a/cli/src/prescale_cli/commands/run.py b/cli/src/prescale_cli/commands/run.py index f260d62..df01f9c 100644 --- a/cli/src/prescale_cli/commands/run.py +++ b/cli/src/prescale_cli/commands/run.py @@ -13,12 +13,17 @@ import click from rich.console import Console +from rich.markup import escape from rich.panel import Panel from prescale_cli.cloud import CloudError, require_credentials from prescale_cli.commands.hostcheck import confirm_host, host_permission_option +from prescale_cli.commands.journeys import dir_option, refuse_missing_env, refuse_unallowed_writes from prescale_cli.commands.replicaopts import app_option, app_timeout_option, start_replica from prescale_cli.commands.upload import header_option, upload_after_run, upload_option +from prescale_cli.journeys.load import JourneyRun, first_to_break, journeys_block, ramp +from prescale_cli.journeys.model import Journey, load_dir +from prescale_cli.journeys.replay import check from prescale_cli.live import LiveRamp, ramp_progress from prescale_cli.loadtest import ( LoadError, @@ -82,6 +87,14 @@ help="Exit non-zero if it survives fewer than N users (a CI gate).") @click.option("--profile", "profile_name", default=None, help="Frame the run as a launch scenario (see `prescale profiles`).") +@click.option("--journey", "only", multiple=True, + help="Replay only this journey (repeatable). See `prescale journeys`.") +@click.option("--urls", "urls_only", is_flag=True, + help="Test URLs, not journeys, even when prescale/journeys/ has some.") +@click.option("--allow-writes", "allow_writes", multiple=True, metavar="HOST", + help="Let journeys that change data run against this non-local host " + "(repeatable).") +@dir_option @app_option @app_timeout_option @host_permission_option @@ -93,10 +106,16 @@ def run(url: str, paths: tuple[str, ...], from_sitemap: bool, max_users: int, repeat: int, think_time: float, ignore_robots: bool, yes: bool, as_json: bool, html_path: str | None, store: str | None, no_save: bool, fail_under: int | None, - profile_name: str | None, app: str | None, app_timeout: float, - host_permission: bool, headers: dict[str, str], upload: bool) -> None: + profile_name: str | None, only: tuple[str, ...], urls_only: bool, + allow_writes: tuple[str, ...], directory: Path, app: str | None, + app_timeout: float, host_permission: bool, headers: dict[str, str], + upload: bool) -> None: """Load test URL and report what breaks first. + When prescale/journeys/ has journeys, each one is replayed at load (users + walk the whole journey: log in, add to cart, order) and the report names the + journey and step that break first. Use --urls to test URLs instead. + \b Examples: prescale run http://localhost:8000 @@ -128,10 +147,16 @@ def run(url: str, paths: tuple[str, ...], from_sitemap: bool, max_users: int, max_users = prof.peak_users think_time = prof.think_time_s + journeys = _journeys_to_run(directory, only, urls_only=urls_only, + url_mode=bool(paths or from_sitemap or method.upper() != "GET")) + host = (parsed.hostname or "").lower() + if journeys: + refuse_unallowed_writes(journeys, host, {h.lower() for h in allow_writes}) + refuse_missing_env(journeys) + if app: start_replica(app, url, timeout=app_timeout, console=console, quiet=as_json) - host = (parsed.hostname or "").lower() is_local = host in _LOCAL_HOSTS if not is_local and not yes: if as_json: @@ -151,6 +176,30 @@ def run(url: str, paths: tuple[str, ...], from_sitemap: bool, max_users: int, confirm_host(url, console=console, permitted=host_permission, as_json=as_json, timeout=timeout, headers=headers) + config = { + "method": method, + "max_users": max_users, + "stage_seconds": stage_seconds, + "latency_wall_s": latency_wall, + "error_threshold": error_threshold, + "max_rps": max_rps, + "warmup": warmup, + "repeat": repeat, + "think_time_s": think_time, + "fail_under": fail_under, + "profile": prof.name if prof else None, + } + finish = dict(prof=prof, no_save=no_save, store=store, html_path=html_path, + as_json=as_json, upload=upload, fail_under=fail_under) + + if journeys: + report, result, live_mode = _run_journeys( + journeys, url, config, latency_wall=latency_wall, + error_threshold=error_threshold, timeout=timeout, headers=headers, + replica=bool(app), as_json=as_json) + _finish(result, report, live_mode=live_mode, **finish) + return + extra: list[str] = [] if from_sitemap: if not as_json: @@ -203,22 +252,15 @@ def run(url: str, paths: tuple[str, ...], from_sitemap: bool, max_users: int, report = analyze(stages, latency_wall=latency_wall, error_threshold=error_threshold, rate_capped=max_rps is not None) - - config = { - "method": method, - "max_users": max_users, - "stage_seconds": stage_seconds, - "latency_wall_s": latency_wall, - "error_threshold": error_threshold, - "max_rps": max_rps, - "warmup": warmup, - "repeat": repeat, - "think_time_s": think_time, - "fail_under": fail_under, - "profile": prof.name if prof else None, - } result = build_result(report, url=url, targets=targets, config=config, warning=warning, replica=bool(app)) + _finish(result, report, live_mode=live_mode, **finish) + + +def _finish(result: dict, report, *, prof, no_save: bool, store: str | None, + html_path: str | None, as_json: bool, upload: bool, fail_under: int | None, + live_mode: bool) -> None: + """Save, render, upload and gate a finished run (URLs or journeys alike).""" if prof is not None: result["profile"] = scenario_block(prof, report.survives_users) saved_path = None if no_save else write_result(result, store=store) @@ -244,3 +286,91 @@ def run(url: str, paths: tuple[str, ...], from_sitemap: bool, max_users: int, Console(stderr=True).print( f"[red]✗ fail-under:[/red] survives ~{report.survives_users} < {fail_under}") raise SystemExit(1) + + +def _journeys_to_run(directory: Path, only: tuple[str, ...], *, urls_only: bool, + url_mode: bool) -> list[Journey]: + """The journeys this run replays: none in URL mode (--urls, --path, + --from-sitemap or a non-GET --method), else every valid file in DIRECTORY, + narrowed by --journey. A broken journey file stops the run rather than + silently falling back to testing URLs.""" + if urls_only or url_mode: + if only: + console.print("[red]Error:[/red] --journey can't be combined with --urls, " + "--path, --from-sitemap or --method.") + raise SystemExit(2) + return [] + found, errors = load_dir(directory) + for error in errors: + console.print(f"[red]Error:[/red] {escape(error)}", highlight=False) + if errors: + raise SystemExit(2) + if only: + missing = sorted(set(only) - {j.name for j in found}) + if missing: + console.print(f"[red]Error:[/red] no journey named {', '.join(missing)} " + f"in {escape(str(directory))}.", highlight=False) + raise SystemExit(2) + found = [j for j in found if j.name in only] + return found + + +def _run_journeys(journeys: list[Journey], url: str, config: dict, *, latency_wall: float, + error_threshold: float, timeout: float, headers: dict[str, str], + replica: bool, as_json: bool) -> tuple: + """The isolation pass: check each journey once, then ramp each one that + replays on its own. Returns (report, result, live_mode) for the first + journey to break, with every journey's verdict in result["journeys"].""" + if not as_json: + console.print(f"\n[bold]PreScale[/bold] — replaying {len(journeys)} " + f"journey{'s' if len(journeys) != 1 else ''} against [cyan]{url}[/cyan]") + checks = asyncio.run(_check_all(journeys, url, timeout=timeout, headers=headers)) + runs = [JourneyRun(j, c) for j, c in zip(journeys, checks)] + levels = default_levels(config["max_users"]) + live_mode = False + for run in runs: + name = escape(run.journey.name) + if not run.check.ok: + failed = run.check.failed + what = "stale" if run.check.stale else "couldn't run" + console.print(f"[yellow]⚠ {name}: {what}, not load-tested[/yellow] " + f"[dim]({escape(failed.step.name)}: {escape(failed.problem or '')})" + "[/dim]", highlight=False) + continue + if not as_json: + console.print(f"\n[bold]Journey {name}[/bold] [dim]({run.journey.type}, " + f"{len(run.journey.steps)} steps)[/dim]", highlight=False) + progress = ramp_progress(console, latency_wall=latency_wall, as_json=as_json) + live_mode = isinstance(progress, LiveRamp) + try: + with progress: + stages, run.warning = asyncio.run(ramp( + run.journey, url, levels=levels, stage_seconds=config["stage_seconds"], + timeout=timeout, max_rps=config["max_rps"], warmup=config["warmup"], + repeat=config["repeat"], headers=headers, + progress_cb=progress.starting, on_stage=progress.finished)) + except LoadError as exc: + console.print(f"[red]Error:[/red] {exc}") + raise SystemExit(1) + run.report = analyze(stages, latency_wall=latency_wall, + error_threshold=error_threshold, + rate_capped=config["max_rps"] is not None) + + worst = first_to_break(runs) + if worst is None: + console.print("[red]Error:[/red] no journey replays against this app, so there's " + "nothing to load-test. Run `prescale journeys verify` for details.") + raise SystemExit(1) + warnings = list(dict.fromkeys(r.warning for r in runs if r.warning)) + result = build_result(worst.report, url=url, + targets=[s.label for s in worst.journey.steps], + config={**config, "mode": "journeys"}, + warning=" ".join(warnings) or None, replica=replica) + result["verdict"]["culprit_journey"] = worst.journey.name + result["journeys"] = journeys_block(runs) + return worst.report, result, live_mode + + +async def _check_all(journeys: list[Journey], url: str, *, timeout: float, + headers: dict[str, str]): + return [await check(j, url, timeout=timeout, headers=headers) for j in journeys] diff --git a/cli/src/prescale_cli/journeys/load.py b/cli/src/prescale_cli/journeys/load.py new file mode 100644 index 0000000..e398ecf --- /dev/null +++ b/cli/src/prescale_cli/journeys/load.py @@ -0,0 +1,122 @@ +"""Load-test journeys: the isolation pass. + +Each journey is first replayed once (`check`). A journey that fails there is +stale or unreachable and isn't ramped: a failure with no load says nothing about +capacity. Every journey that replays is then ramped on its own, so its verdict +(how many users it survives, the step that breaks first, why) doesn't depend on +any traffic mix. The run's overall verdict is the first journey to break. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +import httpx + +from prescale_cli.journeys.model import Journey +from prescale_cli.journeys.replay import JourneyCheck, client_options, journey_vu +from prescale_cli.loadtest import RunReport, StageResult, run_loadtest + + +@dataclass +class JourneyRun: + journey: Journey + check: JourneyCheck + report: RunReport | None = None + warning: str | None = None + + @property + def status(self) -> str: + if self.report is not None: + return "ok" + return "stale" if self.check.stale else "unreachable" + + +async def ramp(journey: Journey, base_url: str, *, levels: list[int], stage_seconds: float, + timeout: float = 10.0, max_rps: float | None = None, warmup: bool = True, + repeat: int = 1, headers: dict[str, str] | None = None, + env: dict[str, str] | None = None, progress_cb=None, on_stage=None, + transport: httpx.AsyncBaseTransport | None = None, + ) -> tuple[list[StageResult], str | None]: + """Ramp virtual users who each walk JOURNEY, through LEVELS.""" + return await run_loadtest( + [base_url], levels=levels, stage_seconds=stage_seconds, timeout=timeout, + max_rps=max_rps, warmup=warmup, repeat=repeat, headers=headers, + host_permission=True, # the caller has already checked the host's rules + progress_cb=progress_cb, on_stage=on_stage, transport=transport, + vu=journey_vu(journey, base_url, env), client_options=client_options(base_url), + ) + + +def first_to_break(runs: list[JourneyRun]) -> JourneyRun | None: + """The ramped journey with the lowest capacity (ties: the one that broke).""" + ramped = [r for r in runs if r.report is not None] + if not ramped: + return None + return min(ramped, key=lambda r: (r.report.survives_users, + r.report.onset_users is None)) + + +def journeys_block(runs: list[JourneyRun]) -> list[dict]: + """The Result's `journeys` field: one entry per journey, ramped or not.""" + return [_entry(run) for run in runs] + + +def _entry(run: JourneyRun) -> dict: + journey = run.journey + entry: dict = {"name": journey.name, "type": journey.type, "weight": journey.weight, + "status": run.status, "reason": None} + if run.report is None: + failed = run.check.failed + index = run.check.steps.index(failed) + 1 if failed else None + entry["reason"] = f"step {index} ({failed.step.name}): {failed.problem}" if failed \ + else None + return entry + + report = run.report + stage = _stage_at(report) + culprit = None + if report.culprit_route: + index = next((i for i, s in enumerate(journey.steps, 1) + if s.label == report.culprit_route), None) + if index is not None: + culprit = {"index": index, "name": journey.steps[index - 1].name, + "route": report.culprit_route} + entry.update({ + "survives_users": report.survives_users, + "max_tested": report.max_tested, + "onset_users": report.onset_users, + "onset_reason": report.onset_reason, + "confidence": {"survives_low": report.survives_low, + "survives_high": report.survives_high, "stable": report.stable}, + "culprit_step": culprit, + "bottleneck": report.bottleneck, + "steps_at_users": stage.users if stage else None, + "steps": [_step_stats(i, s, stage) for i, s in enumerate(journey.steps, 1)], + }) + return entry + + +def _stage_at(report: RunReport) -> StageResult | None: + """The stage the step timings come from: where it broke, else the last one.""" + if not report.stages: + return None + if report.onset_users is not None: + for stage in report.stages: + if stage.users == report.onset_users: + return stage + return report.stages[-1] + + +def _step_stats(index: int, step, stage: StageResult | None) -> dict: + stat = stage.routes.get(step.label) if stage else None + if stat is None: + return {"index": index, "name": step.name, "route": step.label, "requests": 0, + "p50_ms": None, "p95_ms": None, "error_rate": None} + has_latency = bool(stat.latencies) + return { + "index": index, "name": step.name, "route": step.label, "requests": stat.total, + "p50_ms": round(stat.pct(0.50) * 1000) if has_latency else None, + "p95_ms": round(stat.pct(0.95) * 1000) if has_latency else None, + "error_rate": round(stat.error_rate, 4), + } diff --git a/cli/src/prescale_cli/journeys/replay.py b/cli/src/prescale_cli/journeys/replay.py index a5222e0..69e35a3 100644 --- a/cli/src/prescale_cli/journeys/replay.py +++ b/cli/src/prescale_cli/journeys/replay.py @@ -7,15 +7,20 @@ `check` replays a journey once, as a single user, to see whether it still works against the app as it is now. A step that fails its expectation here, with no load at all, means the journey is stale (the app changed), not that the app -broke under load. Each run gets its own client, so cookies and extracted values -never leak between users. +broke under load. `journey_vu` is the same journey as a virtual user for a load +test: it walks the steps over and over until the stage ends. + +Every run of a journey has its own cookies and extracted values, so a login or +a cart never leaks from one user to another, even on a shared client. """ from __future__ import annotations +import asyncio import os import time from dataclasses import dataclass, field +from http.cookiejar import CookieJar, DefaultCookiePolicy from typing import Any import httpx @@ -46,6 +51,22 @@ def _origin(url: httpx.URL) -> tuple[str, str, int | None]: return url.scheme, url.host, url.port +def client_options(base_url: str) -> dict: + """Settings for any client that runs journeys: requests (redirect hops too) + must stay on BASE_URL's origin, and the client itself keeps no cookies, since + each run of a journey carries its own.""" + origin = _origin(httpx.URL(base_url)) + + async def same_origin(request: httpx.Request) -> None: + if _origin(request.url) != origin: + raise OffOriginError(f"redirected to {request.url.host}, outside the app") + + return { + "event_hooks": {"request": [same_origin]}, + "cookies": CookieJar(policy=DefaultCookiePolicy(allowed_domains=[])), + } + + @dataclass class StepCheck: step: Step @@ -53,6 +74,8 @@ class StepCheck: latency: float = 0.0 ok: bool = False problem: str | None = None # why it failed, in plain words + kind: str | None = None # a short failure kind for load reports + headers: httpx.Headers | None = None @dataclass @@ -82,28 +105,68 @@ async def check(journey: Journey, base_url: str, *, timeout: float = 10.0, env = dict(os.environ) if env is None else env result = JourneyCheck(journey) variables: dict[str, Any] = fresh_uniques() - origin = _origin(httpx.URL(base_url)) - - async def same_origin(request: httpx.Request) -> None: - # Runs for redirect hops too, so an app can't bounce a write elsewhere. - if _origin(request.url) != origin: - raise OffOriginError(f"redirected to {request.url.host}, outside the app") - + jar = httpx.Cookies() async with httpx.AsyncClient(timeout=timeout, follow_redirects=True, headers=client_headers(headers), transport=transport, - event_hooks={"request": [same_origin]}) as client: + **client_options(base_url)) as client: for step in journey.steps: - outcome = await run_step(client, step, base_url, variables, env) + outcome = await run_step(client, step, base_url, variables, env, jar) result.steps.append(outcome) if not outcome.ok: break return result +def journey_vu(journey: Journey, base_url: str, env: dict[str, str] | None = None): + """JOURNEY as a load-test virtual user: walk its steps, pausing for each + step's think time, and start again with a fresh identity until the stage + deadline. A failed step ends that walk (after its pause), like a real user + giving up. Each step is recorded under its label, so the report can name the + step that broke.""" + env = dict(os.environ) if env is None else env + + async def vu(client: httpx.AsyncClient, deadline: float, sink, gate=None) -> None: + loop = asyncio.get_running_loop() + while loop.time() < deadline: + variables: dict[str, Any] = fresh_uniques() + jar = httpx.Cookies() + for step in journey.steps: + if gate is not None and not await gate.wait(deadline): + return + if loop.time() >= deadline: + return + outcome = await run_step(client, step, base_url, variables, env, jar) + _record(sink, outcome) + # A user pauses after a step either way; restarting the instant a step + # fails would hammer the app harder than real users who hit an error. + if step.think: + remaining = deadline - loop.time() + if remaining <= 0: + return + await asyncio.sleep(min(step.think, remaining)) + if not outcome.ok: + break + + return vu + + +def _record(sink, outcome: StepCheck) -> None: + """One step's outcome into the stage's numbers. An answer the journey didn't + expect counts as a failure, even a 4xx: under load that's a step going wrong.""" + label = outcome.step.label + if outcome.status is None: + sink.record_error(label, outcome.kind or "network") + elif outcome.ok or outcome.status >= 500 or outcome.status == 429: + sink.record_response(label, outcome.status, outcome.latency, outcome.headers) + else: + sink.record_error(label, outcome.kind or f"unexpected {outcome.status}") + + async def run_step(client: httpx.AsyncClient, step: Step, base_url: str, - variables: dict[str, Any], env: dict[str, str]) -> StepCheck: - """Send one step, judge it against `expect`, and pull its `extract` values - into VARIABLES for the steps after it.""" + variables: dict[str, Any], env: dict[str, str], + jar: httpx.Cookies) -> StepCheck: + """Send one step with this journey run's cookies (JAR), judge it against + `expect`, and pull its `extract` values into VARIABLES for the steps after it.""" outcome = StepCheck(step) try: url = step_url(base_url, str(render(step.path, variables, env))) @@ -111,24 +174,34 @@ async def run_step(client: httpx.AsyncClient, step: Step, base_url: str, payload = render(step.json, variables, env) if step.json is not None else None body = render(step.body, variables, env) if step.body is not None else None except (TemplateError, OffOriginError) as exc: - outcome.problem = str(exc) + outcome.problem, outcome.kind = str(exc), "journey error" return outcome + request = client.build_request(step.method, url, headers=req_headers, json=payload, + content=body) + jar.set_cookie_header(request) start = time.perf_counter() try: - resp = await client.request(step.method, url, headers=req_headers, json=payload, - content=body) + resp = await client.send(request) except OffOriginError as exc: - outcome.problem = str(exc) + outcome.problem, outcome.kind = str(exc), "left the app" return outcome except httpx.TimeoutException: - outcome.problem = "timed out" + outcome.problem, outcome.kind = "timed out", "timeout" + return outcome + except httpx.ConnectError: + outcome.problem, outcome.kind = "couldn't reach the app (ConnectError)", \ + "connection refused" return outcome except httpx.HTTPError as exc: outcome.problem = f"couldn't reach the app ({exc.__class__.__name__})" + outcome.kind = "network" return outcome outcome.latency = time.perf_counter() - start outcome.status = resp.status_code + outcome.headers = resp.headers + for hop in (*resp.history, resp): + jar.extract_cookies(hop) if not step.accepts(resp.status_code): wanted = (" or ".join(map(str, step.expect)) if step.expect else "a status below 400") @@ -142,6 +215,7 @@ async def run_step(client: httpx.AsyncClient, step: Step, base_url: str, value = extract(resp, source) if value is None: outcome.problem = f"couldn't find {source} in the response to extract {name}" + outcome.kind = "missing extract" return outcome variables[name] = value outcome.ok = True diff --git a/cli/src/prescale_cli/loadtest.py b/cli/src/prescale_cli/loadtest.py index f45ed1d..8005efb 100644 --- a/cli/src/prescale_cli/loadtest.py +++ b/cli/src/prescale_cli/loadtest.py @@ -284,13 +284,15 @@ class SaturationInfo: class _Sink: """Per-stage accumulator, keyed by route label. A 5xx/429 is a failure; - everything else with a status code counts toward latency.""" + everything else with a status code counts toward latency. Targets are URLs, + labelled by path; journeys pass their step labels as they are (`labels=False`).""" - def __init__(self) -> None: + def __init__(self, labels: bool = True) -> None: self.routes: dict[str, RouteStat] = {} + self._labels = labels def _stat(self, target: str) -> RouteStat: - label = route_label(target) + label = route_label(target) if self._labels else target stat = self.routes.get(label) if stat is None: stat = RouteStat() @@ -389,8 +391,11 @@ async def _worker(client: httpx.AsyncClient, method: str, deadline: float, async def _run_stage(client: httpx.AsyncClient, targets: list[str], method: str, users: int, duration: float, gate: _RateGate | None = None, - think_time: float = 0.0) -> StageResult: - sink = _Sink() + think_time: float = 0.0, vu=None) -> StageResult: + """Hold USERS virtual users for DURATION. By default each fires requests at + TARGETS; VU, when given, is the virtual user itself: an async callable + (client, deadline, sink, gate) that records its own requests (journeys).""" + sink = _Sink(labels=vu is None) cycle = itertools.cycle(targets) # round-robin spreads load evenly across routes loop = asyncio.get_running_loop() start = loop.time() @@ -399,10 +404,13 @@ async def _run_stage(client: httpx.AsyncClient, targets: list[str], method: str, lags: list[float] = [] watcher = asyncio.create_task(_watch_loop_lag(stop, lags)) try: - await asyncio.gather( - *(_worker(client, method, deadline, sink, lambda: next(cycle), gate, think_time) - for _ in range(users)) - ) + if vu is not None: + await asyncio.gather(*(vu(client, deadline, sink, gate) for _ in range(users))) + else: + await asyncio.gather( + *(_worker(client, method, deadline, sink, lambda: next(cycle), gate, think_time) + for _ in range(users)) + ) finally: stop.set() await watcher @@ -485,13 +493,18 @@ async def run_loadtest( transport: httpx.AsyncBaseTransport | None = None, headers: dict[str, str] | None = None, host_permission: bool = False, + vu=None, + client_options: dict | None = None, ) -> tuple[list[StageResult], str | None]: """Preflight the target, then ramp through `levels`, spreading each stage's load across `targets`. Stops early once a stage is more than `hard_stop_rate` failed. Returns (stages, warning). A host that restricts load testing (Vercel, Netlify, Cloudflare) stops the run - after the preflight unless `host_permission` says the user has permission.""" + after the preflight unless `host_permission` says the user has permission. + + `vu` replaces the default virtual user (see `_run_stage`), and + `client_options` adds settings to the HTTP client (journeys use both).""" if not targets: raise LoadError("No targets to test.") max_conns = max(levels) + 50 @@ -501,7 +514,7 @@ async def run_loadtest( gate = _RateGate(max_rps) if max_rps else None async with httpx.AsyncClient( timeout=timeout, limits=limits, follow_redirects=True, - headers=client_headers(headers), transport=transport, + headers=client_headers(headers), transport=transport, **(client_options or {}), ) as client: try: preflight = await client.request(method, targets[0]) @@ -529,7 +542,7 @@ async def run_loadtest( # level isn't cold. Still rate-gated, since it's real traffic. warmup_users, warmup_seconds = _warmup_plan(levels, stage_seconds) await _run_stage(client, targets, method, warmup_users, warmup_seconds, - gate, think_time) + gate, think_time, vu) # --repeat pools several ramps per level, so run-to-run variance (cache # state, GC, noisy neighbours) widens the confidence band honestly. @@ -539,7 +552,7 @@ async def run_loadtest( if progress_cb: progress_cb(users) stage = await _run_stage(client, targets, method, users, stage_seconds, - gate, think_time) + gate, think_time, vu) by_level.setdefault(users, []).append(stage) if on_stage: on_stage(stage) diff --git a/cli/src/prescale_cli/render.py b/cli/src/prescale_cli/render.py index 49ae8a8..b7dafb3 100644 --- a/cli/src/prescale_cli/render.py +++ b/cli/src/prescale_cli/render.py @@ -4,6 +4,7 @@ from __future__ import annotations from rich.console import Console +from rich.markup import escape from rich.panel import Panel from rich.table import Table @@ -93,6 +94,8 @@ def render_terminal(result: dict, *, show_ramp: bool = True) -> None: lines.append(f"Traffic {traffic} [dim]({VISITOR_NOTE})[/dim]") if onset_users is not None: culprit = f"{verdict['culprit_route']} " if (multi and verdict["culprit_route"]) else "" + if verdict.get("culprit_journey"): + culprit = f"{escape(verdict['culprit_journey'])} › {escape(culprit)}" if verdict["onset_reason"] == "latency": lines.append(f"Latency wall {culprit}p95 crosses " f"{latency_wall:g}s at ~{onset_users} {_u(onset_users)}.") @@ -130,6 +133,41 @@ def render_terminal(result: dict, *, show_ramp: bool = True) -> None: if multi and stages: _render_routes(result) + if result.get("journeys"): + render_journeys(result) + + +def render_journeys(result: dict) -> None: + """One row per journey: how many users it survives and the step that breaks + first. Each was ramped on its own, so no row depends on the others.""" + table = Table(show_header=True, header_style="bold magenta", + title="Journeys (each ramped on its own)") + for col, justify in (("Journey", "left"), ("Survives", "right"), + ("Breaks at step", "left"), ("Why", "left")): + table.add_column(col, justify=justify) + first = result["verdict"].get("culprit_journey") + for j in result["journeys"]: + name = escape(j["name"]) + if j["status"] != "ok": + label = "stale" if j["status"] == "stale" else "couldn't run" + table.add_row(name, f"[yellow]{label}[/yellow]", "—", + f"[dim]{escape(j.get('reason') or '')}[/dim]") + continue + conf = j.get("confidence") or {} + band = "" + if conf.get("survives_low") is not None and conf.get("survives_high") is not None \ + and conf["survives_low"] != conf["survives_high"]: + band = f" ({conf['survives_low']}–{conf['survives_high']})" + if j["onset_users"] is None: + survives = f"{j['max_tested']}+ [dim](held)[/dim]" + else: + survives = f"~{j['survives_users']}{band}" + step = j.get("culprit_step") + where = f"{step['index']}. {escape(step['name'])} [dim]{escape(step['route'])}[/dim]" \ + if step else "—" + style = "bold red" if j["name"] == first and j["onset_users"] is not None else None + table.add_row(name, survives, where, escape(j.get("bottleneck") or ""), style=style) + console.print(table) def _render_routes(result: dict) -> None: diff --git a/cli/src/prescale_cli/schema/result.schema.json b/cli/src/prescale_cli/schema/result.schema.json index bbc89b4..e2e406e 100644 --- a/cli/src/prescale_cli/schema/result.schema.json +++ b/cli/src/prescale_cli/schema/result.schema.json @@ -45,6 +45,11 @@ "replica": { "type": "boolean", "description": "True when PreScale started the app's own build locally (--app) and tested that, not a deployed host." + }, + "mode": { + "type": "string", + "enum": ["journeys"], + "description": "\"journeys\" when the run replayed journeys (prescale/journeys/); absent for a URL run." } } }, @@ -61,6 +66,10 @@ "onset_users": { "type": ["integer", "null"] }, "onset_reason": { "type": ["string", "null"], "enum": ["errors", "latency", null] }, "culprit_route": { "type": ["string", "null"] }, + "culprit_journey": { + "type": ["string", "null"], + "description": "Journey runs: the journey that breaks first. The verdict, stages and culprit_route (a step's method and path) are that journey's." + }, "bottleneck": { "type": ["string", "null"] }, "saturated": { "type": "boolean" }, "saturation_users": { "type": ["integer", "null"] }, @@ -84,6 +93,11 @@ "git_branch": { "type": ["string", "null"] } } }, + "journeys": { + "type": "array", + "description": "Present on journey runs: one entry per journey. Each journey that replays is ramped on its own; a stale or unreachable one isn't load-tested and carries the reason.", + "items": { "$ref": "#/$defs/journey" } + }, "investigation": { "type": "object", "description": "Present when `prescale investigate` diagnosed a failure.", @@ -123,6 +137,55 @@ } }, "$defs": { + "journey": { + "type": "object", + "required": ["name", "type", "weight", "status", "reason"], + "properties": { + "name": { "type": "string" }, + "type": { "type": "string" }, + "weight": { "type": "number" }, + "status": { "enum": ["ok", "stale", "unreachable"] }, + "reason": { "type": ["string", "null"], "description": "Why a stale or unreachable journey wasn't load-tested." }, + "survives_users": { "type": "integer" }, + "max_tested": { "type": "integer" }, + "onset_users": { "type": ["integer", "null"] }, + "onset_reason": { "type": ["string", "null"], "enum": ["errors", "latency", null] }, + "confidence": { + "type": "object", + "properties": { + "survives_low": { "type": ["integer", "null"] }, + "survives_high": { "type": ["integer", "null"] }, + "stable": { "type": "boolean" } + } + }, + "culprit_step": { + "type": ["object", "null"], + "properties": { + "index": { "type": "integer", "description": "1-based step number." }, + "name": { "type": "string" }, + "route": { "type": "string" } + } + }, + "bottleneck": { "type": ["string", "null"] }, + "steps_at_users": { "type": ["integer", "null"], "description": "The level the step timings come from: where it broke, else the highest tested." }, + "steps": { + "type": "array", + "items": { + "type": "object", + "required": ["index", "name", "route", "requests"], + "properties": { + "index": { "type": "integer" }, + "name": { "type": "string" }, + "route": { "type": "string" }, + "requests": { "type": "integer" }, + "p50_ms": { "type": ["integer", "null"] }, + "p95_ms": { "type": ["integer", "null"] }, + "error_rate": { "type": ["number", "null"] } + } + } + } + } + }, "stage": { "type": "object", "required": [ diff --git a/cli/tests/test_journeys_load.py b/cli/tests/test_journeys_load.py new file mode 100644 index 0000000..2c7897f --- /dev/null +++ b/cli/tests/test_journeys_load.py @@ -0,0 +1,152 @@ +"""Journeys at load: virtual users walk whole journeys, each ramped on its own.""" + +import asyncio +import json + +import httpx +from click.testing import CliRunner + +from prescale_cli.journeys.load import JourneyRun, first_to_break, journeys_block, ramp +from prescale_cli.journeys.model import parse +from prescale_cli.journeys.replay import check +from prescale_cli.loadtest import analyze +from prescale_cli.main import cli + +SHOP = { + "version": 1, "name": "checkout", "type": "checkout", + "steps": [ + {"name": "log in", "request": {"method": "POST", "path": "/api/login"}, + "expect": {"status": 200}, "extract": {"token": "body.token"}}, + {"name": "place order", + "request": {"method": "POST", "path": "/api/orders", + "headers": {"Authorization": "Bearer {{token}}"}}, + "expect": {"status": 201}}, + ], +} +BROWSE = {"version": 1, "name": "browse", "steps": [{"request": {"path": "/"}}]} + + +def _shop(pool: int): + """A mock shop whose order step fails once more than POOL orders are in + flight at once, and which checks every order carries its own login's token + and cookie.""" + state = {"in_flight": 0, "logins": 0, "mixed": 0} + + async def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/login": + state["logins"] += 1 + token = f"t{state['logins']}" + return httpx.Response(200, json={"token": token}, + headers={"set-cookie": f"sid={token}; Path=/"}) + if request.url.path == "/api/orders": + token = request.headers["authorization"].removeprefix("Bearer ") + if request.headers.get("cookie") != f"sid={token}": + state["mixed"] += 1 + if state["in_flight"] >= pool: + return httpx.Response(500, text="pool exhausted") + state["in_flight"] += 1 + try: + await asyncio.sleep(0.02) + finally: + state["in_flight"] -= 1 + return httpx.Response(201) + return httpx.Response(200) + + return httpx.MockTransport(handler), state + + +def _ramp(journey, transport, levels=(1, 2, 5, 10, 20)): + return asyncio.run(ramp(journey, "http://shop.test", levels=list(levels), + stage_seconds=0.2, warmup=False, transport=transport)) + + +def test_ramp_records_each_step_and_finds_the_step_that_breaks(): + transport, state = _shop(pool=4) + stages, _ = _ramp(parse(SHOP), transport) + assert set(stages[0].routes) == {"POST /api/login", "POST /api/orders"} + report = analyze(stages, latency_wall=2.0, error_threshold=0.02) + assert report.onset_reason == "errors" + assert report.culprit_route == "POST /api/orders" + assert 1 <= report.survives_users < 20 + # every order went out with its own user's token and cookie + assert state["mixed"] == 0 and state["logins"] > 20 + + +def test_journeys_block_names_the_step_and_skips_stale_journeys(): + transport, _ = _shop(pool=4) + checkout, browse = parse(SHOP), parse(BROWSE) + stages, _ = _ramp(checkout, transport) + ok = JourneyRun(checkout, asyncio.run(check(checkout, "http://shop.test", env={}, + transport=transport)), + report=analyze(stages, latency_wall=2.0, error_threshold=0.02)) + + def gone(request): + return httpx.Response(404, text="moved") + stale = JourneyRun(browse, asyncio.run(check(browse, "http://shop.test", env={}, + transport=httpx.MockTransport(gone)))) + + block = journeys_block([ok, stale]) + assert block[0]["status"] == "ok" + assert block[0]["culprit_step"] == {"index": 2, "name": "place order", + "route": "POST /api/orders"} + assert [s["route"] for s in block[0]["steps"]] == ["POST /api/login", "POST /api/orders"] + assert block[0]["steps"][1]["error_rate"] > 0 + assert block[1] == {"name": "browse", "type": "browse", "weight": 60.0, + "status": "stale", + "reason": "step 1 (step 1): returned 404, expected a status " + "below 400: moved"} + assert first_to_break([ok, stale]) is ok + json.dumps(block) # it goes into the saved Result + + +def test_first_to_break_prefers_the_lowest_capacity(): + transport, _ = _shop(pool=4) + weak = JourneyRun(parse(SHOP), None, + report=analyze(_ramp(parse(SHOP), transport)[0], + latency_wall=2.0, error_threshold=0.02)) + strong = JourneyRun(parse(BROWSE), None, + report=analyze(_ramp(parse(BROWSE), transport)[0], + latency_wall=2.0, error_threshold=0.02)) + assert strong.report.onset_users is None + assert first_to_break([strong, weak]) is weak + assert first_to_break([JourneyRun(parse(BROWSE), None)]) is None + + +# --- `prescale run` picks journeys --------------------------------------------- + +def _dir(tmp_path, *docs): + for doc in docs: + (tmp_path / f"{doc['name']}.yaml").write_text(json.dumps(doc)) + return str(tmp_path) + + +def _out(res): + return " ".join(res.output.split()) + + +def test_run_refuses_journey_writes_on_a_non_local_host(tmp_path): + res = CliRunner().invoke(cli, ["run", "https://shop.example.com", "--i-own-this", + "--dir", _dir(tmp_path, SHOP)]) + assert res.exit_code == 2 + assert "--allow-writes shop.example.com" in _out(res) + + +def test_run_stops_on_a_broken_journey_file(tmp_path): + res = CliRunner().invoke(cli, ["run", "http://localhost:1", + "--dir", _dir(tmp_path, {**BROWSE, "version": 9})]) + assert res.exit_code == 2 + assert "version must be 1" in _out(res) + + +def test_run_journey_flag_conflicts_with_url_mode(tmp_path): + res = CliRunner().invoke(cli, ["run", "http://localhost:1", "--urls", "--journey", "x", + "--dir", _dir(tmp_path, BROWSE)]) + assert res.exit_code == 2 + assert "--journey can't be combined" in _out(res) + + +def test_run_unknown_journey(tmp_path): + res = CliRunner().invoke(cli, ["run", "http://localhost:1", "--journey", "nope", + "--dir", _dir(tmp_path, BROWSE)]) + assert res.exit_code == 2 + assert "no journey named nope" in _out(res) From 1cad16dd8fbf1792798757a239d7ebcbb732aa8e Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Nov 2026 14:34:51 +0000 Subject: [PATCH 4/4] fix(journeys): cookies across redirects, pure --json, no / probe, honest think time and compare --- cli/src/prescale_cli/commands/run.py | 14 +-- cli/src/prescale_cli/compare.py | 25 +++++- cli/src/prescale_cli/journeys/load.py | 10 ++- cli/src/prescale_cli/journeys/replay.py | 38 +++++++-- cli/src/prescale_cli/loadtest.py | 52 +++++++----- cli/tests/test_compare.py | 22 +++++ cli/tests/test_journeys_load.py | 108 ++++++++++++++++++++++++ 7 files changed, 230 insertions(+), 39 deletions(-) diff --git a/cli/src/prescale_cli/commands/run.py b/cli/src/prescale_cli/commands/run.py index df01f9c..078b832 100644 --- a/cli/src/prescale_cli/commands/run.py +++ b/cli/src/prescale_cli/commands/run.py @@ -324,6 +324,8 @@ def _run_journeys(journeys: list[Journey], url: str, config: dict, *, latency_wa if not as_json: console.print(f"\n[bold]PreScale[/bold] — replaying {len(journeys)} " f"journey{'s' if len(journeys) != 1 else ''} against [cyan]{url}[/cyan]") + # --json keeps stdout for the Result alone; notes about journeys go to stderr. + out = Console(stderr=True) if as_json else console checks = asyncio.run(_check_all(journeys, url, timeout=timeout, headers=headers)) runs = [JourneyRun(j, c) for j, c in zip(journeys, checks)] levels = default_levels(config["max_users"]) @@ -333,7 +335,7 @@ def _run_journeys(journeys: list[Journey], url: str, config: dict, *, latency_wa if not run.check.ok: failed = run.check.failed what = "stale" if run.check.stale else "couldn't run" - console.print(f"[yellow]⚠ {name}: {what}, not load-tested[/yellow] " + out.print(f"[yellow]⚠ {name}: {what}, not load-tested[/yellow] " f"[dim]({escape(failed.step.name)}: {escape(failed.problem or '')})" "[/dim]", highlight=False) continue @@ -347,10 +349,10 @@ def _run_journeys(journeys: list[Journey], url: str, config: dict, *, latency_wa stages, run.warning = asyncio.run(ramp( run.journey, url, levels=levels, stage_seconds=config["stage_seconds"], timeout=timeout, max_rps=config["max_rps"], warmup=config["warmup"], - repeat=config["repeat"], headers=headers, + repeat=config["repeat"], headers=headers, think=config["think_time_s"], progress_cb=progress.starting, on_stage=progress.finished)) except LoadError as exc: - console.print(f"[red]Error:[/red] {exc}") + out.print(f"[red]Error:[/red] {exc}") raise SystemExit(1) run.report = analyze(stages, latency_wall=latency_wall, error_threshold=error_threshold, @@ -358,7 +360,7 @@ def _run_journeys(journeys: list[Journey], url: str, config: dict, *, latency_wa worst = first_to_break(runs) if worst is None: - console.print("[red]Error:[/red] no journey replays against this app, so there's " + out.print("[red]Error:[/red] no journey replays against this app, so there's " "nothing to load-test. Run `prescale journeys verify` for details.") raise SystemExit(1) warnings = list(dict.fromkeys(r.warning for r in runs if r.warning)) @@ -366,7 +368,9 @@ def _run_journeys(journeys: list[Journey], url: str, config: dict, *, latency_wa targets=[s.label for s in worst.journey.steps], config={**config, "mode": "journeys"}, warning=" ".join(warnings) or None, replica=replica) - result["verdict"]["culprit_journey"] = worst.journey.name + # Only a journey that broke is a culprit; if all held, there is none. + result["verdict"]["culprit_journey"] = (worst.journey.name + if worst.report.onset_users is not None else None) result["journeys"] = journeys_block(runs) return worst.report, result, live_mode diff --git a/cli/src/prescale_cli/compare.py b/cli/src/prescale_cli/compare.py index d2d1a45..dea8b2e 100644 --- a/cli/src/prescale_cli/compare.py +++ b/cli/src/prescale_cli/compare.py @@ -27,6 +27,8 @@ def _facts(result: dict) -> dict: "branch": env.get("git_branch"), "created_at": result.get("created_at"), "target": result.get("target", {}).get("url"), + "mode": (result.get("config") or {}).get("mode") or "urls", + "culprit_journey": v.get("culprit_journey"), } @@ -35,6 +37,9 @@ def compare_results(old: dict, new: dict, *, threshold: float = _DEFAULT_THRESHO survives_users; it's 'confident' when the M1 bands don't overlap. A new run that held to its max below the old level is 'inconclusive', not a drop.""" o, n = _facts(old), _facts(new) + if o["mode"] != n["mode"]: + # URL capacity and journey capacity measure different things: never a regression. + return _not_comparable(o, n) o_surv, n_surv = o["survives_users"], n["survives_users"] pct = (n_surv - o_surv) / o_surv if o_surv else 0.0 inconclusive = (n["onset_users"] is None and n["max_tested"] is not None @@ -57,10 +62,28 @@ def compare_results(old: dict, new: dict, *, threshold: float = _DEFAULT_THRESHO "improved": improved, "inconclusive": inconclusive, "confident": confident, - "summary": _summary(o, n, pct, regressed, improved, inconclusive), + "summary": _summary(o, n, pct, regressed, improved, inconclusive) + _journey_note(o, n), } +def _not_comparable(o: dict, n: dict) -> dict: + kinds = {"urls": "tested URLs", "journeys": "replayed journeys"} + return { + "old": o, "new": n, "survives_delta": 0, "survives_pct": 0.0, "peak_rps_pct": 0.0, + "regressed": False, "improved": False, "inconclusive": True, "confident": False, + "summary": (f"Not comparable: the old run {kinds.get(o['mode'], o['mode'])}, the new " + f"run {kinds.get(n['mode'], n['mode'])}. Compare runs of the same kind."), + } + + +def _journey_note(o: dict, n: dict) -> str: + if o["culprit_journey"] and n["culprit_journey"] \ + and o["culprit_journey"] != n["culprit_journey"]: + return (f" The journey that breaks first changed: {o['culprit_journey']} → " + f"{n['culprit_journey']}.") + return "" + + def _summary(o: dict, n: dict, pct: float, regressed: bool, improved: bool, inconclusive: bool = False) -> str: arrow = f"{o['survives_users']} → {n['survives_users']}" diff --git a/cli/src/prescale_cli/journeys/load.py b/cli/src/prescale_cli/journeys/load.py index e398ecf..d25554a 100644 --- a/cli/src/prescale_cli/journeys/load.py +++ b/cli/src/prescale_cli/journeys/load.py @@ -35,16 +35,18 @@ def status(self) -> str: async def ramp(journey: Journey, base_url: str, *, levels: list[int], stage_seconds: float, timeout: float = 10.0, max_rps: float | None = None, warmup: bool = True, repeat: int = 1, headers: dict[str, str] | None = None, - env: dict[str, str] | None = None, progress_cb=None, on_stage=None, - transport: httpx.AsyncBaseTransport | None = None, + env: dict[str, str] | None = None, think: float = 0.0, progress_cb=None, + on_stage=None, transport: httpx.AsyncBaseTransport | None = None, ) -> tuple[list[StageResult], str | None]: - """Ramp virtual users who each walk JOURNEY, through LEVELS.""" + """Ramp virtual users who each walk JOURNEY, through LEVELS. THINK is the + pause for steps without their own `think`.""" return await run_loadtest( [base_url], levels=levels, stage_seconds=stage_seconds, timeout=timeout, max_rps=max_rps, warmup=warmup, repeat=repeat, headers=headers, host_permission=True, # the caller has already checked the host's rules progress_cb=progress_cb, on_stage=on_stage, transport=transport, - vu=journey_vu(journey, base_url, env), client_options=client_options(base_url), + vu=journey_vu(journey, base_url, env, think), client_options=client_options(base_url), + preflight=False, # `check` already replayed the journey once ) diff --git a/cli/src/prescale_cli/journeys/replay.py b/cli/src/prescale_cli/journeys/replay.py index 69e35a3..4760ff3 100644 --- a/cli/src/prescale_cli/journeys/replay.py +++ b/cli/src/prescale_cli/journeys/replay.py @@ -29,6 +29,7 @@ from prescale_cli.loadtest import client_headers _SNIPPET = 160 +_MAX_REDIRECTS = 10 class OffOriginError(Exception): @@ -117,12 +118,14 @@ async def check(journey: Journey, base_url: str, *, timeout: float = 10.0, return result -def journey_vu(journey: Journey, base_url: str, env: dict[str, str] | None = None): +def journey_vu(journey: Journey, base_url: str, env: dict[str, str] | None = None, + think: float = 0.0): """JOURNEY as a load-test virtual user: walk its steps, pausing for each step's think time, and start again with a fresh identity until the stage deadline. A failed step ends that walk (after its pause), like a real user giving up. Each step is recorded under its label, so the report can name the - step that broke.""" + step that broke. THINK is the pause for steps that don't set their own + (`--think-time`, or a launch profile's).""" env = dict(os.environ) if env is None else env async def vu(client: httpx.AsyncClient, deadline: float, sink, gate=None) -> None: @@ -139,13 +142,16 @@ async def vu(client: httpx.AsyncClient, deadline: float, sink, gate=None) -> Non _record(sink, outcome) # A user pauses after a step either way; restarting the instant a step # fails would hammer the app harder than real users who hit an error. - if step.think: + pause = step.think or think + if pause: remaining = deadline - loop.time() if remaining <= 0: return - await asyncio.sleep(min(step.think, remaining)) + await asyncio.sleep(min(pause, remaining)) if not outcome.ok: break + # Never spin: a walk that failed without awaiting must still yield. + await asyncio.sleep(0) return vu @@ -179,10 +185,9 @@ async def run_step(client: httpx.AsyncClient, step: Step, base_url: str, request = client.build_request(step.method, url, headers=req_headers, json=payload, content=body) - jar.set_cookie_header(request) start = time.perf_counter() try: - resp = await client.send(request) + resp = await _send(client, request, jar) except OffOriginError as exc: outcome.problem, outcome.kind = str(exc), "left the app" return outcome @@ -200,8 +205,6 @@ async def run_step(client: httpx.AsyncClient, step: Step, base_url: str, outcome.latency = time.perf_counter() - start outcome.status = resp.status_code outcome.headers = resp.headers - for hop in (*resp.history, resp): - jar.extract_cookies(hop) if not step.accepts(resp.status_code): wanted = (" or ".join(map(str, step.expect)) if step.expect else "a status below 400") @@ -222,6 +225,25 @@ async def run_step(client: httpx.AsyncClient, step: Step, base_url: str, return outcome +async def _send(client: httpx.AsyncClient, request: httpx.Request, + jar: httpx.Cookies) -> httpx.Response: + """Send REQUEST with this run's cookies, following redirects one hop at a time + so every hop carries the cookies earlier hops set (a login that redirects to + /account stays logged in). The client keeps no cookies of its own, and its + same-origin hook checks each hop.""" + for _ in range(_MAX_REDIRECTS + 1): + if "cookie" in request.headers: + del request.headers["cookie"] + jar.set_cookie_header(request) + resp = await client.send(request, follow_redirects=False) + jar.extract_cookies(resp) + if resp.next_request is None: + return resp + await resp.aclose() + request = resp.next_request + raise httpx.TooManyRedirects("too many redirects", request=request) + + def extract(resp: httpx.Response, source: str) -> Any: """Read `body.a.b.0` (JSON) or `header.Name` from a response; None if missing.""" kind, _, where = source.partition(".") diff --git a/cli/src/prescale_cli/loadtest.py b/cli/src/prescale_cli/loadtest.py index 8005efb..131d9f7 100644 --- a/cli/src/prescale_cli/loadtest.py +++ b/cli/src/prescale_cli/loadtest.py @@ -495,6 +495,7 @@ async def run_loadtest( host_permission: bool = False, vu=None, client_options: dict | None = None, + preflight: bool = True, ) -> tuple[list[StageResult], str | None]: """Preflight the target, then ramp through `levels`, spreading each stage's load across `targets`. Stops early once a stage is more than `hard_stop_rate` @@ -504,7 +505,9 @@ async def run_loadtest( after the preflight unless `host_permission` says the user has permission. `vu` replaces the default virtual user (see `_run_stage`), and - `client_options` adds settings to the HTTP client (journeys use both).""" + `client_options` adds settings to the HTTP client. Journeys use both, and skip + the preflight (`preflight=False`): each journey was already replayed once, which + checks the app better than a request to targets[0].""" if not targets: raise LoadError("No targets to test.") max_conns = max(levels) + 50 @@ -516,26 +519,8 @@ async def run_loadtest( timeout=timeout, limits=limits, follow_redirects=True, headers=client_headers(headers), transport=transport, **(client_options or {}), ) as client: - try: - preflight = await client.request(method, targets[0]) - except httpx.HTTPError as exc: - raise LoadError(f"Couldn't reach {targets[0]}: {exc}") from exc - blocker = edge_blocker(preflight.headers) - if blocker: - # Load now would only measure the firewall, and deepen the block. - raise LoadError( - f"{_capitalize(blocker)} is blocking requests from this machine (it answered " - f"the first request with HTTP {preflight.status_code}), so a load test would " - f"only measure the firewall. {_INSTEAD}" - ) - rule = detect_host(preflight.headers) - if rule and not host_permission: - raise HostPolicyError(rule) - if preflight.status_code >= 400: - warning = ( - f"First request returned HTTP {preflight.status_code} — " - "results may reflect a broken endpoint, not a load limit." - ) + if preflight: + warning = await _preflight(client, method, targets[0], host_permission) if warmup: # Discarded: warms caches/JIT/connection pools so the first measured @@ -563,6 +548,31 @@ async def run_loadtest( return stages, warning +async def _preflight(client: httpx.AsyncClient, method: str, target: str, + host_permission: bool) -> str | None: + """One request before any load: stop if a firewall already blocks us or the + host forbids load tests; return a warning if the target answers with an error.""" + try: + first = await client.request(method, target) + except httpx.HTTPError as exc: + raise LoadError(f"Couldn't reach {target}: {exc}") from exc + blocker = edge_blocker(first.headers) + if blocker: + # Load now would only measure the firewall, and deepen the block. + raise LoadError( + f"{_capitalize(blocker)} is blocking requests from this machine (it answered " + f"the first request with HTTP {first.status_code}), so a load test would " + f"only measure the firewall. {_INSTEAD}" + ) + rule = detect_host(first.headers) + if rule and not host_permission: + raise HostPolicyError(rule) + if first.status_code >= 400: + return (f"First request returned HTTP {first.status_code} — " + "results may reflect a broken endpoint, not a load limit.") + return None + + def open_client(*, timeout: float = 10.0, max_conns: int = 64, transport: httpx.AsyncBaseTransport | None = None, headers: dict[str, str] | None = None) -> httpx.AsyncClient: diff --git a/cli/tests/test_compare.py b/cli/tests/test_compare.py index 2f4a97c..4dd404a 100644 --- a/cli/tests/test_compare.py +++ b/cli/tests/test_compare.py @@ -103,3 +103,25 @@ def test_held_run_below_the_old_level_is_inconclusive_not_a_regression(): def test_held_run_that_reached_the_old_level_still_compares(): c = compare_results(_result(200, low=180, high=220), _result(190, held_at=190)) assert not c["inconclusive"] and not c["regressed"] + + +def _mode(result, mode=None, culprit=None): + r = json.loads(json.dumps(result)) + if mode: + r["config"]["mode"] = mode + r["verdict"]["culprit_journey"] = culprit + return r + + +def test_url_run_and_journey_run_are_not_comparable(): + old = _result(200) + new = _mode(_result(20), mode="journeys", culprit="checkout") + cmp = compare_results(old, new) + assert not cmp["regressed"] and cmp["inconclusive"] + assert cmp["summary"].startswith("Not comparable") + + +def test_journey_runs_note_a_new_first_to_break(): + old = _mode(_result(100), mode="journeys", culprit="search") + new = _mode(_result(100), mode="journeys", culprit="checkout") + assert "search → checkout" in compare_results(old, new)["summary"] diff --git a/cli/tests/test_journeys_load.py b/cli/tests/test_journeys_load.py index 2c7897f..6d90c15 100644 --- a/cli/tests/test_journeys_load.py +++ b/cli/tests/test_journeys_load.py @@ -150,3 +150,111 @@ def test_run_unknown_journey(tmp_path): "--dir", _dir(tmp_path, BROWSE)]) assert res.exit_code == 2 assert "no journey named nope" in _out(res) + + +# --- review fixes ------------------------------------------------------------- + +LOGIN_REDIRECT = { + "version": 1, "name": "account", "type": "account", + "steps": [ + {"name": "log in", "request": {"method": "POST", "path": "/login"}, + "expect": {"status": 200}, "extract": {"user": "body.user"}}, + {"name": "update cart", "request": {"method": "POST", "path": "/cart"}, + "expect": {"status": 200}, "extract": {"again": "body.user"}}, + ], +} + + +def _login_redirects(): + """POST /login sets a cookie on a 302; POST /cart answers 303. The pages they + redirect to only know the user if the cookie came along.""" + logins = iter(range(10_000)) + + def handler(request): + cookie = request.headers.get("cookie") + if request.url.path == "/login": + return httpx.Response(302, headers={"location": "/me", + "set-cookie": f"sid=u{next(logins)}; Path=/"}) + if request.url.path == "/cart": + return httpx.Response(303, headers={"location": "/cartview"}) + if request.url.path in ("/me", "/cartview"): + return httpx.Response(200, json={"user": cookie} if cookie else {}) + if request.url.path == "/": + return httpx.Response(302, headers={"location": "https://sso.example.com/"}) + return httpx.Response(404) + + return httpx.MockTransport(handler) + + +def test_cookies_survive_redirects_in_verify(): + result = asyncio.run(check(parse(LOGIN_REDIRECT), "http://shop.test", env={}, + transport=_login_redirects())) + assert result.ok, result.failed and result.failed.problem + + +def test_cookies_survive_redirects_at_load_and_root_redirect_is_fine(): + # "/" redirects to an SSO host: journeys don't probe it, so no OffOriginError. + stages, warning = _ramp(parse(LOGIN_REDIRECT), _login_redirects(), levels=(1, 5)) + assert warning is None + for stage in stages: + assert all(stat.errors == 0 for stat in stage.routes.values()), stage.routes + + +def _fake_checks(monkeypatch, stale_names=()): + """Replace the network in `run`'s journey mode: CHECK says which journeys are + stale; RAMP returns a run that holds at every level.""" + from prescale_cli.commands import run as run_cmd + from prescale_cli.journeys.replay import JourneyCheck, StepCheck + from prescale_cli.loadtest import RouteStat, StageResult + + async def fake_check_all(journeys, url, *, timeout, headers): + out = [] + for j in journeys: + c = JourneyCheck(j) + ok = j.name not in stale_names + c.steps = [StepCheck(s, status=200 if ok else 404, ok=ok, + problem=None if ok else "returned 404, expected 200") + for s in j.steps[:1]] if not ok else \ + [StepCheck(s, status=200, ok=True) for s in j.steps] + out.append(c) + return out + + async def fake_ramp(journey, url, *, levels, **kwargs): + routes = {s.label: RouteStat(total=10, latencies=[0.01] * 10) for s in journey.steps} + return [StageResult(users=u, duration=1.0, routes=dict(routes)) for u in levels], None + + monkeypatch.setattr(run_cmd, "_check_all", fake_check_all) + monkeypatch.setattr(run_cmd, "ramp", fake_ramp) + + +def test_run_json_stays_pure_when_a_journey_is_stale(tmp_path, monkeypatch): + _fake_checks(monkeypatch, stale_names={"browse"}) + res = CliRunner().invoke(cli, ["run", "http://localhost:1", "--json", "--no-save", + "-u", "5", "--dir", _dir(tmp_path, SHOP, BROWSE)]) + assert res.exit_code == 0, res.output + result = json.loads(res.stdout) # nothing but the Result on stdout + assert "stale, not load-tested" in res.stderr + statuses = {j["name"]: j["status"] for j in result["journeys"]} + assert statuses == {"browse": "stale", "checkout": "ok"} + # nothing broke, so there's no culprit journey + assert result["verdict"]["onset_users"] is None + assert result["verdict"]["culprit_journey"] is None + assert result["config"]["mode"] == "journeys" + + +def test_run_applies_think_time_to_steps_without_their_own(tmp_path, monkeypatch): + seen = {} + _fake_checks(monkeypatch) + from prescale_cli.commands import run as run_cmd + inner = run_cmd.ramp + + async def spy(journey, url, **kwargs): + seen[journey.name] = kwargs.get("think") + return await inner(journey, url, **kwargs) + + monkeypatch.setattr(run_cmd, "ramp", spy) + res = CliRunner().invoke(cli, ["run", "http://localhost:1", "--json", "--no-save", + "-u", "5", "--think-time", "1.5", + "--dir", _dir(tmp_path, BROWSE)]) + assert res.exit_code == 0, res.output + assert seen == {"browse": 1.5}