| # 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. |
| |
| """Tests for the validate module.""" |
| |
| import os |
| |
| 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 |
| |
| |
| # 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) -> None: |
| """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(_request, _response, _config) -> None: |
| 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) -> None: |
| """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(_request, _response, _config) -> None: |
| pass |
| |
| impl(common_pb2.Chroot(path=path), None, self.api_config) |
| |
| def test_skip_validation(self) -> None: |
| """Test skipping validation case.""" |
| |
| @validate.exists("path") |
| def impl(_request, _response, _config) -> None: |
| pass |
| |
| # This would otherwise raise an error for an invalid path. |
| impl(common_pb2.Chroot(), None, self.no_validate_config) |
| |
| |
| class EqTest(cros_test_lib.TestCase, api_config.ApiConfigMixin): |
| """Tests for the eq validator.""" |
| |
| def test_eq(self) -> None: |
| """Test a valid value.""" |
| |
| @validate.eq("location", common_pb2.Path.Location.OUTSIDE) |
| def impl(_request, _response, _config) -> None: |
| pass |
| |
| impl( |
| common_pb2.Path( |
| path="/", location=common_pb2.Path.Location.OUTSIDE |
| ), |
| None, |
| self.api_config, |
| ) |
| |
| def test_not_eq(self) -> None: |
| """Test an invalid value.""" |
| |
| @validate.eq("location", common_pb2.Path.Location.OUTSIDE) |
| def impl(_request, _response, _config) -> None: |
| pass |
| |
| # Should be failing on the invalid value. |
| with self.assertRaises(cros_build_lib.DieSystemExit): |
| impl( |
| common_pb2.Path( |
| path="/", location=common_pb2.Path.Location.INSIDE |
| ), |
| None, |
| self.api_config, |
| ) |
| |
| def test_not_set(self) -> None: |
| """Test an unset value.""" |
| |
| @validate.eq("location", common_pb2.Path.Location.OUTSIDE) |
| def impl(_request, _response, _config) -> None: |
| pass |
| |
| # Should be failing without a value set. |
| with self.assertRaises(cros_build_lib.DieSystemExit): |
| impl(common_pb2.Path(path="/"), None, self.api_config) |
| |
| def test_skip_validation(self) -> None: |
| """Test skipping validation case.""" |
| |
| @validate.eq("location", common_pb2.Path.Location.OUTSIDE) |
| def impl(_request, _response, _config) -> None: |
| pass |
| |
| # This would otherwise raise an error for an invalid path. |
| impl(common_pb2.Path(path="/"), None, self.no_validate_config) |
| |
| |
| class IsInTest(cros_test_lib.TestCase, api_config.ApiConfigMixin): |
| """Tests for the is_in validator.""" |
| |
| def test_in(self) -> None: |
| """Test a valid value.""" |
| |
| @validate.is_in("path", ["/chroot/path", "/other/chroot/path"]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test an invalid value.""" |
| |
| @validate.is_in("path", ["/chroot/path", "/other/chroot/path"]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test an unset value.""" |
| |
| @validate.is_in("path", ["/chroot/path", "/other/chroot/path"]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test skipping validation case.""" |
| |
| @validate.is_in("path", ["/chroot/path", "/other/chroot/path"]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test valid values.""" |
| |
| @validate.each_in("messages", "name", ["foo", "bar"]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test valid enum values.""" |
| |
| @validate.each_in("test_enums", None, [self.ENUM_FOO, self.ENUM_BAR]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test valid scalar values.""" |
| |
| @validate.each_in("numbers", None, [1, 2]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test an invalid value.""" |
| |
| @validate.each_in("messages", "name", ["foo", "bar"]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test an invalid enum value.""" |
| |
| @validate.each_in("test_enums", None, [self.ENUM_FOO, self.ENUM_BAR]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test invalid scalar value.""" |
| |
| @validate.each_in("numbers", None, [1, 2]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test an unset value.""" |
| |
| @validate.each_in("messages", "name", ["foo", "bar"]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """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(_request, _response, _config) -> None: |
| 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) -> None: |
| """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(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validator fails when given an unset value.""" |
| |
| @validate.require("does.not.exist") |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validator fails when given an unset value.""" |
| |
| @validate.require("env.use_flags") |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validator passes when given set values.""" |
| |
| @validate.require("path", "env.use_flags") |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validator fails when given a set value and an unset value.""" |
| |
| @validate.require("path", "env.use_flags") |
| def impl(_request, _response, _config) -> None: |
| pass |
| |
| with self.assertRaises(cros_build_lib.DieSystemExit): |
| impl(common_pb2.Chroot(path="/chroot/path"), None, self.api_config) |
| |
| def test_skip_validation(self) -> None: |
| """Test skipping validation case.""" |
| |
| @validate.require("path", "env.use_flags") |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validator fails when given an invalid field.""" |
| |
| @validate.require_any("does.not.exist", "also.invalid") |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validator fails when given unset values.""" |
| |
| @validate.require_any("id", "name") |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validator passes when given set values.""" |
| |
| @validate.require_any("id", "name") |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validator fails when given an invalid field.""" |
| |
| @validate.require_each("does.not", ["exist"]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validator fails when given no subfields.""" |
| |
| with self.assertRaises(AssertionError): |
| |
| @validate.require_each("does.not", []) |
| def _(_request, _response, _config) -> None: |
| pass |
| |
| def test_invalid_call_invalid_subfields(self) -> None: |
| """Test validator fails when given subfields incorrectly.""" |
| |
| with self.assertRaises(AssertionError): |
| |
| @validate.require_each("does.not", "exist") |
| def _(_request, _response, _config) -> None: |
| pass |
| |
| def test_not_set(self) -> None: |
| """Test validator fails when given an unset value.""" |
| |
| @validate.require_each("messages", ["id"]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validator fails when given no messages in the repeated field.""" |
| |
| @validate.require_each("messages", ["id"]) |
| def impl(_request, _response, _config) -> None: |
| pass |
| |
| impl(self._request(), None, self.api_config) |
| |
| def test_no_elements_failure(self) -> None: |
| """Test validator fails when given no messages in the repeated field.""" |
| |
| @validate.require_each("messages", ["id"], allow_empty=False) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validator passes when given set values.""" |
| |
| @validate.require_each("messages", ["id"]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validator passes when given set values.""" |
| |
| @validate.require_each("messages", ["id", "name"]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validator passes when all values set.""" |
| |
| @validate.require_each("messages", ["id", "name"]) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test skipping validation case.""" |
| |
| @validate.require_each("messages", ["id"], allow_empty=False) |
| def impl(_request, _response, _config) -> None: |
| 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) -> None: |
| """Test validate only.""" |
| |
| @validate.require("path") |
| @validate.validation_complete |
| def impl(_request, _response, _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) -> None: |
| """Test no use of validate only.""" |
| |
| @validate.validation_complete |
| def impl(_request, _response, _config) -> None: |
| 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) |
