Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit 62b6b4a1 authored by Spencer Low's avatar Spencer Low
Browse files

adb: win32: get test_adb.py running and passing 100%

* Use posixpath instead of os.path, because os.path uses '\' instead of
'/' when running on Windows.

* tempfile.NamedTemporaryFile() does not work right on Windows because
it holds the file open, preventing other processes from accessing the
same file (https://bugs.python.org/issue14243). To work-around this, use
the mechanical transformation described at
http://stackoverflow.com/questions/15169101/how-to-create-a-temporary-file-that-can-be-read-by-a-subprocess



* Use pipes.quote() to quote path arguments, to prevent C:\foo\bar from
turning into C:foobar.

* Open files in binary mode with "b".

* Fix line-ending test to allow for \r\n on Windows, but to still test
for adbd incorrectly sending \r\n (which is then translated to \r\r\n).

Change-Id: Ib6ba94b919b747a878ba3ab54a4dea0801f76947
Signed-off-by: default avatarSpencer Low <CompareAndSwap@gmail.com>
parent 7c789575
Loading
Loading
Loading
Loading
+55 −31
Original line number Original line Diff line number Diff line
@@ -7,6 +7,7 @@ tests that attempt to touch all accessible attached devices.
import hashlib
import hashlib
import os
import os
import pipes
import pipes
import posixpath
import random
import random
import re
import re
import shlex
import shlex
@@ -98,7 +99,7 @@ class DeviceFile(object):
    def __init__(self, md5, full_path):
    def __init__(self, md5, full_path):
        self.md5 = md5
        self.md5 = md5
        self.full_path = full_path
        self.full_path = full_path
        self.base_name = os.path.basename(self.full_path)
        self.base_name = posixpath.basename(self.full_path)




def make_random_host_files(in_dir, num_files, rand_size=True):
def make_random_host_files(in_dir, num_files, rand_size=True):
@@ -108,7 +109,7 @@ def make_random_host_files(in_dir, num_files, rand_size=True):
    fixed_size = min_size
    fixed_size = min_size


    for _ in range(num_files):
    for _ in range(num_files):
        file_handle = tempfile.NamedTemporaryFile(dir=in_dir)
        file_handle = tempfile.NamedTemporaryFile(dir=in_dir, delete=False)


        if rand_size:
        if rand_size:
            size = random.randrange(min_size, max_size, 1024)
            size = random.randrange(min_size, max_size, 1024)
@@ -117,6 +118,7 @@ def make_random_host_files(in_dir, num_files, rand_size=True):
        rand_str = os.urandom(size)
        rand_str = os.urandom(size)
        file_handle.write(rand_str)
        file_handle.write(rand_str)
        file_handle.flush()
        file_handle.flush()
        file_handle.close()


        md5 = compute_md5(rand_str)
        md5 = compute_md5(rand_str)
        files[file_handle.name] = HostFile(file_handle, md5)
        files[file_handle.name] = HostFile(file_handle, md5)
@@ -153,9 +155,9 @@ class AdbWrapper(object):
        self.out_dir = out_dir
        self.out_dir = out_dir
        self.adb_cmd = "adb "
        self.adb_cmd = "adb "
        if self.device:
        if self.device:
            self.adb_cmd += "-s {} ".format(device)
            self.adb_cmd += "-s {} ".format(pipes.quote(device))
        if self.out_dir:
        if self.out_dir:
            self.adb_cmd += "-p {} ".format(out_dir)
            self.adb_cmd += "-p {} ".format(pipes.quote(out_dir))


    def shell(self, cmd):
    def shell(self, cmd):
        return call_checked(self.adb_cmd + "shell " + cmd)
        return call_checked(self.adb_cmd + "shell " + cmd)
@@ -168,13 +170,16 @@ class AdbWrapper(object):
            self.adb_cmd + "install {}".format(pipes.quote(filename)))
            self.adb_cmd + "install {}".format(pipes.quote(filename)))


    def push(self, local, remote):
    def push(self, local, remote):
        return call_checked(self.adb_cmd + "push {} {}".format(local, remote))
        return call_checked(self.adb_cmd + "push {} {}".format(
            pipes.quote(local), pipes.quote(remote)))


    def pull(self, remote, local):
    def pull(self, remote, local):
        return call_checked(self.adb_cmd + "pull {} {}".format(remote, local))
        return call_checked(self.adb_cmd + "pull {} {}".format(
            pipes.quote(remote), pipes.quote(local)))


    def sync(self, directory=""):
    def sync(self, directory=""):
        return call_checked(self.adb_cmd + "sync {}".format(directory))
        return call_checked(self.adb_cmd + "sync {}".format(
            pipes.quote(directory) if directory else directory))


    def forward(self, local, remote):
    def forward(self, local, remote):
        return call_checked(self.adb_cmd + "forward {} {}".format(local,
        return call_checked(self.adb_cmd + "forward {} {}".format(local,
@@ -300,11 +305,11 @@ class AdbBasic(unittest.TestCase):
        adb = AdbWrapper()
        adb = AdbWrapper()


        # http://b/20323053
        # http://b/20323053
        tf = tempfile.NamedTemporaryFile("w", suffix="-text;ls;1.apk")
        tf = tempfile.NamedTemporaryFile("wb", suffix="-text;ls;1.apk")
        self.assertIn("-text;ls;1.apk", adb.install(tf.name))
        self.assertIn("-text;ls;1.apk", adb.install(tf.name))


        # http://b/3090932
        # http://b/3090932
        tf = tempfile.NamedTemporaryFile("w", suffix="-Live Hold'em.apk")
        tf = tempfile.NamedTemporaryFile("wb", suffix="-Live Hold'em.apk")
        self.assertIn("-Live Hold'em.apk", adb.install(tf.name))
        self.assertIn("-Live Hold'em.apk", adb.install(tf.name))


    def test_line_endings(self):
    def test_line_endings(self):
@@ -312,7 +317,16 @@ class AdbBasic(unittest.TestCase):


        Bug: http://b/19735063
        Bug: http://b/19735063
        """
        """
        self.assertFalse(AdbWrapper().shell("uname").endswith("\r\n"))
        output = AdbWrapper().shell("uname");
        if sys.platform == 'win32':
            # adb.exe running on Windows does translation to the Windows \r\n
            # convention, so we should expect those chars.
            self.assertTrue(output.endswith("\r\n"));
            # If the server outputs \r\n and adb.exe translates that to \r\r\n
            # we want to catch that server problem.
            self.assertFalse(output.endswith("\r\r\n"));
        else:
            self.assertFalse(output.endswith("\r\n"))




class AdbFile(unittest.TestCase):
class AdbFile(unittest.TestCase):
@@ -324,10 +338,12 @@ class AdbFile(unittest.TestCase):
        """Push a randomly generated file to specified device."""
        """Push a randomly generated file to specified device."""
        kbytes = 512
        kbytes = 512
        adb = AdbWrapper()
        adb = AdbWrapper()
        with tempfile.NamedTemporaryFile(mode="w") as tmp:
        with tempfile.NamedTemporaryFile(mode="wb", delete=False) as tmp:
            try:
                rand_str = os.urandom(1024 * kbytes)
                rand_str = os.urandom(1024 * kbytes)
                tmp.write(rand_str)
                tmp.write(rand_str)
                tmp.flush()
                tmp.flush()
                tmp.close()


                host_md5 = compute_md5(rand_str)
                host_md5 = compute_md5(rand_str)
                adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_FILE))
                adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_FILE))
@@ -338,6 +354,8 @@ class AdbFile(unittest.TestCase):
                    self.assertEqual(host_md5, dev_md5)
                    self.assertEqual(host_md5, dev_md5)
                finally:
                finally:
                    adb.shell_nocheck("rm {}".format(AdbFile.DEVICE_TEMP_FILE))
                    adb.shell_nocheck("rm {}".format(AdbFile.DEVICE_TEMP_FILE))
            finally:
                os.remove(tmp.name)


    # TODO: write push directory test.
    # TODO: write push directory test.


@@ -352,12 +370,18 @@ class AdbFile(unittest.TestCase):
            dev_md5, _ = adb.shell(
            dev_md5, _ = adb.shell(
                "md5sum {}".format(AdbFile.DEVICE_TEMP_FILE)).split()
                "md5sum {}".format(AdbFile.DEVICE_TEMP_FILE)).split()


            with tempfile.NamedTemporaryFile(mode="w") as tmp_write:
            with tempfile.NamedTemporaryFile(mode="wb", delete=False) \
                adb.pull(remote=AdbFile.DEVICE_TEMP_FILE, local=tmp_write.name)
                as tmp_write:
                with open(tmp_write.name) as tmp_read:
                try:
                    tmp_write.close()
                    adb.pull(remote=AdbFile.DEVICE_TEMP_FILE,
                             local=tmp_write.name)
                    with open(tmp_write.name, "rb") as tmp_read:
                        host_contents = tmp_read.read()
                        host_contents = tmp_read.read()
                        host_md5 = compute_md5(host_contents)
                        host_md5 = compute_md5(host_contents)
                    self.assertEqual(dev_md5, host_md5)
                    self.assertEqual(dev_md5, host_md5)
                finally:
                    os.remove(tmp_write.name)
        finally:
        finally:
            adb.shell_nocheck("rm {}".format(AdbFile.DEVICE_TEMP_FILE))
            adb.shell_nocheck("rm {}".format(AdbFile.DEVICE_TEMP_FILE))


@@ -383,7 +407,7 @@ class AdbFile(unittest.TestCase):
            for device_full_path in temp_files:
            for device_full_path in temp_files:
                host_path = os.path.join(
                host_path = os.path.join(
                    host_dir, temp_files[device_full_path].base_name)
                    host_dir, temp_files[device_full_path].base_name)
                with open(host_path) as host_file:
                with open(host_path, "rb") as host_file:
                    host_md5 = compute_md5(host_file.read())
                    host_md5 = compute_md5(host_file.read())
                    self.assertEqual(host_md5,
                    self.assertEqual(host_md5,
                                     temp_files[device_full_path].md5)
                                     temp_files[device_full_path].md5)
@@ -421,7 +445,7 @@ class AdbFile(unittest.TestCase):


            # confirm that every file on the device mirrors that on the host
            # confirm that every file on the device mirrors that on the host
            for host_full_path in temp_files.keys():
            for host_full_path in temp_files.keys():
                device_full_path = os.path.join(
                device_full_path = posixpath.join(
                    AdbFile.DEVICE_TEMP_DIR,
                    AdbFile.DEVICE_TEMP_DIR,
                    temp_files[host_full_path].base_name)
                    temp_files[host_full_path].base_name)
                dev_md5, _ = adb.shell(
                dev_md5, _ = adb.shell(
@@ -432,7 +456,7 @@ class AdbFile(unittest.TestCase):
            adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_DIR))
            adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_DIR))
            if temp_files:
            if temp_files:
                for tf in temp_files.values():
                for tf in temp_files.values():
                    tf.handle.close()
                    os.remove(tf.full_path)
            if base_dir:
            if base_dir:
                os.removedirs(base_dir + AdbFile.DEVICE_TEMP_DIR)
                os.removedirs(base_dir + AdbFile.DEVICE_TEMP_DIR)