Formatting: Format all python code with black.
This CL is probably not what you're looking for, it's only
automated formatting. Ignore it with
`git blame --ignore-rev <revision>` for this commit.
BUG=b:233893248
TEST=CQ
Change-Id: I66591d7a738d241aed3290138c0f68065ab10a6d
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/chromite/+/3879174
Reviewed-by: Mike Frysinger <vapier@chromium.org>
Tested-by: Alex Klein <saklein@chromium.org>
diff --git a/api/validate_unittest.py b/api/validate_unittest.py
index 6f15a10..2a8f414 100644
--- a/api/validate_unittest.py
+++ b/api/validate_unittest.py
@@ -23,511 +23,548 @@
# 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."""
+ """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')
+ 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.')
+ @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)
+ 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)
+ 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
+ @validate.exists("path")
+ def impl(_input_proto, _output_proto, _config):
+ pass
- impl(common_pb2.Chroot(path=path), None, self.api_config)
+ 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
+ def test_skip_validation(self):
+ """Test skipping validation case."""
- # This would otherwise raise an error for an invalid path.
- impl(common_pb2.Chroot(), None, self.no_validate_config)
+ @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."""
+ """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
+ def test_in(self):
+ """Test a valid value."""
- # 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)
+ @validate.is_in("path", ["/chroot/path", "/other/chroot/path"])
+ def impl(_input_proto, _output_proto, _config):
+ pass
- 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
+ # 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
+ )
- # 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_in(self):
+ """Test an invalid value."""
- 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
+ @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)
+ # 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_skip_validation(self):
- """Test skipping validation case."""
- @validate.is_in('path', ['/chroot/path', '/other/chroot/path'])
- def impl(_input_proto, _output_proto, _config):
- pass
+ def test_not_set(self):
+ """Test an unset value."""
- # This would otherwise raise an error for an invalid path.
- impl(common_pb2.Chroot(), None, self.no_validate_config)
+ @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."""
+ """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
+ # 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.
+ # 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
+ 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
+ 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)
+ 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
+ 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)
+ 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
+ return request
- def test_message_in(self):
- """Test valid values."""
+ def test_message_in(self):
+ """Test valid values."""
- @validate.each_in('messages', 'name', ['foo', 'bar'])
- def impl(_input_proto, _output_proto, _config):
- pass
+ @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)
+ 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."""
+ 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
+ @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)
+ 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."""
+ def test_scalar_in(self):
+ """Test valid scalar values."""
- @validate.each_in('numbers', None, [1, 2])
- def impl(_input_proto, _output_proto, _config):
- pass
+ @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)
+ 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."""
+ 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
+ @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)
+ # 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."""
+ 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
+ @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)
+ # 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."""
+ 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
+ @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)
+ # 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."""
+ def test_not_set(self):
+ """Test an unset value."""
- @validate.each_in('messages', 'name', ['foo', 'bar'])
- def impl(_input_proto, _output_proto, _config):
- pass
+ @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)
+ # 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."""
+ 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
+ @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)
+ # 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)
+ # 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."""
+ 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
+ @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)
+ # 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."""
+ """Tests for the require validator."""
- def test_invalid_field(self):
- """Test validator fails when given an unset value."""
+ 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.')
+ @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)
+ 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."""
+ 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.')
+ @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)
+ 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."""
+ 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
+ @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)
+ 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."""
+ 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
+ @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)
+ 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
+ def test_skip_validation(self):
+ """Test skipping validation case."""
- # This would otherwise raise an error for an invalid path.
- impl(common_pb2.Chroot(), None, self.no_validate_config)
+ @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."""
+ """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()
+ 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
+ if mid:
+ request.id = mid
+ if name:
+ request.name = name
+ if flag:
+ request.flag = flag
- return request
+ return request
- def test_invalid_field(self):
- """Test validator fails when given an invalid field."""
+ 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.')
+ @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)
+ 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."""
+ 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.')
+ @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)
+ 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."""
+ def test_set(self):
+ """Test validator passes when given set values."""
- @validate.require_any('id', 'name')
- def impl(_input_proto, _output_proto, _config):
- pass
+ @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)
+ 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."""
+ """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 _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)]
+ 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)
+ request = build_api_test_pb2.TestRequestMessage()
+ for message in messages:
+ msg = request.messages.add()
+ msg.CopyFrom(message)
- return request
+ return request
- def test_invalid_field(self):
- """Test validator fails when given an invalid field."""
+ 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.')
+ @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)
+ 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."""
+ 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
+ with self.assertRaises(AssertionError):
- def test_invalid_call_invalid_subfields(self):
- """Test validator fails when given subfields incorrectly."""
+ @validate.require_each("does.not", [])
+ def _(_input_proto, _output_proto, _config):
+ pass
- with self.assertRaises(AssertionError):
- @validate.require_each('does.not', 'exist')
- def _(_input_proto, _output_proto, _config):
- pass
+ def test_invalid_call_invalid_subfields(self):
+ """Test validator fails when given subfields incorrectly."""
- def test_not_set(self):
- """Test validator fails when given an unset value."""
+ with self.assertRaises(AssertionError):
- @validate.require_each('messages', ['id'])
- def impl(_input_proto, _output_proto, _config):
- self.fail('Incorrectly allowed method to execute.')
+ @validate.require_each("does.not", "exist")
+ def _(_input_proto, _output_proto, _config):
+ pass
- with self.assertRaises(cros_build_lib.DieSystemExit):
- impl(self._request(count=2), None, self.api_config)
+ def test_not_set(self):
+ """Test validator fails when given an unset value."""
- 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):
+ self.fail("Incorrectly allowed method to execute.")
- @validate.require_each('messages', ['id'])
- def impl(_input_proto, _output_proto, _config):
- pass
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ impl(self._request(count=2), None, self.api_config)
- impl(self._request(), None, self.api_config)
+ def test_no_elements_success(self):
+ """Test validator fails when given no messages in the repeated field."""
- def test_no_elements_failure(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
- @validate.require_each('messages', ['id'], allow_empty=False)
- def impl(_input_proto, _output_proto, _config):
- self.fail('Incorrectly allowed method to execute.')
+ impl(self._request(), None, self.api_config)
- with self.assertRaises(cros_build_lib.DieSystemExit):
- impl(self._request(), None, self.api_config)
+ def test_no_elements_failure(self):
+ """Test validator fails when given no messages in the repeated field."""
- def test_set(self):
- """Test validator passes when given set values."""
+ @validate.require_each("messages", ["id"], allow_empty=False)
+ def impl(_input_proto, _output_proto, _config):
+ self.fail("Incorrectly allowed method to execute.")
- @validate.require_each('messages', ['id'])
- def impl(_input_proto, _output_proto, _config):
- pass
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ impl(self._request(), None, self.api_config)
- 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_set(self):
+ """Test validator passes when given set values."""
- def test_one_set_fails(self):
- """Test validator passes when given set values."""
+ @validate.require_each("messages", ["id"])
+ def impl(_input_proto, _output_proto, _config):
+ pass
- @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)]
+ impl(self._request(messages=messages), None, self.api_config)
- 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_one_set_fails(self):
+ """Test validator passes when given set values."""
- 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
- @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)
- 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_multi_set(self):
+ """Test validator passes when all values set."""
- 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
+ @validate.require_each("messages", ["id", "name"])
+ def impl(_input_proto, _output_proto, _config):
+ pass
- impl(self._request(), None, self.no_validate_config)
+ 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."""
+ """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
+ def test_validate_only(self):
+ """Test validate only."""
- # 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)
+ @validate.require("path")
+ @validate.validation_complete
+ def impl(_input_proto, _output_proto, _config):
+ self.fail("Implementation was called.")
+ return 1
- self.assertEqual(0, rc)
+ # 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,
+ )
- 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.')
+ self.assertEqual(0, rc)
- # 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)
+ 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)