ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 07:09:11 +02:00
Code Issues Projects Releases Packages Wiki Activity

ci: speed up schema crash test (CSafeLoader + xdist-safe aggregation) (#3873)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-04-12 16:34:13 -04:00 • committed by GitHub
commit 7f80f78906
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 190 additions and 115 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/run-schema-crash-test.yml vendored
View file

@ -52,4 +52,4 @@ jobs:
env:
RUN_REAL_WORLD_SCHEMA_TEST: "1"
OPENAPI_DIRECTORY_PATH: /tmp/openapi-directory
run: uv run pytest tests/utilities/json_schema_type/test_real_world_schemas.py -m integration -v -s --timeout-method=thread
run: uv run pytest tests/utilities/json_schema_type/test_real_world_schemas.py -m integration -v -n auto --timeout-method=thread

105
tests/utilities/json_schema_type/conftest.py Normal file
View file

@ -0,0 +1,105 @@
"""Session hooks for the real-world schema crash test.
Per-provider tests in `test_real_world_schemas.py` run under `pytest-xdist` —
each worker is a separate process, so module-level accumulators don't survive
across workers. Each test persists its counts to `SCHEMA_CRASH_RESULTS_DIR`
and the session-finish hook below reads them back on the xdist master after
all workers finish.
These hooks must live in conftest.py (not the test module) because pytest
only picks up `pytest_configure` / `pytest_sessionfinish` from conftest files.
"""
from __future__ import annotations
import json
import os
import shutil
from pathlib import Path
import pytest
SCHEMA_CRASH_RESULTS_DIR = Path(
os.environ.get("SCHEMA_CRASH_RESULTS_DIR", "/tmp/schema_crash_results")
)
def pytest_configure(config: pytest.Config) -> None:
"""Clear prior per-provider results at session start (xdist master only)."""
if hasattr(config, "workerinput"):
return
shutil.rmtree(SCHEMA_CRASH_RESULTS_DIR, ignore_errors=True)
SCHEMA_CRASH_RESULTS_DIR.mkdir(parents=True, exist_ok=True)
def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
"""Aggregate per-provider results once all xdist workers finish."""
if hasattr(session.config, "workerinput"):
return
if not SCHEMA_CRASH_RESULTS_DIR.exists():
return
files = list(SCHEMA_CRASH_RESULTS_DIR.glob("*.json"))
if not files:
return
keys = ("schemas", "type_errors", "schema_errors", "timeouts", "other_errors")
totals = dict.fromkeys(keys, 0)
for f in files:
data = json.loads(f.read_text())
for k in keys:
totals[k] += data.get(k, 0)
crashes = (
totals["type_errors"]
+ totals["schema_errors"]
+ totals["timeouts"]
+ totals["other_errors"]
)
print(f"\n{'=' * 60}")
print("Real-world schema crash test — aggregate results")
print(f"{'=' * 60}")
print(f"Providers tested: {len(files):,}")
print(f"Schemas tested: {totals['schemas']:,}")
print(f"TypeErrors: {totals['type_errors']:,}")
print(f"SchemaErrors: {totals['schema_errors']:,}")
print(f"Timeouts: {totals['timeouts']:,}")
print(f"Other errors: {totals['other_errors']:,}")
print(
f"Total crashes: {crashes:,} ({crashes / max(totals['schemas'], 1) * 100:.2f}%)"
)
# Snapshot baselines (captured 2026-04-10, openapi-directory@f7207cf0,
# origin/main, with JSON round-trip to strip YAML artifacts).
MAX_TYPE_ERRORS = 420 # was 388 — real json_schema_to_type bugs
MAX_SCHEMA_ERRORS = 300 # was 277 — Pydantic regex rejections (not our code)
MAX_TIMEOUTS = 5 # was 0
MAX_OTHER_ERRORS = 50 # was 0
failures: list[str] = []
if totals["schemas"] <= 200_000:
failures.append(
f"Expected >200k schemas but only found {totals['schemas']}. "
f"Is the openapi-directory checkout correct?"
)
if totals["type_errors"] > MAX_TYPE_ERRORS:
failures.append(
f"TypeErrors regressed: {totals['type_errors']} > {MAX_TYPE_ERRORS}"
)
if totals["schema_errors"] > MAX_SCHEMA_ERRORS:
failures.append(
f"SchemaErrors regressed: {totals['schema_errors']} > {MAX_SCHEMA_ERRORS}"
)
if totals["timeouts"] > MAX_TIMEOUTS:
failures.append(f"Timeouts regressed: {totals['timeouts']} > {MAX_TIMEOUTS}")
if totals["other_errors"] > MAX_OTHER_ERRORS:
failures.append(
f"Other errors regressed: {totals['other_errors']} > {MAX_OTHER_ERRORS}"
)
if failures:
print("\nBASELINE VIOLATIONS:")
for msg in failures:
print(f" - {msg}")
# Force a non-zero exit even though all individual tests passed.
session.exitstatus = pytest.ExitCode.TESTS_FAILED

198
tests/utilities/json_schema_type/test_real_world_schemas.py
View file

@ -14,14 +14,16 @@ from __future__ import annotations
import json
import os
import shutil
import signal
import subprocess
from dataclasses import dataclass
from dataclasses import asdict, dataclass
from pathlib import Path
import pytest
import yaml
from pydantic import TypeAdapter
from yaml import CSafeLoader # ty: ignore[possibly-missing-import]
from fastmcp.utilities.json_schema_type import json_schema_to_type
@ -75,58 +77,89 @@ def _is_openapi_directory_clone(path: Path) -> bool:
def _ensure_repo() -> Path:
"""Clone the openapi-directory repo if not already present at the pinned commit."""
if CLONE_DIR.exists() and (CLONE_DIR / ".git").is_dir():
result = subprocess.run(
["git", "-C", str(CLONE_DIR), "rev-parse", "HEAD"],
capture_output=True,
text=True,
)
if result.stdout.strip() == OPENAPI_DIRECTORY_COMMIT:
return CLONE_DIR
"""Clone the openapi-directory repo if not already present at the pinned commit.
if CLONE_DIR.exists():
if not _is_openapi_directory_clone(CLONE_DIR):
raise RuntimeError(
f"{CLONE_DIR} exists but is not an openapi-directory clone. "
f"Remove it manually or set OPENAPI_DIRECTORY_PATH to a different path."
Safe under pytest-xdist: a file lock serializes the rmtree/reclone path so
concurrent workers don't race on the shared CLONE_DIR.
"""
# Fast-path check without a lock — if HEAD already matches, skip locking entirely.
if _head_matches_pinned():
return CLONE_DIR
# fcntl is POSIX-only; imported here (not at module level) so this file
# collects cleanly on Windows where the integration test is skipped.
import fcntl
lock_path = CLONE_DIR.with_suffix(".lock")
lock_path.parent.mkdir(parents=True, exist_ok=True)
with open(lock_path, "w") as lf:
fcntl.flock(lf.fileno(), fcntl.LOCK_EX)
try:
# Re-check under the lock in case another worker already fixed it.
if _head_matches_pinned():
return CLONE_DIR
if CLONE_DIR.exists():
if not _is_openapi_directory_clone(CLONE_DIR):
raise RuntimeError(
f"{CLONE_DIR} exists but is not an openapi-directory clone. "
f"Remove it manually or set OPENAPI_DIRECTORY_PATH to a different path."
)
shutil.rmtree(CLONE_DIR)
subprocess.run(
[
"git",
"clone",
"--depth",
"1",
OPENAPI_DIRECTORY_REPO,
str(CLONE_DIR),
],
check=True,
capture_output=True,
)
import shutil
subprocess.run(
[
"git",
"-C",
str(CLONE_DIR),
"fetch",
"--depth",
"1",
"origin",
OPENAPI_DIRECTORY_COMMIT,
],
check=True,
capture_output=True,
)
subprocess.run(
["git", "-C", str(CLONE_DIR), "checkout", OPENAPI_DIRECTORY_COMMIT],
check=True,
capture_output=True,
)
return CLONE_DIR
finally:
fcntl.flock(lf.fileno(), fcntl.LOCK_UN)
shutil.rmtree(CLONE_DIR)
subprocess.run(
["git", "clone", "--depth", "1", OPENAPI_DIRECTORY_REPO, str(CLONE_DIR)],
check=True,
def _head_matches_pinned() -> bool:
"""Return True when CLONE_DIR is already checked out at the pinned commit."""
if not (CLONE_DIR.exists() and (CLONE_DIR / ".git").is_dir()):
return False
result = subprocess.run(
["git", "-C", str(CLONE_DIR), "rev-parse", "HEAD"],
capture_output=True,
text=True,
)
subprocess.run(
[
"git",
"-C",
str(CLONE_DIR),
"fetch",
"--depth",
"1",
"origin",
OPENAPI_DIRECTORY_COMMIT,
],
check=True,
capture_output=True,
)
subprocess.run(
["git", "-C", str(CLONE_DIR), "checkout", OPENAPI_DIRECTORY_COMMIT],
check=True,
capture_output=True,
)
return CLONE_DIR
return result.stdout.strip() == OPENAPI_DIRECTORY_COMMIT
def _load_spec(spec_file: Path) -> dict | None:
"""Load and return a spec dict, or None on failure."""
try:
if spec_file.suffix == ".yaml":
spec = yaml.safe_load(spec_file.read_text())
spec = yaml.load(spec_file.read_text(), Loader=CSafeLoader)
else:
spec = json.loads(spec_file.read_text())
return spec if isinstance(spec, dict) else None
@ -284,8 +317,14 @@ def _providers_with_timeouts() -> list: # list of pytest.param
return params
# Accumulator: per-provider tests store results here, final test reads them.
_results: dict[str, ProviderResult] = {}
# Per-provider results are persisted to disk so they survive across xdist
# workers (each worker runs in its own process). Aggregation and baseline
# assertions happen in the pytest_sessionfinish hook in conftest.py.
# This path is duplicated in conftest.py by design — tests shouldn't import
# from conftest, which isn't a reliably importable module.
_RESULTS_DIR = Path(
os.environ.get("SCHEMA_CRASH_RESULTS_DIR", "/tmp/schema_crash_results")
)
@pytest.mark.integration
@ -293,78 +332,9 @@ _results: dict[str, ProviderResult] = {}
def test_provider_schemas(provider: str):
"""json_schema_to_type should not infinite-loop on schemas from this provider."""
_ensure_repo()
_RESULTS_DIR.mkdir(parents=True, exist_ok=True)
result = _test_provider(provider)
_results[provider] = result
(_RESULTS_DIR / f"{provider}.json").write_text(json.dumps(asdict(result)))
assert result.timeouts == 0, (
f"{provider}: {result.timeouts} schema(s) timed out (possible infinite loop)"
)
# ── Aggregate baseline test ──────────────────────────────────────────
#
# Runs last (sorted after test_provider_schemas by name).
# Reads accumulated _results instead of re-running all providers.
@pytest.mark.integration
@pytest.mark.timeout(30, method="thread")
def test_z_aggregate_crash_rate():
"""Aggregate crash-rate baseline across all providers.
Asserts that total crash counts haven't regressed beyond known baselines.
As we fix crash patterns, ratchet the baselines down.
Named test_z_* so pytest runs it after all test_provider_schemas.
"""
if not _results:
pytest.skip("No provider results collected — run with -m integration")
total = ProviderResult()
for r in _results.values():
total.schemas += r.schemas
total.type_errors += r.type_errors
total.schema_errors += r.schema_errors
total.timeouts += r.timeouts
total.other_errors += r.other_errors
crashes = (
total.type_errors + total.schema_errors + total.timeouts + total.other_errors
)
print(f"\n{'=' * 60}")
print("Real-world schema crash test — aggregate results")
print(f"{'=' * 60}")
print(f"Providers tested: {len(_results):,}")
print(f"Schemas tested: {total.schemas:,}")
print(f"TypeErrors: {total.type_errors:,}")
print(f"SchemaErrors: {total.schema_errors:,}")
print(f"Timeouts: {total.timeouts:,}")
print(f"Other errors: {total.other_errors:,}")
print(
f"Total crashes: {crashes:,} ({crashes / max(total.schemas, 1) * 100:.2f}%)"
)
assert total.schemas > 200_000, (
f"Expected >200k schemas but only found {total.schemas}. "
f"Is the openapi-directory checkout correct?"
)
# Snapshot baselines (captured 2026-04-10, openapi-directory@f7207cf0,
# origin/main, with JSON round-trip to strip YAML artifacts).
MAX_TYPE_ERRORS = 420 # was 388 — real json_schema_to_type bugs
MAX_SCHEMA_ERRORS = 300 # was 277 — Pydantic regex rejections (not our code)
MAX_TIMEOUTS = 5 # was 0
MAX_OTHER_ERRORS = 50 # was 0
assert total.type_errors <= MAX_TYPE_ERRORS, (
f"TypeErrors regressed: {total.type_errors} > {MAX_TYPE_ERRORS}"
)
assert total.schema_errors <= MAX_SCHEMA_ERRORS, (
f"SchemaErrors regressed: {total.schema_errors} > {MAX_SCHEMA_ERRORS}"
)
assert total.timeouts <= MAX_TIMEOUTS, (
f"Timeouts regressed: {total.timeouts} > {MAX_TIMEOUTS}"
)
assert total.other_errors <= MAX_OTHER_ERRORS, (
f"Other errors regressed: {total.other_errors} > {MAX_OTHER_ERRORS}"
)