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/router_unittest.py b/api/router_unittest.py
index 7d3a755..db0c452 100644
--- a/api/router_unittest.py
+++ b/api/router_unittest.py
@@ -19,367 +19,504 @@
from chromite.lib import osutils
-class RouterTest(cros_test_lib.RunCommandTempDirTestCase,
- api_config.ApiConfigMixin):
- """Test Router functionality."""
+class RouterTest(
+ cros_test_lib.RunCommandTempDirTestCase, api_config.ApiConfigMixin
+):
+ """Test Router functionality."""
- def setUp(self):
- self.router = router.Router()
- self.router.Register(build_api_test_pb2)
+ def setUp(self):
+ self.router = router.Router()
+ self.router.Register(build_api_test_pb2)
- self.chroot_dir = os.path.join(self.tempdir, 'chroot')
- chroot_tmp = os.path.join(self.chroot_dir, 'tmp')
- # Make the tmp dir for the re-exec inside chroot input/output files.
- osutils.SafeMakedirs(chroot_tmp)
+ self.chroot_dir = os.path.join(self.tempdir, "chroot")
+ chroot_tmp = os.path.join(self.chroot_dir, "tmp")
+ # Make the tmp dir for the re-exec inside chroot input/output files.
+ osutils.SafeMakedirs(chroot_tmp)
- # Build the input/output/config paths we'll be using in the tests.
- self.json_input_file = os.path.join(self.tempdir, 'input.json')
- self.json_output_file = os.path.join(self.tempdir, 'output.json')
- self.json_config_file = os.path.join(self.tempdir, 'config.json')
- self.binary_input_file = os.path.join(self.tempdir, 'input.bin')
- self.binary_output_file = os.path.join(self.tempdir, 'output.bin')
- self.binary_config_file = os.path.join(self.tempdir, 'config.bin')
+ # Build the input/output/config paths we'll be using in the tests.
+ self.json_input_file = os.path.join(self.tempdir, "input.json")
+ self.json_output_file = os.path.join(self.tempdir, "output.json")
+ self.json_config_file = os.path.join(self.tempdir, "config.json")
+ self.binary_input_file = os.path.join(self.tempdir, "input.bin")
+ self.binary_output_file = os.path.join(self.tempdir, "output.bin")
+ self.binary_config_file = os.path.join(self.tempdir, "config.bin")
- # The message handlers for the respective files.
- self.json_input_handler = message_util.get_message_handler(
- self.json_input_file, message_util.FORMAT_JSON)
- self.json_output_handler = message_util.get_message_handler(
- self.json_output_file, message_util.FORMAT_JSON)
- self.json_config_handler = message_util.get_message_handler(
- self.json_config_file, message_util.FORMAT_JSON)
- self.binary_input_handler = message_util.get_message_handler(
- self.binary_input_file, message_util.FORMAT_BINARY)
- self.binary_output_handler = message_util.get_message_handler(
- self.binary_output_file, message_util.FORMAT_BINARY)
- self.binary_config_handler = message_util.get_message_handler(
- self.binary_config_file, message_util.FORMAT_BINARY)
+ # The message handlers for the respective files.
+ self.json_input_handler = message_util.get_message_handler(
+ self.json_input_file, message_util.FORMAT_JSON
+ )
+ self.json_output_handler = message_util.get_message_handler(
+ self.json_output_file, message_util.FORMAT_JSON
+ )
+ self.json_config_handler = message_util.get_message_handler(
+ self.json_config_file, message_util.FORMAT_JSON
+ )
+ self.binary_input_handler = message_util.get_message_handler(
+ self.binary_input_file, message_util.FORMAT_BINARY
+ )
+ self.binary_output_handler = message_util.get_message_handler(
+ self.binary_output_file, message_util.FORMAT_BINARY
+ )
+ self.binary_config_handler = message_util.get_message_handler(
+ self.binary_config_file, message_util.FORMAT_BINARY
+ )
- # Build an input message to use.
- self.expected_id = 'input id'
- input_msg = build_api_test_pb2.TestRequestMessage()
- input_msg.id = self.expected_id
- input_msg.chroot.path = self.chroot_dir
+ # Build an input message to use.
+ self.expected_id = "input id"
+ input_msg = build_api_test_pb2.TestRequestMessage()
+ input_msg.id = self.expected_id
+ input_msg.chroot.path = self.chroot_dir
- # Write out base input and config messages.
- osutils.WriteFile(self.json_input_file,
- json_format.MessageToJson(input_msg))
- osutils.WriteFile(
- self.binary_input_file, input_msg.SerializeToString(), mode='wb')
+ # Write out base input and config messages.
+ osutils.WriteFile(
+ self.json_input_file, json_format.MessageToJson(input_msg)
+ )
+ osutils.WriteFile(
+ self.binary_input_file, input_msg.SerializeToString(), mode="wb"
+ )
- config_msg = self.api_config.get_proto()
- osutils.WriteFile(self.json_config_file,
- json_format.MessageToJson(config_msg))
- osutils.WriteFile(
- self.binary_config_file, config_msg.SerializeToString(), mode='wb')
+ config_msg = self.api_config.get_proto()
+ osutils.WriteFile(
+ self.json_config_file, json_format.MessageToJson(config_msg)
+ )
+ osutils.WriteFile(
+ self.binary_config_file, config_msg.SerializeToString(), mode="wb"
+ )
- self.subprocess_tempdir = os.path.join(self.chroot_dir, 'tempdir')
- osutils.SafeMakedirs(self.subprocess_tempdir)
+ self.subprocess_tempdir = os.path.join(self.chroot_dir, "tempdir")
+ osutils.SafeMakedirs(self.subprocess_tempdir)
- def testJsonInputOutputMethod(self):
- """Test json input/output handling."""
- def impl(input_msg, output_msg, config):
- self.assertIsInstance(input_msg, build_api_test_pb2.TestRequestMessage)
- self.assertIsInstance(output_msg, build_api_test_pb2.TestResultMessage)
- self.assertIsInstance(config, api_config.ApiConfig)
- self.assertEqual(config, self.api_config)
+ def testJsonInputOutputMethod(self):
+ """Test json input/output handling."""
- self.PatchObject(self.router, '_GetMethod', return_value=impl)
+ def impl(input_msg, output_msg, config):
+ self.assertIsInstance(
+ input_msg, build_api_test_pb2.TestRequestMessage
+ )
+ self.assertIsInstance(
+ output_msg, build_api_test_pb2.TestResultMessage
+ )
+ self.assertIsInstance(config, api_config.ApiConfig)
+ self.assertEqual(config, self.api_config)
- self.router.Route(
- 'chromite.api.TestApiService',
- 'InputOutputMethod',
- self.api_config,
- self.json_input_handler,
- [self.json_output_handler],
- self.json_config_handler)
+ self.PatchObject(self.router, "_GetMethod", return_value=impl)
- def testBinaryInputOutputMethod(self):
- """Test binary input/output handling."""
- def impl(input_msg, output_msg, config):
- self.assertIsInstance(input_msg, build_api_test_pb2.TestRequestMessage)
- self.assertIsInstance(output_msg, build_api_test_pb2.TestResultMessage)
- self.assertIsInstance(config, api_config.ApiConfig)
- self.assertEqual(config, self.api_config)
+ self.router.Route(
+ "chromite.api.TestApiService",
+ "InputOutputMethod",
+ self.api_config,
+ self.json_input_handler,
+ [self.json_output_handler],
+ self.json_config_handler,
+ )
- self.PatchObject(self.router, '_GetMethod', return_value=impl)
+ def testBinaryInputOutputMethod(self):
+ """Test binary input/output handling."""
- self.router.Route(
- 'chromite.api.TestApiService',
- 'InputOutputMethod',
- self.api_config,
- self.binary_input_handler,
- [self.binary_output_handler],
- self.binary_config_handler)
+ def impl(input_msg, output_msg, config):
+ self.assertIsInstance(
+ input_msg, build_api_test_pb2.TestRequestMessage
+ )
+ self.assertIsInstance(
+ output_msg, build_api_test_pb2.TestResultMessage
+ )
+ self.assertIsInstance(config, api_config.ApiConfig)
+ self.assertEqual(config, self.api_config)
- def testMultipleOutputHandling(self):
- """Test multiple output handling."""
- expected_result = 'Success!'
+ self.PatchObject(self.router, "_GetMethod", return_value=impl)
- def impl(input_msg, output_msg, config):
- self.assertIsInstance(input_msg, build_api_test_pb2.TestRequestMessage)
- self.assertIsInstance(output_msg, build_api_test_pb2.TestResultMessage)
- self.assertIsInstance(config, api_config.ApiConfig)
- self.assertEqual(config, self.api_config)
- # Set the property on the output to test against.
- output_msg.result = expected_result
+ self.router.Route(
+ "chromite.api.TestApiService",
+ "InputOutputMethod",
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
- self.PatchObject(self.router, '_GetMethod', return_value=impl)
+ def testMultipleOutputHandling(self):
+ """Test multiple output handling."""
+ expected_result = "Success!"
- self.router.Route(
- 'chromite.api.TestApiService',
- 'InputOutputMethod',
- self.api_config,
- self.binary_input_handler,
- [self.binary_output_handler, self.json_output_handler],
- self.binary_config_handler)
+ def impl(input_msg, output_msg, config):
+ self.assertIsInstance(
+ input_msg, build_api_test_pb2.TestRequestMessage
+ )
+ self.assertIsInstance(
+ output_msg, build_api_test_pb2.TestResultMessage
+ )
+ self.assertIsInstance(config, api_config.ApiConfig)
+ self.assertEqual(config, self.api_config)
+ # Set the property on the output to test against.
+ output_msg.result = expected_result
- # Make sure it did write out all the expected files.
- self.assertExists(self.binary_output_file)
- self.assertExists(self.json_output_file)
+ self.PatchObject(self.router, "_GetMethod", return_value=impl)
- # Parse the output files back into a message.
- binary_msg = build_api_test_pb2.TestResultMessage()
- json_msg = build_api_test_pb2.TestResultMessage()
- self.binary_output_handler.read_into(binary_msg)
- self.json_output_handler.read_into(json_msg)
+ self.router.Route(
+ "chromite.api.TestApiService",
+ "InputOutputMethod",
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler, self.json_output_handler],
+ self.binary_config_handler,
+ )
- # Make sure the parsed messages have the expected content.
- self.assertEqual(binary_msg.result, expected_result)
- self.assertEqual(json_msg.result, expected_result)
+ # Make sure it did write out all the expected files.
+ self.assertExists(self.binary_output_file)
+ self.assertExists(self.json_output_file)
- def testRenameMethod(self):
- """Test implementation name config."""
- def _GetMethod(_, method_name):
- self.assertEqual('CorrectName', method_name)
- return lambda x, y, z: None
+ # Parse the output files back into a message.
+ binary_msg = build_api_test_pb2.TestResultMessage()
+ json_msg = build_api_test_pb2.TestResultMessage()
+ self.binary_output_handler.read_into(binary_msg)
+ self.json_output_handler.read_into(json_msg)
- self.PatchObject(self.router, '_GetMethod', side_effect=_GetMethod)
+ # Make sure the parsed messages have the expected content.
+ self.assertEqual(binary_msg.result, expected_result)
+ self.assertEqual(json_msg.result, expected_result)
- self.router.Route('chromite.api.TestApiService', 'RenamedMethod',
- self.api_config, self.binary_input_handler,
- [self.binary_output_handler], self.binary_config_handler)
+ def testRenameMethod(self):
+ """Test implementation name config."""
- def _mock_callable(self, expect_called: bool) -> Callable:
- """Helper to create the implementation mock to test chroot assertions.
+ def _GetMethod(_, method_name):
+ self.assertEqual("CorrectName", method_name)
+ return lambda x, y, z: None
- Args:
- expect_called: Whether the implementation should be called.
- When False, an assertion will fail if it is called.
+ self.PatchObject(self.router, "_GetMethod", side_effect=_GetMethod)
- Returns:
- The implementation.
- """
- def impl(_input_msg, _output_msg, _config):
- self.assertTrue(expect_called,
- 'The implementation should not have been called.')
+ self.router.Route(
+ "chromite.api.TestApiService",
+ "RenamedMethod",
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
- return impl
+ def _mock_callable(self, expect_called: bool) -> Callable:
+ """Helper to create the implementation mock to test chroot assertions.
- def _writeChrootCallOutput(self, content='{}', mode='w'):
- def impl(*_args, **_kwargs):
- """Side effect for inside-chroot calls to the API."""
- osutils.WriteFile(
- os.path.join(self.subprocess_tempdir,
- router.Router.REEXEC_OUTPUT_FILE),
- content,
- mode=mode)
+ Args:
+ expect_called: Whether the implementation should be called.
+ When False, an assertion will fail if it is called.
- return impl
+ Returns:
+ The implementation.
+ """
- def testInsideServiceInsideMethodInsideChroot(self):
- """Test inside/inside/inside works correctly."""
- self.PatchObject(self.router, '_GetMethod',
- return_value=self._mock_callable(expect_called=True))
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=True)
- self.router.Route('chromite.api.InsideChrootApiService',
- 'InsideServiceInsideMethod', self.api_config,
- self.binary_input_handler, [self.binary_output_handler],
- self.binary_config_handler)
+ def impl(_input_msg, _output_msg, _config):
+ self.assertTrue(
+ expect_called, "The implementation should not have been called."
+ )
- def testInsideServiceOutsideMethodOutsideChroot(self):
- """Test the outside method override works as expected."""
- self.PatchObject(self.router, '_GetMethod',
- return_value=self._mock_callable(expect_called=True))
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=False)
- self.router.Route('chromite.api.InsideChrootApiService',
- 'InsideServiceOutsideMethod', self.api_config,
- self.binary_input_handler, [self.binary_output_handler],
- self.binary_config_handler)
+ return impl
- def testInsideServiceInsideMethodOutsideChroot(self):
- """Test calling an inside method from outside the chroot."""
- self.PatchObject(self.router, '_GetMethod',
- return_value=self._mock_callable(expect_called=False))
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=False)
+ def _writeChrootCallOutput(self, content="{}", mode="w"):
+ def impl(*_args, **_kwargs):
+ """Side effect for inside-chroot calls to the API."""
+ osutils.WriteFile(
+ os.path.join(
+ self.subprocess_tempdir, router.Router.REEXEC_OUTPUT_FILE
+ ),
+ content,
+ mode=mode,
+ )
- service = 'chromite.api.InsideChrootApiService'
- method = 'InsideServiceInsideMethod'
- service_method = '%s/%s' % (service, method)
- self.router.Route(service, method, self.api_config,
- self.binary_input_handler, [self.binary_output_handler],
- self.binary_config_handler)
+ return impl
- self.assertCommandContains(['build_api', service_method], enter_chroot=True)
+ def testInsideServiceInsideMethodInsideChroot(self):
+ """Test inside/inside/inside works correctly."""
+ self.PatchObject(
+ self.router,
+ "_GetMethod",
+ return_value=self._mock_callable(expect_called=True),
+ )
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=True)
+ self.router.Route(
+ "chromite.api.InsideChrootApiService",
+ "InsideServiceInsideMethod",
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
- def testInsideServiceOutsideMethodInsideChroot(self):
- """Test inside chroot for outside method raises an error."""
- self.PatchObject(self.router, '_GetMethod',
- return_value=self._mock_callable(expect_called=False))
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=True)
- with self.assertRaises(cros_build_lib.DieSystemExit):
- self.router.Route('chromite.api.InsideChrootApiService',
- 'InsideServiceOutsideMethod', self.api_config,
- self.binary_input_handler, [self.binary_output_handler],
- self.binary_config_handler)
+ def testInsideServiceOutsideMethodOutsideChroot(self):
+ """Test the outside method override works as expected."""
+ self.PatchObject(
+ self.router,
+ "_GetMethod",
+ return_value=self._mock_callable(expect_called=True),
+ )
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)
+ self.router.Route(
+ "chromite.api.InsideChrootApiService",
+ "InsideServiceOutsideMethod",
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
- def testOutsideServiceOutsideMethodOutsideChroot(self):
- """Test outside/outside/outside works correctly."""
- self.PatchObject(self.router, '_GetMethod',
- return_value=self._mock_callable(expect_called=True))
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=False)
- self.router.Route('chromite.api.OutsideChrootApiService',
- 'OutsideServiceOutsideMethod', self.api_config,
- self.binary_input_handler, [self.binary_output_handler],
- self.binary_config_handler)
+ def testInsideServiceInsideMethodOutsideChroot(self):
+ """Test calling an inside method from outside the chroot."""
+ self.PatchObject(
+ self.router,
+ "_GetMethod",
+ return_value=self._mock_callable(expect_called=False),
+ )
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)
- def testOutsideServiceInsideMethodInsideChroot(self):
- """Test the inside method assertion override works properly."""
- self.PatchObject(self.router, '_GetMethod',
- return_value=self._mock_callable(expect_called=True))
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=True)
- self.router.Route('chromite.api.OutsideChrootApiService',
- 'OutsideServiceInsideMethod', self.api_config,
- self.binary_input_handler, [self.binary_output_handler],
- self.binary_config_handler)
+ service = "chromite.api.InsideChrootApiService"
+ method = "InsideServiceInsideMethod"
+ service_method = "%s/%s" % (service, method)
+ self.router.Route(
+ service,
+ method,
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
- def testOutsideServiceInsideMethodOutsideChroot(self):
- """Test calling an inside override method from outside the chroot."""
- self.PatchObject(self.router, '_GetMethod',
- return_value=self._mock_callable(expect_called=False))
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=False)
+ self.assertCommandContains(
+ ["build_api", service_method], enter_chroot=True
+ )
- service = 'chromite.api.OutsideChrootApiService'
- method = 'OutsideServiceInsideMethod'
- service_method = '%s/%s' % (service, method)
- self.router.Route(service, method, self.api_config,
- self.binary_input_handler, [self.binary_output_handler],
- self.binary_config_handler)
+ def testInsideServiceOutsideMethodInsideChroot(self):
+ """Test inside chroot for outside method raises an error."""
+ self.PatchObject(
+ self.router,
+ "_GetMethod",
+ return_value=self._mock_callable(expect_called=False),
+ )
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=True)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ self.router.Route(
+ "chromite.api.InsideChrootApiService",
+ "InsideServiceOutsideMethod",
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
- self.assertCommandContains(['build_api', service_method], enter_chroot=True)
+ def testOutsideServiceOutsideMethodOutsideChroot(self):
+ """Test outside/outside/outside works correctly."""
+ self.PatchObject(
+ self.router,
+ "_GetMethod",
+ return_value=self._mock_callable(expect_called=True),
+ )
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)
+ self.router.Route(
+ "chromite.api.OutsideChrootApiService",
+ "OutsideServiceOutsideMethod",
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
- def testReexecNonemptyOutput(self):
- """Test calling an inside chroot method that produced output."""
- self.PatchObject(self.router, '_GetMethod',
- return_value=self._mock_callable(expect_called=False))
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=False)
+ def testOutsideServiceInsideMethodInsideChroot(self):
+ """Test the inside method assertion override works properly."""
+ self.PatchObject(
+ self.router,
+ "_GetMethod",
+ return_value=self._mock_callable(expect_called=True),
+ )
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=True)
+ self.router.Route(
+ "chromite.api.OutsideChrootApiService",
+ "OutsideServiceInsideMethod",
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
- # Patch the chroot tempdir method to return a tempdir with our subprocess
- # tempdir so the output file will be in the expected location.
- tempdir = osutils.TempDir()
- original = tempdir.tempdir
- tempdir.tempdir = self.subprocess_tempdir
- self.PatchObject(chroot_lib.Chroot, 'tempdir', return_value=tempdir)
+ def testOutsideServiceInsideMethodOutsideChroot(self):
+ """Test calling an inside override method from outside the chroot."""
+ self.PatchObject(
+ self.router,
+ "_GetMethod",
+ return_value=self._mock_callable(expect_called=False),
+ )
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)
- expected_output_msg = build_api_test_pb2.TestResultMessage()
- expected_output_msg.result = 'foo'
+ service = "chromite.api.OutsideChrootApiService"
+ method = "OutsideServiceInsideMethod"
+ service_method = "%s/%s" % (service, method)
+ self.router.Route(
+ service,
+ method,
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
- # Set the command side effect to write out our expected output to the
- # output file for the inside the chroot reexecution of the endpoint.
- # This lets us make sure the logic moving everything out works as intended.
- self.rc.SetDefaultCmdResult(
- side_effect=self._writeChrootCallOutput(
- content=expected_output_msg.SerializeToString(), mode='wb'))
+ self.assertCommandContains(
+ ["build_api", service_method], enter_chroot=True
+ )
- service = 'chromite.api.OutsideChrootApiService'
- method = 'OutsideServiceInsideMethod'
- service_method = '%s/%s' % (service, method)
- self.router.Route(service, method, self.api_config,
- self.binary_input_handler, [self.binary_output_handler],
- self.binary_config_handler)
+ def testReexecNonemptyOutput(self):
+ """Test calling an inside chroot method that produced output."""
+ self.PatchObject(
+ self.router,
+ "_GetMethod",
+ return_value=self._mock_callable(expect_called=False),
+ )
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)
- self.assertCommandContains(['build_api', service_method], enter_chroot=True)
+ # Patch the chroot tempdir method to return a tempdir with our subprocess
+ # tempdir so the output file will be in the expected location.
+ tempdir = osutils.TempDir()
+ original = tempdir.tempdir
+ tempdir.tempdir = self.subprocess_tempdir
+ self.PatchObject(chroot_lib.Chroot, "tempdir", return_value=tempdir)
- # It should be writing the result out to our output file.
- output_msg = build_api_test_pb2.TestResultMessage()
- self.binary_output_handler.read_into(output_msg)
- self.assertEqual(expected_output_msg, output_msg)
+ expected_output_msg = build_api_test_pb2.TestResultMessage()
+ expected_output_msg.result = "foo"
- tempdir.tempdir = original
- del tempdir
+ # Set the command side effect to write out our expected output to the
+ # output file for the inside the chroot reexecution of the endpoint.
+ # This lets us make sure the logic moving everything out works as intended.
+ self.rc.SetDefaultCmdResult(
+ side_effect=self._writeChrootCallOutput(
+ content=expected_output_msg.SerializeToString(), mode="wb"
+ )
+ )
- def testReexecEmptyOutput(self):
- """Test calling an inside chroot method that produced no output."""
- self.PatchObject(self.router, '_GetMethod',
- return_value=self._mock_callable(expect_called=False))
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=False)
- expected_output_msg = build_api_test_pb2.TestResultMessage()
+ service = "chromite.api.OutsideChrootApiService"
+ method = "OutsideServiceInsideMethod"
+ service_method = "%s/%s" % (service, method)
+ self.router.Route(
+ service,
+ method,
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
- # Set the command side effect to write out our expected output to the
- # output file for the inside the chroot reexecution of the endpoint.
- # This lets us make sure the logic moving everything out works as intended.
- self.rc.SetDefaultCmdResult(
- side_effect=self._writeChrootCallOutput(
- content=expected_output_msg.SerializeToString(), mode='wb'))
+ self.assertCommandContains(
+ ["build_api", service_method], enter_chroot=True
+ )
- service = 'chromite.api.OutsideChrootApiService'
- method = 'OutsideServiceInsideMethod'
- service_method = '%s/%s' % (service, method)
- self.router.Route(service, method, self.api_config,
- self.binary_input_handler, [self.binary_output_handler],
- self.binary_config_handler)
+ # It should be writing the result out to our output file.
+ output_msg = build_api_test_pb2.TestResultMessage()
+ self.binary_output_handler.read_into(output_msg)
+ self.assertEqual(expected_output_msg, output_msg)
- self.assertCommandContains(['build_api', service_method], enter_chroot=True)
+ tempdir.tempdir = original
+ del tempdir
- output_msg = build_api_test_pb2.TestResultMessage()
- self.binary_output_handler.read_into(output_msg)
- self.assertEqual(expected_output_msg, output_msg)
+ def testReexecEmptyOutput(self):
+ """Test calling an inside chroot method that produced no output."""
+ self.PatchObject(
+ self.router,
+ "_GetMethod",
+ return_value=self._mock_callable(expect_called=False),
+ )
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)
+ expected_output_msg = build_api_test_pb2.TestResultMessage()
- def testReexecNoOutput(self):
- """Test calling an inside chroot method that produced no output."""
- self.PatchObject(self.router, '_GetMethod',
- return_value=self._mock_callable(expect_called=False))
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=False)
- self.rc.SetDefaultCmdResult(returncode=1)
+ # Set the command side effect to write out our expected output to the
+ # output file for the inside the chroot reexecution of the endpoint.
+ # This lets us make sure the logic moving everything out works as intended.
+ self.rc.SetDefaultCmdResult(
+ side_effect=self._writeChrootCallOutput(
+ content=expected_output_msg.SerializeToString(), mode="wb"
+ )
+ )
- service = 'chromite.api.OutsideChrootApiService'
- method = 'OutsideServiceInsideMethod'
- service_method = '%s/%s' % (service, method)
- self.router.Route(service, method, self.api_config,
- self.binary_input_handler, [self.binary_output_handler],
- self.binary_config_handler)
+ service = "chromite.api.OutsideChrootApiService"
+ method = "OutsideServiceInsideMethod"
+ service_method = "%s/%s" % (service, method)
+ self.router.Route(
+ service,
+ method,
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
- self.assertCommandContains(['build_api', service_method], enter_chroot=True)
+ self.assertCommandContains(
+ ["build_api", service_method], enter_chroot=True
+ )
- output_msg = build_api_test_pb2.TestResultMessage()
- empty_msg = build_api_test_pb2.TestResultMessage()
- self.binary_output_handler.read_into(output_msg)
- self.assertEqual(empty_msg, output_msg)
+ output_msg = build_api_test_pb2.TestResultMessage()
+ self.binary_output_handler.read_into(output_msg)
+ self.assertEqual(expected_output_msg, output_msg)
- def testInvalidService(self):
- """Test invalid service call."""
- service = 'chromite.api.DoesNotExist'
- method = 'OutsideServiceInsideMethod'
+ def testReexecNoOutput(self):
+ """Test calling an inside chroot method that produced no output."""
+ self.PatchObject(
+ self.router,
+ "_GetMethod",
+ return_value=self._mock_callable(expect_called=False),
+ )
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)
+ self.rc.SetDefaultCmdResult(returncode=1)
- with self.assertRaises(router.UnknownServiceError):
- self.router.Route(service, method, self.api_config,
- self.binary_input_handler, [self.binary_output_handler],
- self.binary_config_handler)
+ service = "chromite.api.OutsideChrootApiService"
+ method = "OutsideServiceInsideMethod"
+ service_method = "%s/%s" % (service, method)
+ self.router.Route(
+ service,
+ method,
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
- def testInvalidMethod(self):
- """Test invalid method call."""
- service = 'chromite.api.OutsideChrootApiService'
- method = 'DoesNotExist'
+ self.assertCommandContains(
+ ["build_api", service_method], enter_chroot=True
+ )
- with self.assertRaises(router.UnknownMethodError):
- self.router.Route(service, method, self.api_config,
- self.binary_input_handler, [self.binary_output_handler],
- self.binary_config_handler)
+ output_msg = build_api_test_pb2.TestResultMessage()
+ empty_msg = build_api_test_pb2.TestResultMessage()
+ self.binary_output_handler.read_into(output_msg)
+ self.assertEqual(empty_msg, output_msg)
- def testListVisibility(self):
- """Test visibility options."""
- service = 'HiddenService'
- method = 'HiddenMethod'
+ def testInvalidService(self):
+ """Test invalid service call."""
+ service = "chromite.api.DoesNotExist"
+ method = "OutsideServiceInsideMethod"
- for current in self.router.ListMethods():
- self.assertNotIn(service, current)
- self.assertNotIn(method, current)
+ with self.assertRaises(router.UnknownServiceError):
+ self.router.Route(
+ service,
+ method,
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
+
+ def testInvalidMethod(self):
+ """Test invalid method call."""
+ service = "chromite.api.OutsideChrootApiService"
+ method = "DoesNotExist"
+
+ with self.assertRaises(router.UnknownMethodError):
+ self.router.Route(
+ service,
+ method,
+ self.api_config,
+ self.binary_input_handler,
+ [self.binary_output_handler],
+ self.binary_config_handler,
+ )
+
+ def testListVisibility(self):
+ """Test visibility options."""
+ service = "HiddenService"
+ method = "HiddenMethod"
+
+ for current in self.router.ListMethods():
+ self.assertNotIn(service, current)
+ self.assertNotIn(method, current)