codekingpro/portable-devtools

1"""2Python representations of the JSON Schema Test Suite tests.3"""4from __future__ import annotations5 6from contextlib import suppress7from functools import partial8from pathlib import Path9from typing import TYPE_CHECKING, Any10import json11import os12import re13import sys14import unittest15 16from attrs import field, frozen17from referencing import Registry18import referencing.jsonschema19 20if TYPE_CHECKING:21    from collections.abc import Iterable, Mapping, Sequence22 23    from referencing.jsonschema import Schema24    import pyperf25 26from jsonschema.validators import _VALIDATORS27import jsonschema28 29MAGIC_REMOTE_URL = "http://localhost:1234"30 31_DELIMITERS = re.compile(r"[\W\- ]+")32 33 34def _find_suite():35    root = os.environ.get("JSON_SCHEMA_TEST_SUITE")36    if root is not None:37        return Path(root)38 39    root = Path(jsonschema.__file__).parent.parent / "json"40    if not root.is_dir():  # pragma: no cover41        raise ValueError(42            (43                "Can't find the JSON-Schema-Test-Suite directory. "44                "Set the 'JSON_SCHEMA_TEST_SUITE' environment "45                "variable or run the tests from alongside a checkout "46                "of the suite."47            ),48        )49    return root50 51 52@frozen53class Suite:54 55    _root: Path = field(factory=_find_suite)56 57 58    def benchmark(self, runner: pyperf.Runner):  # pragma: no cover59        for name, Validator in _VALIDATORS.items():60            self.version(name=name).benchmark(61                runner=runner,62                Validator=Validator,63            )64 65    def version(self, name) -> Version:66        Validator = _VALIDATORS[name]67        uri: str = Validator.ID_OF(Validator.META_SCHEMA)  # type: ignore[assignment]68        specification = referencing.jsonschema.specification_with(uri)69 70        registry = Registry().with_contents(71            remotes_in(root=self._root / "remotes", name=name, uri=uri),72            default_specification=specification,73        )74        return Version(75            name=name,76            path=self._root / "tests" / name,77            remotes=registry,78        )79 80 81@frozen82class Version:83 84    _path: Path85    _remotes: referencing.jsonschema.SchemaRegistry86 87    name: str88 89    def benchmark(self, **kwargs):  # pragma: no cover90        for case in self.cases():91            case.benchmark(**kwargs)92 93    def cases(self) -> Iterable[_Case]:94        return self._cases_in(paths=self._path.glob("*.json"))95 96    def format_cases(self) -> Iterable[_Case]:97        return self._cases_in(paths=self._path.glob("optional/format/*.json"))98 99    def optional_cases_of(self, name: str) -> Iterable[_Case]:100        return self._cases_in(paths=[self._path / "optional" / f"{name}.json"])101 102    def to_unittest_testcase(self, *groups, **kwargs):103        name = kwargs.pop("name", "Test" + self.name.title().replace("-", ""))104        methods = {105            method.__name__: method106            for method in (107                test.to_unittest_method(**kwargs)108                for group in groups109                for case in group110                for test in case.tests111            )112        }113        cls = type(name, (unittest.TestCase,), methods)114 115        # We're doing crazy things, so if they go wrong, like a function116        # behaving differently on some other interpreter, just make them117        # not happen.118        with suppress(Exception):119            cls.__module__ = _someone_save_us_the_module_of_the_caller()120 121        return cls122 123    def _cases_in(self, paths: Iterable[Path]) -> Iterable[_Case]:124        for path in paths:125            for case in json.loads(path.read_text(encoding="utf-8")):126                yield _Case.from_dict(127                    case,128                    version=self,129                    subject=path.stem,130                    remotes=self._remotes,131                )132 133 134@frozen135class _Case:136 137    version: Version138 139    subject: str140    description: str141    schema: Mapping[str, Any] | bool142    tests: list[_Test]143    comment: str | None = None144    specification: Sequence[dict[str, str]] = ()145 146    @classmethod147    def from_dict(cls, data, remotes, **kwargs):148        data.update(kwargs)149        tests = [150            _Test(151                version=data["version"],152                subject=data["subject"],153                case_description=data["description"],154                schema=data["schema"],155                remotes=remotes,156                **test,157            ) for test in data.pop("tests")158        ]159        return cls(tests=tests, **data)160 161    def benchmark(self, runner: pyperf.Runner, **kwargs):  # pragma: no cover162        for test in self.tests:163            runner.bench_func(164                test.fully_qualified_name,165                partial(test.validate_ignoring_errors, **kwargs),166            )167 168 169def remotes_in(170    root: Path,171    name: str,172    uri: str,173) -> Iterable[tuple[str, Schema]]:174    # This messy logic is because the test suite is terrible at indicating175    # what remotes are needed for what drafts, and mixes in schemas which176    # have no $schema and which are invalid under earlier versions, in with177    # other schemas which are needed for tests.178 179    for each in root.rglob("*.json"):180        schema = json.loads(each.read_text())181 182        relative = str(each.relative_to(root)).replace("\\", "/")183 184        if (185            ( # invalid boolean schema186                name in {"draft3", "draft4"}187                and each.stem == "tree"188            ) or189            (  # draft<NotThisDialect>/*.json190                "$schema" not in schema191                and relative.startswith("draft")192                and not relative.startswith(name)193            )194        ):195            continue196        yield f"{MAGIC_REMOTE_URL}/{relative}", schema197 198 199@frozen(repr=False)200class _Test:201 202    version: Version203 204    subject: str205    case_description: str206    description: str207 208    data: Any209    schema: Mapping[str, Any] | bool210 211    valid: bool212 213    _remotes: referencing.jsonschema.SchemaRegistry214 215    comment: str | None = None216 217    def __repr__(self):  # pragma: no cover218        return f"<Test {self.fully_qualified_name}>"219 220    @property221    def fully_qualified_name(self):  # pragma: no cover222        return " > ".join(  # noqa: FLY002223            [224                self.version.name,225                self.subject,226                self.case_description,227                self.description,228            ],229        )230 231    def to_unittest_method(self, skip=lambda test: None, **kwargs):232        if self.valid:233            def fn(this):234                self.validate(**kwargs)235        else:236            def fn(this):237                with this.assertRaises(jsonschema.ValidationError):238                    self.validate(**kwargs)239 240        fn.__name__ = "_".join(241            [242                "test",243                _DELIMITERS.sub("_", self.subject),244                _DELIMITERS.sub("_", self.case_description),245                _DELIMITERS.sub("_", self.description),246            ],247        )248        reason = skip(self)249        if reason is None or os.environ.get("JSON_SCHEMA_DEBUG", "0") != "0":250            return fn251        elif os.environ.get("JSON_SCHEMA_EXPECTED_FAILURES", "0") != "0":  # pragma: no cover  # noqa: E501252            return unittest.expectedFailure(fn)253        else:254            return unittest.skip(reason)(fn)255 256    def validate(self, Validator, **kwargs):257        Validator.check_schema(self.schema)258        validator = Validator(259            schema=self.schema,260            registry=self._remotes,261            **kwargs,262        )263        if os.environ.get("JSON_SCHEMA_DEBUG", "0") != "0":  # pragma: no cover264            breakpoint()  # noqa: T100265        validator.validate(instance=self.data)266 267    def validate_ignoring_errors(self, Validator):  # pragma: no cover268        with suppress(jsonschema.ValidationError):269            self.validate(Validator=Validator)270 271 272def _someone_save_us_the_module_of_the_caller():273    """274    The FQON of the module 2nd stack frames up from here.275 276    This is intended to allow us to dynamically return test case classes that277    are indistinguishable from being defined in the module that wants them.278 279    Otherwise, trial will mis-print the FQON, and copy pasting it won't re-run280    the class that really is running.281 282    Save us all, this is all so so so so so terrible.283    """284 285    return sys._getframe(2).f_globals["__name__"]286