recipe_modules/testing_requests/examples/full.py

# Copyright 2019 The Fuchsia 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 PB.go.chromium.org.luci.buildbucket.proto import build as build_pb2
from PB.go.fuchsia.dev.fuchsia.tools.integration.fint.proto import (
    set_artifacts as fint_set_artifacts_pb2,
)
from PB.infra.fuchsia import Fuchsia

from recipe_engine.recipe_api import Property

DEPS = [
    "fuchsia/artifacts",
    "fuchsia/build",
    "fuchsia/buildbucket_util",
    "fuchsia/checkout",
    "fuchsia/status_check",
    "fuchsia/testing_requests",
    "fuchsia/testsharder",
    "recipe_engine/buildbucket",
    "recipe_engine/path",
    "recipe_engine/properties",
    "recipe_engine/step",
    "recipe_engine/swarming",
]

PROPERTIES = {
    "build_artifact_hash": Property(
        kind=str, help="Hash for which build artifact may be downloaded", default="abc"
    ),
    "device_type": Property(
        kind=str,
        help="Passed through to spec field Fuchsia.Test.device_type",
        default="QEMU",
    ),
    "pave": Property(
        kind=bool, help="Passed through to spec field Fuchsia.Test.pave", default=True
    ),
    "pool": Property(
        kind=str,
        help="Passed through to spec field Fuchsia.Test.pool",
        default="fuchsia.tests",
    ),
    "product": Property(
        kind=str,
        help="The product configuration of the build",
        default="products/foo.gni",
    ),
    "target_arch": Property(kind=str, help="target CPU for build", default="x64"),
    "per_test_timeout_secs": Property(
        kind=int,
        help="Passed through to spec field Fuchsia.Test.per_test_timeout_secs",
        default=0,
    ),
    "use_runtests": Property(kind=bool, help="Whether to use runtests", default=False),
    "variants": Property(kind=list, help="GN variants", default=[]),
    "zircon_args": Property(
        kind=list, help="kernel command-line arguments", default=[]
    ),
}


def RunSteps(
    api,
    device_type,
    pave,
    target_arch,
    per_test_timeout_secs,
    use_runtests,
    variants,
    zircon_args,
    pool,
    product,
):
    unit_tests(api)

    # Intended to be a minimal amount of code to create a valid FuchsiaBuildResults object.
    checkout_root = api.path["start_dir"]
    checkout = api.checkout.CheckoutResults(
        checkout_root,
        snapshot_file=api.path["start_dir"].join("snapshot"),
        release_branch=None,
        release_version=None,
        source_info={},
    )
    build_dir = checkout_root.join("out", "not-default")
    gn_results = api.build.gn_results(
        build_dir,
        fint_set_artifacts=fint_set_artifacts_pb2.SetArtifacts(
            metadata=dict(
                variants=variants,
                product=product,
                target_arch=target_arch,
                board="boards/bar.gni",
                optimize="release",
            )
        ),
    )
    build_results = api.build.build_results(
        checkout=checkout,
        build_dir=build_dir,
        gn_results=gn_results,
        images=[
            {
                "name": "zircon-a",
                "type": "zbi",
                "path": "fuchsia.zbi",
                "bootserver_pave": ["--boot"],
            }
        ],
    )

    # Configure context of uploaded artifacts for test task construction.
    api.artifacts.gcs_bucket = "fuchsia-artifacts"
    api.artifacts.uuid = api.buildbucket_util.id

    test_spec = Fuchsia.Test(
        pave=pave,
        pool=pool,
        per_test_timeout_secs=per_test_timeout_secs,
        use_runtests=use_runtests,
        default_service_account="default_service_account",
        targets_serial=True,
        zircon_args=zircon_args,
        gce_mediator=Fuchsia.Test.GCEMediator(
            endpoint="gcem-endpoint", cloud_project="gcem-cloud-project",
        ),
    )

    shards = (
        api.testsharder.Shard(
            "NAME",
            tests=[
                {
                    "name": "TEST",
                    "label": "//path/to/test:test(//toolchain)",
                    "os": "linux",
                    "path": "/path/to/test",
                }
            ],
            deps=["path/to/a/dep"],
            dimensions={"device_type": device_type},
            netboot=not pave,
            service_account="foo@bar.com",
        ),
    )
    task_requests = api.testing_requests.task_requests(
        build_results,
        api.buildbucket.build,
        test_spec.per_test_timeout_secs,
        test_spec.pool,
        shards,
        test_spec.swarming_expiration_timeout_secs,
        test_spec.swarming_io_timeout_secs,
        test_spec.use_runtests,
        test_spec.timeout_secs,
        test_spec.default_service_account,
        test_spec.pave,
        test_spec.targets_serial,
        release_version="0.20200531.0.1",
        zircon_args=test_spec.zircon_args,
        gcem_host=test_spec.gce_mediator.endpoint,
        gcem_cloud_project=test_spec.gce_mediator.cloud_project,
    )

    task_requests_data = [r.to_jsonish() for r in task_requests]
    step = api.step("task requests", None)
    step.presentation.step_text = str(task_requests_data)


def assert_raises(expected_exception, func):
    try:
        func()
    except expected_exception:
        pass
    else:
        raise AssertionError("Expected exception was not raised")


def unit_tests(api):
    assert api.testing_requests.get_catapult_dashboard_env_vars(
        "example.fuchsia.global.ci", "example-fuchsia-x64-nuc", "refs/heads/master"
    ) == {
        "CATAPULT_DASHBOARD_MASTER": "example.fuchsia.global.ci",
        "CATAPULT_DASHBOARD_BOT": "example-fuchsia-x64-nuc",
    }

    assert api.testing_requests.get_catapult_dashboard_env_vars(
        "example.fuchsia.global.ci",
        "example-fuchsia-x64-nuc",
        "refs/heads/releases/releasefoo",
    ) == {
        "CATAPULT_DASHBOARD_MASTER": "example.fuchsia.global.ci.releasefoo",
        "CATAPULT_DASHBOARD_BOT": "example-fuchsia-x64-nuc",
    }

    assert (
        api.testing_requests.get_catapult_dashboard_env_vars(
            "example.fuchsia.global.ci",
            "example-fuchsia-x64-nuc",
            "refs/heads/otherbranch",
        )
        == {}
    )

    # If Catapult upload names are not specified in the infra spec, the env
    # vars should not get set.
    assert (
        api.testing_requests.get_catapult_dashboard_env_vars(
            "", "", "refs/heads/master"
        )
        == {}
    )
    assert (
        api.testing_requests.get_catapult_dashboard_env_vars(
            "", "", "refs/heads/releases/releasefoo"
        )
        == {}
    )
    assert (
        api.testing_requests.get_catapult_dashboard_env_vars(
            "", "", "refs/heads/otherbranch"
        )
        == {}
    )

    # Test that the function raises an exception if only one of the two
    # parameters is provided.
    assert_raises(
        ValueError,
        lambda: api.testing_requests.get_catapult_dashboard_env_vars(
            "global.ci", "", "refs/heads/master"
        ),
    )
    assert_raises(
        ValueError,
        lambda: api.testing_requests.get_catapult_dashboard_env_vars(
            "", "fuchsia-x64-nuc", "refs/heads/master"
        ),
    )

    # Test assert_raises() to fulfill code coverage requirements.
    assert_raises(AssertionError, lambda: assert_raises(ValueError, lambda: None))


def GenTests(api):
    # For coverage
    api.testing_requests.task_requests_step_data(
        [api.swarming.example_task_request_jsonish()], ""
    )

    def test(name):
        return api.status_check.test(name)

    yield test("not_emu_pave") + api.properties(
        device_type="DEVICE TYPE",
        use_runtests=True,
        per_test_timeout_secs=1,
        variants=["host-asan", "asan"],
        zircon_args=["zircon_arg=true", "zircon_arg2=true"],
    )

    yield test("not_emu_netboot") + api.properties(
        device_type="DEVICE TYPE",
        product="products/bringup.gni",
        pave=False,
        use_runtests=True,
        per_test_timeout_secs=1,
        variants=["host-asan", "asan"],
    )

    yield test("emu_bringup") + api.properties(
        product="products/bringup.gni",
        pave=False,
        use_runtests=True,
        per_test_timeout_secs=1,
        variants=["host-asan", "asan"],
    )

    yield test("qemu") + api.properties(
        target_arch="x64",
        per_test_timeout_secs=1,
        zircon_args=["zircon_arg=true", "zircon_arg2=true"],
    )

    yield test("qemu_variants") + api.properties(
        per_test_timeout_secs=1,
        zircon_args=["zircon_arg=true", "zircon_arg2=true"],
        variants=["host-asan", "asan"],
    )

    yield test("aemu") + api.properties(device_type="AEMU",)

    yield test("gce_not_enabled") + api.properties(
        **{"device_type": "GCE", "pool": "fuchsia.dev.tests",}
    )
    yield test("gce_dev") + api.properties(
        **{
            "experimental.test_on_gce": True,
            "device_type": "GCE",
            "pool": "fuchsia.dev.tests",
        }
    ) + api.buildbucket.build(
        build_pb2.Build(
            infra=build_pb2.BuildInfra(
                swarming=build_pb2.BuildInfra.Swarming(
                    hostname="chrome-swarming.appspot.com",
                    parent_run_id="50a11839dfdb5911",
                )
            )
        )
    )
    yield test("gce_prod") + api.properties(
        **{
            "experimental.test_on_gce": True,
            "device_type": "GCE",
            "pool": "fuchsia.tests",
        }
    ) + api.buildbucket.build(
        build_pb2.Build(
            infra=build_pb2.BuildInfra(
                swarming=build_pb2.BuildInfra.Swarming(
                    hostname="chrome-swarming.appspot.com",
                    parent_run_id="50a11839dfdb5911",
                )
            )
        )
    )