recipe_modules/testsharder/examples/full.py

# Copyright 2018 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

from recipe_engine.config import List
from recipe_engine.recipe_api import Property

from RECIPE_MODULES.fuchsia.testsharder.api import TestModifier

DEPS = [
    "fuchsia/status_check",
    "fuchsia/testsharder",
    "recipe_engine/path",
    "recipe_engine/properties",
    "recipe_engine/step",
]

PROPERTIES = {
    "multiplied_tests": Property(kind=List(str), help="tests to multiply", default=()),
}


def RunSteps(api, multiplied_tests):
    multiplier_unittests(api)

    modifiers = [TestModifier(name=t) for t in multiplied_tests]
    modifiers.extend(api.testsharder.affected_test_modifiers(multiplied_tests, 2))

    # Run the testsharder.
    shards = api.testsharder.execute(
        "shard test specs",
        testsharder_path="path/to/testsharder",
        build_dir=api.path["start_dir"].join("out"),
        # It's actually invalid to pass both `max_shard_size` and
        # `target_duration_secs` to the testshadder executable; we just do it
        # here for code coverage.
        max_shard_size=200,
        target_duration_secs=10 * 60,
        max_shards_per_env=8,
        mode="restricted",
        modifiers=modifiers,
        output_file=api.path["start_dir"].join("leak_output_here"),
        tags=("one-tag", "two-tag", "red-tag", "blue-tag"),
        use_affected_tests=True,
        affected_tests_file=api.path["start_dir"].join("affected_tests.txt"),
        affected_tests_multiply_threshold=1,
        affected_tests_max_attempts=1,
        affected_only=True,
    )

    # One may access a number of different aspects of the shard easily as it is a
    # Python object.
    # pylint: disable=pointless-statement
    for shard in shards:
        shard.name
        shard.deps
        # One may access all of the underlying swarming dimensions...
        shard.dimensions
        # ...or any of them individually.
        shard.device_type
        shard.os
        shard.any_swarming_dimension
        shard.targets_fuchsia
    # pylint: enable=pointless-statement


def multiplier_unittests(api):
    cases = {
        # Raw JSON
        """MULTIPLY: `[
        {
          "name": "testsharder_tests",
          "os": "linux",
          "total_runs": 5
        }
      ]`""": [
            TestModifier("testsharder_tests", "linux", 5, 0),
        ],
        # Equals instead of colon
        """MULTIPLY = `[
        {
          "name": "testsharder_tests",
          "os": "linux",
          "total_runs": 5
        }
      ]`""": [
            TestModifier("testsharder_tests", "linux", 5, 0),
        ],
        # No total_runs or os
        """MULTIPLY = `[
        {
          "name": "testsharder_tests",
        }
      ]`""": [
            TestModifier("testsharder_tests", "", 0, 0),
        ],
        # Invalid field names
        """MULTIPLY: `[
        {
          "name": "sometest",
          "foo": "testsharder_tests",
          "bar": "linux",
        }
      ]`""": api.step.StepFailure,
        # Friendly format
        "MULTIPLY: testsharder_tests (linux): 123": [
            TestModifier("testsharder_tests", "linux", 123, 0),
        ],
        # Titlecase "Multiply"
        "Multiply: testsharder_tests (linux): 123": [
            TestModifier("testsharder_tests", "linux", 123, 0),
        ],
        # No total_runs
        "Multiply: testsharder_tests (linux)": [
            TestModifier("testsharder_tests", "linux", 0, 0),
        ],
        # No os
        "Multiply: testsharder_tests: 5": [
            TestModifier("testsharder_tests", "", 5, 0),
        ],
        # No total_runs or os
        "Multiply: testsharder_tests": [TestModifier("testsharder_tests", "", 0, 0),],
        # Non-letter characters in name
        "Multiply: fuchsia-pkg://fuchsia.com/tests/foo-test": [
            TestModifier("fuchsia-pkg://fuchsia.com/tests/foo-test", "", 0, 0),
        ],
        # Non-letter characters in name with count
        "Multiply: fuchsia-pkg://fuchsia.com/tests/foo-test: 123": [
            TestModifier("fuchsia-pkg://fuchsia.com/tests/foo-test", "", 123, 0),
        ],
        # Non-letter characters in name with os and count
        "Multiply: fuchsia-pkg://fuchsia.com/tests/foo-test (linux): 123": [
            TestModifier("fuchsia-pkg://fuchsia.com/tests/foo-test", "linux", 123, 0),
        ],
        # Multiple tests on one line
        "Multiply: testsharder_tests, other_tests": [
            TestModifier("testsharder_tests", "", 0, 0),
            TestModifier("other_tests", "", 0, 0),
        ],
        # Multiple tests on one line with fields set and a trailing comma
        "Multiply: testsharder_tests (linux), other_tests: 456,": [
            TestModifier("testsharder_tests", "linux", 0, 0),
            TestModifier("other_tests", "", 456, 0),
        ],
        # Multiple lines
        """
      Multiply: testsharder_tests, foo_tests
      Multiply: other_tests: 123
      """: [
            TestModifier("testsharder_tests", "", 0, 0),
            TestModifier("foo_tests", "", 0, 0),
            TestModifier("other_tests", "", 123, 0),
        ],
        # Invalid multiplier
        "Multiply: (foo), testsharder_tests": api.step.StepFailure,
    }
    for multiply_input, expected_result in cases.items():
        try:
            multipliers = api.testsharder.extract_multipliers(multiply_input)
        except Exception as e:
            assert isinstance(e, expected_result), "bad exception %s, expected %s" % (
                type(e),
                expected_result,
            )
        else:
            assert multipliers == expected_result, "%r != %r" % (
                multipliers,
                expected_result,
            )


def GenTests(api):
    def step_data(name="shard test specs", multiply=False, no_tests=False, stdout=None):
        shards = [
            api.testsharder.shard(
                name="QEMU",
                tests=["test1"],
                dimensions=dict(device_type="QEMU"),
                service_account="myacct@example.iam.gserviceaccount.com",
            ),
            api.testsharder.shard(
                name="Linux", tests=["test2"], dimensions=dict(os="Linux"),
            ),
            api.testsharder.shard(
                name="NUC-netboot",
                tests=["test2"],
                dimensions=dict(device_type="NUC"),
                netboot=True,
            ),
        ]
        properties = {}
        if multiply:
            properties["multiplied_tests"] = ["test1", "test2"]
            shards += [
                api.testsharder.shard(
                    name="QEMU - test1",
                    tests=["test1"] * 2,
                    dimensions=dict(device_type="QEMU"),
                    service_account="myacct@example.iam.gserviceaccount.com",
                ),
                api.testsharder.shard(
                    name="Linux - test2",
                    tests=["test2"] * 2,
                    dimensions=dict(os="Linux"),
                ),
            ]
        if no_tests:
            shards.append(
                api.testsharder.shard(
                    name="QEMU-2", tests=[], dimensions=dict(device_type="QEMU"),
                )
            )
        return api.testsharder.execute(
            step_name=name, shards=shards, stdout=stdout
        ) + api.properties(**properties)

    yield (api.status_check.test("basic") + step_data(stdout="success"))

    yield (
        api.status_check.test("shard_with_multiplied_tests") + step_data(multiply=True)
    )

    yield api.status_check.test("no_tests", status="failure") + step_data(no_tests=True)