api/controller/sdk_unittest.py

# Copyright 2019 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""SDK tests."""

import os
from pathlib import Path
from typing import List, Optional
from unittest import mock

from chromite.api import api_config
from chromite.api.controller import sdk as sdk_controller
from chromite.api.gen.chromite.api import sdk_pb2
from chromite.api.gen.chromiumos import common_pb2
from chromite.lib import chroot_lib
from chromite.lib import constants
from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.service import sdk as sdk_service


class SdkCreateTest(cros_test_lib.MockTestCase, api_config.ApiConfigMixin):
    """Create tests."""

    def setUp(self):
        """Setup method."""
        # We need to run the command outside the chroot.
        self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)
        self.response = sdk_pb2.CreateResponse()

    def _GetRequest(
        self,
        no_replace=False,
        bootstrap=False,
        cache_path=None,
        chroot_path=None,
        sdk_version=None,
        skip_chroot_upgrade=False,
    ):
        """Helper to build a create request message."""
        request = sdk_pb2.CreateRequest()
        request.flags.no_replace = no_replace
        request.flags.bootstrap = bootstrap

        if cache_path:
            request.chroot.cache_dir = cache_path
        if chroot_path:
            request.chroot.path = chroot_path
        if sdk_version:
            request.sdk_version = sdk_version
        if skip_chroot_upgrade:
            request.skip_chroot_upgrade = skip_chroot_upgrade

        return request

    def testValidateOnly(self):
        """Verify a validate-only call does not execute any logic."""
        patch = self.PatchObject(sdk_service, "Create")

        sdk_controller.Create(
            self._GetRequest(), self.response, self.validate_only_config
        )
        patch.assert_not_called()

    def testMockCall(self):
        """Sanity check that a mock call does not execute any logic."""
        patch = self.PatchObject(sdk_service, "Create")

        rc = sdk_controller.Create(
            self._GetRequest(), self.response, self.mock_call_config
        )
        patch.assert_not_called()
        self.assertFalse(rc)
        self.assertTrue(self.response.version.version)

    def testSuccess(self):
        """Test the successful call output handling."""
        self.PatchObject(sdk_service, "Create", return_value=1)

        request = self._GetRequest()

        sdk_controller.Create(request, self.response, self.api_config)

        self.assertEqual(1, self.response.version.version)

    def testFalseArguments(self):
        """Test False argument handling."""
        # Create the patches.
        self.PatchObject(sdk_service, "Create", return_value=1)
        args_patch = self.PatchObject(sdk_service, "CreateArguments")

        # Flag translation tests.
        # Test all false values in the message.
        request = self._GetRequest(
            no_replace=False,
            bootstrap=False,
        )
        sdk_controller.Create(request, self.response, self.api_config)
        args_patch.assert_called_with(
            replace=True,
            bootstrap=False,
            chroot=mock.ANY,
            sdk_version=mock.ANY,
            skip_chroot_upgrade=mock.ANY,
        )

    def testTrueArguments(self):
        """Test True arguments handling."""
        # Create the patches.
        self.PatchObject(sdk_service, "Create", return_value=1)
        args_patch = self.PatchObject(sdk_service, "CreateArguments")

        # Test all True values in the message.
        request = self._GetRequest(
            no_replace=True,
            bootstrap=True,
            sdk_version="foo",
            skip_chroot_upgrade=True,
        )
        sdk_controller.Create(request, self.response, self.api_config)
        args_patch.assert_called_with(
            replace=False,
            bootstrap=True,
            chroot=mock.ANY,
            sdk_version="foo",
            skip_chroot_upgrade=True,
        )


class SdkCleanTest(cros_test_lib.MockTestCase, api_config.ApiConfigMixin):
    """Clean tests."""

    def setUp(self):
        """Setup method."""
        # We need to run the command outside the chroot.
        self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)
        self.response = sdk_pb2.CleanResponse()

    def _GetRequest(self, chroot_path=None, incrementals=False):
        """Helper to build a clean request message."""
        request = sdk_pb2.CleanRequest()
        if chroot_path:
            request.chroot.path = chroot_path

        request.incrementals = incrementals

        return request

    def testMockCall(self):
        """Sanity check that a mock call does not execute any logic."""
        patch = self.PatchObject(sdk_service, "Clean")

        rc = sdk_controller.Clean(
            self._GetRequest(), self.response, self.mock_call_config
        )
        patch.assert_not_called()
        self.assertFalse(rc)

    def testSuccess(self):
        """Test the successful call by verifying service invocation."""
        patch = self.PatchObject(sdk_service, "Clean", return_value=0)

        request = self._GetRequest(incrementals=True)

        sdk_controller.Clean(request, self.response, self.api_config)
        patch.assert_called_once_with(
            mock.ANY,
            safe=False,
            images=False,
            sysroots=False,
            tmp=False,
            cache=False,
            logs=False,
            workdirs=False,
            incrementals=True,
        )

    def testDefaults(self):
        """Test the successful call by verifying service invocation."""
        patch = self.PatchObject(sdk_service, "Clean", return_value=0)

        request = self._GetRequest()

        sdk_controller.Clean(request, self.response, self.api_config)
        patch.assert_called_once_with(mock.ANY, safe=True, sysroots=True)


class SdkDeleteTest(cros_test_lib.MockTestCase, api_config.ApiConfigMixin):
    """Delete tests."""

    def setUp(self):
        """Setup method."""
        # We need to run the command outside the chroot.
        self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)
        self.response = sdk_pb2.DeleteResponse()

    def _GetRequest(self, chroot_path=None):
        """Helper to build a delete request message."""
        request = sdk_pb2.DeleteRequest()
        if chroot_path:
            request.chroot.path = chroot_path

        return request

    def testValidateOnly(self):
        """Verify a validate-only call does not execute any logic."""
        patch = self.PatchObject(sdk_service, "Delete")

        sdk_controller.Delete(
            self._GetRequest(), self.response, self.validate_only_config
        )
        patch.assert_not_called()

    def testMockCall(self):
        """Sanity check that a mock call does not execute any logic."""
        patch = self.PatchObject(sdk_service, "Delete")

        rc = sdk_controller.Delete(
            self._GetRequest(), self.response, self.mock_call_config
        )
        patch.assert_not_called()
        self.assertFalse(rc)

    def testSuccess(self):
        """Test the successful call by verifying service invocation."""
        patch = self.PatchObject(sdk_service, "Delete", return_value=1)

        request = self._GetRequest()

        sdk_controller.Delete(request, self.response, self.api_config)
        # Verify that by default sdk_service.Delete is called with force=True.
        patch.assert_called_once_with(mock.ANY, force=True)


class SdkUnmountTest(cros_test_lib.MockTestCase, api_config.ApiConfigMixin):
    """SDK Unmount tests."""

    def testNoop(self):
        """Unmount is a deprecated noop."""
        request = sdk_pb2.UnmountRequest()
        response = sdk_pb2.UnmountResponse()
        rc = sdk_controller.Unmount(request, response, self.api_config)
        self.assertFalse(rc)


class SdkUnmountPathTest(cros_test_lib.MockTestCase, api_config.ApiConfigMixin):
    """Update tests."""

    def setUp(self):
        """Setup method."""
        self.response = sdk_pb2.UnmountPathResponse()

    def _UnmountPathRequest(self, path=None):
        """Helper to build a delete request message."""
        request = sdk_pb2.UnmountPathRequest()
        if path:
            request.path.path = path
        return request

    def testValidateOnly(self):
        """Verify a validate-only call does not execute any logic."""
        patch = self.PatchObject(sdk_service, "UnmountPath")

        sdk_controller.UnmountPath(
            self._UnmountPathRequest("/test/path"),
            self.response,
            self.validate_only_config,
        )
        patch.assert_not_called()

    def testMockCall(self):
        """Sanity check that a mock call does not execute any logic."""
        patch = self.PatchObject(sdk_service, "UnmountPath")

        rc = sdk_controller.UnmountPath(
            self._UnmountPathRequest(), self.response, self.mock_call_config
        )
        patch.assert_not_called()
        self.assertFalse(rc)

    def testSuccess(self):
        """Test the successful call by verifying service invocation."""
        patch = self.PatchObject(sdk_service, "UnmountPath", return_value=1)

        request = self._UnmountPathRequest("/test/path")
        sdk_controller.UnmountPath(request, self.response, self.api_config)
        patch.assert_called_once_with("/test/path")


class SdkUpdateTest(cros_test_lib.MockTestCase, api_config.ApiConfigMixin):
    """Update tests."""

    def setUp(self):
        """Setup method."""
        # We need to run the command inside the chroot.
        self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=True)

        self.response = sdk_pb2.UpdateResponse()

    def _GetRequest(self, build_source=False, targets=None):
        """Helper to simplify building a request instance."""
        request = sdk_pb2.UpdateRequest()
        request.flags.build_source = build_source

        for target in targets or []:
            added = request.toolchain_targets.add()
            added.name = target

        return request

    def testValidateOnly(self):
        """Verify a validate-only call does not execute any logic."""
        patch = self.PatchObject(sdk_service, "Update")

        sdk_controller.Update(
            self._GetRequest(), self.response, self.validate_only_config
        )
        patch.assert_not_called()

    def testMockCall(self):
        """Sanity check that a mock call does not execute any logic."""
        patch = self.PatchObject(sdk_service, "Update")

        rc = sdk_controller.Create(
            self._GetRequest(), self.response, self.mock_call_config
        )
        patch.assert_not_called()
        self.assertFalse(rc)
        self.assertTrue(self.response.version.version)

    def testSuccess(self):
        """Successful call output handling test."""
        expected_version = 1
        self.PatchObject(sdk_service, "Update", return_value=expected_version)
        request = self._GetRequest()

        sdk_controller.Update(request, self.response, self.api_config)

        self.assertEqual(expected_version, self.response.version.version)

    def testArgumentHandling(self):
        """Test the proto argument handling."""
        args = sdk_service.UpdateArguments()
        self.PatchObject(sdk_service, "Update", return_value=1)
        args_patch = self.PatchObject(
            sdk_service, "UpdateArguments", return_value=args
        )

        # No boards and flags False.
        request = self._GetRequest(build_source=False)
        sdk_controller.Update(request, self.response, self.api_config)
        args_patch.assert_called_with(
            build_source=False, toolchain_targets=[], toolchain_changed=False
        )

        # Multiple boards and flags True.
        targets = ["board1", "board2"]
        request = self._GetRequest(build_source=True, targets=targets)
        sdk_controller.Update(request, self.response, self.api_config)
        args_patch.assert_called_with(
            build_source=True,
            toolchain_targets=targets,
            toolchain_changed=False,
        )


class CreateManifestFromSdkTest(
    cros_test_lib.MockTestCase, api_config.ApiConfigMixin
):
    """Test the SdkService/CreateManifestFromSdk endpoint."""

    _sdk_path = "/build/my_sdk"
    _dest_dir = "/build"
    _manifest_path = "/build/my_sdk.Manifest"

    def _NewRequest(self, inside: bool) -> sdk_pb2.CreateManifestFromSdkRequest:
        return sdk_pb2.CreateManifestFromSdkRequest(
            chroot=common_pb2.Chroot(
                path=self.chroot.path, out_path=str(self.chroot.out_path)
            ),
            sdk_path=common_pb2.Path(
                path=self._sdk_path,
                location=common_pb2.Path.Location.INSIDE
                if inside
                else common_pb2.Path.Location.OUTSIDE,
            ),
            dest_dir=common_pb2.Path(
                path=self._dest_dir,
                location=common_pb2.Path.Location.OUTSIDE,
            ),
        )

    def _NewResponse(self) -> sdk_pb2.CreateManifestFromSdkResponse:
        return sdk_pb2.CreateManifestFromSdkResponse()

    def setUp(self):
        self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)

        self.chroot = chroot_lib.Chroot(
            path=Path("/path/to/chroot"),
            out_path=Path("/path/to/out"),
        )

    def testValidateOnly(self):
        """Check that a validate only call does not execute any logic."""
        impl_patch = self.PatchObject(sdk_service, "CreateManifestFromSdk")
        sdk_controller.CreateManifestFromSdk(
            self._NewRequest(False),
            self._NewResponse(),
            self.validate_only_config,
        )
        impl_patch.assert_not_called()

    def testOutside(self):
        """Check that a call with an outside path succeeds."""
        impl_patch = self.PatchObject(
            sdk_service,
            "CreateManifestFromSdk",
            return_value=Path(self._manifest_path),
        )
        request = self._NewRequest(inside=False)
        response = self._NewResponse()
        sdk_controller.CreateManifestFromSdk(
            request,
            response,
            self.api_config,
        )
        impl_patch.assert_called_with(
            Path(self._sdk_path),
            Path(self._dest_dir),
        )
        self.assertEqual(
            response.manifest_path.location, common_pb2.Path.Location.OUTSIDE
        )
        self.assertEqual(response.manifest_path.path, self._manifest_path)

    def testInside(self):
        """Check that an inside path parses correctly and the call succeeds."""
        impl_patch = self.PatchObject(
            sdk_service,
            "CreateManifestFromSdk",
            return_value=Path(self._manifest_path),
        )
        request = self._NewRequest(inside=True)
        response = self._NewResponse()
        sdk_controller.CreateManifestFromSdk(
            request,
            response,
            self.api_config,
        )
        impl_patch.assert_called_with(
            Path(self.chroot.full_path(self._sdk_path)),
            Path(self._dest_dir),
        )
        self.assertEqual(
            response.manifest_path.location, common_pb2.Path.Location.OUTSIDE
        )
        self.assertEqual(response.manifest_path.path, self._manifest_path)


class BuildSdkToolchainTest(
    cros_test_lib.MockTestCase, api_config.ApiConfigMixin
):
    """Test the SdkService/BuildSdkToolchain endpoint."""

    def setUp(self):
        """Set up the test case."""
        self._chroot_path = "/path/to/chroot"
        self._result_dir = "/out/toolchain-pkgs/"
        self._response = sdk_pb2.BuildSdkToolchainResponse()
        self._generated_filenames = (
            "armv7a-cros-linux-gnueabihf.tar.xz",
            "x86_64-cros-linux-gnu.tar.xz",
        )
        self._paths_for_generated_files = [
            common_pb2.Path(
                path=os.path.join(constants.SDK_TOOLCHAINS_OUTPUT, fname),
                location=common_pb2.Path.Location.INSIDE,
            )
            for fname in self._generated_filenames
        ]

    def _NewRequest(
        self,
        chroot_path: Optional[str] = None,
        use_flags: Optional[List[str]] = None,
    ) -> sdk_pb2.BuildSdkToolchainRequest:
        """Return a new BuildSdkToolchainRequest message."""
        request = sdk_pb2.BuildSdkToolchainRequest(
            result_path=common_pb2.ResultPath(
                path=common_pb2.Path(
                    path=self._result_dir,
                    location=common_pb2.Path.Location.OUTSIDE,
                )
            )
        )
        if chroot_path:
            request.chroot.path = chroot_path
        if use_flags:
            request.use_flags.extend(
                common_pb2.UseFlag(flag=flag) for flag in use_flags
            )
        return request

    def _NewResponse(
        self, generated_filenames: Optional[List[str]] = None
    ) -> sdk_pb2.BuildSdkToolchainResponse:
        """Return a new BuildSdkToolchainResponse message."""
        response = sdk_pb2.BuildSdkToolchainResponse()
        if generated_filenames:
            response.generated_files.extend(
                common_pb2.Path(
                    path=os.path.join(self._result_dir, fname),
                    location=common_pb2.Path.Location.OUTSIDE,
                )
                for fname in generated_filenames
            )
        return response

    def testValidateOnly(self):
        """Check that a validate only call does not execute any logic."""
        impl_patch = self.PatchObject(sdk_service, "BuildSdkToolchain")
        sdk_controller.BuildSdkToolchain(
            self._NewRequest(), self._NewResponse(), self.validate_only_config
        )
        impl_patch.assert_not_called()

    def testSuccess(self):
        """Check that a normal call defers to the SDK service as expected."""
        impl_patch = self.PatchObject(sdk_service, "BuildSdkToolchain")
        request = self._NewRequest(use_flags=[])
        response = self._NewResponse()
        sdk_controller.BuildSdkToolchain(
            request,
            response,
            self.api_config,
        )
        # Can't use assert_called_with, since the chroot objects are equal but
        # not identical.
        impl_patch.assert_called_once()
        self.assertEqual(impl_patch.call_args.kwargs["extra_env"], {})

    def testSuccessWithUseFlags(self):
        """Check that a call with USE flags works as expected."""
        impl_patch = self.PatchObject(sdk_service, "BuildSdkToolchain")
        request = self._NewRequest(use_flags=["llvm-next", "another-flag"])
        response = self._NewResponse()
        sdk_controller.BuildSdkToolchain(
            request,
            response,
            self.api_config,
        )
        # Can't use assert_called_with, since the chroot objects are equal but
        # not identical.
        impl_patch.assert_called_once()
        self.assertEqual(
            impl_patch.call_args.kwargs["extra_env"],
            {"USE": "llvm-next another-flag"},
        )


class uprev_test(cros_test_lib.MockTestCase, api_config.ApiConfigMixin):
    """Test case for SdkService/Uprev() endpoint."""

    _binhost_gs_bucket = "gs://chromiumos-prebuilts/"
    _latest_uprev_target_version = "2023.02.19.112358"

    def setUp(self):
        """Set up the test case."""
        self.PatchObject(
            sdk_service,
            "get_latest_uprev_target_version",
            return_value=self._latest_uprev_target_version,
        )
        self._uprev_patch = self.PatchObject(
            sdk_service,
            "uprev_sdk_and_prebuilts",
        )

    def NewRequest(
        self, version: str = "", toolchain_tarball_template: str = ""
    ):
        """Return a new UprevRequest with standard inputs."""
        return sdk_pb2.UprevRequest(
            binhost_gs_bucket=self._binhost_gs_bucket,
            version=version,
            toolchain_tarball_template=toolchain_tarball_template,
        )

    @staticmethod
    def NewResponse() -> sdk_pb2.UprevResponse:
        """Return a new empty UprevResponse."""
        return sdk_pb2.UprevResponse()

    def testWithVersion(self):
        """Test the endpoint with `version` specified.

        In this case, we expect that sdk_controller.Uprev is called with the
        version specified in the UprevRequest.
        """
        specified_version = "1970.01.01.000000"
        toolchain_tarball_template = "path/to/%(version)s/toolchain"
        request = self.NewRequest(
            version=specified_version,
            toolchain_tarball_template=toolchain_tarball_template,
        )
        response = self.NewResponse()
        sdk_controller.Uprev(request, response, self.api_config)
        self._uprev_patch.assert_called_with(
            binhost_gs_bucket=self._binhost_gs_bucket,
            version=specified_version,
            toolchain_tarball_template=toolchain_tarball_template,
        )

    def testWithoutVersion(self):
        """Test the endpoint with `version` not specified.

        In this case, we expect that sdk_controller.Uprev is called with the
        latest uprev target version, based on the remote file in gs://. This is
        fetched via sdk_controller.GetLatestUprevTargetVersionVersion
        (mocked here in setUp()).
        """
        toolchain_tarball_template = "path/to/%(version)s/toolchain"
        request = self.NewRequest(
            toolchain_tarball_template=toolchain_tarball_template
        )
        response = self.NewResponse()
        sdk_controller.Uprev(request, response, self.api_config)
        self._uprev_patch.assert_called_with(
            binhost_gs_bucket=self._binhost_gs_bucket,
            version=self._latest_uprev_target_version,
            toolchain_tarball_template=toolchain_tarball_template,
        )

    def testWithoutToolchainTarballTemplate(self):
        """Test the endpoint with `toolchain_tarball_template` not specified."""
        request = self.NewRequest(version="1234")
        response = self.NewResponse()
        with self.assertRaises(cros_build_lib.DieSystemExit):
            sdk_controller.Uprev(request, response, self.api_config)