Assertions and Functional Correctness Under Load: Guided Hands-On Workflow

1. Safety envelope

2. Start the synthetic assertion fixture

Save as fixtures/assertion_fixture.py:

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse
from pathlib import Path
import argparse
import json
import threading
import time

lock = threading.Lock()
active = 0
max_active = 0
total = 0
by_path = {}
event_log = None

def write_event(event):
    if event_log is None:
        return
    with lock:
        with event_log.open("a", encoding="utf-8") as handle:
            handle.write(json.dumps(event, sort_keys=True) + "\n")

class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def _send(self, status, payload, *, content_type="application/json"):
        if isinstance(payload, (dict, list)):
            body = json.dumps(payload, sort_keys=True).encode("utf-8")
        elif isinstance(payload, str):
            body = payload.encode("utf-8")
        else:
            body = payload
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("X-Lab-Variant", urlparse(self.path).path)
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        global active, max_active, total
        path = urlparse(self.path).path

        if path == "/health":
            self._send(200, {"status": "ok"})
            return

        if path == "/stats":
            with lock:
                snapshot = {
                    "active": active,
                    "max_active": max_active,
                    "total": total,
                    "by_path": dict(by_path),
                }
            self._send(200, snapshot)
            return

        with lock:
            total += 1
            active += 1
            max_active = max(max_active, active)
            by_path[path] = by_path.get(path, 0) + 1
            request_no = total
            active_now = active

        started_ms = int(time.time() * 1000)
        status = 200
        try:
            if path == "/success":
                time.sleep(0.025)
                payload = {
                    "transport": "ok",
                    "business": "accepted",
                    "order_id": f"ORD-{request_no:05d}",
                    "amount": 42,
                    "items": ["synthetic-a", "synthetic-b"],
                }

            elif path == "/business-error":
                time.sleep(0.025)
                payload = {
                    "transport": "ok",
                    "business": "rejected",
                    "error": {"code": "SYNTHETIC_RULE", "message": "Synthetic business rejection"},
                    "order_id": None,
                }

            elif path == "/protocol-error":
                time.sleep(0.025)
                status = 503
                payload = {
                    "transport": "unavailable",
                    "business": "not_evaluated",
                    "error": {"code": "SYNTHETIC_503"},
                }

            elif path == "/invalid-json":
                time.sleep(0.025)
                payload = "this is intentionally not json"
                self._send(200, payload, content_type="text/plain")
                return

            elif path == "/slow-success":
                time.sleep(0.300)
                payload = {
                    "transport": "ok",
                    "business": "accepted",
                    "order_id": f"SLOW-{request_no:05d}",
                }

            elif path == "/small":
                time.sleep(0.025)
                payload = {"business": "accepted"}

            elif path == "/large":
                time.sleep(0.025)
                payload = {
                    "transport": "ok",
                    "business": "accepted",
                    "blob": "x" * 131072,
                    "marker": "ASSERTION-LAB-END",
                }

            else:
                status = 404
                payload = {"error": "not_found", "path": path}

            self._send(status, payload)

        finally:
            finished_ms = int(time.time() * 1000)
            write_event(
                {
                    "path": path,
                    "status": status,
                    "request_no": request_no,
                    "started_ms": started_ms,
                    "finished_ms": finished_ms,
                    "service_wall_ms": finished_ms - started_ms,
                    "active_at_start": active_now,
                }
            )
            with lock:
                active -= 1

    def log_message(self, format, *args):
        return

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--log", default="results/server-events.jsonl")
    args = parser.parse_args()
    event_log = Path(args.log).resolve()
    event_log.parent.mkdir(parents=True, exist_ok=True)
    event_log.write_text("", encoding="utf-8")
    print("fixture=http://127.0.0.1:8000")
    print(f"event_log={event_log}")
    ThreadingHTTPServer(("127.0.0.1", 8000), Handler).serve_forever()

Start and preflight:

python fixtures/assertion_fixture.py --log results/server-events.jsonl
curl --fail --silent http://127.0.0.1:8000/health
curl --fail --silent http://127.0.0.1:8000/success
curl --silent --output /dev/null --write-out "%{http_code}\n" http://127.0.0.1:8000/protocol-error

PowerShell:

(Invoke-WebRequest -UseBasicParsing http://127.0.0.1:8000/health).Content
(Invoke-WebRequest -UseBasicParsing http://127.0.0.1:8000/success).Content
try {
  Invoke-WebRequest -UseBasicParsing http://127.0.0.1:8000/protocol-error
} catch {
  $_.Exception.Response.StatusCode.value__
}

3. Save the JTL classifier

import csv
import math
import sys
from collections import Counter, defaultdict
from pathlib import Path

path = Path(sys.argv[1] if len(sys.argv) > 1 else "results/run/results.jtl")
rows = list(csv.DictReader(path.open(encoding="utf-8")))

if not rows:
    raise SystemExit("No sample rows found")

required = {"timeStamp", "elapsed", "label", "success", "responseCode", "responseMessage"}
missing = required.difference(rows[0])
if missing:
    raise SystemExit(f"Missing JTL columns: {sorted(missing)}")

def pct(values, p):
    values = sorted(values)
    rank = max(1, math.ceil((p / 100.0) * len(values)))
    return values[rank - 1]

def start_ms(row):
    return int(row["timeStamp"]) - int(row["elapsed"])

starts = [start_ms(r) for r in rows]
ends = [int(r["timeStamp"]) for r in rows]
span_s = max((max(ends) - min(starts)) / 1000.0, 0.001)
elapsed = [int(r["elapsed"]) for r in rows]
failures = [r for r in rows if r["success"].strip().lower() != "true"]

print(f"samples={len(rows)}")
print(f"failures={len(failures)}")
print(f"error_rate_pct={100.0 * len(failures) / len(rows):.2f}")
print(f"observed_sample_rps={len(rows) / span_s:.3f}")
print(f"p50_elapsed_ms={pct(elapsed, 50)}")
print(f"p95_elapsed_ms={pct(elapsed, 95)}")
print(f"response_codes={dict(Counter(r['responseCode'] for r in rows))}")

by_label = defaultdict(list)
for row in rows:
    by_label[row["label"]].append(row)

for label, items in sorted(by_label.items()):
    label_fail = sum(r["success"].strip().lower() != "true" for r in items)
    vals = [int(r["elapsed"]) for r in items]
    print(
        f"label={label!r} samples={len(items)} failures={label_fail} "
        f"p50_ms={pct(vals, 50)} p95_ms={pct(vals, 95)}"
    )

if failures:
    print("failure_details:")
    for row in failures[:20]:
        failure_message = row.get("failureMessage", "")
        print(
            f"  label={row['label']!r} code={row['responseCode']} "
            f"responseMessage={row['responseMessage']!r} "
            f"failureMessage={failure_message!r}"
        )

The script intentionally reports response code and assertion failureMessage separately. A business assertion can fail while HTTP response code remains 200.

4. Build the bounded debug tree

Test Plan
└── Thread Group — 1 user × 1 loop
    ├── HTTP Request Defaults — HttpClient4 / http / 127.0.0.1 / 8000
    ├── Success — GET /success
    ├── Business Error — GET /business-error
    ├── Protocol Error — GET /protocol-error
    ├── Invalid JSON — GET /invalid-json
    ├── Slow Success — GET /slow-success
    └── Small Payload — GET /small

Authoring listeners only:
    View Results Tree
    Assertion Results

Run this once before adding assertions. Expected raw state: Success, Business Error, Invalid JSON, Slow Success, and Small Payload are transport-successful HTTP 200 samples; Protocol Error is unsuccessful because it returns HTTP 503.

5. Add explicit transport assertion to Success

Under Success, add Response Assertion:

• Apply to: Main sample only
• Field to Test: Response Code
• Pattern rule: Equals
• Pattern: 200
• Ignore Status: off
• Custom failure message: expected HTTP 200 for Success

This does not add new knowledge to the successful case, but it makes the expected transport contract explicit.

6. Add JSON JMESPath business assertion

Under Success and Business Error, add JSON JMESPath Assertion:

• JMESPath: business
• Additionally assert value: enabled
• Expected value: accepted
• Match as regular expression: off

Expected: Success stays green. Business Error changes from transport-successful to failed even though its HTTP code remains 200. This is the central before/after demonstration.

7. Let JSON parsing become correctness evidence

Add the same JSON JMESPath Assertion to Invalid JSON. The endpoint remains HTTP 200, but JMeter cannot parse the body as JSON, so the assertion fails. This distinguishes “status 200” from “valid JSON contract.”

8. Classify Protocol Error without hiding it

Under Protocol Error, add a Response Assertion expecting code 200 with Ignore Status off. The HTTP sampler is already unsuccessful for 503; the assertion can add an explicit failure message, but it must not reclassify the unexpected 503 as a pass.

9. Add Duration Assertion to Slow Success

Under Slow Success, add Duration Assertion:

• Duration in Milliseconds: 200

The fixture deliberately takes ~300 ms. Expected: HTTP 200 plus correct business JSON, but the Duration Assertion marks the sample failed because it violates the synthetic response-time budget.

10. Add a Size Assertion where size is actually meaningful

Under Small Payload, add Size Assertion:

• Apply to: Main sample only
• Size in bytes: 40
• Comparison: Greater than

The tiny JSON body is deliberately below the threshold, so the assertion fails. The lesson is not that 40 bytes is a universal rule—it demonstrates that a byte-count invariant is coarse and must be attached only where it represents a real contract.

11. Create the CLI/load copy

Disable View Results Tree and Assertion Results. Use a short classification plan with each sampler executed once per loop and 1 thread × 3 loops. Preserve dashboard-compatible CSV fields explicitly:

mkdir -p results/classification
jmeter -n   -t plans/assertions-load.jmx   -l results/classification/results.jtl   -j results/classification/jmeter.log   -e -o results/classification/report   -Jjmeter.save.saveservice.print_field_names=true   -Jjmeter.save.saveservice.successful=true   -Jjmeter.save.saveservice.response_code=true   -Jjmeter.save.saveservice.response_message=true   -Jjmeter.save.saveservice.thread_counts=true   -Jjmeter.save.saveservice.assertion_results_failure_message=true
python tools/analyze_assertions.py results/classification/results.jtl

PowerShell uses jmeter.bat with the same options and backtick continuation. The report output directory must be empty/nonexistent before generation.

12. Verify dashboard error evidence

Open results/classification/report/index.html and inspect:

• Success versus failed-request percentage;
• Error table;
• Top 5 Errors by Sampler;
• Statistics table per label.

Cross-check dashboard counts against raw JTL. The dashboard is a derived view; raw JTL remains the source evidence for exact response code and assertion failure message.

13. Minimal assertion-overhead comparison

Create two plans against /success only, both 2 threads × 25 loops (50 local samples):

• Baseline: no assertions.
• Asserted: Response Code = 200 + JSON JMESPath business=accepted.

Run each into a fresh result directory, record whole-command wall time and Task Manager/top CPU/memory, then compare JTL:

import csv
import math
import sys
from pathlib import Path

def load(path):
    rows = list(csv.DictReader(Path(path).open(encoding="utf-8")))
    if not rows:
        raise SystemExit(f"No rows: {path}")
    starts = [int(r["timeStamp"]) - int(r["elapsed"]) for r in rows]
    ends = [int(r["timeStamp"]) for r in rows]
    span = max((max(ends) - min(starts)) / 1000.0, 0.001)
    elapsed = sorted(int(r["elapsed"]) for r in rows)
    failures = sum(r["success"].lower() != "true" for r in rows)
    def p(values, pct):
        rank = max(1, math.ceil((pct / 100) * len(values)))
        return values[rank - 1]
    return {
        "samples": len(rows),
        "failures": failures,
        "span_s": span,
        "rps": len(rows) / span,
        "p50_ms": p(elapsed, 50),
        "p95_ms": p(elapsed, 95),
    }

if len(sys.argv) != 3:
    raise SystemExit("usage: compare_runs.py BASELINE.jtl ASSERTED.jtl")

base = load(sys.argv[1])
asserted = load(sys.argv[2])

print(f"baseline={base}")
print(f"asserted={asserted}")
if base["rps"] > 0:
    print(f"throughput_delta_pct={100 * (asserted['rps'] - base['rps']) / base['rps']:.2f}")
print(
    "Interpretation: sampler elapsed primarily represents target/protocol time. "
    "Assertion CPU cost is generator-side post-sample work, so compare achieved throughput, "
    "whole-run duration, and generator CPU/GC as well as sampler percentiles."
)

Do not expect a dramatic difference from two narrow assertions on a tiny payload. If the delta is within run-to-run noise, report that honestly. The purpose is to establish the method: same target/workload, one assertion change, generator evidence, repeated runs if precision matters.

14. Controlled contrast: why giant-body regex is different

The fixture's /large endpoint returns about 128 KiB. In a one-thread × five-loop debug-only comparison, a body-wide regex must scan much more data than the narrow JSON JMESPath check on /success. Do not scale this variant. It exists only to show why assertion depth can become injector cost.

15. Challenge: which assertion belongs where?

A Submit operation returns HTTP 200, JSON {"business":"accepted","order_id":"ORD-123"}, and a 2 KB payload. Which checks are most meaningful?

Prefer transport status + narrow business invariant (for example business=accepted and perhaps an order ID existence/pattern when required). Do not add a 2 KB Size Assertion unless payload size itself is part of the contract.

Knowledge check

Official references and version notes

• Component Reference — current Response, Duration, Size, JSON, JSON JMESPath, JSR223, and other assertion semantics.
• Elements of a Test Plan — assertion scope and execution order after the sampler/Post-Processors and before listeners.
• Generating Dashboard Report — failed-request summary, error table, Top 5 Errors by Sampler, and required CSV fields.
• Properties Reference — result-save configuration, including assertion failure messages.
• Best Practices — CLI execution, lean listeners, and generator-validity guidance.
• Apache JMeter downloads — current stable release and Java requirement.