diff --git a/src/adcp/validation/schema_loader.py b/src/adcp/validation/schema_loader.py index 2a0e02570..be607edd8 100644 --- a/src/adcp/validation/schema_loader.py +++ b/src/adcp/validation/schema_loader.py @@ -1,7 +1,7 @@ """JSON Schema loader for AdCP tool request/response validation. Loads the bundled per-tool schemas shipped with the SDK plus the ``core/`` -schemas that async response variants ``$ref``, then compiles validators +schemas that async response variants ``$ref``, then prepares validator specs lazily by ``(tool_name, direction, bundle_key)``. Schemas live under a per-version bundle key (see @@ -9,7 +9,7 @@ versions can coexist. Callers pass an optional ``version`` to :func:`get_validator`; ``None`` defaults to the SDK's compile-time pin (``ADCP_VERSION``). Each bundle key gets its own ``_LoaderState`` — file -index, compiled validators, core registry — so cross-version traffic +index, validator specs, core registry — so cross-version traffic doesn't share compilation state. Discovery paths (first hit wins, per bundle key): @@ -31,11 +31,12 @@ import threading import warnings from copy import deepcopy +from dataclasses import dataclass from datetime import date from importlib.resources import as_file, files from pathlib import Path from typing import Any, Literal, cast -from urllib.parse import unquote, urlparse +from urllib.parse import unquote, urldefrag, urljoin, urlparse, urlsplit from pydantic import AnyUrl, TypeAdapter, ValidationError @@ -52,12 +53,14 @@ def _uri_adapter() -> TypeAdapter[AnyUrl]: return TypeAdapter(AnyUrl) -# Serialize first-time init and validator compilation. Concurrent callers -# on a fresh process can otherwise both walk the schema tree or compile -# the same validator twice. Result is the same either way, but the lock +# Serialize first-time init and validator-spec preparation. Concurrent callers +# on a fresh process can otherwise both walk the schema tree or prepare +# the same spec twice. Result is the same either way, but the lock # keeps behaviour deterministic and avoids redundant filesystem walks. _init_lock = threading.Lock() _compile_lock = threading.Lock() +_resolver_import_lock = threading.Lock() +_ref_resolver_type: Any = None ResponseVariant = Literal["sync", "submitted", "working", "input-required"] Direction = Literal["request", "sync", "submitted", "working", "input-required"] @@ -161,6 +164,28 @@ def _build_format_checker() -> Any: return checker +@dataclass(frozen=True) +class _ValidatorSpec: + """Cached schema inputs; each caller gets its own resolver scope stack.""" + + schema: dict[str, Any] + base_uri: str + store: dict[str, dict[str, Any]] + format_checker: Any + bundle_key: str | None = None + + def make(self) -> Any: + from jsonschema import Draft7Validator + + return Draft7Validator( + self.schema, + resolver=_ref_resolver_from_store( + self.base_uri, self.schema, self.store, bundle_key=self.bundle_key + ), + format_checker=self.format_checker, + ) + + class _SchemaRoot: """Filesystem view of the schema tree, regardless of packaged vs dev.""" @@ -220,8 +245,8 @@ def __init__(self, root: _SchemaRoot, bundle_key: str) -> None: self.file_index: dict[tuple[str, Direction], Path] = {} self.source_index: dict[tuple[str, Direction], Path] = {} self.mcp_index: dict[tuple[str, Direction], Path] = {} - self.compiled: dict[tuple[str, Direction], Any] = {} - self.named_compiled: dict[str, Any] = {} + self.compiled: dict[tuple[str, Direction], _ValidatorSpec] = {} + self.named_compiled: dict[str, _ValidatorSpec] = {} self.portable: dict[tuple[str, Direction], dict[str, Any]] = {} # Serialized JSON keeps the immutable cached value private and makes # every returned tree independent, including any repeated branches. @@ -441,12 +466,46 @@ def _load_schema_registry(state: _LoaderState) -> None: def _make_ref_resolver(state: _LoaderState, base_file: Path, schema: dict[str, Any]) -> Any: + """Build a task resolver with the version-scoped local schema registry.""" + _load_schema_registry(state) + return _ref_resolver_from_store( + base_file.resolve().as_uri(), schema, state.registry, bundle_key=state.bundle_key + ) + + +def _get_ref_resolver_type() -> Any: + """Resolve the deprecated class once, without racing warning filters.""" + global _ref_resolver_type + if _ref_resolver_type is not None: + return _ref_resolver_type + with _resolver_import_lock: + if _ref_resolver_type is None: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + try: + from jsonschema import RefResolver + except ImportError as exc: # pragma: no cover - guarded by dep install + raise RuntimeError( + "jsonschema is required for AdCP schema validation. " + "Install with: pip install 'jsonschema>=4.20.0'" + ) from exc + _ref_resolver_type = RefResolver + return _ref_resolver_type + + +def _ref_resolver_from_store( + base_uri: str, + schema: dict[str, Any], + store: dict[str, dict[str, Any]], + *, + bundle_key: str | None = None, +) -> Any: """Build a jsonschema ``RefResolver`` rooted at the file's directory. Async variant schemas use relative refs like ``../core/context.json``; giving the resolver a ``file://`` base URI lets those resolve against - disk. Also seeds the core ``$id``-keyed registry so bundled schemas - that reference a core type by canonical id still resolve. + disk. Uses the cached local schema store so bundled schemas that + reference a core type by canonical id still resolve. Sets ``referrer=schema`` (not ``{}``) so fragment-only refs like ``#/$defs/MediaChannel`` inside the bundled per-tool schema resolve @@ -457,32 +516,99 @@ def _make_ref_resolver(state: _LoaderState, base_file: Path, schema: dict[str, A capabilities-style schema does). ``RefResolver`` is deprecated in jsonschema 4.18+ (to be replaced by - the ``referencing`` library). Suppress the warning locally so + the ``referencing`` library). Suppress the warning on first import so downstream projects running ``-W error::DeprecationWarning`` don't - crash on import; migration tracked as a follow-up. + crash on import. Cached lookups do not change process warning filters; + migration tracked as a follow-up. """ - with warnings.catch_warnings(): - warnings.simplefilter("ignore", DeprecationWarning) - try: - from jsonschema import RefResolver - except ImportError as exc: # pragma: no cover - guarded by dep install - raise RuntimeError( - "jsonschema is required for AdCP schema validation. " - "Install with: pip install 'jsonschema>=4.20.0'" - ) from exc - - _load_schema_registry(state) - base_uri = base_file.resolve().as_uri() - - def missing_local_reference(uri: str) -> Any: - raise ValueError(f"schema reference is not in bundle {state.bundle_key}: {uri}") - - return RefResolver( - base_uri=base_uri, - referrer=schema, - store=dict(state.registry), - handlers={"http": missing_local_reference, "https": missing_local_reference}, - ) + resolver_type = _get_ref_resolver_type() + + def missing_local_reference(uri: str) -> Any: + raise ValueError(f"schema reference is not in bundle {bundle_key}: {uri}") + + return resolver_type( + base_uri=base_uri, + referrer=schema, + store=dict(store), + handlers=( + {"http": missing_local_reference, "https": missing_local_reference} + if bundle_key is not None + else {} + ), + ) + + +def _has_external_refs(value: Any) -> bool: + """Whether a schema needs documents beyond its own fragment references.""" + if isinstance(value, dict): + reference = value.get("$ref") + if isinstance(reference, str) and not reference.startswith("#"): + return True + return any(_has_external_refs(child) for child in value.values()) + if isinstance(value, list): + return any(_has_external_refs(child) for child in value) + return False + + +def _reachable_registry_store( + registry: dict[str, dict[str, Any]], base_uri: str, schema: dict[str, Any] | bool +) -> dict[str, dict[str, Any]]: + """Keep every alias of documents reached through absolute references. + + Relative references and nested IDs depend on resolution scopes, so leave + their registry intact. Unknown targets also retain the original store and + fail at validation through the existing offline resolver handlers. + """ + if not isinstance(schema, dict): + return dict(registry) + # Match RefResolver's normalized alias pass, document-ID pass, and final + # referrer override. Filtering below preserves that same insertion order. + lookup = {urlsplit(uri).geturl(): document for uri, document in registry.items()} + for document in registry.values(): + identifier = document.get("$id") + if isinstance(identifier, str): + lookup[urlsplit(identifier).geturl()] = document + lookup[urlsplit(base_uri).geturl()] = schema + + selected: set[int] = set() + pending = [schema] + while pending: + document = pending.pop() + if id(document) in selected: + continue + selected.add(id(document)) + identifier = document.get("$id") + if isinstance(identifier, str): + parsed_id = urlsplit(identifier) + if not parsed_id.scheme or not parsed_id.netloc or parsed_id.fragment: + return dict(registry) + # Retain an ID's winning document even if this document was + # reached through another alias, including on fragment-only refs. + target = lookup.get(parsed_id.geturl()) + if target is not None: + pending.append(target) + + values: list[Any] = [document] + while values: + value = values.pop() + if isinstance(value, dict): + if value is not document and "$id" in value: + return dict(registry) + reference = value.get("$ref") + if isinstance(reference, str) and reference and not reference.startswith("#"): + parsed = urlsplit(reference) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + return dict(registry) + uri, _ = urldefrag(reference.rstrip("/")) + target = lookup.get(urlsplit(uri).geturl()) + if target is None: + return dict(registry) + pending.append(target) + values.extend(value.values()) + elif isinstance(value, list): + values.extend(value) + + return {uri: document for uri, document in registry.items() if id(document) in selected} def _bundle_relative_path(state: _LoaderState, path: str) -> Path: @@ -633,7 +759,11 @@ def get_validator( *, version: str | None = None, ) -> Any | None: - """Return a compiled validator for ``(tool_name, direction, version)``. + """Return a fresh validator for ``(tool_name, direction, version)``. + + Schema loading and reference discovery are cached. Each lookup creates + an independent resolver so callers can validate concurrently. Obtain a + validator per caller rather than sharing one instance between threads. Returns ``None`` when no schema ships for this pair — callers should skip validation (e.g., custom tools outside the AdCP catalog, or @@ -652,7 +782,7 @@ def get_validator( key = (tool_name, direction) cached = state.compiled.get(key) if cached is not None: - return cached + return cached.make() file = state.file_index.get(key) if file is None: return None @@ -666,7 +796,6 @@ def get_validator( schema = _effective_task_schema(schema, tool_name, direction, bundle_key=state.bundle_key) try: - from jsonschema import Draft7Validator from jsonschema.exceptions import SchemaError except ImportError as exc: # pragma: no cover raise RuntimeError( @@ -675,22 +804,42 @@ def get_validator( ) from exc with _compile_lock: - # Re-check: another thread may have compiled the validator for + # Re-check: another thread may have prepared the validator spec for # this key while we were loading the schema off disk. cached = state.compiled.get(key) if cached is not None: - return cached + return cached.make() try: - resolver = _make_ref_resolver(state, file, schema) - validator = Draft7Validator( - schema, - resolver=resolver, + _load_schema_registry(state) + base_uri = file.resolve().as_uri() + schema_id = schema.get("$id") if isinstance(schema, dict) else None + # Flattened bundles have no nested $id scopes after normalization. + # Their fragment refs only need the root document. Avoid copying + # the entire modular registry into every fresh resolver. + if ( + file.is_relative_to(state.root.bundled) + and isinstance(schema, dict) + and not _has_external_refs(schema) + and not (isinstance(schema_id, str) and urlsplit(schema_id).fragment) + ): + store = {base_uri: schema} + if isinstance(schema_id, str): + store[schema_id] = schema + store[urljoin(base_uri, schema_id)] = schema + else: + store = _reachable_registry_store(state.registry, base_uri, schema) + spec = _ValidatorSpec( + schema=schema, + base_uri=base_uri, + store=store, format_checker=_build_format_checker(), + bundle_key=state.bundle_key, ) + validator = spec.make() except SchemaError as exc: logger.warning("Invalid schema %s for %s: %s", file, key, exc) return None - state.compiled[key] = validator + state.compiled[key] = spec return validator @@ -699,7 +848,11 @@ def get_named_validator( *, version: str | None = None, ) -> Any | None: - """Return an offline validator for a non-task schema in the bundle. + """Return a fresh offline validator for a non-task schema in the bundle. + + Schema loading and reference discovery are cached. Each lookup creates + an independent resolver so callers can validate concurrently. Obtain a + validator per caller rather than sharing one instance between threads. Task validation normally goes through :func:`get_validator`. Some SDK helpers also consume standalone protocol documents (for example a @@ -721,7 +874,7 @@ def get_named_validator( cache_key = path.as_posix() cached = state.named_compiled.get(cache_key) if cached is not None: - return cached + return cached.make() file = state.root.root.joinpath(*path.parts) try: if not file.is_file() or not file.resolve().is_relative_to(state.root.root.resolve()): @@ -733,7 +886,6 @@ def get_named_validator( return None try: - from jsonschema import Draft7Validator from jsonschema.exceptions import SchemaError except ImportError as exc: # pragma: no cover raise RuntimeError( @@ -744,25 +896,18 @@ def get_named_validator( with _compile_lock: cached = state.named_compiled.get(cache_key) if cached is not None: - return cached + return cached.make() try: - with warnings.catch_warnings(): - warnings.simplefilter("ignore", DeprecationWarning) - from jsonschema import RefResolver - - resolver = RefResolver( - base_uri=file.resolve().as_uri(), - referrer=schema, - store=_reachable_schema_store(state, file, schema), - ) - validator = Draft7Validator( - schema, - resolver=resolver, + spec = _ValidatorSpec( + schema=schema, + base_uri=file.resolve().as_uri(), + store=_reachable_schema_store(state, file, schema), format_checker=_build_format_checker(), ) + validator = spec.make() except (OSError, json.JSONDecodeError, SchemaError, ValueError): return None - state.named_compiled[cache_key] = validator + state.named_compiled[cache_key] = spec return validator diff --git a/tests/test_rc7_version_bridge.py b/tests/test_rc7_version_bridge.py index 540659acf..510a84dcc 100644 --- a/tests/test_rc7_version_bridge.py +++ b/tests/test_rc7_version_bridge.py @@ -34,12 +34,11 @@ def test_rc7_wire_claim_resolves_to_stable(version: str) -> None: def test_alias_is_advertised_and_has_stable_validators() -> None: assert "3.2-rc.7" in get_supported_adcp_versions() assert "3.2-rc.7" in capabilities_response(["media_buy"])["adcp"]["supported_versions"] - assert get_validator("get_products", "request", version="3.2-rc.7") is get_validator( - "get_products", "request", version="3.2" - ) - assert get_validator("get_products", "sync", version="3.2-rc.7") is get_validator( - "get_products", "sync", version="3.2" - ) + for direction in ("request", "sync"): + aliased = get_validator("get_products", direction, version="3.2-rc.7") + stable = get_validator("get_products", direction, version="3.2") + assert aliased is not None and stable is not None + assert aliased.schema == stable.schema assert get_bundle_adcp_version(version="3.2-rc.7") == "3.2.1" assert get_bundle_adcp_version(version="3.2.0-rc.7") == "3.2.0-rc.7" diff --git a/tests/test_schema_loader_per_version.py b/tests/test_schema_loader_per_version.py index c04bbcd6a..8bc0362ef 100644 --- a/tests/test_schema_loader_per_version.py +++ b/tests/test_schema_loader_per_version.py @@ -161,6 +161,37 @@ def test_root_relative_legacy_refs_resolve_from_offline_registry( assert not invalid.valid +def test_bundled_root_id_fragment_retains_the_defragmented_registry_document( + synthetic_legacy_bundle: tuple[str, Path], +) -> None: + version, root = synthetic_legacy_bundle + identifier = "https://adcontextprotocol.org/schemas/2.5/core/fragment-root.json" + schema = { + "$id": f"{identifier}#root", + "properties": {"item": {"$ref": "#/$defs/Value"}}, + "$defs": {"Value": {"type": "integer"}}, + } + (root / "bundled" / "synthetic-tool-request.json").write_text(json.dumps(schema)) + (root / "core" / "fragment-root.json").write_text( + json.dumps({"$id": identifier, "$defs": {"Value": {"type": "string"}}}) + ) + validator = get_validator("synthetic_tool", "request", version=version) + assert validator is not None + assert validator.is_valid({"item": "valid"}) + assert not validator.is_valid({"item": 1}) + + +@pytest.mark.parametrize("schema", [True, False]) +def test_boolean_task_schema_retains_its_validation_result( + synthetic_legacy_bundle: tuple[str, Path], schema: bool +) -> None: + version, root = synthetic_legacy_bundle + (root / "bundled" / "synthetic-tool-request.json").write_text(json.dumps(schema)) + validator = get_validator("synthetic_tool", "request", version=version) + assert validator is not None + assert validator.is_valid({}) is schema + + def test_schedule_correction_does_not_change_a_different_version( synthetic_legacy_bundle: tuple[str, Path], ) -> None: @@ -374,7 +405,8 @@ def test_default_version_unchanged_when_arg_omitted( # Pick a tool that ships in the SDK pin. explicit_default = get_validator("get_products", "request", version=None) omitted = get_validator("get_products", "request") - assert explicit_default is omitted + assert explicit_default is not None and omitted is not None + assert explicit_default.schema == omitted.schema def _write_exact_release_bundle(root: Path, declared: str, referenced: str) -> None: diff --git a/tests/test_schema_loader_reachable_registry.py b/tests/test_schema_loader_reachable_registry.py new file mode 100644 index 000000000..08a319bfc --- /dev/null +++ b/tests/test_schema_loader_reachable_registry.py @@ -0,0 +1,123 @@ +"""Compact task stores preserve the full registry's reference behavior.""" + +from __future__ import annotations + +from typing import Any + +import pytest +from jsonschema import Draft7Validator + +from adcp.validation import schema_loader + +BASE = "file:///bundle/request.json" +ROOT = "https://example.test/request.json" +CHILD = "https://example.test/child.json" +LEAF = "https://example.test/leaf.json" + + +def _validator(schema: dict[str, Any], store: dict[str, dict[str, Any]]) -> Draft7Validator: + return Draft7Validator( + schema, + resolver=schema_loader._ref_resolver_from_store(BASE, schema, store, bundle_key="test"), + ) + + +def test_external_fragments_keep_local_and_transitive_refs_and_all_aliases() -> None: + schema = { + "$id": ROOT, + "properties": {"item": {"$ref": f"{CHILD}#/$defs/Entry"}}, + "required": ["item"], + } + child = { + "$id": "https://example.test/canonical-child.json", + "$defs": { + "Entry": { + "properties": {"value": {"$ref": "#/$defs/Leaf"}}, + "required": ["value"], + }, + "Leaf": {"$ref": LEAF}, + "Cycle": {"$ref": ROOT}, + }, + } + leaf = {"$id": LEAF, "type": "integer", "minimum": 1} + unrelated = {"type": "boolean"} + registry = { + ROOT: schema, + CHILD: child, + "HTTPS://example.test/child.json#": child, + "file:///bundle/child.json": child, + LEAF: leaf, + "https://example.test/unrelated.json": unrelated, + } + compact = schema_loader._reachable_registry_store(registry, BASE, schema) + assert list(compact) == list(registry)[:-1] + for payload, expected in (({"item": {"value": 1}}, True), ({"item": {"value": 0}}, False)): + assert _validator(schema, registry).is_valid(payload) is expected + assert _validator(schema, compact).is_valid(payload) is expected + + +def test_document_id_collision_preserves_last_document_winner() -> None: + shared_id = "https://example.test/shared.json" + schema = {"$id": ROOT, "$ref": CHILD} + child = { + "$id": shared_id, + "properties": {"value": {"$ref": "#/$defs/Value"}}, + "$defs": {"Value": {"type": "integer"}}, + } + winner = {"$id": shared_id, "$defs": {"Value": {"type": "string"}}} + registry = { + CHILD: child, + "https://example.test/winner.json": winner, + "https://example.test/unrelated.json": {"type": "null"}, + } + compact = schema_loader._reachable_registry_store(registry, BASE, schema) + assert list(compact) == list(registry)[:-1] + for payload, expected in (({"value": "valid"}, True), ({"value": 1}, False)): + assert _validator(schema, registry).is_valid(payload) is expected + assert _validator(schema, compact).is_valid(payload) is expected + + +def test_root_id_fragment_keeps_the_defragmented_document() -> None: + schema = { + "$id": f"{ROOT}#root", + "properties": {"item": {"$ref": "#/$defs/Value"}}, + "$defs": {"Value": {"type": "integer"}}, + } + registry = {ROOT: {"$defs": {"Value": {"type": "string"}}}, LEAF: {"type": "integer"}} + compact = schema_loader._reachable_registry_store(registry, BASE, schema) + assert compact == registry + for payload, expected in (({"item": "valid"}, True), ({"item": 1}, False)): + assert _validator(schema, registry).is_valid(payload) is expected + assert _validator(schema, compact).is_valid(payload) is expected + + +@pytest.mark.parametrize( + "child", + [ + {"$ref": "relative.json"}, + {"$ref": "file:///bundle/leaf.json"}, + {"$id": "/schemas/test/child.json", "type": "integer"}, + {"$id": f"{CHILD}#child", "type": "integer"}, + {"$defs": {"Scoped": {"$id": "nested.json", "type": "integer"}}}, + {"$ref": "https://example.test/missing.json"}, + ], +) +def test_scope_dependent_or_missing_refs_keep_the_full_registry(child: dict[str, Any]) -> None: + schema = {"$id": ROOT, "$ref": CHILD} + registry = {CHILD: child, LEAF: {"type": "integer"}} + assert schema_loader._reachable_registry_store(registry, BASE, schema) == registry + + +def test_missing_reference_still_fails_at_validation_without_network( + monkeypatch: pytest.MonkeyPatch, +) -> None: + schema = {"$id": ROOT, "$ref": "https://example.test/missing.json"} + registry = {CHILD: {"type": "integer"}} + compact = schema_loader._reachable_registry_store(registry, BASE, schema) + + def unexpected_network(*args: Any, **kwargs: Any) -> None: + pytest.fail("validation attempted a network request") + + monkeypatch.setattr("requests.get", unexpected_network) + with pytest.raises(Exception, match="schema reference is not in bundle"): + _validator(schema, compact).is_valid(1) diff --git a/tests/test_schema_loader_thread_safety.py b/tests/test_schema_loader_thread_safety.py new file mode 100644 index 000000000..d3846b35e --- /dev/null +++ b/tests/test_schema_loader_thread_safety.py @@ -0,0 +1,91 @@ +"""Cached schemas must not share mutable reference-resolution state (#1434).""" + +from __future__ import annotations + +import warnings +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from threading import Barrier +from types import SimpleNamespace +from typing import Any + +import pytest + +from adcp.validation import schema_loader + + +@pytest.fixture(params=["named", "task", "modular_task"]) +def validator_case(request: pytest.FixtureRequest) -> tuple[Any, dict[str, Any]]: + if request.param == "named": + return ( + lambda: schema_loader.get_named_validator( + "core/targeting-overlay-support.json", version="3.2" + ), + { + "geo_countries": True, + "geo_metros": {"systems": ["nielsen_dma"]}, + "geo_postal_areas": {"US": ["zip"]}, + "frequency_cap": True, + }, + ) + if request.param == "modular_task": + return ( + lambda: schema_loader.get_validator("get_products", "submitted", version="3.2"), + {"status": "submitted", "task_id": "task_1", "context": {"campaign": "sports"}}, + ) + return ( + lambda: schema_loader.get_validator("get_products", "request", version="3.2"), + {"buying_mode": "brief", "brief": "Sports campaign"}, + ) + + +def test_cached_lookups_return_independent_resolvers_without_schema_io( + validator_case: tuple[Any, dict[str, Any]], monkeypatch: pytest.MonkeyPatch +) -> None: + get_validator, payload = validator_case + first = get_validator() + assert first is not None + + def unexpected_read(*args: Any, **kwargs: Any) -> None: + pytest.fail("cached validator lookup reloaded schemas") + + monkeypatch.setattr(Path, "read_text", unexpected_read) + + def unexpected_warning_context(*args: Any, **kwargs: Any) -> None: + pytest.fail("cached validator lookup changed process warning filters") + + monkeypatch.setattr( + schema_loader, "warnings", SimpleNamespace(catch_warnings=unexpected_warning_context) + ) + second = get_validator() + assert second is not None + assert first is not second + assert first._ref_resolver is not second._ref_resolver + assert first._ref_resolver.store is not second._ref_resolver.store + assert not list(first.iter_errors(payload)) + assert not list(second.iter_errors(payload)) + assert list(second.iter_errors({"geo_countries": False})) + + +def test_concurrent_validation_uses_independent_resolvers( + validator_case: tuple[Any, dict[str, Any]], +) -> None: + get_validator, payload = validator_case + workers = 8 + start = Barrier(workers) + + def validate() -> None: + validator = get_validator() + assert validator is not None + start.wait(timeout=30) + for _ in range(150): + assert not list(validator.iter_errors(payload)) + validator = get_validator() + assert validator is not None + + with warnings.catch_warnings(): + warnings.simplefilter("error", DeprecationWarning) + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = [executor.submit(validate) for _ in range(workers)] + for future in futures: + future.result(timeout=60)