api/validate_unittest.py

# -*- coding: utf-8 -*-
# Copyright 2019 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Tests for the validate module."""

from __future__ import print_function

import os
import sys

from chromite.api import api_config
from chromite.api import validate
from chromite.api.gen.chromite.api import build_api_test_pb2
from chromite.api.gen.chromiumos import common_pb2
from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import osutils

assert sys.version_info >= (3, 6), 'This module requires Python 3.6+'


# These tests test the validators by defining a local `impl` function that
# has the same parameters as a controller function and the validator being
# tested. The validators don't care that they aren't actually controller
# functions, they just need the function to look like one, so it works
# to pass an arbitrary message; i.e. passing one of the Request messages
# we'd usually expect in a controller is not required. The validator
# just needs to be checking one of the fields on the message being used.
class ExistsTest(cros_test_lib.TempDirTestCase, api_config.ApiConfigMixin):
  """Tests for the exists validator."""

  def test_not_exists(self):
    """Test the validator fails when given a path that doesn't exist."""
    path = os.path.join(self.tempdir, 'DOES_NOT_EXIST')

    @validate.exists('path')
    def impl(_input_proto, _output_proto, _config):
      self.fail('Incorrectly allowed method to execute.')

    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(common_pb2.Chroot(path=path), None, self.api_config)

  def test_exists(self):
    """Test the validator fails when given a path that doesn't exist."""
    path = os.path.join(self.tempdir, 'chroot')
    osutils.SafeMakedirs(path)

    @validate.exists('path')
    def impl(_input_proto, _output_proto, _config):
      pass

    impl(common_pb2.Chroot(path=path), None, self.api_config)

  def test_skip_validation(self):
    """Test skipping validation case."""
    @validate.exists('path')
    def impl(_input_proto, _output_proto, _config):
      pass

    # This would otherwise raise an error for an invalid path.
    impl(common_pb2.Chroot(), None, self.no_validate_config)


class IsInTest(cros_test_lib.TestCase, api_config.ApiConfigMixin):
  """Tests for the is_in validator."""

  def test_in(self):
    """Test a valid value."""
    @validate.is_in('path', ['/chroot/path', '/other/chroot/path'])
    def impl(_input_proto, _output_proto, _config):
      pass

    # Make sure all of the values work.
    impl(common_pb2.Chroot(path='/chroot/path'), None, self.api_config)
    impl(common_pb2.Chroot(path='/other/chroot/path'), None, self.api_config)

  def test_not_in(self):
    """Test an invalid value."""
    @validate.is_in('path', ['/chroot/path', '/other/chroot/path'])
    def impl(_input_proto, _output_proto, _config):
      pass

    # Should be failing on the invalid value.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(common_pb2.Chroot(path='/bad/value'), None, self.api_config)

  def test_not_set(self):
    """Test an unset value."""
    @validate.is_in('path', ['/chroot/path', '/other/chroot/path'])
    def impl(_input_proto, _output_proto, _config):
      pass

    # Should be failing without a value set.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(common_pb2.Chroot(), None, self.api_config)

  def test_skip_validation(self):
    """Test skipping validation case."""
    @validate.is_in('path', ['/chroot/path', '/other/chroot/path'])
    def impl(_input_proto, _output_proto, _config):
      pass

    # This would otherwise raise an error for an invalid path.
    impl(common_pb2.Chroot(), None, self.no_validate_config)


class EachInTest(cros_test_lib.TestCase, api_config.ApiConfigMixin):
  """Tests for the each_in validator."""

  # Easier access to the enum values.
  ENUM_FOO = build_api_test_pb2.TEST_ENUM_FOO
  ENUM_BAR = build_api_test_pb2.TEST_ENUM_BAR
  ENUM_BAZ = build_api_test_pb2.TEST_ENUM_BAZ

  # pylint: disable=docstring-misnamed-args
  def _message_request(self, *messages):
    """Build a request instance, filling out the messages field.

    Args:
      messages: Each messages data (id, name, flag, enum) as lists. Only
        requires as many as are set. e.g. _request([1], [2]) will create two
        messages with only ids set. _request([1, 'name']) will create one with
        id and name set, but not flag or enum.
    """
    request = build_api_test_pb2.TestRequestMessage()
    for message in messages or []:
      msg = request.messages.add()
      try:
        msg.id = message[0]
        msg.name = message[1]
        msg.flag = message[2]
      except IndexError:
        pass

    return request

  def _enums_request(self, *enum_values):
    """Build a request instance, setting the test_enums field."""
    request = build_api_test_pb2.TestRequestMessage()
    for value in enum_values:
      request.test_enums.append(value)

    return request

  def _numbers_request(self, *numbers):
    """Build a request instance, setting the numbers field."""
    request = build_api_test_pb2.TestRequestMessage()
    request.numbers.extend(numbers)

    return request

  def test_message_in(self):
    """Test valid values."""

    @validate.each_in('messages', 'name', ['foo', 'bar'])
    def impl(_input_proto, _output_proto, _config):
      pass

    impl(self._message_request([1, 'foo']), None, self.api_config)
    impl(self._message_request([1, 'foo'], [2, 'bar']), None, self.api_config)

  def test_enum_in(self):
    """Test valid enum values."""

    @validate.each_in('test_enums', None, [self.ENUM_FOO, self.ENUM_BAR])
    def impl(_input_proto, _output_proto, _config):
      pass

    impl(self._enums_request(self.ENUM_FOO), None, self.api_config)
    impl(self._enums_request(self.ENUM_FOO, self.ENUM_BAR), None,
         self.api_config)

  def test_scalar_in(self):
    """Test valid scalar values."""

    @validate.each_in('numbers', None, [1, 2])
    def impl(_input_proto, _output_proto, _config):
      pass

    impl(self._numbers_request(1), None, self.api_config)
    impl(self._numbers_request(1, 2), None, self.api_config)

  def test_message_not_in(self):
    """Test an invalid value."""

    @validate.each_in('messages', 'name', ['foo', 'bar'])
    def impl(_input_proto, _output_proto, _config):
      pass

    # Should be failing on the invalid value.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._message_request([1, 'invalid']), None, self.api_config)
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._message_request([1, 'invalid'], [2, 'invalid']), None,
           self.api_config)
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._message_request([1, 'foo'], [2, 'invalid']), None,
           self.api_config)

  def test_enum_not_in(self):
    """Test an invalid enum value."""

    @validate.each_in('test_enums', None, [self.ENUM_FOO, self.ENUM_BAR])
    def impl(_input_proto, _output_proto, _config):
      pass

    # Only invalid values.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._enums_request(self.ENUM_BAZ), None, self.api_config)
    # Mixed valid/invalid values.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._enums_request(self.ENUM_FOO, self.ENUM_BAZ), None,
           self.api_config)

  def test_scalar_not_in(self):
    """Test invalid scalar value."""

    @validate.each_in('numbers', None, [1, 2])
    def impl(_input_proto, _output_proto, _config):
      pass

    # Only invalid values.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._numbers_request(3), None, self.api_config)
    # Mixed valid/invalid values.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._numbers_request(1, 2, 3), None, self.api_config)

  def test_not_set(self):
    """Test an unset value."""

    @validate.each_in('messages', 'name', ['foo', 'bar'])
    def impl(_input_proto, _output_proto, _config):
      pass

    # Should be failing without a value set.
    # No entries in the field.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._message_request(), None, self.api_config)
    # No value set on lone entry.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._message_request([1]), None, self.api_config)
    # No value set on multiple entries.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._message_request([1], [2]), None, self.api_config)
    # Some valid and some invalid entries.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._message_request([1, 'foo'], [2]), None, self.api_config)

  def test_optional(self):
    """Test optional argument."""

    @validate.each_in('messages', 'name', ['foo', 'bar'], optional=True)
    @validate.each_in('test_enums', None, [self.ENUM_FOO, self.ENUM_BAR],
                      optional=True)
    @validate.each_in('numbers', None, [1, 2], optional=True)
    def impl(_input_proto, _output_proto, _config):
      pass

    # No entries in the field succeeds.
    impl(self._message_request(), None, self.api_config)

    # Still fails when entries exist but value unset cases.
    # No value set on lone entry.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._message_request([1]), None, self.api_config)
    # No value set on multiple entries.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._message_request([1], [2]), None, self.api_config)
    # Some valid and some invalid entries.
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._message_request([1, 'foo'], [2]), None, self.api_config)

  def test_skip_validation(self):
    """Test skipping validation case."""

    @validate.each_in('messages', 'name', ['foo', 'bar'])
    @validate.each_in('test_enums', None, [self.ENUM_FOO, self.ENUM_BAR])
    @validate.each_in('numbers', None, [1, 2])
    def impl(_input_proto, _output_proto, _config):
      pass

    # This would otherwise raise an error for multiple invalid fields.
    impl(self._message_request([1, 'invalid']), None, self.no_validate_config)


class RequireTest(cros_test_lib.TestCase, api_config.ApiConfigMixin):
  """Tests for the require validator."""

  def test_invalid_field(self):
    """Test validator fails when given an unset value."""

    @validate.require('does.not.exist')
    def impl(_input_proto, _output_proto, _config):
      self.fail('Incorrectly allowed method to execute.')

    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(common_pb2.Chroot(), None, self.api_config)

  def test_not_set(self):
    """Test validator fails when given an unset value."""

    @validate.require('env.use_flags')
    def impl(_input_proto, _output_proto, _config):
      self.fail('Incorrectly allowed method to execute.')

    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(common_pb2.Chroot(), None, self.api_config)

  def test_set(self):
    """Test validator passes when given set values."""

    @validate.require('path', 'env.use_flags')
    def impl(_input_proto, _output_proto, _config):
      pass

    in_proto = common_pb2.Chroot(path='/chroot/path',
                                 env={'use_flags': [{'flag': 'test'}]})
    impl(in_proto, None, self.api_config)

  def test_mixed(self):
    """Test validator fails when given a set value and an unset value."""

    @validate.require('path', 'env.use_flags')
    def impl(_input_proto, _output_proto, _config):
      pass

    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(common_pb2.Chroot(path='/chroot/path'), None, self.api_config)

  def test_skip_validation(self):
    """Test skipping validation case."""
    @validate.require('path', 'env.use_flags')
    def impl(_input_proto, _output_proto, _config):
      pass

    # This would otherwise raise an error for an invalid path.
    impl(common_pb2.Chroot(), None, self.no_validate_config)


class RequireAnyTest(cros_test_lib.TestCase, api_config.ApiConfigMixin):
  """Tests for the require_any validator."""

  def _get_request(self, mid: int = None, name: str = None, flag: bool = None):
    """Build a request instance from the given data."""
    request = build_api_test_pb2.MultiFieldMessage()

    if mid:
      request.id = mid
    if name:
      request.name = name
    if flag:
      request.flag = flag

    return request

  def test_invalid_field(self):
    """Test validator fails when given an invalid field."""

    @validate.require_any('does.not.exist', 'also.invalid')
    def impl(_input_proto, _output_proto, _config):
      self.fail('Incorrectly allowed method to execute.')

    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._get_request(), None, self.api_config)

  def test_not_set(self):
    """Test validator fails when given unset values."""

    @validate.require_any('id', 'name')
    def impl(_input_proto, _output_proto, _config):
      self.fail('Incorrectly allowed method to execute.')

    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._get_request(flag=True), None, self.api_config)

  def test_set(self):
    """Test validator passes when given set values."""

    @validate.require_any('id', 'name')
    def impl(_input_proto, _output_proto, _config):
      pass

    impl(self._get_request(1), None, self.api_config)
    impl(self._get_request(name='foo'), None, self.api_config)
    impl(self._get_request(1, name='foo'), None, self.api_config)


class RequireEachTest(cros_test_lib.TestCase, api_config.ApiConfigMixin):
  """Tests for the require_each validator."""

  def _multi_field_message(self, msg_id=None, name=None, flag=None):
    msg = build_api_test_pb2.MultiFieldMessage()
    if msg_id is not None:
      msg.id = int(msg_id)
    if name is not None:
      msg.name = str(name)
    if flag is not None:
      msg.flag = bool(flag)
    return msg

  def _request(self, messages=None, count=0):
    """Build the request."""
    if messages is None:
      messages = [self._multi_field_message() for _ in range(count)]

    request = build_api_test_pb2.TestRequestMessage()
    for message in messages:
      msg = request.messages.add()
      msg.CopyFrom(message)

    return request

  def test_invalid_field(self):
    """Test validator fails when given an invalid field."""

    @validate.require_each('does.not', ['exist'])
    def impl(_input_proto, _output_proto, _config):
      self.fail('Incorrectly allowed method to execute.')

    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._request(), None, self.api_config)

  def test_invalid_call_no_subfields(self):
    """Test validator fails when given no subfields."""

    with self.assertRaises(AssertionError):
      @validate.require_each('does.not', [])
      def _(_input_proto, _output_proto, _config):
        pass

  def test_invalid_call_invalid_subfields(self):
    """Test validator fails when given subfields incorrectly."""

    with self.assertRaises(AssertionError):
      @validate.require_each('does.not', 'exist')
      def _(_input_proto, _output_proto, _config):
        pass

  def test_not_set(self):
    """Test validator fails when given an unset value."""

    @validate.require_each('messages', ['id'])
    def impl(_input_proto, _output_proto, _config):
      self.fail('Incorrectly allowed method to execute.')

    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._request(count=2), None, self.api_config)

  def test_no_elements_success(self):
    """Test validator fails when given no messages in the repeated field."""

    @validate.require_each('messages', ['id'])
    def impl(_input_proto, _output_proto, _config):
      pass

    impl(self._request(), None, self.api_config)

  def test_no_elements_failure(self):
    """Test validator fails when given no messages in the repeated field."""

    @validate.require_each('messages', ['id'], allow_empty=False)
    def impl(_input_proto, _output_proto, _config):
      self.fail('Incorrectly allowed method to execute.')

    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._request(), None, self.api_config)

  def test_set(self):
    """Test validator passes when given set values."""

    @validate.require_each('messages', ['id'])
    def impl(_input_proto, _output_proto, _config):
      pass

    messages = [self._multi_field_message(msg_id=i) for i in range(1, 5)]
    impl(self._request(messages=messages), None, self.api_config)

  def test_one_set_fails(self):
    """Test validator passes when given set values."""

    @validate.require_each('messages', ['id', 'name'])
    def impl(_input_proto, _output_proto, _config):
      pass

    messages = [self._multi_field_message(msg_id=i) for i in range(1, 5)]
    with self.assertRaises(cros_build_lib.DieSystemExit):
      impl(self._request(messages=messages), None, self.api_config)

  def test_multi_set(self):
    """Test validator passes when all values set."""

    @validate.require_each('messages', ['id', 'name'])
    def impl(_input_proto, _output_proto, _config):
      pass

    messages = [self._multi_field_message(msg_id=i, name=i)
                for i in range(1, 5)]
    impl(self._request(messages=messages), None, self.api_config)

  def test_skip_validation(self):
    """Test skipping validation case."""
    @validate.require_each('messages', ['id'], allow_empty=False)
    def impl(_input_proto, _output_proto, _config):
      pass

    impl(self._request(), None, self.no_validate_config)


class ValidateOnlyTest(cros_test_lib.TestCase, api_config.ApiConfigMixin):
  """validate_only decorator tests."""

  def test_validate_only(self):
    """Test validate only."""
    @validate.require('path')
    @validate.validation_complete
    def impl(_input_proto, _output_proto, _config):
      self.fail('Implementation was called.')
      return 1

    # Just using arbitrary messages, we just need the
    # (request, response, config) arguments so it can check the config.
    rc = impl(common_pb2.Chroot(path='/chroot/path'), common_pb2.Chroot(),
              self.validate_only_config)

    self.assertEqual(0, rc)

  def test_no_validate_only(self):
    """Test no use of validate only."""
    @validate.validation_complete
    def impl(_input_proto, _output_proto, _config):
      self.fail('Incorrectly allowed method to execute.')

    # We will get an assertion error unless validate_only prevents the function
    # from being called.
    with self.assertRaises(AssertionError):
      impl(common_pb2.Chroot(), common_pb2.Chroot(), self.api_config)