adb: add interfaces for Encoder/Decoder.
More groundwork to support more compression algorithms.
Bug: https://issuetracker.google.com/150827486
Test: python3 -m unittest test_device.FileOperationsTest{Uncompressed,Brotli}
Change-Id: I638493083b83e3f6c6854b631471e9d6b50bd79f
diff --git a/adb/test_device.py b/adb/test_device.py
index 6a9ff89..496a0ff 100755
--- a/adb/test_device.py
+++ b/adb/test_device.py
@@ -77,8 +77,7 @@
class DeviceTest(unittest.TestCase):
- def setUp(self):
- self.device = adb.get_device()
+ device = adb.get_device()
class AbbTest(DeviceTest):
@@ -753,535 +752,554 @@
return files
-class FileOperationsTest(DeviceTest):
- SCRATCH_DIR = '/data/local/tmp'
- DEVICE_TEMP_FILE = SCRATCH_DIR + '/adb_test_file'
- DEVICE_TEMP_DIR = SCRATCH_DIR + '/adb_test_dir'
+class FileOperationsTest:
+ class Base(DeviceTest):
+ SCRATCH_DIR = '/data/local/tmp'
+ DEVICE_TEMP_FILE = SCRATCH_DIR + '/adb_test_file'
+ DEVICE_TEMP_DIR = SCRATCH_DIR + '/adb_test_dir'
- def _verify_remote(self, checksum, remote_path):
- dev_md5, _ = self.device.shell([get_md5_prog(self.device),
- remote_path])[0].split()
- self.assertEqual(checksum, dev_md5)
+ def setUp(self):
+ self.previous_env = os.environ.get("ADB_COMPRESSION")
+ os.environ["ADB_COMPRESSION"] = self.compression
- def _verify_local(self, checksum, local_path):
- with open(local_path, 'rb') as host_file:
- host_md5 = compute_md5(host_file.read())
- self.assertEqual(host_md5, checksum)
+ def tearDown(self):
+ if self.previous_env is None:
+ del os.environ["ADB_COMPRESSION"]
+ else:
+ os.environ["ADB_COMPRESSION"] = self.previous_env
- def test_push(self):
- """Push a randomly generated file to specified device."""
- kbytes = 512
- tmp = tempfile.NamedTemporaryFile(mode='wb', delete=False)
- rand_str = os.urandom(1024 * kbytes)
- tmp.write(rand_str)
- tmp.close()
+ def _verify_remote(self, checksum, remote_path):
+ dev_md5, _ = self.device.shell([get_md5_prog(self.device),
+ remote_path])[0].split()
+ self.assertEqual(checksum, dev_md5)
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_FILE])
- self.device.push(local=tmp.name, remote=self.DEVICE_TEMP_FILE)
+ def _verify_local(self, checksum, local_path):
+ with open(local_path, 'rb') as host_file:
+ host_md5 = compute_md5(host_file.read())
+ self.assertEqual(host_md5, checksum)
- self._verify_remote(compute_md5(rand_str), self.DEVICE_TEMP_FILE)
- self.device.shell(['rm', '-f', self.DEVICE_TEMP_FILE])
+ def test_push(self):
+ """Push a randomly generated file to specified device."""
+ kbytes = 512
+ tmp = tempfile.NamedTemporaryFile(mode='wb', delete=False)
+ rand_str = os.urandom(1024 * kbytes)
+ tmp.write(rand_str)
+ tmp.close()
- os.remove(tmp.name)
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_FILE])
+ self.device.push(local=tmp.name, remote=self.DEVICE_TEMP_FILE)
- def test_push_dir(self):
- """Push a randomly generated directory of files to the device."""
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- self.device.shell(['mkdir', self.DEVICE_TEMP_DIR])
+ self._verify_remote(compute_md5(rand_str), self.DEVICE_TEMP_FILE)
+ self.device.shell(['rm', '-f', self.DEVICE_TEMP_FILE])
- try:
- host_dir = tempfile.mkdtemp()
+ os.remove(tmp.name)
- # Make sure the temp directory isn't setuid, or else adb will complain.
- os.chmod(host_dir, 0o700)
-
- # Create 32 random files.
- temp_files = make_random_host_files(in_dir=host_dir, num_files=32)
- self.device.push(host_dir, self.DEVICE_TEMP_DIR)
-
- for temp_file in temp_files:
- remote_path = posixpath.join(self.DEVICE_TEMP_DIR,
- os.path.basename(host_dir),
- temp_file.base_name)
- self._verify_remote(temp_file.checksum, remote_path)
+ def test_push_dir(self):
+ """Push a randomly generated directory of files to the device."""
self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- finally:
- if host_dir is not None:
- shutil.rmtree(host_dir)
+ self.device.shell(['mkdir', self.DEVICE_TEMP_DIR])
- def disabled_test_push_empty(self):
- """Push an empty directory to the device."""
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- self.device.shell(['mkdir', self.DEVICE_TEMP_DIR])
+ try:
+ host_dir = tempfile.mkdtemp()
- try:
- host_dir = tempfile.mkdtemp()
+ # Make sure the temp directory isn't setuid, or else adb will complain.
+ os.chmod(host_dir, 0o700)
- # Make sure the temp directory isn't setuid, or else adb will complain.
- os.chmod(host_dir, 0o700)
+ # Create 32 random files.
+ temp_files = make_random_host_files(in_dir=host_dir, num_files=32)
+ self.device.push(host_dir, self.DEVICE_TEMP_DIR)
- # Create an empty directory.
- empty_dir_path = os.path.join(host_dir, 'empty')
- os.mkdir(empty_dir_path);
+ for temp_file in temp_files:
+ remote_path = posixpath.join(self.DEVICE_TEMP_DIR,
+ os.path.basename(host_dir),
+ temp_file.base_name)
+ self._verify_remote(temp_file.checksum, remote_path)
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ finally:
+ if host_dir is not None:
+ shutil.rmtree(host_dir)
- self.device.push(empty_dir_path, self.DEVICE_TEMP_DIR)
-
- remote_path = os.path.join(self.DEVICE_TEMP_DIR, "empty")
- test_empty_cmd = ["[", "-d", remote_path, "]"]
- rc, _, _ = self.device.shell_nocheck(test_empty_cmd)
-
- self.assertEqual(rc, 0)
+ def disabled_test_push_empty(self):
+ """Push an empty directory to the device."""
self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- finally:
- if host_dir is not None:
- shutil.rmtree(host_dir)
+ self.device.shell(['mkdir', self.DEVICE_TEMP_DIR])
- @unittest.skipIf(sys.platform == "win32", "symlinks require elevated privileges on windows")
- def test_push_symlink(self):
- """Push a symlink.
+ try:
+ host_dir = tempfile.mkdtemp()
- Bug: http://b/31491920
- """
- try:
- host_dir = tempfile.mkdtemp()
+ # Make sure the temp directory isn't setuid, or else adb will complain.
+ os.chmod(host_dir, 0o700)
- # Make sure the temp directory isn't setuid, or else adb will
- # complain.
- os.chmod(host_dir, 0o700)
+ # Create an empty directory.
+ empty_dir_path = os.path.join(host_dir, 'empty')
+ os.mkdir(empty_dir_path);
- with open(os.path.join(host_dir, 'foo'), 'w') as f:
- f.write('foo')
+ self.device.push(empty_dir_path, self.DEVICE_TEMP_DIR)
- symlink_path = os.path.join(host_dir, 'symlink')
- os.symlink('foo', symlink_path)
+ remote_path = os.path.join(self.DEVICE_TEMP_DIR, "empty")
+ test_empty_cmd = ["[", "-d", remote_path, "]"]
+ rc, _, _ = self.device.shell_nocheck(test_empty_cmd)
+
+ self.assertEqual(rc, 0)
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ finally:
+ if host_dir is not None:
+ shutil.rmtree(host_dir)
+
+ @unittest.skipIf(sys.platform == "win32", "symlinks require elevated privileges on windows")
+ def test_push_symlink(self):
+ """Push a symlink.
+
+ Bug: http://b/31491920
+ """
+ try:
+ host_dir = tempfile.mkdtemp()
+
+ # Make sure the temp directory isn't setuid, or else adb will
+ # complain.
+ os.chmod(host_dir, 0o700)
+
+ with open(os.path.join(host_dir, 'foo'), 'w') as f:
+ f.write('foo')
+
+ symlink_path = os.path.join(host_dir, 'symlink')
+ os.symlink('foo', symlink_path)
+
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ self.device.shell(['mkdir', self.DEVICE_TEMP_DIR])
+ self.device.push(symlink_path, self.DEVICE_TEMP_DIR)
+ rc, out, _ = self.device.shell_nocheck(
+ ['cat', posixpath.join(self.DEVICE_TEMP_DIR, 'symlink')])
+ self.assertEqual(0, rc)
+ self.assertEqual(out.strip(), 'foo')
+ finally:
+ if host_dir is not None:
+ shutil.rmtree(host_dir)
+
+ def test_multiple_push(self):
+ """Push multiple files to the device in one adb push command.
+
+ Bug: http://b/25324823
+ """
self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
self.device.shell(['mkdir', self.DEVICE_TEMP_DIR])
- self.device.push(symlink_path, self.DEVICE_TEMP_DIR)
- rc, out, _ = self.device.shell_nocheck(
- ['cat', posixpath.join(self.DEVICE_TEMP_DIR, 'symlink')])
- self.assertEqual(0, rc)
- self.assertEqual(out.strip(), 'foo')
- finally:
- if host_dir is not None:
- shutil.rmtree(host_dir)
- def test_multiple_push(self):
- """Push multiple files to the device in one adb push command.
-
- Bug: http://b/25324823
- """
-
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- self.device.shell(['mkdir', self.DEVICE_TEMP_DIR])
-
- try:
- host_dir = tempfile.mkdtemp()
-
- # Create some random files and a subdirectory containing more files.
- temp_files = make_random_host_files(in_dir=host_dir, num_files=4)
-
- subdir = os.path.join(host_dir, 'subdir')
- os.mkdir(subdir)
- subdir_temp_files = make_random_host_files(in_dir=subdir,
- num_files=4)
-
- paths = [x.full_path for x in temp_files]
- paths.append(subdir)
- self.device._simple_call(['push'] + paths + [self.DEVICE_TEMP_DIR])
-
- for temp_file in temp_files:
- remote_path = posixpath.join(self.DEVICE_TEMP_DIR,
- temp_file.base_name)
- self._verify_remote(temp_file.checksum, remote_path)
-
- for subdir_temp_file in subdir_temp_files:
- remote_path = posixpath.join(self.DEVICE_TEMP_DIR,
- # BROKEN: http://b/25394682
- # 'subdir';
- temp_file.base_name)
- self._verify_remote(temp_file.checksum, remote_path)
-
-
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- finally:
- if host_dir is not None:
- shutil.rmtree(host_dir)
-
- @requires_non_root
- def test_push_error_reporting(self):
- """Make sure that errors that occur while pushing a file get reported
-
- Bug: http://b/26816782
- """
- with tempfile.NamedTemporaryFile() as tmp_file:
- tmp_file.write(b'\0' * 1024 * 1024)
- tmp_file.flush()
try:
- self.device.push(local=tmp_file.name, remote='/system/')
- self.fail('push should not have succeeded')
+ host_dir = tempfile.mkdtemp()
+
+ # Create some random files and a subdirectory containing more files.
+ temp_files = make_random_host_files(in_dir=host_dir, num_files=4)
+
+ subdir = os.path.join(host_dir, 'subdir')
+ os.mkdir(subdir)
+ subdir_temp_files = make_random_host_files(in_dir=subdir,
+ num_files=4)
+
+ paths = [x.full_path for x in temp_files]
+ paths.append(subdir)
+ self.device._simple_call(['push'] + paths + [self.DEVICE_TEMP_DIR])
+
+ for temp_file in temp_files:
+ remote_path = posixpath.join(self.DEVICE_TEMP_DIR,
+ temp_file.base_name)
+ self._verify_remote(temp_file.checksum, remote_path)
+
+ for subdir_temp_file in subdir_temp_files:
+ remote_path = posixpath.join(self.DEVICE_TEMP_DIR,
+ # BROKEN: http://b/25394682
+ # 'subdir';
+ temp_file.base_name)
+ self._verify_remote(temp_file.checksum, remote_path)
+
+
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ finally:
+ if host_dir is not None:
+ shutil.rmtree(host_dir)
+
+ @requires_non_root
+ def test_push_error_reporting(self):
+ """Make sure that errors that occur while pushing a file get reported
+
+ Bug: http://b/26816782
+ """
+ with tempfile.NamedTemporaryFile() as tmp_file:
+ tmp_file.write(b'\0' * 1024 * 1024)
+ tmp_file.flush()
+ try:
+ self.device.push(local=tmp_file.name, remote='/system/')
+ self.fail('push should not have succeeded')
+ except subprocess.CalledProcessError as e:
+ output = e.output
+
+ self.assertTrue(b'Permission denied' in output or
+ b'Read-only file system' in output)
+
+ @requires_non_root
+ def test_push_directory_creation(self):
+ """Regression test for directory creation.
+
+ Bug: http://b/110953234
+ """
+ with tempfile.NamedTemporaryFile() as tmp_file:
+ tmp_file.write(b'\0' * 1024 * 1024)
+ tmp_file.flush()
+ remote_path = self.DEVICE_TEMP_DIR + '/test_push_directory_creation'
+ self.device.shell(['rm', '-rf', remote_path])
+
+ remote_path += '/filename'
+ self.device.push(local=tmp_file.name, remote=remote_path)
+
+ def disabled_test_push_multiple_slash_root(self):
+ """Regression test for pushing to //data/local/tmp.
+
+ Bug: http://b/141311284
+
+ Disabled because this broken on the adbd side as well: b/141943968
+ """
+ with tempfile.NamedTemporaryFile() as tmp_file:
+ tmp_file.write('\0' * 1024 * 1024)
+ tmp_file.flush()
+ remote_path = '/' + self.DEVICE_TEMP_DIR + '/test_push_multiple_slash_root'
+ self.device.shell(['rm', '-rf', remote_path])
+ self.device.push(local=tmp_file.name, remote=remote_path)
+
+ def _test_pull(self, remote_file, checksum):
+ tmp_write = tempfile.NamedTemporaryFile(mode='wb', delete=False)
+ tmp_write.close()
+ self.device.pull(remote=remote_file, local=tmp_write.name)
+ with open(tmp_write.name, 'rb') as tmp_read:
+ host_contents = tmp_read.read()
+ host_md5 = compute_md5(host_contents)
+ self.assertEqual(checksum, host_md5)
+ os.remove(tmp_write.name)
+
+ @requires_non_root
+ def test_pull_error_reporting(self):
+ self.device.shell(['touch', self.DEVICE_TEMP_FILE])
+ self.device.shell(['chmod', 'a-rwx', self.DEVICE_TEMP_FILE])
+
+ try:
+ output = self.device.pull(remote=self.DEVICE_TEMP_FILE, local='x')
except subprocess.CalledProcessError as e:
output = e.output
- self.assertTrue(b'Permission denied' in output or
- b'Read-only file system' in output)
+ self.assertIn(b'Permission denied', output)
- @requires_non_root
- def test_push_directory_creation(self):
- """Regression test for directory creation.
+ self.device.shell(['rm', '-f', self.DEVICE_TEMP_FILE])
- Bug: http://b/110953234
- """
- with tempfile.NamedTemporaryFile() as tmp_file:
- tmp_file.write(b'\0' * 1024 * 1024)
- tmp_file.flush()
- remote_path = self.DEVICE_TEMP_DIR + '/test_push_directory_creation'
- self.device.shell(['rm', '-rf', remote_path])
+ def test_pull(self):
+ """Pull a randomly generated file from specified device."""
+ kbytes = 512
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_FILE])
+ cmd = ['dd', 'if=/dev/urandom',
+ 'of={}'.format(self.DEVICE_TEMP_FILE), 'bs=1024',
+ 'count={}'.format(kbytes)]
+ self.device.shell(cmd)
+ dev_md5, _ = self.device.shell(
+ [get_md5_prog(self.device), self.DEVICE_TEMP_FILE])[0].split()
+ self._test_pull(self.DEVICE_TEMP_FILE, dev_md5)
+ self.device.shell_nocheck(['rm', self.DEVICE_TEMP_FILE])
- remote_path += '/filename'
- self.device.push(local=tmp_file.name, remote=remote_path)
+ def test_pull_dir(self):
+ """Pull a randomly generated directory of files from the device."""
+ try:
+ host_dir = tempfile.mkdtemp()
- def disabled_test_push_multiple_slash_root(self):
- """Regression test for pushing to //data/local/tmp.
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ self.device.shell(['mkdir', '-p', self.DEVICE_TEMP_DIR])
- Bug: http://b/141311284
+ # Populate device directory with random files.
+ temp_files = make_random_device_files(
+ self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=32)
- Disabled because this broken on the adbd side as well: b/141943968
- """
- with tempfile.NamedTemporaryFile() as tmp_file:
- tmp_file.write('\0' * 1024 * 1024)
- tmp_file.flush()
- remote_path = '/' + self.DEVICE_TEMP_DIR + '/test_push_multiple_slash_root'
- self.device.shell(['rm', '-rf', remote_path])
- self.device.push(local=tmp_file.name, remote=remote_path)
+ self.device.pull(remote=self.DEVICE_TEMP_DIR, local=host_dir)
- def _test_pull(self, remote_file, checksum):
- tmp_write = tempfile.NamedTemporaryFile(mode='wb', delete=False)
- tmp_write.close()
- self.device.pull(remote=remote_file, local=tmp_write.name)
- with open(tmp_write.name, 'rb') as tmp_read:
- host_contents = tmp_read.read()
- host_md5 = compute_md5(host_contents)
- self.assertEqual(checksum, host_md5)
- os.remove(tmp_write.name)
+ for temp_file in temp_files:
+ host_path = os.path.join(
+ host_dir, posixpath.basename(self.DEVICE_TEMP_DIR),
+ temp_file.base_name)
+ self._verify_local(temp_file.checksum, host_path)
- @requires_non_root
- def test_pull_error_reporting(self):
- self.device.shell(['touch', self.DEVICE_TEMP_FILE])
- self.device.shell(['chmod', 'a-rwx', self.DEVICE_TEMP_FILE])
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ finally:
+ if host_dir is not None:
+ shutil.rmtree(host_dir)
- try:
- output = self.device.pull(remote=self.DEVICE_TEMP_FILE, local='x')
- except subprocess.CalledProcessError as e:
- output = e.output
+ def test_pull_dir_symlink(self):
+ """Pull a directory into a symlink to a directory.
- self.assertIn(b'Permission denied', output)
+ Bug: http://b/27362811
+ """
+ if os.name != 'posix':
+ raise unittest.SkipTest('requires POSIX')
- self.device.shell(['rm', '-f', self.DEVICE_TEMP_FILE])
+ try:
+ host_dir = tempfile.mkdtemp()
+ real_dir = os.path.join(host_dir, 'dir')
+ symlink = os.path.join(host_dir, 'symlink')
+ os.mkdir(real_dir)
+ os.symlink(real_dir, symlink)
- def test_pull(self):
- """Pull a randomly generated file from specified device."""
- kbytes = 512
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_FILE])
- cmd = ['dd', 'if=/dev/urandom',
- 'of={}'.format(self.DEVICE_TEMP_FILE), 'bs=1024',
- 'count={}'.format(kbytes)]
- self.device.shell(cmd)
- dev_md5, _ = self.device.shell(
- [get_md5_prog(self.device), self.DEVICE_TEMP_FILE])[0].split()
- self._test_pull(self.DEVICE_TEMP_FILE, dev_md5)
- self.device.shell_nocheck(['rm', self.DEVICE_TEMP_FILE])
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ self.device.shell(['mkdir', '-p', self.DEVICE_TEMP_DIR])
- def test_pull_dir(self):
- """Pull a randomly generated directory of files from the device."""
- try:
- host_dir = tempfile.mkdtemp()
+ # Populate device directory with random files.
+ temp_files = make_random_device_files(
+ self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=32)
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- self.device.shell(['mkdir', '-p', self.DEVICE_TEMP_DIR])
+ self.device.pull(remote=self.DEVICE_TEMP_DIR, local=symlink)
- # Populate device directory with random files.
- temp_files = make_random_device_files(
- self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=32)
+ for temp_file in temp_files:
+ host_path = os.path.join(
+ real_dir, posixpath.basename(self.DEVICE_TEMP_DIR),
+ temp_file.base_name)
+ self._verify_local(temp_file.checksum, host_path)
- self.device.pull(remote=self.DEVICE_TEMP_DIR, local=host_dir)
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ finally:
+ if host_dir is not None:
+ shutil.rmtree(host_dir)
+ def test_pull_dir_symlink_collision(self):
+ """Pull a directory into a colliding symlink to directory."""
+ if os.name != 'posix':
+ raise unittest.SkipTest('requires POSIX')
+
+ try:
+ host_dir = tempfile.mkdtemp()
+ real_dir = os.path.join(host_dir, 'real')
+ tmp_dirname = os.path.basename(self.DEVICE_TEMP_DIR)
+ symlink = os.path.join(host_dir, tmp_dirname)
+ os.mkdir(real_dir)
+ os.symlink(real_dir, symlink)
+
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ self.device.shell(['mkdir', '-p', self.DEVICE_TEMP_DIR])
+
+ # Populate device directory with random files.
+ temp_files = make_random_device_files(
+ self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=32)
+
+ self.device.pull(remote=self.DEVICE_TEMP_DIR, local=host_dir)
+
+ for temp_file in temp_files:
+ host_path = os.path.join(real_dir, temp_file.base_name)
+ self._verify_local(temp_file.checksum, host_path)
+
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ finally:
+ if host_dir is not None:
+ shutil.rmtree(host_dir)
+
+ def test_pull_dir_nonexistent(self):
+ """Pull a directory of files from the device to a nonexistent path."""
+ try:
+ host_dir = tempfile.mkdtemp()
+ dest_dir = os.path.join(host_dir, 'dest')
+
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ self.device.shell(['mkdir', '-p', self.DEVICE_TEMP_DIR])
+
+ # Populate device directory with random files.
+ temp_files = make_random_device_files(
+ self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=32)
+
+ self.device.pull(remote=self.DEVICE_TEMP_DIR, local=dest_dir)
+
+ for temp_file in temp_files:
+ host_path = os.path.join(dest_dir, temp_file.base_name)
+ self._verify_local(temp_file.checksum, host_path)
+
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ finally:
+ if host_dir is not None:
+ shutil.rmtree(host_dir)
+
+ # selinux prevents adbd from accessing symlinks on /data/local/tmp.
+ def disabled_test_pull_symlink_dir(self):
+ """Pull a symlink to a directory of symlinks to files."""
+ try:
+ host_dir = tempfile.mkdtemp()
+
+ remote_dir = posixpath.join(self.DEVICE_TEMP_DIR, 'contents')
+ remote_links = posixpath.join(self.DEVICE_TEMP_DIR, 'links')
+ remote_symlink = posixpath.join(self.DEVICE_TEMP_DIR, 'symlink')
+
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ self.device.shell(['mkdir', '-p', remote_dir, remote_links])
+ self.device.shell(['ln', '-s', remote_links, remote_symlink])
+
+ # Populate device directory with random files.
+ temp_files = make_random_device_files(
+ self.device, in_dir=remote_dir, num_files=32)
+
+ for temp_file in temp_files:
+ self.device.shell(
+ ['ln', '-s', '../contents/{}'.format(temp_file.base_name),
+ posixpath.join(remote_links, temp_file.base_name)])
+
+ self.device.pull(remote=remote_symlink, local=host_dir)
+
+ for temp_file in temp_files:
+ host_path = os.path.join(
+ host_dir, 'symlink', temp_file.base_name)
+ self._verify_local(temp_file.checksum, host_path)
+
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ finally:
+ if host_dir is not None:
+ shutil.rmtree(host_dir)
+
+ def test_pull_empty(self):
+ """Pull a directory containing an empty directory from the device."""
+ try:
+ host_dir = tempfile.mkdtemp()
+
+ remote_empty_path = posixpath.join(self.DEVICE_TEMP_DIR, 'empty')
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ self.device.shell(['mkdir', '-p', remote_empty_path])
+
+ self.device.pull(remote=remote_empty_path, local=host_dir)
+ self.assertTrue(os.path.isdir(os.path.join(host_dir, 'empty')))
+ finally:
+ if host_dir is not None:
+ shutil.rmtree(host_dir)
+
+ def test_multiple_pull(self):
+ """Pull a randomly generated directory of files from the device."""
+
+ try:
+ host_dir = tempfile.mkdtemp()
+
+ subdir = posixpath.join(self.DEVICE_TEMP_DIR, 'subdir')
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ self.device.shell(['mkdir', '-p', subdir])
+
+ # Create some random files and a subdirectory containing more files.
+ temp_files = make_random_device_files(
+ self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=4)
+
+ subdir_temp_files = make_random_device_files(
+ self.device, in_dir=subdir, num_files=4, prefix='subdir_')
+
+ paths = [x.full_path for x in temp_files]
+ paths.append(subdir)
+ self.device._simple_call(['pull'] + paths + [host_dir])
+
+ for temp_file in temp_files:
+ local_path = os.path.join(host_dir, temp_file.base_name)
+ self._verify_local(temp_file.checksum, local_path)
+
+ for subdir_temp_file in subdir_temp_files:
+ local_path = os.path.join(host_dir,
+ 'subdir',
+ subdir_temp_file.base_name)
+ self._verify_local(subdir_temp_file.checksum, local_path)
+
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ finally:
+ if host_dir is not None:
+ shutil.rmtree(host_dir)
+
+ def verify_sync(self, device, temp_files, device_dir):
+ """Verifies that a list of temp files was synced to the device."""
+ # Confirm that every file on the device mirrors that on the host.
for temp_file in temp_files:
- host_path = os.path.join(
- host_dir, posixpath.basename(self.DEVICE_TEMP_DIR),
- temp_file.base_name)
- self._verify_local(temp_file.checksum, host_path)
+ device_full_path = posixpath.join(
+ device_dir, temp_file.base_name)
+ dev_md5, _ = device.shell(
+ [get_md5_prog(self.device), device_full_path])[0].split()
+ self.assertEqual(temp_file.checksum, dev_md5)
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- finally:
- if host_dir is not None:
- shutil.rmtree(host_dir)
+ def test_sync(self):
+ """Sync a host directory to the data partition."""
- def test_pull_dir_symlink(self):
- """Pull a directory into a symlink to a directory.
+ try:
+ base_dir = tempfile.mkdtemp()
- Bug: http://b/27362811
- """
- if os.name != 'posix':
- raise unittest.SkipTest('requires POSIX')
+ # Create mirror device directory hierarchy within base_dir.
+ full_dir_path = base_dir + self.DEVICE_TEMP_DIR
+ os.makedirs(full_dir_path)
- try:
- host_dir = tempfile.mkdtemp()
- real_dir = os.path.join(host_dir, 'dir')
- symlink = os.path.join(host_dir, 'symlink')
- os.mkdir(real_dir)
- os.symlink(real_dir, symlink)
+ # Create 32 random files within the host mirror.
+ temp_files = make_random_host_files(
+ in_dir=full_dir_path, num_files=32)
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- self.device.shell(['mkdir', '-p', self.DEVICE_TEMP_DIR])
+ # Clean up any stale files on the device.
+ device = adb.get_device() # pylint: disable=no-member
+ device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- # Populate device directory with random files.
- temp_files = make_random_device_files(
- self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=32)
+ old_product_out = os.environ.get('ANDROID_PRODUCT_OUT')
+ os.environ['ANDROID_PRODUCT_OUT'] = base_dir
+ device.sync('data')
+ if old_product_out is None:
+ del os.environ['ANDROID_PRODUCT_OUT']
+ else:
+ os.environ['ANDROID_PRODUCT_OUT'] = old_product_out
- self.device.pull(remote=self.DEVICE_TEMP_DIR, local=symlink)
+ self.verify_sync(device, temp_files, self.DEVICE_TEMP_DIR)
- for temp_file in temp_files:
- host_path = os.path.join(
- real_dir, posixpath.basename(self.DEVICE_TEMP_DIR),
- temp_file.base_name)
- self._verify_local(temp_file.checksum, host_path)
+ #self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ finally:
+ if base_dir is not None:
+ shutil.rmtree(base_dir)
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- finally:
- if host_dir is not None:
- shutil.rmtree(host_dir)
+ def test_push_sync(self):
+ """Sync a host directory to a specific path."""
- def test_pull_dir_symlink_collision(self):
- """Pull a directory into a colliding symlink to directory."""
- if os.name != 'posix':
- raise unittest.SkipTest('requires POSIX')
+ try:
+ temp_dir = tempfile.mkdtemp()
+ temp_files = make_random_host_files(in_dir=temp_dir, num_files=32)
- try:
- host_dir = tempfile.mkdtemp()
- real_dir = os.path.join(host_dir, 'real')
- tmp_dirname = os.path.basename(self.DEVICE_TEMP_DIR)
- symlink = os.path.join(host_dir, tmp_dirname)
- os.mkdir(real_dir)
- os.symlink(real_dir, symlink)
+ device_dir = posixpath.join(self.DEVICE_TEMP_DIR, 'sync_src_dst')
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- self.device.shell(['mkdir', '-p', self.DEVICE_TEMP_DIR])
+ # Clean up any stale files on the device.
+ device = adb.get_device() # pylint: disable=no-member
+ device.shell(['rm', '-rf', device_dir])
- # Populate device directory with random files.
- temp_files = make_random_device_files(
- self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=32)
+ device.push(temp_dir, device_dir, sync=True)
- self.device.pull(remote=self.DEVICE_TEMP_DIR, local=host_dir)
+ self.verify_sync(device, temp_files, device_dir)
- for temp_file in temp_files:
- host_path = os.path.join(real_dir, temp_file.base_name)
- self._verify_local(temp_file.checksum, host_path)
+ self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
+ finally:
+ if temp_dir is not None:
+ shutil.rmtree(temp_dir)
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- finally:
- if host_dir is not None:
- shutil.rmtree(host_dir)
+ def test_unicode_paths(self):
+ """Ensure that we can support non-ASCII paths, even on Windows."""
+ name = u'로보카 폴리'
- def test_pull_dir_nonexistent(self):
- """Pull a directory of files from the device to a nonexistent path."""
- try:
- host_dir = tempfile.mkdtemp()
- dest_dir = os.path.join(host_dir, 'dest')
+ self.device.shell(['rm', '-f', '/data/local/tmp/adb-test-*'])
+ remote_path = u'/data/local/tmp/adb-test-{}'.format(name)
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- self.device.shell(['mkdir', '-p', self.DEVICE_TEMP_DIR])
+ ## push.
+ tf = tempfile.NamedTemporaryFile('wb', suffix=name, delete=False)
+ tf.close()
+ self.device.push(tf.name, remote_path)
+ os.remove(tf.name)
+ self.assertFalse(os.path.exists(tf.name))
- # Populate device directory with random files.
- temp_files = make_random_device_files(
- self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=32)
+ # Verify that the device ended up with the expected UTF-8 path
+ output = self.device.shell(
+ ['ls', '/data/local/tmp/adb-test-*'])[0].strip()
+ self.assertEqual(remote_path, output)
- self.device.pull(remote=self.DEVICE_TEMP_DIR, local=dest_dir)
+ # pull.
+ self.device.pull(remote_path, tf.name)
+ self.assertTrue(os.path.exists(tf.name))
+ os.remove(tf.name)
+ self.device.shell(['rm', '-f', '/data/local/tmp/adb-test-*'])
- for temp_file in temp_files:
- host_path = os.path.join(dest_dir, temp_file.base_name)
- self._verify_local(temp_file.checksum, host_path)
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- finally:
- if host_dir is not None:
- shutil.rmtree(host_dir)
+class FileOperationsTestUncompressed(FileOperationsTest.Base):
+ compression = "none"
- # selinux prevents adbd from accessing symlinks on /data/local/tmp.
- def disabled_test_pull_symlink_dir(self):
- """Pull a symlink to a directory of symlinks to files."""
- try:
- host_dir = tempfile.mkdtemp()
- remote_dir = posixpath.join(self.DEVICE_TEMP_DIR, 'contents')
- remote_links = posixpath.join(self.DEVICE_TEMP_DIR, 'links')
- remote_symlink = posixpath.join(self.DEVICE_TEMP_DIR, 'symlink')
-
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- self.device.shell(['mkdir', '-p', remote_dir, remote_links])
- self.device.shell(['ln', '-s', remote_links, remote_symlink])
-
- # Populate device directory with random files.
- temp_files = make_random_device_files(
- self.device, in_dir=remote_dir, num_files=32)
-
- for temp_file in temp_files:
- self.device.shell(
- ['ln', '-s', '../contents/{}'.format(temp_file.base_name),
- posixpath.join(remote_links, temp_file.base_name)])
-
- self.device.pull(remote=remote_symlink, local=host_dir)
-
- for temp_file in temp_files:
- host_path = os.path.join(
- host_dir, 'symlink', temp_file.base_name)
- self._verify_local(temp_file.checksum, host_path)
-
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- finally:
- if host_dir is not None:
- shutil.rmtree(host_dir)
-
- def test_pull_empty(self):
- """Pull a directory containing an empty directory from the device."""
- try:
- host_dir = tempfile.mkdtemp()
-
- remote_empty_path = posixpath.join(self.DEVICE_TEMP_DIR, 'empty')
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- self.device.shell(['mkdir', '-p', remote_empty_path])
-
- self.device.pull(remote=remote_empty_path, local=host_dir)
- self.assertTrue(os.path.isdir(os.path.join(host_dir, 'empty')))
- finally:
- if host_dir is not None:
- shutil.rmtree(host_dir)
-
- def test_multiple_pull(self):
- """Pull a randomly generated directory of files from the device."""
-
- try:
- host_dir = tempfile.mkdtemp()
-
- subdir = posixpath.join(self.DEVICE_TEMP_DIR, 'subdir')
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- self.device.shell(['mkdir', '-p', subdir])
-
- # Create some random files and a subdirectory containing more files.
- temp_files = make_random_device_files(
- self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=4)
-
- subdir_temp_files = make_random_device_files(
- self.device, in_dir=subdir, num_files=4, prefix='subdir_')
-
- paths = [x.full_path for x in temp_files]
- paths.append(subdir)
- self.device._simple_call(['pull'] + paths + [host_dir])
-
- for temp_file in temp_files:
- local_path = os.path.join(host_dir, temp_file.base_name)
- self._verify_local(temp_file.checksum, local_path)
-
- for subdir_temp_file in subdir_temp_files:
- local_path = os.path.join(host_dir,
- 'subdir',
- subdir_temp_file.base_name)
- self._verify_local(subdir_temp_file.checksum, local_path)
-
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- finally:
- if host_dir is not None:
- shutil.rmtree(host_dir)
-
- def verify_sync(self, device, temp_files, device_dir):
- """Verifies that a list of temp files was synced to the device."""
- # Confirm that every file on the device mirrors that on the host.
- for temp_file in temp_files:
- device_full_path = posixpath.join(
- device_dir, temp_file.base_name)
- dev_md5, _ = device.shell(
- [get_md5_prog(self.device), device_full_path])[0].split()
- self.assertEqual(temp_file.checksum, dev_md5)
-
- def test_sync(self):
- """Sync a host directory to the data partition."""
-
- try:
- base_dir = tempfile.mkdtemp()
-
- # Create mirror device directory hierarchy within base_dir.
- full_dir_path = base_dir + self.DEVICE_TEMP_DIR
- os.makedirs(full_dir_path)
-
- # Create 32 random files within the host mirror.
- temp_files = make_random_host_files(
- in_dir=full_dir_path, num_files=32)
-
- # Clean up any stale files on the device.
- device = adb.get_device() # pylint: disable=no-member
- device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
-
- old_product_out = os.environ.get('ANDROID_PRODUCT_OUT')
- os.environ['ANDROID_PRODUCT_OUT'] = base_dir
- device.sync('data')
- if old_product_out is None:
- del os.environ['ANDROID_PRODUCT_OUT']
- else:
- os.environ['ANDROID_PRODUCT_OUT'] = old_product_out
-
- self.verify_sync(device, temp_files, self.DEVICE_TEMP_DIR)
-
- #self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- finally:
- if base_dir is not None:
- shutil.rmtree(base_dir)
-
- def test_push_sync(self):
- """Sync a host directory to a specific path."""
-
- try:
- temp_dir = tempfile.mkdtemp()
- temp_files = make_random_host_files(in_dir=temp_dir, num_files=32)
-
- device_dir = posixpath.join(self.DEVICE_TEMP_DIR, 'sync_src_dst')
-
- # Clean up any stale files on the device.
- device = adb.get_device() # pylint: disable=no-member
- device.shell(['rm', '-rf', device_dir])
-
- device.push(temp_dir, device_dir, sync=True)
-
- self.verify_sync(device, temp_files, device_dir)
-
- self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
- finally:
- if temp_dir is not None:
- shutil.rmtree(temp_dir)
-
- def test_unicode_paths(self):
- """Ensure that we can support non-ASCII paths, even on Windows."""
- name = u'로보카 폴리'
-
- self.device.shell(['rm', '-f', '/data/local/tmp/adb-test-*'])
- remote_path = u'/data/local/tmp/adb-test-{}'.format(name)
-
- ## push.
- tf = tempfile.NamedTemporaryFile('wb', suffix=name, delete=False)
- tf.close()
- self.device.push(tf.name, remote_path)
- os.remove(tf.name)
- self.assertFalse(os.path.exists(tf.name))
-
- # Verify that the device ended up with the expected UTF-8 path
- output = self.device.shell(
- ['ls', '/data/local/tmp/adb-test-*'])[0].strip()
- self.assertEqual(remote_path, output)
-
- # pull.
- self.device.pull(remote_path, tf.name)
- self.assertTrue(os.path.exists(tf.name))
- os.remove(tf.name)
- self.device.shell(['rm', '-f', '/data/local/tmp/adb-test-*'])
+class FileOperationsTestBrotli(FileOperationsTest.Base):
+ compression = "brotli"
class DeviceOfflineTest(DeviceTest):