Function test_object_wrapper

qa/workunits/dencoder/test_readable.py:98–133

(type, vdir, arversion, current_ver)

    return 0  # File passed the test

def test_object_wrapper(type, vdir, arversion, current_ver):
    _numtests = 0
    _failed = 0
    unrecognized = ""

    if subprocess.call([CEPH_DENCODER, "type", type], stderr=subprocess.DEVNULL) == 0:

        if should_skip_object(type, arversion, current_ver):
            debug_print(f"skipping object of type {type} due to backward incompatibility")
            return (_numtests, _failed, unrecognized)
        if type in incompat_paths:
            debug_print(f"skipping object of type {type} it is in incompat paths")
            return (_numtests, _failed, unrecognized)

        debug_print(f"        {vdir}/objects/{type}")
        files = list(vdir.joinpath("objects", type).glob('*'))
        files_without_incompat = []

        # Check symbolic links
        if type in incompat_paths:
            incompatible_files = set(incompat_paths[type])
            files_without_incompat = [f for f in files if f.name not in incompatible_files]
        else:
            files_without_incompat = files

        with concurrent.futures.ThreadPoolExecutor() as executor:
            results = [executor.submit(process_type, f, type) for f in files_without_incompat]

            for result in concurrent.futures.as_completed(results):
                _numtests += 1
                _failed += result.result()
    else:
        unrecognized = type
        debug_print("skipping unrecognized type {} return {}".format(type, (_numtests, _failed, unrecognized)))
        
    return (_numtests, _failed, unrecognized)

def parse_version(version):
    """