From 384a48522461bb486fd75483341baaebfefc9d09 Mon Sep 17 00:00:00 2001 From: Rudolf Polzer Date: Tue, 27 Nov 2018 09:57:41 -0800 Subject: [PATCH] Add annotations to pass both mypy and pytype. --- adb-sync | 304 +++++++++++++++++++++++++++++++------------------------ 1 file changed, 172 insertions(+), 132 deletions(-) diff --git a/adb-sync b/adb-sync index c5c0a67..891868e 100755 --- a/adb-sync +++ b/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 .*)) $""", 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( - time.strptime( - match.group('st_mtime').decode('utf-8'), '%Y-%m-%d %H:%M')) + 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('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')