Add annotations to pass both mypy and pytype.

adb-sync

@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/python3
 
 # Copyright 2014 Google Inc. All rights reserved.
 #
@@ -26,13 +26,65 @@ import stat
 import subprocess
 import sys
 import time
+from typing import Callable, cast, Dict, List, IO, Iterable, Tuple
 
 
-class AdbFileSystem(object):
+class OSLike(object):
+
+  def listdir(self, path: bytes) -> Iterable[bytes]:
+    raise NotImplementedError('Abstract')
+
+  def lstat(self, path: bytes) -> os.stat_result:
+    raise NotImplementedError('Abstract')
+
+  def unlink(self, path: bytes) -> None:
+    raise NotImplementedError('Abstract')
+
+  def rmdir(self, path: bytes) -> None:
+    raise NotImplementedError('Abstract')
+
+  def makedirs(self, path: bytes) -> None:
+    raise NotImplementedError('Abstract')
+
+  def utime(self, path: bytes, times: Tuple[float, float]) -> None:
+    raise NotImplementedError('Abstract')
+
+  def glob(self, path: bytes) -> Iterable[bytes]:
+    raise NotImplementedError('Abstract')
+
+
+class Stdout(object):
+
+  def __init__(self, args: List[bytes]) -> None:
+    """Closes the process's stdout when done.
+
+    Usage:
+      with Stdout(...) as stdout:
+        DoSomething(stdout)
+
+    Args:
+      args: Which program to run.
+
+    Returns:
+      An object for use by 'with'.
+    """
+    self.popen = subprocess.Popen(args, stdout=subprocess.PIPE)
+
+  def __enter__(self) -> IO:
+    return self.popen.stdout
+
+  def __exit__(self, exc_type, exc_value, traceback) -> bool:
+    self.popen.stdout.close()
+    if self.popen.wait() != 0:
+      raise OSError('Subprocess exited with nonzero status.')
+    return False
+
+
+class AdbFileSystem(OSLike):
   """Mimics os's file interface but uses the adb utility."""
 
-  def __init__(self, adb):
-    self.stat_cache = {}
+  def __init__(self, adb: List[bytes]) -> None:
+    self.stat_cache = {}  # type: Dict[bytes, os.stat_result]
     self.adb = adb
 
   # Regarding parsing stat results, we only care for the following fields:
@@ -78,7 +130,7 @@ class AdbFileSystem(object):
                              (?(S_IFLNK) .* | (?P<filename> .*))
                              $""", re.DOTALL | re.VERBOSE)
 
-  def LsToStat(self, line):
+  def LsToStat(self, line) -> Tuple[os.stat_result, bytes]:
     """Convert a line from 'ls -l' output to a stat result.
 
     Args:
@@ -116,12 +168,11 @@ class AdbFileSystem(object):
       st_mode |= stat.S_IFLNK
     if groups['S_IFSOCK']:
       st_mode |= stat.S_IFSOCK
-    st_size = groups['st_size']
-    if st_size is not None:
-      st_size = int(st_size)
-    st_mtime = time.mktime(
+    st_size = None if groups['st_size'] is None else int(groups['st_size'])
+    st_mtime = int(
+        time.mktime(
             time.strptime(
-            match.group('st_mtime').decode('utf-8'), '%Y-%m-%d %H:%M'))
+                match.group('st_mtime').decode('ascii'), '%Y-%m-%d %H:%M')))
 
     # Fill the rest with dummy values.
     st_ino = 1
@@ -136,38 +187,7 @@ class AdbFileSystem(object):
     filename = groups['filename']
     return stbuf, filename
 
-  def Stdout(self, *popen_args):
-    """Closes the process's stdout when done.
-
-    Usage:
-      with Stdout(...) as stdout:
-        DoSomething(stdout)
-
-    Args:
-      popen_args: Arguments for subprocess.Popen; stdout=PIPE is implicitly
-        added.
-
-    Returns:
-      An object for use by 'with'.
-    """
-
-    class Stdout(object):
-
-      def __init__(self, popen):
-        self.popen = popen
-
-      def __enter__(self):
-        return self.popen.stdout
-
-      def __exit__(self, exc_type, exc_value, traceback):
-        self.popen.stdout.close()
-        if self.popen.wait() != 0:
-          raise OSError('Subprocess exited with nonzero status.')
-        return False
-
-    return Stdout(subprocess.Popen(*popen_args, stdout=subprocess.PIPE))
-
-  def QuoteArgument(self, arg):
+  def QuoteArgument(self, arg: bytes) -> bytes:
     # Quotes an argument for use by adb shell.
     # Usually, arguments in 'adb shell' use are put in double quotes by adb,
     # but not in any way escaped.
@@ -178,7 +198,7 @@ class AdbFileSystem(object):
     arg = b'"' + arg + b'"'
     return arg
 
-  def IsWorking(self):
+  def IsWorking(self) -> bool:
     """Tests the adb connection."""
     # This string should contain all possible evil, but no percent signs.
     # Note this code uses 'date' and not 'echo', as date just calls strftime
@@ -189,10 +209,9 @@ class AdbFileSystem(object):
     ]
     for test_string in test_strings:
       good = False
-      with self.Stdout(
-          self.adb +
-          [b'shell', b'date +%s' %
-           (self.QuoteArgument(test_string),)]) as stdout:
+      with Stdout(self.adb +
+                  [b'shell',
+                   b'date +%s' % (self.QuoteArgument(test_string),)]) as stdout:
         for line in stdout:
           line = line.rstrip(b'\r\n')
           if line == test_string:
@@ -201,12 +220,11 @@ class AdbFileSystem(object):
         return False
     return True
 
-  def listdir(self, path):  # os's name, so pylint: disable=g-bad-name
+  def listdir(self, path: bytes) -> Iterable[bytes]:  # os's name, so pylint: disable=g-bad-name
     """List the contents of a directory, caching them for later lstat calls."""
-    with self.Stdout(
-        self.adb +
-        [b'shell', b'ls -al %s' %
-         (self.QuoteArgument(path + b'/'),)]) as stdout:
+    with Stdout(self.adb +
+                [b'shell',
+                 b'ls -al %s' % (self.QuoteArgument(path + b'/'),)]) as stdout:
       for line in stdout:
         if line.startswith(b'total '):
           continue
@@ -221,11 +239,11 @@ class AdbFileSystem(object):
           self.stat_cache[path + b'/' + filename] = statdata
           yield filename
 
-  def lstat(self, path):  # os's name, so pylint: disable=g-bad-name
+  def lstat(self, path: bytes) -> os.stat_result:  # os's name, so pylint: disable=g-bad-name
     """Stat a file."""
     if path in self.stat_cache:
       return self.stat_cache[path]
-    with self.Stdout(
+    with Stdout(
         self.adb +
         [b'shell', b'ls -ald %s' % (self.QuoteArgument(path),)]) as stdout:
       for line in stdout:
@@ -237,62 +255,65 @@ class AdbFileSystem(object):
         return statdata
     raise OSError('No such file or directory')
 
-  def unlink(self, path):  # os's name, so pylint: disable=g-bad-name
+  def unlink(self, path: bytes) -> None:  # os's name, so pylint: disable=g-bad-name
     """Delete a file."""
     if subprocess.call(
         self.adb + [b'shell', b'rm %s' % (self.QuoteArgument(path),)]) != 0:
       raise OSError('unlink failed')
 
-  def rmdir(self, path):  # os's name, so pylint: disable=g-bad-name
+  def rmdir(self, path: bytes) -> None:  # os's name, so pylint: disable=g-bad-name
     """Delete a directory."""
     if subprocess.call(
         self.adb +
         [b'shell', b'rmdir %s' % (self.QuoteArgument(path),)]) != 0:
       raise OSError('rmdir failed')
 
-  def makedirs(self, path):  # os's name, so pylint: disable=g-bad-name
+  def makedirs(self, path: bytes) -> None:  # os's name, so pylint: disable=g-bad-name
     """Create a directory."""
     if subprocess.call(
         self.adb +
         [b'shell', b'mkdir -p %s' % (self.QuoteArgument(path),)]) != 0:
       raise OSError('mkdir failed')
 
-  def utime(self, path, times):
+  def utime(self, path: bytes, times: Tuple[float, float]) -> None:
     # TODO(rpolzer): Find out why this does not work (returns status 255).
     """Set the time of a file to a specified unix time."""
     atime, mtime = times
-    timestr = time.strftime(b'%Y%m%d.%H%M%S', time.localtime(mtime))
+    timestr = time.strftime('%Y%m%d.%H%M%S',
+                            time.localtime(mtime)).encode('ascii')
     if subprocess.call(
         self.adb +
         [b'shell',
          b'touch -mt %s %s' % (timestr, self.QuoteArgument(path))]) != 0:
       raise OSError('touch failed')
-    timestr = time.strftime(b'%Y%m%d.%H%M%S', time.localtime(atime))
+    timestr = time.strftime('%Y%m%d.%H%M%S',
+                            time.localtime(atime)).encode('ascii')
     if subprocess.call(
         self.adb +
         [b'shell',
          b'touch -at %s %s' % (timestr, self.QuoteArgument(path))]) != 0:
       raise OSError('touch failed')
 
-  def glob(self, path):
-    with self.Stdout(
+  def glob(self, path) -> Iterable[bytes]:
+    with Stdout(
         self.adb +
         [b'shell', b'for p in %s; do echo "$p"; done' % (path,)]) as stdout:
       for line in stdout:
         yield line.rstrip(b'\r\n')
 
-  def Push(self, src, dst):
+  def Push(self, src: bytes, dst: bytes):
     """Push a file from the local file system to the Android device."""
     if subprocess.call(self.adb + [b'push', src, dst]) != 0:
       raise OSError('push failed')
 
-  def Pull(self, src, dst):
+  def Pull(self, src: bytes, dst: bytes):
     """Pull a file from the Android device to the local file system."""
     if subprocess.call(self.adb + [b'pull', src, dst]) != 0:
       raise OSError('pull failed')
 
 
-def BuildFileList(fs, path, prefix=b''):
+def BuildFileList(fs, path: bytes,
+                  prefix: bytes) -> Iterable[Tuple[bytes, os.stat_result]]:
   """Builds a file list.
 
   Args:
@@ -311,10 +332,9 @@ def BuildFileList(fs, path, prefix=b''):
   if stat.S_ISDIR(statresult.st_mode):
     yield prefix, statresult
     try:
-      files = list(fs.listdir(path))
+      files = fs.listdir(path)
     except OSError:
       return
-    files.sort()
     for n in files:
       if n == b'.' or n == b'..':
         continue
@@ -326,7 +346,11 @@ def BuildFileList(fs, path, prefix=b''):
     logging.info('Unsupported file: %r.', path)
 
 
-def DiffLists(a, b):
+def DiffLists(a: Iterable[Tuple[bytes, os.stat_result]],
+              b: Iterable[Tuple[bytes, os.stat_result]]
+             ) -> Tuple[List[Tuple[bytes, os.stat_result]], List[
+                 Tuple[bytes, os.stat_result, os
+                       .stat_result]], List[Tuple[bytes, os.stat_result]]]:
   """Compares two lists.
 
   Args:
@@ -338,12 +362,12 @@ def DiffLists(a, b):
     both: the items from both list, with the remaining tuple items combined.
     b_only: the items from list b.
   """
-  a_only = []
-  b_only = []
-  both = []
+  a_only = []  # type: List[Tuple[bytes, os.stat_result]]
+  b_only = []  # type: List[Tuple[bytes, os.stat_result]]
+  both = []  # type: List[Tuple[bytes, os.stat_result, os.stat_result]]
 
-  a_iter = iter(a)
-  b_iter = iter(b)
+  a_iter = iter(sorted(a))
+  b_iter = iter(sorted(b))
   a_active = True
   b_active = True
   a_available = False
@@ -367,7 +391,7 @@ def DiffLists(a, b):
         b_active = False
         break
     if a_item[0] == b_item[0]:
-      both.append(tuple([a_item[0]] + list(a_item[1:]) + list(b_item[1:])))
+      both.append((a_item[0], a_item[1], b_item[1]))
       a_available = False
       b_available = False
     elif a_item[0] < b_item[0]:
@@ -381,24 +405,62 @@ def DiffLists(a, b):
 
   if a_active:
     if a_available:
-      a_only.append(a_item)
+      a_only.append(cast(Tuple[bytes, os.stat_result], a_item))
     for item in a_iter:
       a_only.append(item)
   if b_active:
     if b_available:
-      b_only.append(b_item)
+      b_only.append(cast(Tuple[bytes, os.stat_result], b_item))
     for item in b_iter:
       b_only.append(item)
 
   return a_only, both, b_only
 
 
+class DeleteInterruptedFile(object):
+
+  def __init__(self, dry_run: bool, fs: OSLike, name: bytes) -> None:
+    """Sets up interrupt protection.
+
+    Usage:
+      with DeleteInterruptedFile(False, fs, name):
+        DoSomething()
+
+      If DoSomething() should get interrupted, the file 'name' will be deleted.
+      The exception otherwise will be passed on.
+
+    Args:
+      dry_run: If true, we don't actually delete.
+      fs: File system object.
+      name: File name to delete.
+
+    Returns:
+      An object for use by 'with'.
+    """
+    self.dry_run = dry_run
+    self.fs = fs
+    self.name = name
+
+  def __enter__(self) -> None:
+    pass
+
+  def __exit__(self, exc_type, exc_value, traceback) -> bool:
+    if exc_type is not None:
+      logging.info('Interrupted-%s-Delete: %r',
+                   'Pull' if self.fs == os else 'Push', self.name)
+      if not self.dry_run:
+        self.fs.unlink(self.name)
+    return False
+
+
 class FileSyncer(object):
   """File synchronizer."""
 
-  def __init__(self, adb, local_path, remote_path, local_to_remote,
-               remote_to_local, preserve_times, delete_missing, allow_overwrite,
-               allow_replace, dry_run):
+  def __init__(self, adb: AdbFileSystem, local_path: bytes, remote_path: bytes,
+               local_to_remote: bool, remote_to_local: bool,
+               preserve_times: bool, delete_missing: bool,
+               allow_overwrite: bool, allow_replace: bool,
+               dry_run: bool) -> None:
     self.local = local_path
     self.remote = remote_path
     self.adb = adb
@@ -409,21 +471,32 @@ class FileSyncer(object):
     self.allow_overwrite = allow_overwrite
     self.allow_replace = allow_replace
     self.dry_run = dry_run
-    self.local_only = None
-    self.both = None
-    self.remote_only = None
     self.num_bytes = 0
     self.start_time = time.time()
 
-  def IsWorking(self):
+  # Attributes filled in later.
+  local_only = None  # type: List[Tuple[bytes, os.stat_result]]
+  both = None  # type: List[Tuple[bytes, os.stat_result, os.stat_result]]
+  remote_only = None  # type: List[Tuple[bytes, os.stat_result]]
+  src_to_dst = None  # type: Tuple[bool, bool]
+  dst_to_src = None  # type: Tuple[bool, bool]
+  src_only = None  # type: Tuple[List[Tuple[bytes, os.stat_result]], List[Tuple[bytes, os.stat_result]]]
+  dst_only = None  # type: Tuple[List[Tuple[bytes, os.stat_result]], List[Tuple[bytes, os.stat_result]]]
+  src = None  # type: Tuple[bytes, bytes]
+  dst = None  # type: Tuple[bytes, bytes]
+  dst_fs = None  # type: Tuple[OSLike, OSLike]
+  push = None  # type: Tuple[str, str]
+  copy = None  # type: Tuple[Callable[[bytes, bytes], None], Callable[[bytes, bytes], None]]
+
+  def IsWorking(self) -> bool:
     """Tests the adb connection."""
     return self.adb.IsWorking()
 
-  def ScanAndDiff(self):
+  def ScanAndDiff(self) -> None:
     """Scans the local and remote locations and identifies differences."""
     logging.info('Scanning and diffing...')
-    locallist = BuildFileList(os, self.local)
-    remotelist = BuildFileList(self.adb, self.remote)
+    locallist = BuildFileList(os, self.local, b'')
+    remotelist = BuildFileList(self.adb, self.remote, b'')
     self.local_only, self.both, self.remote_only = DiffLists(
         locallist, remotelist)
     if not self.local_only and not self.both and not self.remote_only:
@@ -434,46 +507,11 @@ class FileSyncer(object):
     self.dst_only = (self.remote_only, self.local_only)
     self.src = (self.local, self.remote)
     self.dst = (self.remote, self.local)
-    self.dst_fs = (self.adb, os)
+    self.dst_fs = (self.adb, cast(OSLike, os))
     self.push = ('Push', 'Pull')
     self.copy = (self.adb.Push, self.adb.Pull)
 
-  def InterruptProtection(self, fs, name):
-    """Sets up interrupt protection.
-
-    Usage:
-      with self.InterruptProtection(fs, name):
-        DoSomething()
-
-      If DoSomething() should get interrupted, the file 'name' will be deleted.
-      The exception otherwise will be passed on.
-
-    Args:
-      fs: File system object.
-      name: File name to delete.
-
-    Returns:
-      An object for use by 'with'.
-    """
-
-    dry_run = self.dry_run
-
-    class DeleteInterruptedFile(object):
-
-      def __enter__(self):
-        pass
-
-      def __exit__(self, exc_type, exc_value, traceback):
-        if exc_type is not None:
-          logging.info(b'Interrupted-%s-Delete: %r',
-                       'Pull' if fs == os else 'Push', name)
-          if not dry_run:
-            fs.unlink(name)
-        return False
-
-    return DeleteInterruptedFile()
-
-  def PerformDeletions(self):
+  def PerformDeletions(self) -> None:
     """Perform all deleting necessary for the file sync operation."""
     if not self.delete_missing:
       return
@@ -493,9 +531,11 @@ class FileSyncer(object):
                 self.dst_fs[i].unlink(dst_name)
           del self.dst_only[i][:]
 
-  def PerformOverwrites(self):
+  def PerformOverwrites(self) -> None:
     """Delete files/directories that are in the way for overwriting."""
-    src_only_prepend = ([], [])
+    src_only_prepend = (
+        [], []
+    )  # type: Tuple[List[Tuple[bytes, os.stat_result]], List[Tuple[bytes, os.stat_result]]]
     for name, localstat, remotestat in self.both:
       if stat.S_ISDIR(localstat.st_mode) and stat.S_ISDIR(remotestat.st_mode):
         # A dir is a dir is a dir.
@@ -566,7 +606,7 @@ class FileSyncer(object):
     for i in [0, 1]:
       self.src_only[i][:0] = src_only_prepend[i]
 
-  def PerformCopies(self):
+  def PerformCopies(self) -> None:
     """Perform all copying necessary for the file sync operation."""
     for i in [0, 1]:
       if self.src_to_dst[i]:
@@ -578,7 +618,7 @@ class FileSyncer(object):
             if not self.dry_run:
               self.dst_fs[i].makedirs(dst_name)
           else:
-            with self.InterruptProtection(self.dst_fs[i], dst_name):
+            with DeleteInterruptedFile(self.dry_run, self.dst_fs[i], dst_name):
               if not self.dry_run:
                 self.copy[i](src_name, dst_name)
               if stat.S_ISREG(s.st_mode):
@@ -590,7 +630,7 @@ class FileSyncer(object):
                            time.asctime(time.localtime(s.st_mtime)))
               self.dst_fs[i].utime(dst_name, (s.st_atime, s.st_mtime))
 
-  def TimeReport(self):
+  def TimeReport(self) -> None:
     """Report time and amount of data transferred."""
     if self.dry_run:
       logging.info('Total: %d bytes', self.num_bytes)
@@ -602,13 +642,13 @@ class FileSyncer(object):
                    dt)
 
 
-def ExpandWildcards(globber, path):
+def ExpandWildcards(globber: OSLike, path: bytes) -> Iterable[bytes]:
   if path.find(b'?') == -1 and path.find(b'*') == -1 and path.find(b'[') == -1:
     return [path]
   return globber.glob(path)
 
 
-def FixPath(src, dst):
+def FixPath(src: bytes, dst: bytes) -> Tuple[bytes, bytes]:
   # rsync-like path munging to make remote specifications shorter.
   append = b''
   pos = src.rfind(b'/')
@@ -629,7 +669,7 @@ def FixPath(src, dst):
   return (src, dst)
 
 
-def main(*args):
+def main(*unused_args) -> None:
   parser = argparse.ArgumentParser(
       description='Synchronize a directory between an Android device and the ' +
       'local file system')