Function generate_test_classes

test/asynchronous/unified_format.py:1591–1669

Method for generating test classes. Returns a dictionary where keys are the names of test classes and values are the test class objects.

(
    test_path,
    module=__name__,
    class_name_prefix="",
    expected_failures=[],  # noqa: B006
    bypass_test_generation_errors=False,
    **kwargs,
)

def generate_test_classes(
    test_path,
    module=__name__,
    class_name_prefix="",
    expected_failures=[],  # noqa: B006
    bypass_test_generation_errors=False,
    **kwargs,
):
    """Method for generating test classes. Returns a dictionary where keys are
    the names of test classes and values are the test class objects.
    """
    test_klasses = {}

    def test_base_class_factory(test_spec):
        """Utility that creates the base class to use for test generation.
        This is needed to ensure that cls.TEST_SPEC is appropriately set when
        the metaclass __init__ is invoked.
        """

        class SpecTestBase(with_metaclass(UnifiedSpecTestMeta)):  # type: ignore
            TEST_SPEC = test_spec
            EXPECTED_FAILURES = expected_failures

        base = SpecTestBase

        # Add "encryption" marker if the "csfle" runOnRequirement is set.
        for req in test_spec.get("runOnRequirements", []):
            if "csfle" in req:
                base = pytest.mark.encryption(base)

        return base

    found_any = False
    for dirpath, _, filenames in os.walk(test_path):
        dirname = os.path.split(dirpath)[-1]

        for filename in filenames:
            found_any = True
            fpath = os.path.join(dirpath, filename)
            with open(fpath) as scenario_stream:
                # Use tz_aware=False to match how CodecOptions decodes
                # dates.
                opts = json_util.JSONOptions(tz_aware=False)
                scenario_def = json_util.loads(scenario_stream.read(), json_options=opts)

            test_type = os.path.splitext(filename)[0]
            snake_class_name = "Test{}_{}_{}".format(
                class_name_prefix,
                dirname.replace("-", "_"),
                test_type.replace("-", "_").replace(".", "_"),
            )
            class_name = snake_to_camel(snake_class_name)

            try:
                schema_version = Version.from_string(scenario_def["schemaVersion"])
                mixin_class = _SCHEMA_VERSION_MAJOR_TO_MIXIN_CLASS.get(schema_version[0])
                if mixin_class is None:
                    raise ValueError(