recipe_modules/testing_requests/tests/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/release",
    "fuchsia/status_check",
    "fuchsia/testing_requests",
    "fuchsia/testsharder",
    "recipe_engine/buildbucket",
    "recipe_engine/json",
    "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"),
    "task_timeout_secs": Property(
        kind=int,
        help="Passed through to spec field Fuchsia.Test.timeout_secs",
        default=0,
    ),
    "test_on_gce": Property(kind=bool, help="Whether to run GCE shards", default=False),
    "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=[]
    ),
    "use_ffx": Property(kind=bool, help="Whether to enable using ffx", default=False),
    "use_ffx_max": Property(
        kind=bool, help="Whether to use the max ffx experiment level", default=False
    ),
    "use_cas": Property(
        kind=bool,
        help="Whether to enable using CAS delivery of images and blobs",
        default=False,
    ),
    "use_image_overrides": Property(
        kind=bool, help="Whether to use image_overrides", default=False
    ),
}


def RunSteps(
    api,
    device_type,
    pave,
    target_arch,
    task_timeout_secs,
    test_on_gce,
    use_runtests,
    variants,
    zircon_args,
    pool,
    product,
    use_ffx,
    use_ffx_max,
    use_cas,
    use_image_overrides,
):
    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.namespace = api.buildbucket_util.id

    ffx_experiment_level = 0
    if use_ffx:
        ffx_experiment_level = 1
    if use_ffx_max:
        ffx_experiment_level = 2
    test_spec = Fuchsia.Test(
        pave=pave,
        pool=pool,
        timeout_secs=task_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",
            machine_shape="n2-standard-4",
        ),
        use_ffx=use_ffx,
        ffx_experiment_level=ffx_experiment_level,
        use_cas=use_cas,
        enable_sandboxing=True,
    )

    image_overrides = {}
    if use_image_overrides:
        image_overrides = {
            "zbi_image": "zbi-image",
            "vbmeta_image": "vbmeta-image",
            "qemu_kernel": "qemu-kernel",
        }
    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",
            timeout_secs=1800,
            pkg_repo="path/to/pkg/repo",
            image_overrides=image_overrides,
        ),
        api.testsharder.Shard(
            "NAME2",
            tests=[
                {
                    "name": "TEST2",
                    "label": "//path/to/test:test(//toolchain)",
                    "os": "linux",
                    "path": "/path/to/test",
                }
            ],
            deps=["path/to/a/dep"],
            dimensions={"device_type": device_type, "os": "Linux"},
            netboot=not pave,
            service_account="foo@bar.com",
            timeout_secs=1800,
            pkg_repo="path/to/pkg/repo",
            image_overrides=image_overrides,
        ),
    )
    task_requests = api.testing_requests.task_requests(
        build_results,
        api.buildbucket.build,
        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=api.release.ReleaseVersion.from_string("0.20200531.0.1"),
        test_on_gce=test_on_gce,
        zircon_args=test_spec.zircon_args,
        gcem_host=test_spec.gce_mediator.endpoint,
        gcem_cloud_project=test_spec.gce_mediator.cloud_project,
        gcem_machine_shape=test_spec.gce_mediator.machine_shape,
        use_ffx=test_spec.use_ffx,
        ffx_experiment_level=test_spec.ffx_experiment_level,
        use_cas=test_spec.use_cas,
        enable_sandboxing=test_spec.enable_sandboxing,
    )

    task_requests_data = [r.to_jsonish() for r in task_requests]
    api.step.empty("task requests").presentation.logs["requests"] = api.json.dumps(
        task_requests_data, indent=2
    )


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-core.x64-nuc", "refs/heads/main"
    ) == {
        "CATAPULT_DASHBOARD_MASTER": "example.fuchsia.global.ci",
        "CATAPULT_DASHBOARD_BOT": "example-core.x64-nuc",
    }

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

    assert (
        api.testing_requests.get_catapult_dashboard_env_vars(
            "example.fuchsia.global.ci",
            "example-core.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/main")
        == {}
    )
    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/main"
        ),
    )
    assert_raises(
        ValueError,
        lambda: api.testing_requests.get_catapult_dashboard_env_vars(
            "", "core.x64-nuc", "refs/heads/main"
        ),
    )

    # 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,
        task_timeout_secs=50,  # Should override testsharder's computed shard timeout.
        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,
        variants=["host-asan", "asan"],
    )

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

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

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

    yield test("aemu_ffx") + api.properties(**{"device_type": "AEMU", "use_ffx": True})
    yield test("device_ffx") + api.properties(
        **{"device_type": "DEVICE TYPE", "use_ffx": True}
    )
    yield test("device_ffx_max_level") + api.properties(
        **{"device_type": "DEVICE TYPE", "use_ffx_max": True}
    )

    yield test("gce_not_enabled") + api.properties(
        **{"device_type": "GCE", "pool": "fuchsia.dev.tests"}
    )
    yield test("gce_dev") + api.properties(
        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(
        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",
                )
            )
        )
    )
    yield test("cas_delivery_enabled") + api.properties(**{"use_cas": True})