mirror of
https://github.com/Stirling-Tools/Stirling-PDF.git
synced 2026-09-03 05:10:16 +03:00
Ludy87
444f6ea01f
Reorder and simplify imports in testing/cucumber feature files: remove noqa comments in environment.py, import parallel_support consistently, remove an unused os import in parallel_support.py, and consolidate/add parallel_support imports in step definition modules to satisfy linters and remove redundancy.
478 lines
17 KiB
Python
478 lines
17 KiB
Python
"""Re-issues a request concurrently and asserts every response matches the baseline."""
|
|
|
|
import io
|
|
import json as json_module
|
|
import re
|
|
import sys
|
|
import zipfile
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from hashlib import sha256
|
|
|
|
import requests
|
|
from pypdf import PdfReader
|
|
|
|
_UUID_RE = re.compile(r"[0-9a-fA-F]{8}-(?:[0-9a-fA-F]{4}-){3}[0-9a-fA-F]{12}")
|
|
_LONG_NUM_RE = re.compile(r"\d{10,}")
|
|
_DATE_RE = re.compile(r"\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:\.\d+)?Z?")
|
|
_VOLATILE_KEY_RE = re.compile(
|
|
r"^(id|uuid|.*Id|.*_id|.*[Tt]ime|.*[Dd]ate|timestamp|created.*|modified.*|"
|
|
r"updated.*|expires.*|token|traceId|requestId|duration.*|elapsed.*)$"
|
|
)
|
|
|
|
# Only hash text for reasonably sized documents so the check stays cheap.
|
|
_MAX_TEXT_PAGES = 25
|
|
|
|
# Sequential samples taken to separate inherent nondeterminism from a concurrency bug.
|
|
NOISE_PROBE_SAMPLES = 3
|
|
|
|
# Response size is allowed to drift this far before it counts as a difference.
|
|
SIZE_TOLERANCE = 0.05
|
|
|
|
# Collected per-scenario results, printed as a summary in after_all.
|
|
VALIDATIONS = []
|
|
|
|
|
|
# A spec replaces open file handles with bytes so it can be replayed from many threads.
|
|
|
|
|
|
def materialize(spec):
|
|
"""Turn a captured spec into a fresh `files=` list for one request."""
|
|
parts = []
|
|
for key, filename, payload, mime in spec:
|
|
if filename is None:
|
|
parts.append((key, (None, payload) if mime is None else (None, payload, mime)))
|
|
else:
|
|
parts.append((key, (filename, io.BytesIO(payload), mime)))
|
|
return parts
|
|
|
|
|
|
def send(url, spec, headers, timeout=300):
|
|
return requests.post(url, files=materialize(spec), headers=headers, timeout=timeout)
|
|
|
|
|
|
def _normalize_name(name):
|
|
name = _UUID_RE.sub("<uuid>", name)
|
|
return _LONG_NUM_RE.sub("<num>", name)
|
|
|
|
|
|
def _strip_volatile(value):
|
|
"""Drop keys whose values legitimately differ between two identical requests."""
|
|
if isinstance(value, dict):
|
|
return {
|
|
k: _strip_volatile(v)
|
|
for k, v in sorted(value.items())
|
|
if not _VOLATILE_KEY_RE.match(k)
|
|
}
|
|
if isinstance(value, list):
|
|
return [_strip_volatile(v) for v in value]
|
|
if isinstance(value, str):
|
|
return _normalize_name(value)
|
|
return value
|
|
|
|
|
|
def _pdf_fingerprint(body, prefix=""):
|
|
parts = {}
|
|
try:
|
|
reader = PdfReader(io.BytesIO(body))
|
|
except Exception as exc:
|
|
parts[prefix + "pdf"] = f"unreadable: {type(exc).__name__}"
|
|
return parts
|
|
parts[prefix + "encrypted"] = reader.is_encrypted
|
|
if reader.is_encrypted:
|
|
return parts
|
|
try:
|
|
pages = reader.pages
|
|
parts[prefix + "pages"] = len(pages)
|
|
except Exception as exc:
|
|
parts[prefix + "pdf"] = f"unreadable pages: {type(exc).__name__}"
|
|
return parts
|
|
if len(pages) <= _MAX_TEXT_PAGES:
|
|
try:
|
|
text = "\n".join((page.extract_text() or "") for page in pages)
|
|
parts[prefix + "text_sha"] = sha256(text.encode("utf-8")).hexdigest()[:16]
|
|
except Exception:
|
|
pass
|
|
return parts
|
|
|
|
|
|
def _entry_sha(entry):
|
|
"""Hash an archive entry, normalizing the ids and timestamps EPUB/ODF restamp each run."""
|
|
try:
|
|
text = entry.decode("utf-8")
|
|
if "\x00" not in text:
|
|
entry = _DATE_RE.sub("<date>", _normalize_name(text)).encode("utf-8")
|
|
except UnicodeDecodeError:
|
|
pass
|
|
return sha256(entry).hexdigest()[:16]
|
|
|
|
|
|
def _zip_fingerprint(body):
|
|
parts = {}
|
|
try:
|
|
with zipfile.ZipFile(io.BytesIO(body)) as archive:
|
|
names = sorted(_normalize_name(n) for n in archive.namelist())
|
|
parts["zip_entries"] = len(names)
|
|
parts["zip_names"] = names
|
|
for index, name in enumerate(sorted(archive.namelist())):
|
|
entry = archive.read(name)
|
|
if entry[:5] == b"%PDF-":
|
|
parts.update(_pdf_fingerprint(entry, prefix=f"zip[{index}]."))
|
|
else:
|
|
parts[f"zip[{index}].sha"] = _entry_sha(entry)
|
|
except Exception as exc:
|
|
parts["zip"] = f"unreadable: {type(exc).__name__}"
|
|
return parts
|
|
|
|
|
|
def fingerprint(response):
|
|
"""Structural signature of a response, ignoring benign per-request variance."""
|
|
body = response.content
|
|
content_type = (response.headers.get("Content-Type") or "").split(";")[0].strip()
|
|
parts = {"status": response.status_code, "content_type": content_type, "size": len(body)}
|
|
|
|
|
|
if "json" in content_type:
|
|
try:
|
|
parts["json"] = _strip_volatile(json_module.loads(body.decode("utf-8")))
|
|
except Exception:
|
|
parts["body_sha"] = sha256(body).hexdigest()[:16]
|
|
elif body[:5] == b"%PDF-":
|
|
parts.update(_pdf_fingerprint(body))
|
|
elif body[:2] == b"PK":
|
|
parts.update(_zip_fingerprint(body))
|
|
elif content_type.startswith("text/") or response.status_code >= 400:
|
|
try:
|
|
parts["text"] = _normalize_name(body.decode("utf-8", "replace"))[:2000]
|
|
except Exception:
|
|
parts["body_sha"] = sha256(body).hexdigest()[:16]
|
|
else:
|
|
parts["body_sha"] = sha256(body).hexdigest()[:16]
|
|
return parts
|
|
|
|
|
|
def differing_keys(baseline, other):
|
|
return {
|
|
key
|
|
for key in set(baseline) | set(other)
|
|
if key != "size" and baseline.get(key) != other.get(key)
|
|
}
|
|
|
|
|
|
def size_differs(baseline, other):
|
|
base_size, other_size = baseline.get("size", 0), other.get("size", 0)
|
|
return abs(other_size - base_size) > max(64, base_size * SIZE_TOLERANCE)
|
|
|
|
|
|
def compare(baseline, other, ignore=frozenset(), ignore_size=False):
|
|
"""Return a list of human-readable differences between two fingerprints."""
|
|
diffs = []
|
|
for key in sorted(differing_keys(baseline, other) - set(ignore)):
|
|
diffs.append(
|
|
f"{key}: baseline={_short(baseline.get(key))} parallel={_short(other.get(key))}"
|
|
)
|
|
if not ignore_size and size_differs(baseline, other):
|
|
diffs.append(
|
|
f"size: baseline={baseline.get('size', 0)} parallel={other.get('size', 0)} "
|
|
f"(differs by more than {SIZE_TOLERANCE:.0%})"
|
|
)
|
|
return diffs
|
|
|
|
|
|
def _short(value):
|
|
text = repr(value)
|
|
return text if len(text) <= 160 else text[:157] + "..."
|
|
|
|
|
|
def build_decoy_spec(spec):
|
|
"""Clone a spec with each PDF stamped with unique text, or None if not possible.
|
|
|
|
Structure is preserved so parameters stay valid; only the text differs, which
|
|
is what makes bleed visible.
|
|
"""
|
|
try:
|
|
from pypdf import PdfWriter
|
|
from reportlab.pdfgen import canvas
|
|
except ImportError:
|
|
return None
|
|
|
|
decoy = []
|
|
stamped_any = False
|
|
for key, filename, payload, mime in spec:
|
|
if filename is None or not isinstance(payload, bytes) or payload[:5] != b"%PDF-":
|
|
decoy.append((key, filename, payload, mime))
|
|
continue
|
|
try:
|
|
reader = PdfReader(io.BytesIO(payload))
|
|
if reader.is_encrypted:
|
|
return None
|
|
writer = PdfWriter()
|
|
for index, page in enumerate(reader.pages):
|
|
box = page.mediabox
|
|
width, height = float(box.width), float(box.height)
|
|
overlay_buffer = io.BytesIO()
|
|
overlay_canvas = canvas.Canvas(overlay_buffer, pagesize=(width, height))
|
|
overlay_canvas.drawString(
|
|
20, max(20.0, height - 20), f"DECOY-MARKER-{index}-do-not-mix"
|
|
)
|
|
overlay_canvas.showPage()
|
|
overlay_canvas.save()
|
|
overlay_buffer.seek(0)
|
|
page.merge_page(PdfReader(overlay_buffer).pages[0])
|
|
writer.add_page(page)
|
|
out = io.BytesIO()
|
|
writer.write(out)
|
|
decoy.append((key, filename, out.getvalue(), mime))
|
|
stamped_any = True
|
|
except Exception:
|
|
return None
|
|
return decoy if stamped_any else None
|
|
|
|
|
|
def _run_concurrently(url, specs, headers, timeout):
|
|
"""Fire every spec at once and return (response, error) in submission order."""
|
|
results = [None] * len(specs)
|
|
|
|
def _worker(index):
|
|
try:
|
|
results[index] = (send(url, specs[index], headers, timeout=timeout), None)
|
|
except Exception as exc:
|
|
results[index] = (None, exc)
|
|
|
|
with ThreadPoolExecutor(max_workers=len(specs)) as pool:
|
|
list(pool.map(_worker, range(len(specs))))
|
|
return results
|
|
|
|
|
|
def validate(context, url, spec, headers, baseline, label, timeout=300):
|
|
"""Re-issue the request concurrently; no-op unless the repeat count is above 1."""
|
|
repeat = getattr(context, "parallel_repeat", 1)
|
|
if repeat < 2 or getattr(context, "parallel_validated", False):
|
|
return
|
|
context.parallel_validated = True
|
|
context.parallel_ran_at = repeat
|
|
|
|
decoy_spec = build_decoy_spec(spec) if getattr(context, "parallel_decoy", False) else None
|
|
specs = [spec] * repeat + ([decoy_spec] * repeat if decoy_spec else [])
|
|
results = _run_concurrently(url, specs, headers, timeout)
|
|
|
|
main_results = results[:repeat]
|
|
decoy_results = results[repeat:]
|
|
baseline_fp = fingerprint(baseline)
|
|
|
|
noise, noisy_size = frozenset(), False
|
|
failures = _collect_failures(
|
|
main_results, decoy_results, baseline_fp, repeat, noise, noisy_size
|
|
)
|
|
if failures:
|
|
# Some endpoints are inherently nondeterministic (embedded ids, timestamps,
|
|
# deliberate randomness). Re-run sequentially to tell that apart from a real bug.
|
|
noise, noisy_size = _probe_noise(url, spec, headers, baseline_fp, timeout)
|
|
failures = _collect_failures(
|
|
main_results, decoy_results, baseline_fp, repeat, noise, noisy_size
|
|
)
|
|
|
|
VALIDATIONS.append(
|
|
{
|
|
"label": label,
|
|
"repeat": repeat,
|
|
"decoy": bool(decoy_spec),
|
|
"failed": bool(failures),
|
|
"noise": sorted(noise) + (["size"] if noisy_size else []),
|
|
}
|
|
)
|
|
|
|
if failures:
|
|
raise AssertionError(
|
|
f"Parallel consistency failed for {label} at concurrency {repeat}.\n"
|
|
f"Two sequential runs agreed on these fields, so the differences below are "
|
|
f"caused by running the same operation concurrently.\n"
|
|
f"Baseline fingerprint: {_short(baseline_fp)}\n - " + "\n - ".join(failures)
|
|
)
|
|
|
|
|
|
def _collect_failures(main_results, decoy_results, baseline_fp, repeat, noise, noisy_size):
|
|
decoy_fp = _decoy_reference(decoy_results, baseline_fp, noise, noisy_size)
|
|
failures = []
|
|
|
|
for index, (response, error) in enumerate(main_results):
|
|
if error is not None:
|
|
failures.append(f"copy {index + 1}/{repeat} raised {type(error).__name__}: {error}")
|
|
continue
|
|
actual_fp = fingerprint(response)
|
|
diffs = compare(baseline_fp, actual_fp, noise, noisy_size)
|
|
if not diffs:
|
|
continue
|
|
if decoy_fp is not None and not compare(
|
|
decoy_fp, actual_fp, noise, noisy_size
|
|
):
|
|
failures.append(
|
|
f"copy {index + 1}/{repeat} returned the CONCURRENT DECOY REQUEST'S response "
|
|
f"(cross-request bleed)"
|
|
)
|
|
else:
|
|
failures.append(f"copy {index + 1}/{repeat} diverged: " + "; ".join(diffs))
|
|
|
|
failures.extend(_check_decoys(decoy_results, repeat, noise, noisy_size))
|
|
return failures
|
|
|
|
|
|
def _probe_noise(url, spec, headers, baseline_fp, timeout, samples=NOISE_PROBE_SAMPLES):
|
|
"""Fields that already vary between uncontended runs, so they prove nothing.
|
|
|
|
Several samples: high-variance output can look stable across any single pair.
|
|
"""
|
|
probes = []
|
|
for _ in range(samples):
|
|
try:
|
|
probes.append(fingerprint(send(url, spec, headers, timeout=timeout)))
|
|
except Exception:
|
|
break
|
|
return _noise_from_samples(baseline_fp, probes)
|
|
|
|
|
|
def _noise_from_samples(baseline_fp, probes):
|
|
noise = set()
|
|
noisy_size = False
|
|
for index, probe in enumerate(probes):
|
|
for other in [baseline_fp] + probes[:index]:
|
|
noise |= differing_keys(other, probe)
|
|
noisy_size = noisy_size or size_differs(other, probe)
|
|
return frozenset(noise), noisy_size
|
|
|
|
|
|
def validate_get(context, url, params, headers, baseline, label, timeout=60):
|
|
"""Concurrency check for read-only GET endpoints."""
|
|
repeat = getattr(context, "parallel_repeat", 1)
|
|
if repeat < 2 or getattr(context, "parallel_validated", False):
|
|
return
|
|
context.parallel_validated = True
|
|
context.parallel_ran_at = repeat
|
|
|
|
results = [None] * repeat
|
|
|
|
def _worker(index):
|
|
try:
|
|
results[index] = (
|
|
requests.get(url, params=params, headers=headers, timeout=timeout),
|
|
None,
|
|
)
|
|
except Exception as exc:
|
|
results[index] = (None, exc)
|
|
|
|
with ThreadPoolExecutor(max_workers=repeat) as pool:
|
|
list(pool.map(_worker, range(repeat)))
|
|
|
|
baseline_fp = fingerprint(baseline)
|
|
|
|
def _failures(noise, noisy_size):
|
|
found = []
|
|
for index, (response, error) in enumerate(results):
|
|
if error is not None:
|
|
found.append(f"copy {index + 1}/{repeat} raised {type(error).__name__}: {error}")
|
|
continue
|
|
diffs = compare(
|
|
baseline_fp,
|
|
fingerprint(response),
|
|
noise,
|
|
noisy_size,
|
|
)
|
|
if diffs:
|
|
found.append(f"copy {index + 1}/{repeat} diverged: " + "; ".join(diffs))
|
|
return found
|
|
|
|
noise, noisy_size = frozenset(), False
|
|
failures = _failures(noise, noisy_size)
|
|
if failures:
|
|
probes = []
|
|
for _ in range(NOISE_PROBE_SAMPLES):
|
|
try:
|
|
probes.append(
|
|
fingerprint(
|
|
requests.get(url, params=params, headers=headers, timeout=timeout)
|
|
)
|
|
)
|
|
except Exception:
|
|
break
|
|
noise, noisy_size = _noise_from_samples(baseline_fp, probes)
|
|
failures = _failures(noise, noisy_size)
|
|
|
|
VALIDATIONS.append(
|
|
{
|
|
"label": label,
|
|
"repeat": repeat,
|
|
"decoy": False,
|
|
"failed": bool(failures),
|
|
"noise": sorted(noise) + (["size"] if noisy_size else []),
|
|
}
|
|
)
|
|
if failures:
|
|
raise AssertionError(
|
|
f"Parallel consistency failed for GET {label} at concurrency {repeat}.\n - "
|
|
+ "\n - ".join(failures)
|
|
)
|
|
|
|
|
|
def _decoy_reference(decoy_results, baseline_fp, noise, noisy_size):
|
|
"""Fingerprint of the decoy response, or None when it is not distinguishable."""
|
|
live = [r for r, _e in decoy_results if r is not None]
|
|
if not live:
|
|
return None
|
|
decoy_fp = fingerprint(live[0])
|
|
# Some endpoints ignore page content, so the decoy cannot prove anything there.
|
|
if not compare(baseline_fp, decoy_fp, noise, noisy_size):
|
|
return None
|
|
return decoy_fp
|
|
|
|
|
|
def _check_decoys(decoy_results, repeat, noise, noisy_size):
|
|
"""Assert the decoy load stayed self-consistent while contending with the main copies."""
|
|
if not decoy_results:
|
|
return []
|
|
live = [(i, r) for i, (r, _e) in enumerate(decoy_results) if r is not None]
|
|
if not live:
|
|
return ["every decoy request failed to complete"]
|
|
|
|
failures = []
|
|
reference_fp = fingerprint(live[0][1])
|
|
for index, response in live[1:]:
|
|
diffs = compare(
|
|
reference_fp,
|
|
fingerprint(response),
|
|
noise,
|
|
noisy_size,
|
|
)
|
|
if diffs:
|
|
failures.append(f"decoy {index + 1}/{repeat} diverged: " + "; ".join(diffs))
|
|
return failures
|
|
|
|
|
|
def print_summary():
|
|
if not VALIDATIONS:
|
|
return
|
|
total = len(VALIDATIONS)
|
|
failed = sum(1 for v in VALIDATIONS if v["failed"])
|
|
max_repeat = max(v["repeat"] for v in VALIDATIONS)
|
|
requests_sent = sum(v["repeat"] * (2 if v["decoy"] else 1) for v in VALIDATIONS)
|
|
|
|
lines = [
|
|
f"\n[PARALLEL] {total - failed}/{total} operations stayed consistent under "
|
|
f"concurrency (up to {max_repeat} at once, {requests_sent} concurrent requests sent)."
|
|
]
|
|
noisy = {}
|
|
for entry in VALIDATIONS:
|
|
if entry["noise"]:
|
|
noisy.setdefault(entry["label"], set()).update(entry["noise"])
|
|
if noisy:
|
|
lines.append(
|
|
f"[PARALLEL] {len(noisy)} endpoint(s) produce nondeterministic output. Those "
|
|
f"fields were excluded only after confirming they also vary across "
|
|
f"{NOISE_PROBE_SAMPLES} sequential runs:"
|
|
)
|
|
for label, fields in sorted(noisy.items()):
|
|
lines.append(f" {label} -> {', '.join(sorted(fields))}")
|
|
|
|
# behave's --junit reporter swallows after_all stdout, so bypass any capture.
|
|
stream = getattr(sys, "__stderr__", None) or sys.stderr
|
|
stream.write("\n".join(lines) + "\n")
|
|
stream.flush()
|