def get_sanity_tests_for_slice()

in mozetl/graphics/graphics_telemetry_dashboard.py [0:0]

• 69 lines of code
• 2 McCabe index (conditional complexity)

def get_sanity_tests_for_slice(sanity_test_pings):
    data = sanity_test_pings.filter(lambda p: get_sanity_test_result(p) is not None)

    # Aggregate the sanity test data.
    with Prof("initial map"):
        sanity_test_results = data.map(
            lambda p: (get_sanity_test_result(p),)
        ).countByKey()

    with Prof("share resolve"):
        os_share = map_x_to_count(data, "OSVersion")

    with Prof("ping_count"):
        sanity_test_count = data.count()
        ping_count = sanity_test_pings.count()

    sanity_test_by_vendor = None
    sanity_test_by_os = None
    sanity_test_by_device = None
    sanity_test_by_driver = None
    for value in range(SANITY_TEST_FAILED_RENDER, SANITY_TEST_LAST_VALUE):
        subset = data.filter(lambda p: get_sanity_test_result(p) == value)

        tuples = subset.map(
            lambda p: (value, {p["vendorID"]: int(p[SANITY_TEST][value])})
        )
        sanity_test_by_vendor = union_pipelines(sanity_test_by_vendor, tuples)

        tuples = subset.map(lambda p: (value, {p["OS"]: int(p[SANITY_TEST][value])}))
        sanity_test_by_os = union_pipelines(sanity_test_by_os, tuples)

        tuples = subset.map(
            lambda p: (value, {p["deviceID"]: int(p[SANITY_TEST][value])})
        )
        sanity_test_by_device = union_pipelines(sanity_test_by_device, tuples)

        tuples = subset.map(
            lambda p: (value, {p["driverVersion"]: int(p[SANITY_TEST][value])})
        )
        sanity_test_by_driver = union_pipelines(sanity_test_by_driver, tuples)

    sanity_test_by_vendor = repartition(sanity_test_by_vendor)
    sanity_test_by_os = repartition(sanity_test_by_os)
    sanity_test_by_device = repartition(sanity_test_by_device)
    sanity_test_by_driver = repartition(sanity_test_by_driver)

    with Prof("vendor resolve"):
        sanity_test_by_vendor = sanity_test_by_vendor.reduceByKey(combiner)
    with Prof("os resolve"):
        sanity_test_by_os = sanity_test_by_os.reduceByKey(combiner)
    with Prof("device resolve"):
        sanity_test_by_device = sanity_test_by_device.reduceByKey(combiner)
    with Prof("driver resolve"):
        sanity_test_by_driver = sanity_test_by_driver.reduceByKey(combiner)

    print(
        "Partitions: {0},{1},{2},{3}".format(
            sanity_test_by_vendor.getNumPartitions(),
            sanity_test_by_os.getNumPartitions(),
            sanity_test_by_device.getNumPartitions(),
            sanity_test_by_driver.getNumPartitions(),
        )
    )

    with Prof("vendor collect"):
        by_vendor = sanity_test_by_vendor.collect()
    with Prof("os collect"):
        by_os = sanity_test_by_os.collect()
    with Prof("device collect"):
        by_device = sanity_test_by_device.collect()
    with Prof("driver collect"):
        by_driver = sanity_test_by_driver.collect()

    return {
        "sanityTestPings": sanity_test_count,
        "totalPings": ping_count,
        "results": sanity_test_results,
        "byVendor": by_vendor,
        "byOS": by_os,
        "byDevice": coalesce_to_n_items(by_device, 10),
        "byDriver": coalesce_to_n_items(by_driver, 10),
        "windows": os_share,
    }