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/controller/artifacts_unittest.py b/api/controller/artifacts_unittest.py
index 356af07..12b2541 100644
--- a/api/controller/artifacts_unittest.py
+++ b/api/controller/artifacts_unittest.py
@@ -25,921 +25,1145 @@
class BundleRequestMixin(object):
- """Mixin to provide bundle request methods."""
+ """Mixin to provide bundle request methods."""
- def EmptyRequest(self):
- return artifacts_pb2.BundleRequest()
+ def EmptyRequest(self):
+ return artifacts_pb2.BundleRequest()
- def BuildTargetRequest(self, build_target=None, output_dir=None, chroot=None):
- """Get a build target format request instance."""
- request = self.EmptyRequest()
- if build_target:
- request.build_target.name = build_target
- if output_dir:
- request.output_dir = output_dir
- if chroot:
- request.chroot.path = chroot
+ def BuildTargetRequest(
+ self, build_target=None, output_dir=None, chroot=None
+ ):
+ """Get a build target format request instance."""
+ request = self.EmptyRequest()
+ if build_target:
+ request.build_target.name = build_target
+ if output_dir:
+ request.output_dir = output_dir
+ if chroot:
+ request.chroot.path = chroot
- return request
+ return request
- def SysrootRequest(self,
- sysroot=None,
- build_target=None,
- output_dir=None,
- chroot=None):
- """Get a sysroot format request instance."""
- request = self.EmptyRequest()
- if sysroot:
- request.sysroot.path = sysroot
- if build_target:
- request.sysroot.build_target.name = build_target
- if output_dir:
- request.output_dir = output_dir
- if chroot:
- request.chroot.path = chroot
+ def SysrootRequest(
+ self, sysroot=None, build_target=None, output_dir=None, chroot=None
+ ):
+ """Get a sysroot format request instance."""
+ request = self.EmptyRequest()
+ if sysroot:
+ request.sysroot.path = sysroot
+ if build_target:
+ request.sysroot.build_target.name = build_target
+ if output_dir:
+ request.output_dir = output_dir
+ if chroot:
+ request.chroot.path = chroot
- return request
+ return request
-class BundleTestCase(cros_test_lib.MockTempDirTestCase,
- api_config.ApiConfigMixin, BundleRequestMixin):
- """Basic setup for all artifacts unittests."""
+class BundleTestCase(
+ cros_test_lib.MockTempDirTestCase,
+ api_config.ApiConfigMixin,
+ BundleRequestMixin,
+):
+ """Basic setup for all artifacts unittests."""
- def setUp(self):
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=False)
- self.output_dir = os.path.join(self.tempdir, 'artifacts')
- osutils.SafeMakedirs(self.output_dir)
- self.sysroot_path = '/build/target'
- self.sysroot = sysroot_lib.Sysroot(self.sysroot_path)
- self.chroot_path = os.path.join(self.tempdir, 'chroot')
- full_sysroot_path = os.path.join(self.chroot_path,
- self.sysroot_path.lstrip(os.sep))
- osutils.SafeMakedirs(full_sysroot_path)
+ def setUp(self):
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)
+ self.output_dir = os.path.join(self.tempdir, "artifacts")
+ osutils.SafeMakedirs(self.output_dir)
+ self.sysroot_path = "/build/target"
+ self.sysroot = sysroot_lib.Sysroot(self.sysroot_path)
+ self.chroot_path = os.path.join(self.tempdir, "chroot")
+ full_sysroot_path = os.path.join(
+ self.chroot_path, self.sysroot_path.lstrip(os.sep)
+ )
+ osutils.SafeMakedirs(full_sysroot_path)
- # All requests use same response type.
- self.response = artifacts_pb2.BundleResponse()
+ # All requests use same response type.
+ self.response = artifacts_pb2.BundleResponse()
- # Build target request.
- self.target_request = self.BuildTargetRequest(
- build_target='target',
- output_dir=self.output_dir,
- chroot=self.chroot_path)
+ # Build target request.
+ self.target_request = self.BuildTargetRequest(
+ build_target="target",
+ output_dir=self.output_dir,
+ chroot=self.chroot_path,
+ )
- # Sysroot request.
- self.sysroot_request = self.SysrootRequest(
- sysroot=self.sysroot_path,
- build_target='target',
- output_dir=self.output_dir,
- chroot=self.chroot_path)
+ # Sysroot request.
+ self.sysroot_request = self.SysrootRequest(
+ sysroot=self.sysroot_path,
+ build_target="target",
+ output_dir=self.output_dir,
+ chroot=self.chroot_path,
+ )
- self.source_root = self.tempdir
- self.PatchObject(constants, 'SOURCE_ROOT', new=self.tempdir)
+ self.source_root = self.tempdir
+ self.PatchObject(constants, "SOURCE_ROOT", new=self.tempdir)
class BundleImageArchivesTest(BundleTestCase):
- """BundleImageArchives tests."""
+ """BundleImageArchives tests."""
- def testValidateOnly(self):
- """Quick check that a validate only call does not execute any logic."""
- patch = self.PatchObject(artifacts_svc, 'ArchiveImages')
- artifacts.BundleImageArchives(self.target_request, self.response,
- self.validate_only_config)
- patch.assert_not_called()
+ def testValidateOnly(self):
+ """Quick check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(artifacts_svc, "ArchiveImages")
+ artifacts.BundleImageArchives(
+ self.target_request, self.response, self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(artifacts_svc, 'ArchiveImages')
- artifacts.BundleImageArchives(self.target_request, self.response,
- self.mock_call_config)
- patch.assert_not_called()
- self.assertEqual(len(self.response.artifacts), 2)
- self.assertEqual(self.response.artifacts[0].path,
- os.path.join(self.output_dir, 'path0.tar.xz'))
- self.assertEqual(self.response.artifacts[1].path,
- os.path.join(self.output_dir, 'path1.tar.xz'))
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(artifacts_svc, "ArchiveImages")
+ artifacts.BundleImageArchives(
+ self.target_request, self.response, self.mock_call_config
+ )
+ patch.assert_not_called()
+ self.assertEqual(len(self.response.artifacts), 2)
+ self.assertEqual(
+ self.response.artifacts[0].path,
+ os.path.join(self.output_dir, "path0.tar.xz"),
+ )
+ self.assertEqual(
+ self.response.artifacts[1].path,
+ os.path.join(self.output_dir, "path1.tar.xz"),
+ )
- def testNoBuildTarget(self):
- """Test that no build target fails."""
- request = self.BuildTargetRequest(output_dir=str(self.tempdir))
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleImageArchives(request, self.response, self.api_config)
+ def testNoBuildTarget(self):
+ """Test that no build target fails."""
+ request = self.BuildTargetRequest(output_dir=str(self.tempdir))
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleImageArchives(
+ request, self.response, self.api_config
+ )
- def testNoOutputDir(self):
- """Test no output dir fails."""
- request = self.BuildTargetRequest(build_target='board')
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleImageArchives(request, self.response, self.api_config)
+ def testNoOutputDir(self):
+ """Test no output dir fails."""
+ request = self.BuildTargetRequest(build_target="board")
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleImageArchives(
+ request, self.response, self.api_config
+ )
- def testInvalidOutputDir(self):
- """Test invalid output dir fails."""
- request = self.BuildTargetRequest(
- build_target='board', output_dir=os.path.join(self.tempdir, 'DNE'))
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleImageArchives(request, self.response, self.api_config)
+ def testInvalidOutputDir(self):
+ """Test invalid output dir fails."""
+ request = self.BuildTargetRequest(
+ build_target="board", output_dir=os.path.join(self.tempdir, "DNE")
+ )
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleImageArchives(
+ request, self.response, self.api_config
+ )
- def testOutputHandling(self):
- """Test the artifact output handling."""
- expected = [os.path.join(self.output_dir, f) for f in ('a', 'b', 'c')]
- self.PatchObject(artifacts_svc, 'ArchiveImages', return_value=expected)
- self.PatchObject(os.path, 'exists', return_value=True)
+ def testOutputHandling(self):
+ """Test the artifact output handling."""
+ expected = [os.path.join(self.output_dir, f) for f in ("a", "b", "c")]
+ self.PatchObject(artifacts_svc, "ArchiveImages", return_value=expected)
+ self.PatchObject(os.path, "exists", return_value=True)
- artifacts.BundleImageArchives(self.target_request, self.response,
- self.api_config)
+ artifacts.BundleImageArchives(
+ self.target_request, self.response, self.api_config
+ )
- self.assertCountEqual(expected, [a.path for a in self.response.artifacts])
+ self.assertCountEqual(
+ expected, [a.path for a in self.response.artifacts]
+ )
class BundleImageZipTest(BundleTestCase):
- """Unittests for BundleImageZip."""
+ """Unittests for BundleImageZip."""
- def testValidateOnly(self):
- """Quick check that a validate only call does not execute any logic."""
- patch = self.PatchObject(commands, 'BuildImageZip')
- artifacts.BundleImageZip(self.target_request, self.response,
- self.validate_only_config)
- patch.assert_not_called()
+ def testValidateOnly(self):
+ """Quick check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(commands, "BuildImageZip")
+ artifacts.BundleImageZip(
+ self.target_request, self.response, self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(commands, 'BuildImageZip')
- artifacts.BundleImageZip(self.target_request, self.response,
- self.mock_call_config)
- patch.assert_not_called()
- self.assertEqual(len(self.response.artifacts), 1)
- self.assertEqual(self.response.artifacts[0].path,
- os.path.join(self.output_dir, 'image.zip'))
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(commands, "BuildImageZip")
+ artifacts.BundleImageZip(
+ self.target_request, self.response, self.mock_call_config
+ )
+ patch.assert_not_called()
+ self.assertEqual(len(self.response.artifacts), 1)
+ self.assertEqual(
+ self.response.artifacts[0].path,
+ os.path.join(self.output_dir, "image.zip"),
+ )
- def testBundleImageZip(self):
- """BundleImageZip calls cbuildbot/commands with correct args."""
- bundle_image_zip = self.PatchObject(
- artifacts_svc, 'BundleImageZip', return_value='image.zip')
- self.PatchObject(os.path, 'exists', return_value=True)
- artifacts.BundleImageZip(self.target_request, self.response,
- self.api_config)
- self.assertEqual(
- [artifact.path for artifact in self.response.artifacts],
- [os.path.join(self.output_dir, 'image.zip')])
+ def testBundleImageZip(self):
+ """BundleImageZip calls cbuildbot/commands with correct args."""
+ bundle_image_zip = self.PatchObject(
+ artifacts_svc, "BundleImageZip", return_value="image.zip"
+ )
+ self.PatchObject(os.path, "exists", return_value=True)
+ artifacts.BundleImageZip(
+ self.target_request, self.response, self.api_config
+ )
+ self.assertEqual(
+ [artifact.path for artifact in self.response.artifacts],
+ [os.path.join(self.output_dir, "image.zip")],
+ )
- latest = os.path.join(self.source_root, 'src/build/images/target/latest')
- self.assertEqual(
- bundle_image_zip.call_args_list,
- [mock.call(self.output_dir, latest)])
+ latest = os.path.join(
+ self.source_root, "src/build/images/target/latest"
+ )
+ self.assertEqual(
+ bundle_image_zip.call_args_list,
+ [mock.call(self.output_dir, latest)],
+ )
- def testBundleImageZipNoImageDir(self):
- """BundleImageZip dies when image dir does not exist."""
- self.PatchObject(os.path, 'exists', return_value=False)
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleImageZip(self.target_request, self.response,
- self.api_config)
+ def testBundleImageZipNoImageDir(self):
+ """BundleImageZip dies when image dir does not exist."""
+ self.PatchObject(os.path, "exists", return_value=False)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleImageZip(
+ self.target_request, self.response, self.api_config
+ )
class BundleAutotestFilesTest(BundleTestCase):
- """Unittests for BundleAutotestFiles."""
+ """Unittests for BundleAutotestFiles."""
- def testValidateOnly(self):
- """Quick check that a validate only call does not execute any logic."""
- patch = self.PatchObject(artifacts_svc, 'BundleAutotestFiles')
- artifacts.BundleAutotestFiles(self.sysroot_request, self.response,
- self.validate_only_config)
- patch.assert_not_called()
+ def testValidateOnly(self):
+ """Quick check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(artifacts_svc, "BundleAutotestFiles")
+ artifacts.BundleAutotestFiles(
+ self.sysroot_request, self.response, self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(artifacts_svc, 'BundleAutotestFiles')
- artifacts.BundleAutotestFiles(self.sysroot_request, self.response,
- self.mock_call_config)
- patch.assert_not_called()
- self.assertEqual(len(self.response.artifacts), 1)
- self.assertEqual(self.response.artifacts[0].path,
- os.path.join(self.output_dir, 'autotest-a.tar.gz'))
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(artifacts_svc, "BundleAutotestFiles")
+ artifacts.BundleAutotestFiles(
+ self.sysroot_request, self.response, self.mock_call_config
+ )
+ patch.assert_not_called()
+ self.assertEqual(len(self.response.artifacts), 1)
+ self.assertEqual(
+ self.response.artifacts[0].path,
+ os.path.join(self.output_dir, "autotest-a.tar.gz"),
+ )
- def testBundleAutotestFiles(self):
- """BundleAutotestFiles calls service correctly."""
- files = {
- artifacts_svc.ARCHIVE_CONTROL_FILES: '/tmp/artifacts/autotest-a.tar.gz',
- artifacts_svc.ARCHIVE_PACKAGES: '/tmp/artifacts/autotest-b.tar.gz',
- }
- patch = self.PatchObject(artifacts_svc, 'BundleAutotestFiles',
- return_value=files)
+ def testBundleAutotestFiles(self):
+ """BundleAutotestFiles calls service correctly."""
+ files = {
+ artifacts_svc.ARCHIVE_CONTROL_FILES: "/tmp/artifacts/autotest-a.tar.gz",
+ artifacts_svc.ARCHIVE_PACKAGES: "/tmp/artifacts/autotest-b.tar.gz",
+ }
+ patch = self.PatchObject(
+ artifacts_svc, "BundleAutotestFiles", return_value=files
+ )
- artifacts.BundleAutotestFiles(self.sysroot_request, self.response,
- self.api_config)
+ artifacts.BundleAutotestFiles(
+ self.sysroot_request, self.response, self.api_config
+ )
- # Verify the arguments are being passed through.
- patch.assert_called_with(mock.ANY, self.sysroot, self.output_dir)
+ # Verify the arguments are being passed through.
+ patch.assert_called_with(mock.ANY, self.sysroot, self.output_dir)
- # Verify the output proto is being populated correctly.
- self.assertTrue(self.response.artifacts)
- paths = [artifact.path for artifact in self.response.artifacts]
- self.assertCountEqual(list(files.values()), paths)
+ # Verify the output proto is being populated correctly.
+ self.assertTrue(self.response.artifacts)
+ paths = [artifact.path for artifact in self.response.artifacts]
+ self.assertCountEqual(list(files.values()), paths)
- def testInvalidOutputDir(self):
- """Test invalid output directory argument."""
- request = self.SysrootRequest(chroot=self.chroot_path,
- sysroot=self.sysroot_path)
+ def testInvalidOutputDir(self):
+ """Test invalid output directory argument."""
+ request = self.SysrootRequest(
+ chroot=self.chroot_path, sysroot=self.sysroot_path
+ )
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleAutotestFiles(request, self.response, self.api_config)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleAutotestFiles(
+ request, self.response, self.api_config
+ )
- def testInvalidSysroot(self):
- """Test no sysroot directory."""
- request = self.SysrootRequest(chroot=self.chroot_path,
- output_dir=self.output_dir)
+ def testInvalidSysroot(self):
+ """Test no sysroot directory."""
+ request = self.SysrootRequest(
+ chroot=self.chroot_path, output_dir=self.output_dir
+ )
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleAutotestFiles(request, self.response, self.api_config)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleAutotestFiles(
+ request, self.response, self.api_config
+ )
- def testSysrootDoesNotExist(self):
- """Test dies when no sysroot does not exist."""
- request = self.SysrootRequest(chroot=self.chroot_path,
- sysroot='/does/not/exist',
- output_dir=self.output_dir)
+ def testSysrootDoesNotExist(self):
+ """Test dies when no sysroot does not exist."""
+ request = self.SysrootRequest(
+ chroot=self.chroot_path,
+ sysroot="/does/not/exist",
+ output_dir=self.output_dir,
+ )
- artifacts.BundleAutotestFiles(request, self.response, self.api_config)
- self.assertFalse(self.response.artifacts)
+ artifacts.BundleAutotestFiles(request, self.response, self.api_config)
+ self.assertFalse(self.response.artifacts)
class BundleTastFilesTest(BundleTestCase):
- """Unittests for BundleTastFiles."""
+ """Unittests for BundleTastFiles."""
- def testValidateOnly(self):
- """Quick check that a validate only call does not execute any logic."""
- patch = self.PatchObject(artifacts_svc, 'BundleTastFiles')
- artifacts.BundleTastFiles(self.sysroot_request, self.response,
- self.validate_only_config)
- patch.assert_not_called()
+ def testValidateOnly(self):
+ """Quick check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(artifacts_svc, "BundleTastFiles")
+ artifacts.BundleTastFiles(
+ self.sysroot_request, self.response, self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(artifacts_svc, 'BundleTastFiles')
- artifacts.BundleTastFiles(self.sysroot_request, self.response,
- self.mock_call_config)
- patch.assert_not_called()
- self.assertEqual(len(self.response.artifacts), 1)
- self.assertEqual(self.response.artifacts[0].path,
- os.path.join(self.output_dir, 'tast_bundles.tar.gz'))
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(artifacts_svc, "BundleTastFiles")
+ artifacts.BundleTastFiles(
+ self.sysroot_request, self.response, self.mock_call_config
+ )
+ patch.assert_not_called()
+ self.assertEqual(len(self.response.artifacts), 1)
+ self.assertEqual(
+ self.response.artifacts[0].path,
+ os.path.join(self.output_dir, "tast_bundles.tar.gz"),
+ )
- def testBundleTastFilesNoLogs(self):
- """BundleTasteFiles succeeds when no tast files found."""
- self.PatchObject(commands, 'BuildTastBundleTarball',
- return_value=None)
- artifacts.BundleTastFiles(self.sysroot_request, self.response,
- self.api_config)
- self.assertFalse(self.response.artifacts)
+ def testBundleTastFilesNoLogs(self):
+ """BundleTasteFiles succeeds when no tast files found."""
+ self.PatchObject(commands, "BuildTastBundleTarball", return_value=None)
+ artifacts.BundleTastFiles(
+ self.sysroot_request, self.response, self.api_config
+ )
+ self.assertFalse(self.response.artifacts)
- def testBundleTastFiles(self):
- """BundleTastFiles calls service correctly."""
- chroot = chroot_lib.Chroot(self.chroot_path)
+ def testBundleTastFiles(self):
+ """BundleTastFiles calls service correctly."""
+ chroot = chroot_lib.Chroot(self.chroot_path)
- expected_archive = os.path.join(self.output_dir,
- artifacts_svc.TAST_BUNDLE_NAME)
- # Patch the service being called.
- bundle_patch = self.PatchObject(artifacts_svc, 'BundleTastFiles',
- return_value=expected_archive)
+ expected_archive = os.path.join(
+ self.output_dir, artifacts_svc.TAST_BUNDLE_NAME
+ )
+ # Patch the service being called.
+ bundle_patch = self.PatchObject(
+ artifacts_svc, "BundleTastFiles", return_value=expected_archive
+ )
- artifacts.BundleTastFiles(self.sysroot_request, self.response,
- self.api_config)
+ artifacts.BundleTastFiles(
+ self.sysroot_request, self.response, self.api_config
+ )
- # Make sure the artifact got recorded successfully.
- self.assertTrue(self.response.artifacts)
- self.assertEqual(expected_archive, self.response.artifacts[0].path)
- # Make sure the service got called correctly.
- bundle_patch.assert_called_once_with(chroot, self.sysroot, self.output_dir)
+ # Make sure the artifact got recorded successfully.
+ self.assertTrue(self.response.artifacts)
+ self.assertEqual(expected_archive, self.response.artifacts[0].path)
+ # Make sure the service got called correctly.
+ bundle_patch.assert_called_once_with(
+ chroot, self.sysroot, self.output_dir
+ )
class BundleFirmwareTest(BundleTestCase):
- """Unittests for BundleFirmware."""
+ """Unittests for BundleFirmware."""
- def testValidateOnly(self):
- """Quick check that a validate only call does not execute any logic."""
- patch = self.PatchObject(artifacts_svc, 'BundleTastFiles')
- artifacts.BundleFirmware(self.sysroot_request, self.response,
- self.validate_only_config)
- patch.assert_not_called()
+ def testValidateOnly(self):
+ """Quick check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(artifacts_svc, "BundleTastFiles")
+ artifacts.BundleFirmware(
+ self.sysroot_request, self.response, self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(artifacts_svc, 'BundleTastFiles')
- artifacts.BundleFirmware(self.sysroot_request, self.response,
- self.mock_call_config)
- patch.assert_not_called()
- self.assertEqual(len(self.response.artifacts), 1)
- self.assertEqual(self.response.artifacts[0].path,
- os.path.join(self.output_dir, 'firmware.tar.gz'))
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(artifacts_svc, "BundleTastFiles")
+ artifacts.BundleFirmware(
+ self.sysroot_request, self.response, self.mock_call_config
+ )
+ patch.assert_not_called()
+ self.assertEqual(len(self.response.artifacts), 1)
+ self.assertEqual(
+ self.response.artifacts[0].path,
+ os.path.join(self.output_dir, "firmware.tar.gz"),
+ )
- def testBundleFirmware(self):
- """BundleFirmware calls cbuildbot/commands with correct args."""
- self.PatchObject(
- artifacts_svc,
- 'BuildFirmwareArchive',
- return_value=os.path.join(self.output_dir, 'firmware.tar.gz'))
+ def testBundleFirmware(self):
+ """BundleFirmware calls cbuildbot/commands with correct args."""
+ self.PatchObject(
+ artifacts_svc,
+ "BuildFirmwareArchive",
+ return_value=os.path.join(self.output_dir, "firmware.tar.gz"),
+ )
- artifacts.BundleFirmware(self.sysroot_request, self.response,
- self.api_config)
- self.assertEqual(
- [artifact.path for artifact in self.response.artifacts],
- [os.path.join(self.output_dir, 'firmware.tar.gz')])
+ artifacts.BundleFirmware(
+ self.sysroot_request, self.response, self.api_config
+ )
+ self.assertEqual(
+ [artifact.path for artifact in self.response.artifacts],
+ [os.path.join(self.output_dir, "firmware.tar.gz")],
+ )
- def testBundleFirmwareNoLogs(self):
- """BundleFirmware dies when no firmware found."""
- self.PatchObject(commands, 'BuildFirmwareArchive', return_value=None)
- artifacts.BundleFirmware(self.sysroot_request, self.response,
- self.api_config)
- self.assertEqual(len(self.response.artifacts), 0)
+ def testBundleFirmwareNoLogs(self):
+ """BundleFirmware dies when no firmware found."""
+ self.PatchObject(commands, "BuildFirmwareArchive", return_value=None)
+ artifacts.BundleFirmware(
+ self.sysroot_request, self.response, self.api_config
+ )
+ self.assertEqual(len(self.response.artifacts), 0)
class BundleFpmcuUnittestsTest(BundleTestCase):
- """Unittests for BundleFpmcuUnittests."""
+ """Unittests for BundleFpmcuUnittests."""
- def testValidateOnly(self):
- """Quick check that a validate only call does not execute any logic."""
- patch = self.PatchObject(artifacts_svc, 'BundleFpmcuUnittests')
- artifacts.BundleFpmcuUnittests(self.sysroot_request, self.response,
- self.validate_only_config)
- patch.assert_not_called()
+ def testValidateOnly(self):
+ """Quick check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(artifacts_svc, "BundleFpmcuUnittests")
+ artifacts.BundleFpmcuUnittests(
+ self.sysroot_request, self.response, self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(artifacts_svc, 'BundleFpmcuUnittests')
- artifacts.BundleFpmcuUnittests(self.sysroot_request, self.response,
- self.mock_call_config)
- patch.assert_not_called()
- self.assertEqual(len(self.response.artifacts), 1)
- self.assertEqual(self.response.artifacts[0].path,
- os.path.join(self.output_dir,
- 'fpmcu_unittests.tar.gz'))
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(artifacts_svc, "BundleFpmcuUnittests")
+ artifacts.BundleFpmcuUnittests(
+ self.sysroot_request, self.response, self.mock_call_config
+ )
+ patch.assert_not_called()
+ self.assertEqual(len(self.response.artifacts), 1)
+ self.assertEqual(
+ self.response.artifacts[0].path,
+ os.path.join(self.output_dir, "fpmcu_unittests.tar.gz"),
+ )
- def testBundleFpmcuUnittests(self):
- """BundleFpmcuUnittests calls cbuildbot/commands with correct args."""
- self.PatchObject(
- artifacts_svc,
- 'BundleFpmcuUnittests',
- return_value=os.path.join(self.output_dir, 'fpmcu_unittests.tar.gz'))
- artifacts.BundleFpmcuUnittests(self.sysroot_request, self.response,
- self.api_config)
- self.assertEqual(
- [artifact.path for artifact in self.response.artifacts],
- [os.path.join(self.output_dir, 'fpmcu_unittests.tar.gz')])
+ def testBundleFpmcuUnittests(self):
+ """BundleFpmcuUnittests calls cbuildbot/commands with correct args."""
+ self.PatchObject(
+ artifacts_svc,
+ "BundleFpmcuUnittests",
+ return_value=os.path.join(
+ self.output_dir, "fpmcu_unittests.tar.gz"
+ ),
+ )
+ artifacts.BundleFpmcuUnittests(
+ self.sysroot_request, self.response, self.api_config
+ )
+ self.assertEqual(
+ [artifact.path for artifact in self.response.artifacts],
+ [os.path.join(self.output_dir, "fpmcu_unittests.tar.gz")],
+ )
- def testBundleFpmcuUnittestsNoLogs(self):
- """BundleFpmcuUnittests does not die when no fpmcu unittests found."""
- self.PatchObject(artifacts_svc, 'BundleFpmcuUnittests',
- return_value=None)
- artifacts.BundleFpmcuUnittests(self.sysroot_request, self.response,
- self.api_config)
- self.assertFalse(self.response.artifacts)
+ def testBundleFpmcuUnittestsNoLogs(self):
+ """BundleFpmcuUnittests does not die when no fpmcu unittests found."""
+ self.PatchObject(
+ artifacts_svc, "BundleFpmcuUnittests", return_value=None
+ )
+ artifacts.BundleFpmcuUnittests(
+ self.sysroot_request, self.response, self.api_config
+ )
+ self.assertFalse(self.response.artifacts)
class BundleEbuildLogsTest(BundleTestCase):
- """Unittests for BundleEbuildLogs."""
+ """Unittests for BundleEbuildLogs."""
- def testValidateOnly(self):
- """Quick check that a validate only call does not execute any logic."""
- patch = self.PatchObject(commands, 'BuildEbuildLogsTarball')
- artifacts.BundleEbuildLogs(self.sysroot_request, self.response,
- self.validate_only_config)
- patch.assert_not_called()
+ def testValidateOnly(self):
+ """Quick check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(commands, "BuildEbuildLogsTarball")
+ artifacts.BundleEbuildLogs(
+ self.sysroot_request, self.response, self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(commands, 'BuildEbuildLogsTarball')
- artifacts.BundleEbuildLogs(self.sysroot_request, self.response,
- self.mock_call_config)
- patch.assert_not_called()
- self.assertEqual(len(self.response.artifacts), 1)
- self.assertEqual(self.response.artifacts[0].path,
- os.path.join(self.output_dir, 'ebuild-logs.tar.gz'))
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(commands, "BuildEbuildLogsTarball")
+ artifacts.BundleEbuildLogs(
+ self.sysroot_request, self.response, self.mock_call_config
+ )
+ patch.assert_not_called()
+ self.assertEqual(len(self.response.artifacts), 1)
+ self.assertEqual(
+ self.response.artifacts[0].path,
+ os.path.join(self.output_dir, "ebuild-logs.tar.gz"),
+ )
- def testBundleEbuildLogs(self):
- """BundleEbuildLogs calls cbuildbot/commands with correct args."""
- bundle_ebuild_logs_tarball = self.PatchObject(
- artifacts_svc, 'BundleEBuildLogsTarball',
- return_value='ebuild-logs.tar.gz')
- artifacts.BundleEbuildLogs(self.sysroot_request, self.response,
- self.api_config)
- self.assertEqual(
- [artifact.path for artifact in self.response.artifacts],
- [os.path.join(self.output_dir, 'ebuild-logs.tar.gz')])
- self.assertEqual(
- bundle_ebuild_logs_tarball.call_args_list,
- [mock.call(mock.ANY, self.sysroot, self.output_dir)])
+ def testBundleEbuildLogs(self):
+ """BundleEbuildLogs calls cbuildbot/commands with correct args."""
+ bundle_ebuild_logs_tarball = self.PatchObject(
+ artifacts_svc,
+ "BundleEBuildLogsTarball",
+ return_value="ebuild-logs.tar.gz",
+ )
+ artifacts.BundleEbuildLogs(
+ self.sysroot_request, self.response, self.api_config
+ )
+ self.assertEqual(
+ [artifact.path for artifact in self.response.artifacts],
+ [os.path.join(self.output_dir, "ebuild-logs.tar.gz")],
+ )
+ self.assertEqual(
+ bundle_ebuild_logs_tarball.call_args_list,
+ [mock.call(mock.ANY, self.sysroot, self.output_dir)],
+ )
- def testBundleEbuildLogsNoLogs(self):
- """BundleEbuildLogs dies when no logs found."""
- self.PatchObject(commands, 'BuildEbuildLogsTarball', return_value=None)
- artifacts.BundleEbuildLogs(self.sysroot_request, self.response,
- self.api_config)
+ def testBundleEbuildLogsNoLogs(self):
+ """BundleEbuildLogs dies when no logs found."""
+ self.PatchObject(commands, "BuildEbuildLogsTarball", return_value=None)
+ artifacts.BundleEbuildLogs(
+ self.sysroot_request, self.response, self.api_config
+ )
- self.assertFalse(self.response.artifacts)
+ self.assertFalse(self.response.artifacts)
class BundleChromeOSConfigTest(BundleTestCase):
- """Unittests for BundleChromeOSConfig"""
+ """Unittests for BundleChromeOSConfig"""
- def testValidateOnly(self):
- """Quick check that a validate only call does not execute any logic."""
- patch = self.PatchObject(artifacts_svc, 'BundleChromeOSConfig')
- artifacts.BundleChromeOSConfig(self.sysroot_request, self.response,
- self.validate_only_config)
- patch.assert_not_called()
+ def testValidateOnly(self):
+ """Quick check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(artifacts_svc, "BundleChromeOSConfig")
+ artifacts.BundleChromeOSConfig(
+ self.sysroot_request, self.response, self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(artifacts_svc, 'BundleChromeOSConfig')
- artifacts.BundleChromeOSConfig(self.sysroot_request, self.response,
- self.mock_call_config)
- patch.assert_not_called()
- self.assertEqual(len(self.response.artifacts), 1)
- self.assertEqual(self.response.artifacts[0].path,
- os.path.join(self.output_dir, 'config.yaml'))
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(artifacts_svc, "BundleChromeOSConfig")
+ artifacts.BundleChromeOSConfig(
+ self.sysroot_request, self.response, self.mock_call_config
+ )
+ patch.assert_not_called()
+ self.assertEqual(len(self.response.artifacts), 1)
+ self.assertEqual(
+ self.response.artifacts[0].path,
+ os.path.join(self.output_dir, "config.yaml"),
+ )
- def testBundleChromeOSConfigSuccess(self):
- """Test standard success case."""
- bundle_chromeos_config = self.PatchObject(
- artifacts_svc, 'BundleChromeOSConfig', return_value='config.yaml')
- artifacts.BundleChromeOSConfig(self.sysroot_request, self.response,
- self.api_config)
- self.assertEqual(
- [artifact.path for artifact in self.response.artifacts],
- [os.path.join(self.output_dir, 'config.yaml')])
+ def testBundleChromeOSConfigSuccess(self):
+ """Test standard success case."""
+ bundle_chromeos_config = self.PatchObject(
+ artifacts_svc, "BundleChromeOSConfig", return_value="config.yaml"
+ )
+ artifacts.BundleChromeOSConfig(
+ self.sysroot_request, self.response, self.api_config
+ )
+ self.assertEqual(
+ [artifact.path for artifact in self.response.artifacts],
+ [os.path.join(self.output_dir, "config.yaml")],
+ )
- self.assertEqual(bundle_chromeos_config.call_args_list,
- [mock.call(mock.ANY, self.sysroot, self.output_dir)])
+ self.assertEqual(
+ bundle_chromeos_config.call_args_list,
+ [mock.call(mock.ANY, self.sysroot, self.output_dir)],
+ )
- def testBundleChromeOSConfigNoConfigFound(self):
- """Empty results when the config payload isn't found."""
- self.PatchObject(artifacts_svc, 'BundleChromeOSConfig', return_value=None)
+ def testBundleChromeOSConfigNoConfigFound(self):
+ """Empty results when the config payload isn't found."""
+ self.PatchObject(
+ artifacts_svc, "BundleChromeOSConfig", return_value=None
+ )
- artifacts.BundleChromeOSConfig(self.sysroot_request, self.response,
- self.api_config)
- self.assertFalse(self.response.artifacts)
+ artifacts.BundleChromeOSConfig(
+ self.sysroot_request, self.response, self.api_config
+ )
+ self.assertFalse(self.response.artifacts)
-class BundleTestUpdatePayloadsTest(cros_test_lib.MockTempDirTestCase,
- api_config.ApiConfigMixin):
- """Unittests for BundleTestUpdatePayloads."""
+class BundleTestUpdatePayloadsTest(
+ cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
+):
+ """Unittests for BundleTestUpdatePayloads."""
- def setUp(self):
- self.source_root = os.path.join(self.tempdir, 'cros')
- osutils.SafeMakedirs(self.source_root)
+ def setUp(self):
+ self.source_root = os.path.join(self.tempdir, "cros")
+ osutils.SafeMakedirs(self.source_root)
- self.archive_root = os.path.join(self.tempdir, 'output')
- osutils.SafeMakedirs(self.archive_root)
+ self.archive_root = os.path.join(self.tempdir, "output")
+ osutils.SafeMakedirs(self.archive_root)
- self.target = 'target'
- self.image_root = os.path.join(self.source_root,
- 'src/build/images/target/latest')
+ self.target = "target"
+ self.image_root = os.path.join(
+ self.source_root, "src/build/images/target/latest"
+ )
- self.input_proto = artifacts_pb2.BundleRequest()
- self.input_proto.build_target.name = self.target
- self.input_proto.output_dir = self.archive_root
- self.output_proto = artifacts_pb2.BundleResponse()
+ self.input_proto = artifacts_pb2.BundleRequest()
+ self.input_proto.build_target.name = self.target
+ self.input_proto.output_dir = self.archive_root
+ self.output_proto = artifacts_pb2.BundleResponse()
- self.PatchObject(constants, 'SOURCE_ROOT', new=self.source_root)
+ self.PatchObject(constants, "SOURCE_ROOT", new=self.source_root)
- def MockPayloads(image_path, archive_dir):
- osutils.WriteFile(os.path.join(archive_dir, 'payload1.bin'), image_path)
- osutils.WriteFile(os.path.join(archive_dir, 'payload2.bin'), image_path)
- return [os.path.join(archive_dir, 'payload1.bin'),
- os.path.join(archive_dir, 'payload2.bin')]
+ def MockPayloads(image_path, archive_dir):
+ osutils.WriteFile(
+ os.path.join(archive_dir, "payload1.bin"), image_path
+ )
+ osutils.WriteFile(
+ os.path.join(archive_dir, "payload2.bin"), image_path
+ )
+ return [
+ os.path.join(archive_dir, "payload1.bin"),
+ os.path.join(archive_dir, "payload2.bin"),
+ ]
- self.bundle_patch = self.PatchObject(
- artifacts_svc, 'BundleTestUpdatePayloads', side_effect=MockPayloads)
+ self.bundle_patch = self.PatchObject(
+ artifacts_svc, "BundleTestUpdatePayloads", side_effect=MockPayloads
+ )
- def testValidateOnly(self):
- """Quick check that a validate only call does not execute any logic."""
- patch = self.PatchObject(artifacts_svc, 'BundleTestUpdatePayloads')
- artifacts.BundleTestUpdatePayloads(self.input_proto, self.output_proto,
- self.validate_only_config)
- patch.assert_not_called()
+ def testValidateOnly(self):
+ """Quick check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(artifacts_svc, "BundleTestUpdatePayloads")
+ artifacts.BundleTestUpdatePayloads(
+ self.input_proto, self.output_proto, self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(artifacts_svc, 'BundleTestUpdatePayloads')
- artifacts.BundleTestUpdatePayloads(self.input_proto, self.output_proto,
- self.mock_call_config)
- patch.assert_not_called()
- self.assertEqual(len(self.output_proto.artifacts), 3)
- self.assertEqual(self.output_proto.artifacts[0].path,
- os.path.join(self.archive_root, 'payload1.bin'))
- self.assertEqual(self.output_proto.artifacts[1].path,
- os.path.join(self.archive_root, 'payload1.json'))
- self.assertEqual(self.output_proto.artifacts[2].path,
- os.path.join(self.archive_root, 'payload1.log'))
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(artifacts_svc, "BundleTestUpdatePayloads")
+ artifacts.BundleTestUpdatePayloads(
+ self.input_proto, self.output_proto, self.mock_call_config
+ )
+ patch.assert_not_called()
+ self.assertEqual(len(self.output_proto.artifacts), 3)
+ self.assertEqual(
+ self.output_proto.artifacts[0].path,
+ os.path.join(self.archive_root, "payload1.bin"),
+ )
+ self.assertEqual(
+ self.output_proto.artifacts[1].path,
+ os.path.join(self.archive_root, "payload1.json"),
+ )
+ self.assertEqual(
+ self.output_proto.artifacts[2].path,
+ os.path.join(self.archive_root, "payload1.log"),
+ )
- def testBundleTestUpdatePayloads(self):
- """BundleTestUpdatePayloads calls cbuildbot/commands with correct args."""
- image_path = os.path.join(self.image_root, constants.BASE_IMAGE_BIN)
- osutils.WriteFile(image_path, 'image!', makedirs=True)
+ def testBundleTestUpdatePayloads(self):
+ """BundleTestUpdatePayloads calls cbuildbot/commands with correct args."""
+ image_path = os.path.join(self.image_root, constants.BASE_IMAGE_BIN)
+ osutils.WriteFile(image_path, "image!", makedirs=True)
- artifacts.BundleTestUpdatePayloads(self.input_proto, self.output_proto,
- self.api_config)
+ artifacts.BundleTestUpdatePayloads(
+ self.input_proto, self.output_proto, self.api_config
+ )
- actual = [
- os.path.relpath(artifact.path, self.archive_root)
- for artifact in self.output_proto.artifacts
- ]
- expected = ['payload1.bin', 'payload2.bin']
- self.assertCountEqual(actual, expected)
+ actual = [
+ os.path.relpath(artifact.path, self.archive_root)
+ for artifact in self.output_proto.artifacts
+ ]
+ expected = ["payload1.bin", "payload2.bin"]
+ self.assertCountEqual(actual, expected)
- actual = [
- os.path.relpath(path, self.archive_root)
- for path in osutils.DirectoryIterator(self.archive_root)
- ]
- self.assertCountEqual(actual, expected)
+ actual = [
+ os.path.relpath(path, self.archive_root)
+ for path in osutils.DirectoryIterator(self.archive_root)
+ ]
+ self.assertCountEqual(actual, expected)
- def testBundleTestUpdatePayloadsNoImageDir(self):
- """BundleTestUpdatePayloads dies if no image dir is found."""
- # Intentionally do not write image directory.
- artifacts.BundleTestUpdatePayloads(self.input_proto, self.output_proto,
- self.api_config)
- self.assertFalse(self.output_proto.artifacts)
+ def testBundleTestUpdatePayloadsNoImageDir(self):
+ """BundleTestUpdatePayloads dies if no image dir is found."""
+ # Intentionally do not write image directory.
+ artifacts.BundleTestUpdatePayloads(
+ self.input_proto, self.output_proto, self.api_config
+ )
+ self.assertFalse(self.output_proto.artifacts)
- def testBundleTestUpdatePayloadsNoImage(self):
- """BundleTestUpdatePayloads dies if no usable image is found for target."""
- # Intentionally do not write image, but create the directory.
- osutils.SafeMakedirs(self.image_root)
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleTestUpdatePayloads(self.input_proto, self.output_proto,
- self.api_config)
+ def testBundleTestUpdatePayloadsNoImage(self):
+ """BundleTestUpdatePayloads dies if no usable image is found for target."""
+ # Intentionally do not write image, but create the directory.
+ osutils.SafeMakedirs(self.image_root)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleTestUpdatePayloads(
+ self.input_proto, self.output_proto, self.api_config
+ )
-class BundleSimpleChromeArtifactsTest(cros_test_lib.MockTempDirTestCase,
- api_config.ApiConfigMixin):
- """BundleSimpleChromeArtifacts tests."""
+class BundleSimpleChromeArtifactsTest(
+ cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
+):
+ """BundleSimpleChromeArtifacts tests."""
- def setUp(self):
- self.chroot_dir = os.path.join(self.tempdir, 'chroot_dir')
- self.sysroot_path = '/sysroot'
- self.sysroot_dir = os.path.join(self.chroot_dir, 'sysroot')
- osutils.SafeMakedirs(self.sysroot_dir)
- self.output_dir = os.path.join(self.tempdir, 'output_dir')
- osutils.SafeMakedirs(self.output_dir)
+ def setUp(self):
+ self.chroot_dir = os.path.join(self.tempdir, "chroot_dir")
+ self.sysroot_path = "/sysroot"
+ self.sysroot_dir = os.path.join(self.chroot_dir, "sysroot")
+ osutils.SafeMakedirs(self.sysroot_dir)
+ self.output_dir = os.path.join(self.tempdir, "output_dir")
+ osutils.SafeMakedirs(self.output_dir)
- self.does_not_exist = os.path.join(self.tempdir, 'does_not_exist')
+ self.does_not_exist = os.path.join(self.tempdir, "does_not_exist")
- self.response = artifacts_pb2.BundleResponse()
+ self.response = artifacts_pb2.BundleResponse()
- def _GetRequest(
- self,
- chroot: Optional[str] = None,
- sysroot: Optional[str] = None,
- build_target: Optional[str] = None,
- output_dir: Optional[str] = None) -> artifacts_pb2.BundleRequest:
- """Helper to create a request message instance.
+ def _GetRequest(
+ self,
+ chroot: Optional[str] = None,
+ sysroot: Optional[str] = None,
+ build_target: Optional[str] = None,
+ output_dir: Optional[str] = None,
+ ) -> artifacts_pb2.BundleRequest:
+ """Helper to create a request message instance.
- Args:
- chroot: The chroot path.
- sysroot: The sysroot path.
- build_target: The build target name.
- output_dir: The output directory.
- """
- return artifacts_pb2.BundleRequest(
- sysroot={'path': sysroot, 'build_target': {'name': build_target}},
- chroot={'path': chroot}, output_dir=output_dir)
+ Args:
+ chroot: The chroot path.
+ sysroot: The sysroot path.
+ build_target: The build target name.
+ output_dir: The output directory.
+ """
+ return artifacts_pb2.BundleRequest(
+ sysroot={"path": sysroot, "build_target": {"name": build_target}},
+ chroot={"path": chroot},
+ output_dir=output_dir,
+ )
- def testValidateOnly(self):
- """Quick check that a validate only call does not execute any logic."""
- patch = self.PatchObject(artifacts_svc, 'BundleSimpleChromeArtifacts')
- request = self._GetRequest(chroot=self.chroot_dir,
- sysroot=self.sysroot_path,
- build_target='board', output_dir=self.output_dir)
- artifacts.BundleSimpleChromeArtifacts(request, self.response,
- self.validate_only_config)
- patch.assert_not_called()
+ def testValidateOnly(self):
+ """Quick check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(artifacts_svc, "BundleSimpleChromeArtifacts")
+ request = self._GetRequest(
+ chroot=self.chroot_dir,
+ sysroot=self.sysroot_path,
+ build_target="board",
+ output_dir=self.output_dir,
+ )
+ artifacts.BundleSimpleChromeArtifacts(
+ request, self.response, self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(artifacts_svc, 'BundleSimpleChromeArtifacts')
- request = self._GetRequest(chroot=self.chroot_dir,
- sysroot=self.sysroot_path,
- build_target='board', output_dir=self.output_dir)
- artifacts.BundleSimpleChromeArtifacts(request, self.response,
- self.mock_call_config)
- patch.assert_not_called()
- self.assertEqual(len(self.response.artifacts), 1)
- self.assertEqual(self.response.artifacts[0].path,
- os.path.join(self.output_dir, 'simple_chrome.txt'))
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(artifacts_svc, "BundleSimpleChromeArtifacts")
+ request = self._GetRequest(
+ chroot=self.chroot_dir,
+ sysroot=self.sysroot_path,
+ build_target="board",
+ output_dir=self.output_dir,
+ )
+ artifacts.BundleSimpleChromeArtifacts(
+ request, self.response, self.mock_call_config
+ )
+ patch.assert_not_called()
+ self.assertEqual(len(self.response.artifacts), 1)
+ self.assertEqual(
+ self.response.artifacts[0].path,
+ os.path.join(self.output_dir, "simple_chrome.txt"),
+ )
- def testNoBuildTarget(self):
- """Test no build target fails."""
- request = self._GetRequest(chroot=self.chroot_dir,
- sysroot=self.sysroot_path,
- output_dir=self.output_dir)
- response = self.response
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleSimpleChromeArtifacts(request, response, self.api_config)
+ def testNoBuildTarget(self):
+ """Test no build target fails."""
+ request = self._GetRequest(
+ chroot=self.chroot_dir,
+ sysroot=self.sysroot_path,
+ output_dir=self.output_dir,
+ )
+ response = self.response
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleSimpleChromeArtifacts(
+ request, response, self.api_config
+ )
- def testNoSysroot(self):
- """Test no sysroot fails."""
- request = self._GetRequest(build_target='board', output_dir=self.output_dir)
- response = self.response
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleSimpleChromeArtifacts(request, response, self.api_config)
+ def testNoSysroot(self):
+ """Test no sysroot fails."""
+ request = self._GetRequest(
+ build_target="board", output_dir=self.output_dir
+ )
+ response = self.response
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleSimpleChromeArtifacts(
+ request, response, self.api_config
+ )
- def testSysrootDoesNotExist(self):
- """Test no sysroot fails."""
- request = self._GetRequest(build_target='board', output_dir=self.output_dir,
- sysroot=self.does_not_exist)
- response = self.response
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleSimpleChromeArtifacts(request, response, self.api_config)
+ def testSysrootDoesNotExist(self):
+ """Test no sysroot fails."""
+ request = self._GetRequest(
+ build_target="board",
+ output_dir=self.output_dir,
+ sysroot=self.does_not_exist,
+ )
+ response = self.response
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleSimpleChromeArtifacts(
+ request, response, self.api_config
+ )
- def testNoOutputDir(self):
- """Test no output dir fails."""
- request = self._GetRequest(chroot=self.chroot_dir,
- sysroot=self.sysroot_path,
- build_target='board')
- response = self.response
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleSimpleChromeArtifacts(request, response, self.api_config)
+ def testNoOutputDir(self):
+ """Test no output dir fails."""
+ request = self._GetRequest(
+ chroot=self.chroot_dir,
+ sysroot=self.sysroot_path,
+ build_target="board",
+ )
+ response = self.response
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleSimpleChromeArtifacts(
+ request, response, self.api_config
+ )
- def testOutputDirDoesNotExist(self):
- """Test no output dir fails."""
- request = self._GetRequest(chroot=self.chroot_dir,
- sysroot=self.sysroot_path,
- build_target='board',
- output_dir=self.does_not_exist)
- response = self.response
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleSimpleChromeArtifacts(request, response, self.api_config)
+ def testOutputDirDoesNotExist(self):
+ """Test no output dir fails."""
+ request = self._GetRequest(
+ chroot=self.chroot_dir,
+ sysroot=self.sysroot_path,
+ build_target="board",
+ output_dir=self.does_not_exist,
+ )
+ response = self.response
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleSimpleChromeArtifacts(
+ request, response, self.api_config
+ )
- def testOutputHandling(self):
- """Test response output."""
- files = ['file1', 'file2', 'file3']
- expected_files = [os.path.join(self.output_dir, f) for f in files]
- self.PatchObject(artifacts_svc, 'BundleSimpleChromeArtifacts',
- return_value=expected_files)
- request = self._GetRequest(chroot=self.chroot_dir,
- sysroot=self.sysroot_path,
- build_target='board', output_dir=self.output_dir)
- response = self.response
+ def testOutputHandling(self):
+ """Test response output."""
+ files = ["file1", "file2", "file3"]
+ expected_files = [os.path.join(self.output_dir, f) for f in files]
+ self.PatchObject(
+ artifacts_svc,
+ "BundleSimpleChromeArtifacts",
+ return_value=expected_files,
+ )
+ request = self._GetRequest(
+ chroot=self.chroot_dir,
+ sysroot=self.sysroot_path,
+ build_target="board",
+ output_dir=self.output_dir,
+ )
+ response = self.response
- artifacts.BundleSimpleChromeArtifacts(request, response, self.api_config)
+ artifacts.BundleSimpleChromeArtifacts(
+ request, response, self.api_config
+ )
- self.assertTrue(response.artifacts)
- self.assertCountEqual(expected_files, [a.path for a in response.artifacts])
+ self.assertTrue(response.artifacts)
+ self.assertCountEqual(
+ expected_files, [a.path for a in response.artifacts]
+ )
-class BundleVmFilesTest(cros_test_lib.MockTempDirTestCase,
- api_config.ApiConfigMixin):
- """BuildVmFiles tests."""
+class BundleVmFilesTest(
+ cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
+):
+ """BuildVmFiles tests."""
- def setUp(self):
- self.output_dir = os.path.join(self.tempdir, 'output')
- osutils.SafeMakedirs(self.output_dir)
+ def setUp(self):
+ self.output_dir = os.path.join(self.tempdir, "output")
+ osutils.SafeMakedirs(self.output_dir)
- self.response = artifacts_pb2.BundleResponse()
+ self.response = artifacts_pb2.BundleResponse()
- def _GetInput(
- self,
- chroot: Optional[str] = None,
- sysroot: Optional[str] = None,
- test_results_dir: Optional[str] = None,
- output_dir: Optional[str] = None) -> artifacts_pb2.BundleVmFilesRequest:
- """Helper to build out an input message instance.
+ def _GetInput(
+ self,
+ chroot: Optional[str] = None,
+ sysroot: Optional[str] = None,
+ test_results_dir: Optional[str] = None,
+ output_dir: Optional[str] = None,
+ ) -> artifacts_pb2.BundleVmFilesRequest:
+ """Helper to build out an input message instance.
- Args:
- chroot: The chroot path.
- sysroot: The sysroot path relative to the chroot.
- test_results_dir: The test results directory relative to the sysroot.
- output_dir: The directory where the results tarball should be saved.
- """
- return artifacts_pb2.BundleVmFilesRequest(
- chroot={'path': chroot}, sysroot={'path': sysroot},
- test_results_dir=test_results_dir, output_dir=output_dir,
- )
+ Args:
+ chroot: The chroot path.
+ sysroot: The sysroot path relative to the chroot.
+ test_results_dir: The test results directory relative to the sysroot.
+ output_dir: The directory where the results tarball should be saved.
+ """
+ return artifacts_pb2.BundleVmFilesRequest(
+ chroot={"path": chroot},
+ sysroot={"path": sysroot},
+ test_results_dir=test_results_dir,
+ output_dir=output_dir,
+ )
- def testValidateOnly(self):
- """Quick check that a validate only call does not execute any logic."""
- patch = self.PatchObject(artifacts_svc, 'BundleVmFiles')
- in_proto = self._GetInput(chroot='/chroot/dir', sysroot='/build/board',
- test_results_dir='/test/results',
- output_dir=self.output_dir)
- artifacts.BundleVmFiles(in_proto, self.response, self.validate_only_config)
- patch.assert_not_called()
+ def testValidateOnly(self):
+ """Quick check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(artifacts_svc, "BundleVmFiles")
+ in_proto = self._GetInput(
+ chroot="/chroot/dir",
+ sysroot="/build/board",
+ test_results_dir="/test/results",
+ output_dir=self.output_dir,
+ )
+ artifacts.BundleVmFiles(
+ in_proto, self.response, self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(artifacts_svc, 'BundleVmFiles')
- in_proto = self._GetInput(chroot='/chroot/dir', sysroot='/build/board',
- test_results_dir='/test/results',
- output_dir=self.output_dir)
- artifacts.BundleVmFiles(in_proto, self.response, self.mock_call_config)
- patch.assert_not_called()
- self.assertEqual(len(self.response.artifacts), 1)
- self.assertEqual(self.response.artifacts[0].path,
- os.path.join(self.output_dir, 'f1.tar'))
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(artifacts_svc, "BundleVmFiles")
+ in_proto = self._GetInput(
+ chroot="/chroot/dir",
+ sysroot="/build/board",
+ test_results_dir="/test/results",
+ output_dir=self.output_dir,
+ )
+ artifacts.BundleVmFiles(in_proto, self.response, self.mock_call_config)
+ patch.assert_not_called()
+ self.assertEqual(len(self.response.artifacts), 1)
+ self.assertEqual(
+ self.response.artifacts[0].path,
+ os.path.join(self.output_dir, "f1.tar"),
+ )
- def testChrootMissing(self):
- """Test error handling for missing chroot."""
- in_proto = self._GetInput(sysroot='/build/board',
- test_results_dir='/test/results',
- output_dir=self.output_dir)
+ def testChrootMissing(self):
+ """Test error handling for missing chroot."""
+ in_proto = self._GetInput(
+ sysroot="/build/board",
+ test_results_dir="/test/results",
+ output_dir=self.output_dir,
+ )
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleVmFiles(in_proto, self.response, self.api_config)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleVmFiles(in_proto, self.response, self.api_config)
- def testTestResultsDirMissing(self):
- """Test error handling for missing test results directory."""
- in_proto = self._GetInput(chroot='/chroot/dir', sysroot='/build/board',
- output_dir=self.output_dir)
+ def testTestResultsDirMissing(self):
+ """Test error handling for missing test results directory."""
+ in_proto = self._GetInput(
+ chroot="/chroot/dir",
+ sysroot="/build/board",
+ output_dir=self.output_dir,
+ )
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleVmFiles(in_proto, self.response, self.api_config)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleVmFiles(in_proto, self.response, self.api_config)
- def testOutputDirMissing(self):
- """Test error handling for missing output directory."""
- in_proto = self._GetInput(chroot='/chroot/dir', sysroot='/build/board',
- test_results_dir='/test/results')
+ def testOutputDirMissing(self):
+ """Test error handling for missing output directory."""
+ in_proto = self._GetInput(
+ chroot="/chroot/dir",
+ sysroot="/build/board",
+ test_results_dir="/test/results",
+ )
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleVmFiles(in_proto, self.response, self.api_config)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleVmFiles(in_proto, self.response, self.api_config)
- def testOutputDirDoesNotExist(self):
- """Test error handling for output directory that does not exist."""
- in_proto = self._GetInput(chroot='/chroot/dir', sysroot='/build/board',
- output_dir=os.path.join(self.tempdir, 'dne'),
- test_results_dir='/test/results')
+ def testOutputDirDoesNotExist(self):
+ """Test error handling for output directory that does not exist."""
+ in_proto = self._GetInput(
+ chroot="/chroot/dir",
+ sysroot="/build/board",
+ output_dir=os.path.join(self.tempdir, "dne"),
+ test_results_dir="/test/results",
+ )
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleVmFiles(in_proto, self.response, self.api_config)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleVmFiles(in_proto, self.response, self.api_config)
- def testValidCall(self):
- """Test image dir building."""
- in_proto = self._GetInput(chroot='/chroot/dir', sysroot='/build/board',
- test_results_dir='/test/results',
- output_dir=self.output_dir)
+ def testValidCall(self):
+ """Test image dir building."""
+ in_proto = self._GetInput(
+ chroot="/chroot/dir",
+ sysroot="/build/board",
+ test_results_dir="/test/results",
+ output_dir=self.output_dir,
+ )
- expected_files = ['/tmp/output/f1.tar', '/tmp/output/f2.tar']
- patch = self.PatchObject(artifacts_svc, 'BundleVmFiles',
- return_value=expected_files)
+ expected_files = ["/tmp/output/f1.tar", "/tmp/output/f2.tar"]
+ patch = self.PatchObject(
+ artifacts_svc, "BundleVmFiles", return_value=expected_files
+ )
- artifacts.BundleVmFiles(in_proto, self.response, self.api_config)
+ artifacts.BundleVmFiles(in_proto, self.response, self.api_config)
- patch.assert_called_with(mock.ANY, '/test/results', self.output_dir)
+ patch.assert_called_with(mock.ANY, "/test/results", self.output_dir)
- # Make sure we have artifacts, and that every artifact is an expected file.
- self.assertTrue(self.response.artifacts)
- for artifact in self.response.artifacts:
- self.assertIn(artifact.path, expected_files)
- expected_files.remove(artifact.path)
+ # Make sure we have artifacts, and that every artifact is an expected file.
+ self.assertTrue(self.response.artifacts)
+ for artifact in self.response.artifacts:
+ self.assertIn(artifact.path, expected_files)
+ expected_files.remove(artifact.path)
- # Make sure we've seen all of the expected files.
- self.assertFalse(expected_files)
-
+ # Make sure we've seen all of the expected files.
+ self.assertFalse(expected_files)
class ExportCpeReportTest(BundleTestCase):
- """ExportCpeReport tests."""
+ """ExportCpeReport tests."""
- def testValidateOnly(self):
- """Quick check validate only calls don't execute."""
- patch = self.PatchObject(artifacts_svc, 'GenerateCpeReport')
+ def testValidateOnly(self):
+ """Quick check validate only calls don't execute."""
+ patch = self.PatchObject(artifacts_svc, "GenerateCpeReport")
- artifacts.ExportCpeReport(self.sysroot_request, self.response,
- self.validate_only_config)
+ artifacts.ExportCpeReport(
+ self.sysroot_request, self.response, self.validate_only_config
+ )
- patch.assert_not_called()
+ patch.assert_not_called()
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(artifacts_svc, 'GenerateCpeReport')
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(artifacts_svc, "GenerateCpeReport")
- artifacts.ExportCpeReport(self.sysroot_request, self.response,
- self.mock_call_config)
+ artifacts.ExportCpeReport(
+ self.sysroot_request, self.response, self.mock_call_config
+ )
- patch.assert_not_called()
- self.assertEqual(len(self.response.artifacts), 2)
- self.assertEqual(self.response.artifacts[0].path,
- os.path.join(self.output_dir, 'cpe_report.txt'))
- self.assertEqual(self.response.artifacts[1].path,
- os.path.join(self.output_dir, 'cpe_warnings.txt'))
+ patch.assert_not_called()
+ self.assertEqual(len(self.response.artifacts), 2)
+ self.assertEqual(
+ self.response.artifacts[0].path,
+ os.path.join(self.output_dir, "cpe_report.txt"),
+ )
+ self.assertEqual(
+ self.response.artifacts[1].path,
+ os.path.join(self.output_dir, "cpe_warnings.txt"),
+ )
- def testSuccess(self):
- """Test success case."""
- expected = artifacts_svc.CpeResult(
- report='/output/report.json', warnings='/output/warnings.json')
- self.PatchObject(artifacts_svc, 'GenerateCpeReport', return_value=expected)
+ def testSuccess(self):
+ """Test success case."""
+ expected = artifacts_svc.CpeResult(
+ report="/output/report.json", warnings="/output/warnings.json"
+ )
+ self.PatchObject(
+ artifacts_svc, "GenerateCpeReport", return_value=expected
+ )
- artifacts.ExportCpeReport(self.sysroot_request, self.response,
- self.api_config)
+ artifacts.ExportCpeReport(
+ self.sysroot_request, self.response, self.api_config
+ )
- for artifact in self.response.artifacts:
- self.assertIn(artifact.path, [expected.report, expected.warnings])
+ for artifact in self.response.artifacts:
+ self.assertIn(artifact.path, [expected.report, expected.warnings])
class BundleGceTarballTest(BundleTestCase):
- """Unittests for BundleGceTarball."""
+ """Unittests for BundleGceTarball."""
- def testValidateOnly(self):
- """Check that a validate only call does not execute any logic."""
- patch = self.PatchObject(artifacts_svc, 'BundleGceTarball')
- artifacts.BundleGceTarball(self.target_request, self.response,
- self.validate_only_config)
- patch.assert_not_called()
+ def testValidateOnly(self):
+ """Check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(artifacts_svc, "BundleGceTarball")
+ artifacts.BundleGceTarball(
+ self.target_request, self.response, self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(artifacts_svc, 'BundleGceTarball')
- artifacts.BundleGceTarball(self.target_request, self.response,
- self.mock_call_config)
- patch.assert_not_called()
- self.assertEqual(len(self.response.artifacts), 1)
- self.assertEqual(self.response.artifacts[0].path,
- os.path.join(self.output_dir,
- constants.TEST_IMAGE_GCE_TAR))
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(artifacts_svc, "BundleGceTarball")
+ artifacts.BundleGceTarball(
+ self.target_request, self.response, self.mock_call_config
+ )
+ patch.assert_not_called()
+ self.assertEqual(len(self.response.artifacts), 1)
+ self.assertEqual(
+ self.response.artifacts[0].path,
+ os.path.join(self.output_dir, constants.TEST_IMAGE_GCE_TAR),
+ )
- def testBundleGceTarball(self):
- """BundleGceTarball calls cbuildbot/commands with correct args."""
- bundle_gce_tarball = self.PatchObject(
- artifacts_svc, 'BundleGceTarball',
- return_value=os.path.join(self.output_dir,
- constants.TEST_IMAGE_GCE_TAR))
- self.PatchObject(os.path, 'exists', return_value=True)
- artifacts.BundleGceTarball(self.target_request, self.response,
- self.api_config)
- self.assertEqual(
- [artifact.path for artifact in self.response.artifacts],
- [os.path.join(self.output_dir, constants.TEST_IMAGE_GCE_TAR)])
+ def testBundleGceTarball(self):
+ """BundleGceTarball calls cbuildbot/commands with correct args."""
+ bundle_gce_tarball = self.PatchObject(
+ artifacts_svc,
+ "BundleGceTarball",
+ return_value=os.path.join(
+ self.output_dir, constants.TEST_IMAGE_GCE_TAR
+ ),
+ )
+ self.PatchObject(os.path, "exists", return_value=True)
+ artifacts.BundleGceTarball(
+ self.target_request, self.response, self.api_config
+ )
+ self.assertEqual(
+ [artifact.path for artifact in self.response.artifacts],
+ [os.path.join(self.output_dir, constants.TEST_IMAGE_GCE_TAR)],
+ )
- latest = os.path.join(self.source_root, 'src/build/images/target/latest')
- self.assertEqual(
- bundle_gce_tarball.call_args_list,
- [mock.call(self.output_dir, latest)])
+ latest = os.path.join(
+ self.source_root, "src/build/images/target/latest"
+ )
+ self.assertEqual(
+ bundle_gce_tarball.call_args_list,
+ [mock.call(self.output_dir, latest)],
+ )
- def testBundleGceTarballNoImageDir(self):
- """BundleGceTarball dies when image dir does not exist."""
- self.PatchObject(os.path, 'exists', return_value=False)
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.BundleGceTarball(self.target_request, self.response,
- self.api_config)
+ def testBundleGceTarballNoImageDir(self):
+ """BundleGceTarball dies when image dir does not exist."""
+ self.PatchObject(os.path, "exists", return_value=False)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.BundleGceTarball(
+ self.target_request, self.response, self.api_config
+ )
-class FetchMetadataTestCase(cros_test_lib.MockTempDirTestCase,
- api_config.ApiConfigMixin):
- """Unittests for FetchMetadata."""
- sysroot_path = '/build/coral'
- chroot_name = 'chroot'
+class FetchMetadataTestCase(
+ cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
+):
+ """Unittests for FetchMetadata."""
- def setUp(self):
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=False)
- self.chroot_path = os.path.join(self.tempdir, 'chroot')
- pathlib.Path(self.chroot_path).touch()
- self.expected_filepaths = [os.path.join(self.chroot_path, fp) for fp in (
- 'build/coral/usr/local/build/autotest/autotest_metadata.pb',
- 'build/coral/usr/share/tast/metadata/local/cros.pb',
- 'build/coral/build/share/tast/metadata/local/crosint.pb',
- 'usr/share/tast/metadata/remote/cros.pb',
- )]
- self.PatchObject(cros_build_lib, 'AssertOutsideChroot')
+ sysroot_path = "/build/coral"
+ chroot_name = "chroot"
- def createFetchMetadataRequest(self, use_sysroot_path=True, use_chroot=True):
- """Construct a FetchMetadataRequest for use in test cases."""
- request = artifacts_pb2.FetchMetadataRequest()
- if use_sysroot_path:
- request.sysroot.path = self.sysroot_path
- if use_chroot:
- request.chroot.path = self.chroot_path
- return request
+ def setUp(self):
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)
+ self.chroot_path = os.path.join(self.tempdir, "chroot")
+ pathlib.Path(self.chroot_path).touch()
+ self.expected_filepaths = [
+ os.path.join(self.chroot_path, fp)
+ for fp in (
+ "build/coral/usr/local/build/autotest/autotest_metadata.pb",
+ "build/coral/usr/share/tast/metadata/local/cros.pb",
+ "build/coral/build/share/tast/metadata/local/crosint.pb",
+ "usr/share/tast/metadata/remote/cros.pb",
+ )
+ ]
+ self.PatchObject(cros_build_lib, "AssertOutsideChroot")
- def testValidateOnly(self):
- """Check that a validate only call does not execute any logic."""
- patch = self.PatchObject(controller_util, 'ParseSysroot')
- request = self.createFetchMetadataRequest()
- response = artifacts_pb2.FetchMetadataResponse()
- artifacts.FetchMetadata(request, response, self.validate_only_config)
- patch.assert_not_called()
+ def createFetchMetadataRequest(
+ self, use_sysroot_path=True, use_chroot=True
+ ):
+ """Construct a FetchMetadataRequest for use in test cases."""
+ request = artifacts_pb2.FetchMetadataRequest()
+ if use_sysroot_path:
+ request.sysroot.path = self.sysroot_path
+ if use_chroot:
+ request.chroot.path = self.chroot_path
+ return request
- def testMockCall(self):
- """Test that a mock call does not execute logic, returns mocked value."""
- patch = self.PatchObject(controller_util, 'ParseSysroot')
- request = self.createFetchMetadataRequest()
- response = artifacts_pb2.FetchMetadataResponse()
- artifacts.FetchMetadata(request, response, self.mock_call_config)
- patch.assert_not_called()
- self.assertGreater(len(response.filepaths), 0)
+ def testValidateOnly(self):
+ """Check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(controller_util, "ParseSysroot")
+ request = self.createFetchMetadataRequest()
+ response = artifacts_pb2.FetchMetadataResponse()
+ artifacts.FetchMetadata(request, response, self.validate_only_config)
+ patch.assert_not_called()
- def testNoSysrootPath(self):
- """Check that a request with no sysroot.path results in failure."""
- request = self.createFetchMetadataRequest(use_sysroot_path=False)
- response = artifacts_pb2.FetchMetadataResponse()
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.FetchMetadata(request, response, self.api_config)
+ def testMockCall(self):
+ """Test that a mock call does not execute logic, returns mocked value."""
+ patch = self.PatchObject(controller_util, "ParseSysroot")
+ request = self.createFetchMetadataRequest()
+ response = artifacts_pb2.FetchMetadataResponse()
+ artifacts.FetchMetadata(request, response, self.mock_call_config)
+ patch.assert_not_called()
+ self.assertGreater(len(response.filepaths), 0)
- def testNoChroot(self):
- """Check that a request with no chroot results in failure."""
- request = self.createFetchMetadataRequest(use_chroot=False)
- response = artifacts_pb2.FetchMetadataResponse()
- with self.assertRaises(cros_build_lib.DieSystemExit):
- artifacts.FetchMetadata(request, response, self.api_config)
+ def testNoSysrootPath(self):
+ """Check that a request with no sysroot.path results in failure."""
+ request = self.createFetchMetadataRequest(use_sysroot_path=False)
+ response = artifacts_pb2.FetchMetadataResponse()
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.FetchMetadata(request, response, self.api_config)
- def testSuccess(self):
- """Check that a well-formed request yields the expected results."""
- request = self.createFetchMetadataRequest(use_chroot=True)
- response = artifacts_pb2.FetchMetadataResponse()
- artifacts.FetchMetadata(request, response, self.api_config)
- actual_filepaths = [fp.path.path for fp in response.filepaths]
- self.assertEqual(sorted(actual_filepaths), sorted(self.expected_filepaths))
- self.assertTrue(all(fp.path.location == common_pb2.Path.OUTSIDE
- for fp in response.filepaths))
+ def testNoChroot(self):
+ """Check that a request with no chroot results in failure."""
+ request = self.createFetchMetadataRequest(use_chroot=False)
+ response = artifacts_pb2.FetchMetadataResponse()
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ artifacts.FetchMetadata(request, response, self.api_config)
+
+ def testSuccess(self):
+ """Check that a well-formed request yields the expected results."""
+ request = self.createFetchMetadataRequest(use_chroot=True)
+ response = artifacts_pb2.FetchMetadataResponse()
+ artifacts.FetchMetadata(request, response, self.api_config)
+ actual_filepaths = [fp.path.path for fp in response.filepaths]
+ self.assertEqual(
+ sorted(actual_filepaths), sorted(self.expected_filepaths)
+ )
+ self.assertTrue(
+ all(
+ fp.path.location == common_pb2.Path.OUTSIDE
+ for fp in response.filepaths
+ )
+ )