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/field_handler_unittest.py b/api/field_handler_unittest.py
index 786136a..852a1d7 100644
--- a/api/field_handler_unittest.py
+++ b/api/field_handler_unittest.py
@@ -16,467 +16,513 @@
class ChrootHandlerTest(cros_test_lib.TestCase):
- """ChrootHandler tests."""
+ """ChrootHandler tests."""
- def setUp(self):
- self.path = '/chroot/dir'
- self.cache_dir = '/cache/dir'
- self.chrome_dir = '/chrome/dir'
- self.env = {'FEATURES': 'thing', 'CHROME_ORIGIN': 'LOCAL_SOURCE'}
- self.expected_chroot = chroot_lib.Chroot(path=self.path,
- cache_dir=self.cache_dir,
- chrome_root=self.chrome_dir,
- env=self.env)
+ def setUp(self):
+ self.path = "/chroot/dir"
+ self.cache_dir = "/cache/dir"
+ self.chrome_dir = "/chrome/dir"
+ self.env = {"FEATURES": "thing", "CHROME_ORIGIN": "LOCAL_SOURCE"}
+ self.expected_chroot = chroot_lib.Chroot(
+ path=self.path,
+ cache_dir=self.cache_dir,
+ chrome_root=self.chrome_dir,
+ env=self.env,
+ )
- def test_parse_chroot_success(self):
- """Test successful Chroot message parse."""
- chroot_msg = common_pb2.Chroot()
- chroot_msg.path = self.path
- chroot_msg.cache_dir = self.cache_dir
- chroot_msg.chrome_dir = self.chrome_dir
- chroot_msg.env.features.add().feature = 'thing'
+ def test_parse_chroot_success(self):
+ """Test successful Chroot message parse."""
+ chroot_msg = common_pb2.Chroot()
+ chroot_msg.path = self.path
+ chroot_msg.cache_dir = self.cache_dir
+ chroot_msg.chrome_dir = self.chrome_dir
+ chroot_msg.env.features.add().feature = "thing"
- chroot_handler = field_handler.ChrootHandler(clear_field=False)
- parsed_chroot = chroot_handler.parse_chroot(chroot_msg)
+ chroot_handler = field_handler.ChrootHandler(clear_field=False)
+ parsed_chroot = chroot_handler.parse_chroot(chroot_msg)
- self.assertEqual(self.expected_chroot, parsed_chroot)
+ self.assertEqual(self.expected_chroot, parsed_chroot)
- def test_handle_success(self):
- """Test a successful Chroot message parse from a parent message."""
- message = build_api_test_pb2.TestRequestMessage()
- message.chroot.path = self.path
- message.chroot.cache_dir = self.cache_dir
- message.chroot.chrome_dir = self.chrome_dir
- message.chroot.env.features.add().feature = 'thing'
+ def test_handle_success(self):
+ """Test a successful Chroot message parse from a parent message."""
+ message = build_api_test_pb2.TestRequestMessage()
+ message.chroot.path = self.path
+ message.chroot.cache_dir = self.cache_dir
+ message.chroot.chrome_dir = self.chrome_dir
+ message.chroot.env.features.add().feature = "thing"
- # First a no-clear parse.
- chroot_handler = field_handler.ChrootHandler(clear_field=False)
- chroot = chroot_handler.handle(message)
+ # First a no-clear parse.
+ chroot_handler = field_handler.ChrootHandler(clear_field=False)
+ chroot = chroot_handler.handle(message)
- self.assertEqual(self.expected_chroot, chroot)
- self.assertEqual(message.chroot.path, self.path)
+ self.assertEqual(self.expected_chroot, chroot)
+ self.assertEqual(message.chroot.path, self.path)
- # A clear field parse.
- clear_chroot_handler = field_handler.ChrootHandler(clear_field=True)
- chroot = clear_chroot_handler.handle(message)
+ # A clear field parse.
+ clear_chroot_handler = field_handler.ChrootHandler(clear_field=True)
+ chroot = clear_chroot_handler.handle(message)
- self.assertEqual(self.expected_chroot, chroot)
- self.assertFalse(message.chroot.path)
+ self.assertEqual(self.expected_chroot, chroot)
+ self.assertFalse(message.chroot.path)
- def test_handle_empty_chroot_message(self):
- """Test handling of an empty chroot message."""
- message = build_api_test_pb2.TestRequestMessage()
- empty_chroot = chroot_lib.Chroot()
+ def test_handle_empty_chroot_message(self):
+ """Test handling of an empty chroot message."""
+ message = build_api_test_pb2.TestRequestMessage()
+ empty_chroot = chroot_lib.Chroot()
- chroot_handler = field_handler.ChrootHandler(clear_field=False)
- chroot = chroot_handler.handle(message)
+ chroot_handler = field_handler.ChrootHandler(clear_field=False)
+ chroot = chroot_handler.handle(message)
- self.assertEqual(empty_chroot, chroot)
+ self.assertEqual(empty_chroot, chroot)
class HandleRemoteexec(cros_test_lib.TempDirTestCase):
- """Tests for handling remoteexec."""
+ """Tests for handling remoteexec."""
- def test_handle_remoteexec(self):
- """Test handling remoteexec when there is a RemoteexecConfig."""
- reclient_dir = os.path.join(self.tempdir, 'cipd/rbe')
- reproxy_cfg_file = os.path.join(self.tempdir,
- 'reclient_cfgs/reproxy_config.cfg')
+ def test_handle_remoteexec(self):
+ """Test handling remoteexec when there is a RemoteexecConfig."""
+ reclient_dir = os.path.join(self.tempdir, "cipd/rbe")
+ reproxy_cfg_file = os.path.join(
+ self.tempdir, "reclient_cfgs/reproxy_config.cfg"
+ )
- osutils.SafeMakedirs(reclient_dir)
- osutils.Touch(reproxy_cfg_file, makedirs=True)
- remoteexec_config = common_pb2.RemoteexecConfig(
- reclient_dir=reclient_dir, reproxy_cfg_file=reproxy_cfg_file)
- message = build_api_test_pb2.TestRequestMessage(
- remoteexec_config=remoteexec_config)
+ osutils.SafeMakedirs(reclient_dir)
+ osutils.Touch(reproxy_cfg_file, makedirs=True)
+ remoteexec_config = common_pb2.RemoteexecConfig(
+ reclient_dir=reclient_dir, reproxy_cfg_file=reproxy_cfg_file
+ )
+ message = build_api_test_pb2.TestRequestMessage(
+ remoteexec_config=remoteexec_config
+ )
- expected = remoteexec_util.Remoteexec(reclient_dir, reproxy_cfg_file)
- self.assertEqual(expected, field_handler.handle_remoteexec(message))
+ expected = remoteexec_util.Remoteexec(reclient_dir, reproxy_cfg_file)
+ self.assertEqual(expected, field_handler.handle_remoteexec(message))
- def test_handle_remoteexec_no_config(self):
- """Test handling remoteexec when there is no RmoteexecConfig."""
- message = build_api_test_pb2.TestRequestMessage()
- self.assertIsNone(field_handler.handle_remoteexec(message))
+ def test_handle_remoteexec_no_config(self):
+ """Test handling remoteexec when there is no RmoteexecConfig."""
+ message = build_api_test_pb2.TestRequestMessage()
+ self.assertIsNone(field_handler.handle_remoteexec(message))
class CopyPathInTest(cros_test_lib.TempDirTestCase):
- """PathHandler tests."""
+ """PathHandler tests."""
- def setUp(self):
- self.source_dir = os.path.join(self.tempdir, 'source')
- self.dest_dir = os.path.join(self.tempdir, 'destination')
- osutils.SafeMakedirs(self.source_dir)
- osutils.SafeMakedirs(self.dest_dir)
+ def setUp(self):
+ self.source_dir = os.path.join(self.tempdir, "source")
+ self.dest_dir = os.path.join(self.tempdir, "destination")
+ osutils.SafeMakedirs(self.source_dir)
+ osutils.SafeMakedirs(self.dest_dir)
- self.source_file1 = os.path.join(self.source_dir, 'file1')
- self.file1_contents = 'file 1'
- osutils.WriteFile(self.source_file1, self.file1_contents)
+ self.source_file1 = os.path.join(self.source_dir, "file1")
+ self.file1_contents = "file 1"
+ osutils.WriteFile(self.source_file1, self.file1_contents)
- self.file2_contents = 'some data'
- self.source_file2 = os.path.join(self.source_dir, 'file2')
- osutils.WriteFile(self.source_file2, self.file2_contents)
+ self.file2_contents = "some data"
+ self.source_file2 = os.path.join(self.source_dir, "file2")
+ osutils.WriteFile(self.source_file2, self.file2_contents)
- def _path_checks(self, source_file, dest_file, contents=None):
- """Set of common checks for the copied files/directories."""
- # Message should now reflect the new path.
- self.assertNotEqual(source_file, dest_file)
- # The new path should be in the destination directory.
- self.assertStartsWith(dest_file, self.dest_dir)
- # The new file should exist.
- self.assertExists(dest_file)
+ def _path_checks(self, source_file, dest_file, contents=None):
+ """Set of common checks for the copied files/directories."""
+ # Message should now reflect the new path.
+ self.assertNotEqual(source_file, dest_file)
+ # The new path should be in the destination directory.
+ self.assertStartsWith(dest_file, self.dest_dir)
+ # The new file should exist.
+ self.assertExists(dest_file)
- if contents:
- # The contents should be the same as the source file.
- self.assertFileContents(dest_file, contents)
+ if contents:
+ # The contents should be the same as the source file.
+ self.assertFileContents(dest_file, contents)
- def test_handle_file(self):
- """Test handling of a single file."""
- message = build_api_test_pb2.TestRequestMessage()
- message.path.path = self.source_file1
- message.path.location = common_pb2.Path.OUTSIDE
+ def test_handle_file(self):
+ """Test handling of a single file."""
+ message = build_api_test_pb2.TestRequestMessage()
+ message.path.path = self.source_file1
+ message.path.location = common_pb2.Path.OUTSIDE
- with field_handler.copy_paths_in(message, self.dest_dir, delete=True):
- new_path = message.path.path
- self._path_checks(self.source_file1, new_path, self.file1_contents)
+ with field_handler.copy_paths_in(message, self.dest_dir, delete=True):
+ new_path = message.path.path
+ self._path_checks(self.source_file1, new_path, self.file1_contents)
- # The file should have been deleted on exit with delete=True.
- self.assertNotExists(new_path)
- # The original should still exist.
- self.assertExists(self.source_file1)
- # The path should get reset.
- self.assertEqual(message.path.path, self.source_file1)
+ # The file should have been deleted on exit with delete=True.
+ self.assertNotExists(new_path)
+ # The original should still exist.
+ self.assertExists(self.source_file1)
+ # The path should get reset.
+ self.assertEqual(message.path.path, self.source_file1)
- def test_handle_files(self):
- """Test handling of multiple files."""
- message = build_api_test_pb2.TestRequestMessage()
- message.path.path = self.source_file1
- message.path.location = common_pb2.Path.OUTSIDE
- message.another_path.path = self.source_file2
- message.another_path.location = common_pb2.Path.OUTSIDE
+ def test_handle_files(self):
+ """Test handling of multiple files."""
+ message = build_api_test_pb2.TestRequestMessage()
+ message.path.path = self.source_file1
+ message.path.location = common_pb2.Path.OUTSIDE
+ message.another_path.path = self.source_file2
+ message.another_path.location = common_pb2.Path.OUTSIDE
- with field_handler.copy_paths_in(message, self.dest_dir, delete=False):
- new_path1 = message.path.path
- new_path2 = message.another_path.path
+ with field_handler.copy_paths_in(message, self.dest_dir, delete=False):
+ new_path1 = message.path.path
+ new_path2 = message.another_path.path
- self._path_checks(self.source_file1, new_path1, self.file1_contents)
- self._path_checks(self.source_file2, new_path2, self.file2_contents)
+ self._path_checks(self.source_file1, new_path1, self.file1_contents)
+ self._path_checks(self.source_file2, new_path2, self.file2_contents)
- # The files should still exist with delete=False.
- self.assertExists(new_path1)
- self.assertExists(new_path2)
+ # The files should still exist with delete=False.
+ self.assertExists(new_path1)
+ self.assertExists(new_path2)
- def test_handle_nested_file(self):
- """Test the nested path handling."""
- message = build_api_test_pb2.TestRequestMessage()
- message.nested_path.path.path = self.source_file1
- message.nested_path.path.location = common_pb2.Path.OUTSIDE
+ def test_handle_nested_file(self):
+ """Test the nested path handling."""
+ message = build_api_test_pb2.TestRequestMessage()
+ message.nested_path.path.path = self.source_file1
+ message.nested_path.path.location = common_pb2.Path.OUTSIDE
- with field_handler.copy_paths_in(message, self.dest_dir):
- new_path = message.nested_path.path.path
- self._path_checks(self.source_file1, new_path, self.file1_contents)
+ with field_handler.copy_paths_in(message, self.dest_dir):
+ new_path = message.nested_path.path.path
+ self._path_checks(self.source_file1, new_path, self.file1_contents)
- def test_handle_directory(self):
- """Test handling of a directory."""
- message = build_api_test_pb2.TestRequestMessage()
- message.path.path = self.source_dir
- message.path.location = common_pb2.Path.OUTSIDE
+ def test_handle_directory(self):
+ """Test handling of a directory."""
+ message = build_api_test_pb2.TestRequestMessage()
+ message.path.path = self.source_dir
+ message.path.location = common_pb2.Path.OUTSIDE
- with field_handler.copy_paths_in(message, self.dest_dir):
- new_path = message.path.path
+ with field_handler.copy_paths_in(message, self.dest_dir):
+ new_path = message.path.path
- self._path_checks(self.source_dir, self.dest_dir)
- # Make sure both directories have the same files.
- self.assertCountEqual(os.listdir(self.source_dir), os.listdir(new_path))
+ self._path_checks(self.source_dir, self.dest_dir)
+ # Make sure both directories have the same files.
+ self.assertCountEqual(
+ os.listdir(self.source_dir), os.listdir(new_path)
+ )
- def test_direction(self):
- """Test the direction argument preventing copies."""
- message = build_api_test_pb2.TestRequestMessage()
- message.path.path = self.source_file1
- message.path.location = common_pb2.Path.INSIDE
+ def test_direction(self):
+ """Test the direction argument preventing copies."""
+ message = build_api_test_pb2.TestRequestMessage()
+ message.path.path = self.source_file1
+ message.path.location = common_pb2.Path.INSIDE
- with field_handler.copy_paths_in(message, self.dest_dir, delete=True):
- self.assertEqual(self.source_file1, message.path.path)
+ with field_handler.copy_paths_in(message, self.dest_dir, delete=True):
+ self.assertEqual(self.source_file1, message.path.path)
- # It should not be deleting the file when it doesn't need to copy it even
- # with delete=True.
- self.assertExists(self.source_file1)
+ # It should not be deleting the file when it doesn't need to copy it even
+ # with delete=True.
+ self.assertExists(self.source_file1)
- def test_prefix_inside(self):
- """Test the transfer inside prefix handling."""
- message = build_api_test_pb2.TestRequestMessage()
- message.path.path = self.source_dir
- message.path.location = common_pb2.Path.OUTSIDE
+ def test_prefix_inside(self):
+ """Test the transfer inside prefix handling."""
+ message = build_api_test_pb2.TestRequestMessage()
+ message.path.path = self.source_dir
+ message.path.location = common_pb2.Path.OUTSIDE
- with field_handler.copy_paths_in(message, self.dest_dir,
- prefix=self.tempdir):
- new_path = message.path.path
- # The prefix should be removed.
- self.assertFalse(new_path.startswith(str(self.tempdir)))
+ with field_handler.copy_paths_in(
+ message, self.dest_dir, prefix=self.tempdir
+ ):
+ new_path = message.path.path
+ # The prefix should be removed.
+ self.assertFalse(new_path.startswith(str(self.tempdir)))
class SyncDirsTest(cros_test_lib.TempDirTestCase):
- """Tests for sync_dirs."""
+ """Tests for sync_dirs."""
- def setUp(self):
- D = cros_test_lib.Directory
- filesystem = (
- D('chroot', (
- D('tmp', (
- D('tempdir', ()),
- )),
- )),
- D('sources', (
- D('single_file', ('single_file.txt',)),
- D('nested_directories', (
- 'basedir_file.log',
- D('nested1', (
- 'nested1.txt',
- D('nested2', ('nested2.txt',)),
- )),
- )),
- )),
- )
- cros_test_lib.CreateOnDiskHierarchy(self.tempdir, filesystem)
+ def setUp(self):
+ D = cros_test_lib.Directory
+ filesystem = (
+ D("chroot", (D("tmp", (D("tempdir", ()),)),)),
+ D(
+ "sources",
+ (
+ D("single_file", ("single_file.txt",)),
+ D(
+ "nested_directories",
+ (
+ "basedir_file.log",
+ D(
+ "nested1",
+ (
+ "nested1.txt",
+ D("nested2", ("nested2.txt",)),
+ ),
+ ),
+ ),
+ ),
+ ),
+ ),
+ )
+ cros_test_lib.CreateOnDiskHierarchy(self.tempdir, filesystem)
- self.chroot = os.path.join(self.tempdir, 'chroot')
- self.chroot_tmp = os.path.join(self.chroot, 'tmp')
- self.destination = os.path.join(self.chroot_tmp, 'tempdir')
- self.inside_path = '/tmp/tempdir'
+ self.chroot = os.path.join(self.tempdir, "chroot")
+ self.chroot_tmp = os.path.join(self.chroot, "tmp")
+ self.destination = os.path.join(self.chroot_tmp, "tempdir")
+ self.inside_path = "/tmp/tempdir"
- self.single_file_src = os.path.join(self.tempdir, 'sources', 'single_file')
- self.sf_src_file = os.path.join(self.single_file_src, 'single_file.txt')
- self.sf_dest_file = os.path.join(self.destination, 'single_file.txt')
+ self.single_file_src = os.path.join(
+ self.tempdir, "sources", "single_file"
+ )
+ self.sf_src_file = os.path.join(self.single_file_src, "single_file.txt")
+ self.sf_dest_file = os.path.join(self.destination, "single_file.txt")
- self.nested_dirs_src = (
- os.path.join(self.tempdir, 'sources', 'nested_directories'))
- self.nested_src_files = (
- os.path.join(self.nested_dirs_src, 'basedir_file.log'),
- os.path.join(self.nested_dirs_src, 'nested1', 'nested1.txt'),
- os.path.join(self.nested_dirs_src, 'nested1', 'nested2', 'nested2.txt'),
- )
- self.nested_dest_files = (
- os.path.join(self.destination, 'basedir_file.log'),
- os.path.join(self.destination, 'nested1', 'nested1.txt'),
- os.path.join(self.destination, 'nested1', 'nested2', 'nested2.txt'),
- )
+ self.nested_dirs_src = os.path.join(
+ self.tempdir, "sources", "nested_directories"
+ )
+ self.nested_src_files = (
+ os.path.join(self.nested_dirs_src, "basedir_file.log"),
+ os.path.join(self.nested_dirs_src, "nested1", "nested1.txt"),
+ os.path.join(
+ self.nested_dirs_src, "nested1", "nested2", "nested2.txt"
+ ),
+ )
+ self.nested_dest_files = (
+ os.path.join(self.destination, "basedir_file.log"),
+ os.path.join(self.destination, "nested1", "nested1.txt"),
+ os.path.join(self.destination, "nested1", "nested2", "nested2.txt"),
+ )
- self.message = build_api_test_pb2.TestRequestMessage()
+ self.message = build_api_test_pb2.TestRequestMessage()
- def _assertExist(self, files):
- for f in files:
- self.assertExists(f)
+ def _assertExist(self, files):
+ for f in files:
+ self.assertExists(f)
- def _assertNotExist(self, files):
- for f in files:
- self.assertNotExists(f)
+ def _assertNotExist(self, files):
+ for f in files:
+ self.assertNotExists(f)
- def testSingleFileTransfer(self):
- """Single source file syncs."""
- self.message.synced_dir.dir = self.single_file_src
+ def testSingleFileTransfer(self):
+ """Single source file syncs."""
+ self.message.synced_dir.dir = self.single_file_src
- # Verify source files exist and destination files do not.
- self.assertExists(self.sf_src_file)
- self.assertNotExists(self.sf_dest_file)
+ # Verify source files exist and destination files do not.
+ self.assertExists(self.sf_src_file)
+ self.assertNotExists(self.sf_dest_file)
- with field_handler.sync_dirs(self.message, self.destination, self.chroot):
- # Verify the prefix is getting correctly stripped.
- self.assertEqual(self.message.synced_dir.dir, self.inside_path)
- # Verify the files have all been correctly copied in.
- self.assertExists(self.sf_dest_file)
+ with field_handler.sync_dirs(
+ self.message, self.destination, self.chroot
+ ):
+ # Verify the prefix is getting correctly stripped.
+ self.assertEqual(self.message.synced_dir.dir, self.inside_path)
+ # Verify the files have all been correctly copied in.
+ self.assertExists(self.sf_dest_file)
- self.assertEqual(self.message.synced_dir.dir, self.single_file_src)
- # Verify the files have all been copied out.
- self.assertExists(self.sf_src_file)
+ self.assertEqual(self.message.synced_dir.dir, self.single_file_src)
+ # Verify the files have all been copied out.
+ self.assertExists(self.sf_src_file)
- def testNestedFileSync(self):
- """Nested directories and files sync."""
- self.message.synced_dir.dir = self.nested_dirs_src
+ def testNestedFileSync(self):
+ """Nested directories and files sync."""
+ self.message.synced_dir.dir = self.nested_dirs_src
- self._assertExist(self.nested_src_files)
- self._assertNotExist(self.nested_dest_files)
+ self._assertExist(self.nested_src_files)
+ self._assertNotExist(self.nested_dest_files)
- with field_handler.sync_dirs(self.message, self.destination, self.chroot):
- self.assertEqual(self.message.synced_dir.dir, self.inside_path)
- self._assertExist(self.nested_dest_files)
+ with field_handler.sync_dirs(
+ self.message, self.destination, self.chroot
+ ):
+ self.assertEqual(self.message.synced_dir.dir, self.inside_path)
+ self._assertExist(self.nested_dest_files)
- self.assertEqual(self.message.synced_dir.dir, self.nested_dirs_src)
- self._assertExist(self.nested_src_files)
+ self.assertEqual(self.message.synced_dir.dir, self.nested_dirs_src)
+ self._assertExist(self.nested_src_files)
- def testDeletion(self):
- """Test file deletions are exported correctly."""
- self.message.synced_dir.dir = self.nested_dirs_src
+ def testDeletion(self):
+ """Test file deletions are exported correctly."""
+ self.message.synced_dir.dir = self.nested_dirs_src
- deleted_src = os.path.join(self.nested_dirs_src, 'nested1', 'nested1.txt')
- deleted_dest = os.path.join(self.destination, 'nested1', 'nested1.txt')
+ deleted_src = os.path.join(
+ self.nested_dirs_src, "nested1", "nested1.txt"
+ )
+ deleted_dest = os.path.join(self.destination, "nested1", "nested1.txt")
- self._assertExist(self.nested_src_files)
- self._assertNotExist(self.nested_dest_files)
+ self._assertExist(self.nested_src_files)
+ self._assertNotExist(self.nested_dest_files)
- with field_handler.sync_dirs(self.message, self.destination, self.chroot):
- self._assertExist(self.nested_dest_files)
- osutils.SafeUnlink(deleted_dest)
+ with field_handler.sync_dirs(
+ self.message, self.destination, self.chroot
+ ):
+ self._assertExist(self.nested_dest_files)
+ osutils.SafeUnlink(deleted_dest)
- self._assertExist(set(self.nested_src_files) - {deleted_src})
- self.assertNotExists(deleted_src)
+ self._assertExist(set(self.nested_src_files) - {deleted_src})
+ self.assertNotExists(deleted_src)
- def testCreation(self):
- """Test file creations are exported correctly."""
- self.message.synced_dir.dir = self.nested_dirs_src
+ def testCreation(self):
+ """Test file creations are exported correctly."""
+ self.message.synced_dir.dir = self.nested_dirs_src
- new_src = os.path.join(self.nested_dirs_src, 'new_dir', 'new_file')
- new_dest = os.path.join(self.destination, 'new_dir', 'new_file')
+ new_src = os.path.join(self.nested_dirs_src, "new_dir", "new_file")
+ new_dest = os.path.join(self.destination, "new_dir", "new_file")
- self._assertExist(self.nested_src_files)
- self._assertNotExist(self.nested_dest_files)
+ self._assertExist(self.nested_src_files)
+ self._assertNotExist(self.nested_dest_files)
- with field_handler.sync_dirs(self.message, self.destination, self.chroot):
- self._assertExist(self.nested_dest_files)
- osutils.Touch(new_dest, makedirs=True)
+ with field_handler.sync_dirs(
+ self.message, self.destination, self.chroot
+ ):
+ self._assertExist(self.nested_dest_files)
+ osutils.Touch(new_dest, makedirs=True)
- self._assertExist(self.nested_src_files)
- self.assertExists(new_src)
+ self._assertExist(self.nested_src_files)
+ self.assertExists(new_src)
- def testModification(self):
- """Test file modifications are exported correctly."""
- self.message.synced_dir.dir = self.single_file_src
+ def testModification(self):
+ """Test file modifications are exported correctly."""
+ self.message.synced_dir.dir = self.single_file_src
- self.assertExists(self.sf_src_file)
- self.assertNotExists(self.sf_dest_file)
+ self.assertExists(self.sf_src_file)
+ self.assertNotExists(self.sf_dest_file)
- self.assertEqual('', osutils.ReadFile(self.sf_src_file))
- file_content = 'Content!'
+ self.assertEqual("", osutils.ReadFile(self.sf_src_file))
+ file_content = "Content!"
- with field_handler.sync_dirs(self.message, self.destination, self.chroot):
- self.assertExists(self.sf_dest_file)
- osutils.WriteFile(self.sf_dest_file, file_content)
+ with field_handler.sync_dirs(
+ self.message, self.destination, self.chroot
+ ):
+ self.assertExists(self.sf_dest_file)
+ osutils.WriteFile(self.sf_dest_file, file_content)
- self.assertExists(self.sf_src_file)
- self.assertEqual(file_content, osutils.ReadFile(self.sf_src_file))
+ self.assertExists(self.sf_src_file)
+ self.assertEqual(file_content, osutils.ReadFile(self.sf_src_file))
class ExtractResultsTest(cros_test_lib.TempDirTestCase):
- """Tests for extract_results."""
+ """Tests for extract_results."""
- def setUp(self):
- # Setup the directories.
- self.chroot_dir = os.path.join(self.tempdir, 'chroot')
- self.source_dir = '/source'
- self.chroot_source = os.path.join(self.chroot_dir,
- self.source_dir.lstrip(os.sep))
- self.source_dir2 = '/source2'
- self.chroot_source2 = os.path.join(self.chroot_dir,
- self.source_dir2.lstrip(os.sep))
- self.dest_dir = os.path.join(self.tempdir, 'destination')
- osutils.SafeMakedirs(self.chroot_source)
- osutils.SafeMakedirs(self.chroot_source2)
- osutils.SafeMakedirs(self.dest_dir)
+ def setUp(self):
+ # Setup the directories.
+ self.chroot_dir = os.path.join(self.tempdir, "chroot")
+ self.source_dir = "/source"
+ self.chroot_source = os.path.join(
+ self.chroot_dir, self.source_dir.lstrip(os.sep)
+ )
+ self.source_dir2 = "/source2"
+ self.chroot_source2 = os.path.join(
+ self.chroot_dir, self.source_dir2.lstrip(os.sep)
+ )
+ self.dest_dir = os.path.join(self.tempdir, "destination")
+ osutils.SafeMakedirs(self.chroot_source)
+ osutils.SafeMakedirs(self.chroot_source2)
+ osutils.SafeMakedirs(self.dest_dir)
- # Two files in the same directory inside the chroot.
- self.source_file1 = os.path.join(self.chroot_source, 'file1')
- self.source_file1_inside = os.path.join(self.source_dir, 'file1')
- self.file1_contents = 'file 1'
- osutils.WriteFile(self.source_file1, self.file1_contents)
+ # Two files in the same directory inside the chroot.
+ self.source_file1 = os.path.join(self.chroot_source, "file1")
+ self.source_file1_inside = os.path.join(self.source_dir, "file1")
+ self.file1_contents = "file 1"
+ osutils.WriteFile(self.source_file1, self.file1_contents)
- self.file2_contents = 'some data'
- self.source_file2 = os.path.join(self.chroot_source, 'file2')
- self.source_file2_inside = os.path.join(self.source_dir, 'file2')
- osutils.WriteFile(self.source_file2, self.file2_contents)
+ self.file2_contents = "some data"
+ self.source_file2 = os.path.join(self.chroot_source, "file2")
+ self.source_file2_inside = os.path.join(self.source_dir, "file2")
+ osutils.WriteFile(self.source_file2, self.file2_contents)
- # Third file in a different location.
- self.file3_contents = 'another file'
- self.source_file3 = os.path.join(self.chroot_source2, 'file3')
- self.source_file3_inside = os.path.join(self.source_dir2, 'file3')
- osutils.WriteFile(self.source_file3, self.file3_contents)
+ # Third file in a different location.
+ self.file3_contents = "another file"
+ self.source_file3 = os.path.join(self.chroot_source2, "file3")
+ self.source_file3_inside = os.path.join(self.source_dir2, "file3")
+ osutils.WriteFile(self.source_file3, self.file3_contents)
- self.request = build_api_test_pb2.TestRequestMessage()
- self.request.result_path.path.path = self.dest_dir
- self.request.result_path.path.location = common_pb2.Path.OUTSIDE
- self.response = build_api_test_pb2.TestResultMessage()
- self.chroot = chroot_lib.Chroot(path=self.chroot_dir)
+ self.request = build_api_test_pb2.TestRequestMessage()
+ self.request.result_path.path.path = self.dest_dir
+ self.request.result_path.path.location = common_pb2.Path.OUTSIDE
+ self.response = build_api_test_pb2.TestResultMessage()
+ self.chroot = chroot_lib.Chroot(path=self.chroot_dir)
- def _path_checks(self, path, destination, contents=None):
- self.assertTrue(path)
- self.assertStartsWith(path, destination)
- self.assertExists(path)
- if contents:
- self.assertFileContents(path, contents)
+ def _path_checks(self, path, destination, contents=None):
+ self.assertTrue(path)
+ self.assertStartsWith(path, destination)
+ self.assertExists(path)
+ if contents:
+ self.assertFileContents(path, contents)
- def test_single_file(self):
- """Test a single file.
+ def test_single_file(self):
+ """Test a single file.
- Verify:
- /path/to/chroot/file -> /path/to/destination/file
- """
- self.response.artifact.path = self.source_file1_inside
- self.response.artifact.location = common_pb2.Path.INSIDE
+ Verify:
+ /path/to/chroot/file -> /path/to/destination/file
+ """
+ self.response.artifact.path = self.source_file1_inside
+ self.response.artifact.location = common_pb2.Path.INSIDE
- field_handler.extract_results(self.request, self.response, self.chroot)
+ field_handler.extract_results(self.request, self.response, self.chroot)
- self._path_checks(self.response.artifact.path, self.dest_dir,
- contents=self.file1_contents)
+ self._path_checks(
+ self.response.artifact.path,
+ self.dest_dir,
+ contents=self.file1_contents,
+ )
- def test_single_directory(self):
- """Test a single directory.
+ def test_single_directory(self):
+ """Test a single directory.
- Verify:
- /path/to/chroot/directory/* -> /path/to/destination/directory/*
- """
- self.response.artifact.path = self.source_dir
- self.response.artifact.location = common_pb2.Path.INSIDE
+ Verify:
+ /path/to/chroot/directory/* -> /path/to/destination/directory/*
+ """
+ self.response.artifact.path = self.source_dir
+ self.response.artifact.location = common_pb2.Path.INSIDE
- field_handler.extract_results(self.request, self.response, self.chroot)
+ field_handler.extract_results(self.request, self.response, self.chroot)
- self._path_checks(self.response.artifact.path, self.dest_dir)
- self.assertCountEqual(os.listdir(self.chroot_source),
- os.listdir(self.response.artifact.path))
+ self._path_checks(self.response.artifact.path, self.dest_dir)
+ self.assertCountEqual(
+ os.listdir(self.chroot_source),
+ os.listdir(self.response.artifact.path),
+ )
- def test_multiple_files(self):
- """Test multiple files.
+ def test_multiple_files(self):
+ """Test multiple files.
- Verify:
- /path/to/chroot/some/path/file1 -> /path/to/destination/file1
- /path/to/chroot/different/path/file2 -> /path/to/destination/file2
- etc.
- """
- self.response.artifact.path = self.source_file1_inside
- self.response.artifact.location = common_pb2.Path.INSIDE
- self.response.nested_artifact.path.path = self.source_file2_inside
- self.response.nested_artifact.path.location = common_pb2.Path.INSIDE
+ Verify:
+ /path/to/chroot/some/path/file1 -> /path/to/destination/file1
+ /path/to/chroot/different/path/file2 -> /path/to/destination/file2
+ etc.
+ """
+ self.response.artifact.path = self.source_file1_inside
+ self.response.artifact.location = common_pb2.Path.INSIDE
+ self.response.nested_artifact.path.path = self.source_file2_inside
+ self.response.nested_artifact.path.location = common_pb2.Path.INSIDE
- artifact3 = self.response.artifacts.add()
- artifact3.path = self.source_file3_inside
- artifact3.location = common_pb2.Path.INSIDE
+ artifact3 = self.response.artifacts.add()
+ artifact3.path = self.source_file3_inside
+ artifact3.location = common_pb2.Path.INSIDE
- field_handler.extract_results(self.request, self.response, self.chroot)
+ field_handler.extract_results(self.request, self.response, self.chroot)
- self._path_checks(self.response.artifact.path, self.dest_dir,
- contents=self.file1_contents)
- self._path_checks(self.response.nested_artifact.path.path, self.dest_dir,
- contents=self.file2_contents)
+ self._path_checks(
+ self.response.artifact.path,
+ self.dest_dir,
+ contents=self.file1_contents,
+ )
+ self._path_checks(
+ self.response.nested_artifact.path.path,
+ self.dest_dir,
+ contents=self.file2_contents,
+ )
- self.assertEqual(1, len(self.response.artifacts))
- for artifact in self.response.artifacts:
- self._path_checks(artifact.path, self.dest_dir,
- contents=self.file3_contents)
+ self.assertEqual(1, len(self.response.artifacts))
+ for artifact in self.response.artifacts:
+ self._path_checks(
+ artifact.path, self.dest_dir, contents=self.file3_contents
+ )
- def test_multiple_directories(self):
- """Test multiple directories.
+ def test_multiple_directories(self):
+ """Test multiple directories.
- Verify:
- /path/to/chroot/some/directory -> /path/to/destination/directory
- /path/to/chroot/another/directory2 -> /path/to/destination/directory2
- etc.
- """
- self.response.artifact.path = self.source_dir
- self.response.artifact.location = common_pb2.Path.INSIDE
- self.response.nested_artifact.path.path = self.source_dir2
- self.response.nested_artifact.path.location = common_pb2.Path.INSIDE
+ Verify:
+ /path/to/chroot/some/directory -> /path/to/destination/directory
+ /path/to/chroot/another/directory2 -> /path/to/destination/directory2
+ etc.
+ """
+ self.response.artifact.path = self.source_dir
+ self.response.artifact.location = common_pb2.Path.INSIDE
+ self.response.nested_artifact.path.path = self.source_dir2
+ self.response.nested_artifact.path.location = common_pb2.Path.INSIDE
- field_handler.extract_results(self.request, self.response, self.chroot)
+ field_handler.extract_results(self.request, self.response, self.chroot)
- self._path_checks(self.response.artifact.path, self.dest_dir)
- self._path_checks(self.response.nested_artifact.path.path, self.dest_dir)
+ self._path_checks(self.response.artifact.path, self.dest_dir)
+ self._path_checks(
+ self.response.nested_artifact.path.path, self.dest_dir
+ )
- expected = os.listdir(self.chroot_source)
- expected.extend(os.listdir(self.chroot_source2))
- self.assertCountEqual(expected, os.listdir(self.response.artifact.path))
+ expected = os.listdir(self.chroot_source)
+ expected.extend(os.listdir(self.chroot_source2))
+ self.assertCountEqual(expected, os.listdir(self.response.artifact.path))