1
0
mirror of https://github.com/google/adb-sync.git synced 2026-01-03 01:48:02 +00:00

Also pass mypy --strict.

This commit is contained in:
Rudolf Polzer
2018-11-27 11:41:45 -08:00
parent 384a485224
commit 270bfe6b3a

View File

@@ -26,7 +26,8 @@ import stat
import subprocess import subprocess
import sys import sys
import time import time
from typing import Callable, cast, Dict, List, IO, Iterable, Tuple from types import TracebackType
from typing import Callable, cast, Dict, List, IO, Iterable, Optional, Tuple, Type
class OSLike(object): class OSLike(object):
@@ -73,7 +74,9 @@ class Stdout(object):
def __enter__(self) -> IO: def __enter__(self) -> IO:
return self.popen.stdout return self.popen.stdout
def __exit__(self, exc_type, exc_value, traceback) -> bool: def __exit__(self, exc_type: Optional[Type[BaseException]],
exc_val: Optional[Exception],
exc_tb: Optional[TracebackType]) -> bool:
self.popen.stdout.close() self.popen.stdout.close()
if self.popen.wait() != 0: if self.popen.wait() != 0:
raise OSError('Subprocess exited with nonzero status.') raise OSError('Subprocess exited with nonzero status.')
@@ -130,7 +133,7 @@ class AdbFileSystem(OSLike):
(?(S_IFLNK) .* | (?P<filename> .*)) (?(S_IFLNK) .* | (?P<filename> .*))
$""", re.DOTALL | re.VERBOSE) $""", re.DOTALL | re.VERBOSE)
def LsToStat(self, line) -> Tuple[os.stat_result, bytes]: def LsToStat(self, line: bytes) -> Tuple[os.stat_result, bytes]:
"""Convert a line from 'ls -l' output to a stat result. """Convert a line from 'ls -l' output to a stat result.
Args: Args:
@@ -294,25 +297,25 @@ class AdbFileSystem(OSLike):
b'touch -at %s %s' % (timestr, self.QuoteArgument(path))]) != 0: b'touch -at %s %s' % (timestr, self.QuoteArgument(path))]) != 0:
raise OSError('touch failed') raise OSError('touch failed')
def glob(self, path) -> Iterable[bytes]: def glob(self, path: bytes) -> Iterable[bytes]:
with Stdout( with Stdout(
self.adb + self.adb +
[b'shell', b'for p in %s; do echo "$p"; done' % (path,)]) as stdout: [b'shell', b'for p in %s; do echo "$p"; done' % (path,)]) as stdout:
for line in stdout: for line in stdout:
yield line.rstrip(b'\r\n') yield line.rstrip(b'\r\n')
def Push(self, src: bytes, dst: bytes): def Push(self, src: bytes, dst: bytes) -> None:
"""Push a file from the local file system to the Android device.""" """Push a file from the local file system to the Android device."""
if subprocess.call(self.adb + [b'push', src, dst]) != 0: if subprocess.call(self.adb + [b'push', src, dst]) != 0:
raise OSError('push failed') raise OSError('push failed')
def Pull(self, src: bytes, dst: bytes): def Pull(self, src: bytes, dst: bytes) -> None:
"""Pull a file from the Android device to the local file system.""" """Pull a file from the Android device to the local file system."""
if subprocess.call(self.adb + [b'pull', src, dst]) != 0: if subprocess.call(self.adb + [b'pull', src, dst]) != 0:
raise OSError('pull failed') raise OSError('pull failed')
def BuildFileList(fs, path: bytes, def BuildFileList(fs: OSLike, path: bytes,
prefix: bytes) -> Iterable[Tuple[bytes, os.stat_result]]: prefix: bytes) -> Iterable[Tuple[bytes, os.stat_result]]:
"""Builds a file list. """Builds a file list.
@@ -366,54 +369,33 @@ def DiffLists(a: Iterable[Tuple[bytes, os.stat_result]],
b_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]] both = [] # type: List[Tuple[bytes, os.stat_result, os.stat_result]]
a_iter = iter(sorted(a)) a_revlist = sorted(a)
b_iter = iter(sorted(b)) a_revlist.reverse()
a_active = True b_revlist = sorted(b)
b_active = True b_revlist.reverse()
a_available = False
b_available = False
a_item = None
b_item = None
while a_active and b_active: while True:
if not a_available: if not a_revlist:
try: b_only.extend(reversed(b_revlist))
a_item = next(a_iter) break
a_available = True if not b_revlist:
except StopIteration: a_only.extend(reversed(a_revlist))
a_active = False break
break a_item = a_revlist[len(a_revlist) - 1]
if not b_available: b_item = b_revlist[len(b_revlist) - 1]
try:
b_item = next(b_iter)
b_available = True
except StopIteration:
b_active = False
break
if a_item[0] == b_item[0]: if a_item[0] == b_item[0]:
both.append((a_item[0], a_item[1], b_item[1])) both.append((a_item[0], a_item[1], b_item[1]))
a_available = False a_revlist.pop()
b_available = False b_revlist.pop()
elif a_item[0] < b_item[0]: elif a_item[0] < b_item[0]:
a_only.append(a_item) a_only.append(a_item)
a_available = False a_revlist.pop()
elif a_item[0] > b_item[0]: elif a_item[0] > b_item[0]:
b_only.append(b_item) b_only.append(b_item)
b_available = False b_revlist.pop()
else: else:
raise raise
if a_active:
if a_available:
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(cast(Tuple[bytes, os.stat_result], b_item))
for item in b_iter:
b_only.append(item)
return a_only, both, b_only return a_only, both, b_only
@@ -444,7 +426,9 @@ class DeleteInterruptedFile(object):
def __enter__(self) -> None: def __enter__(self) -> None:
pass pass
def __exit__(self, exc_type, exc_value, traceback) -> bool: def __exit__(self, exc_type: Optional[Type[BaseException]],
exc_val: Optional[Exception],
exc_tb: Optional[TracebackType]) -> bool:
if exc_type is not None: if exc_type is not None:
logging.info('Interrupted-%s-Delete: %r', logging.info('Interrupted-%s-Delete: %r',
'Pull' if self.fs == os else 'Push', self.name) 'Pull' if self.fs == os else 'Push', self.name)
@@ -495,7 +479,7 @@ class FileSyncer(object):
def ScanAndDiff(self) -> None: def ScanAndDiff(self) -> None:
"""Scans the local and remote locations and identifies differences.""" """Scans the local and remote locations and identifies differences."""
logging.info('Scanning and diffing...') logging.info('Scanning and diffing...')
locallist = BuildFileList(os, self.local, b'') locallist = BuildFileList(cast(OSLike, os), self.local, b'')
remotelist = BuildFileList(self.adb, self.remote, b'') remotelist = BuildFileList(self.adb, self.remote, b'')
self.local_only, self.both, self.remote_only = DiffLists( self.local_only, self.both, self.remote_only = DiffLists(
locallist, remotelist) locallist, remotelist)
@@ -669,7 +653,7 @@ def FixPath(src: bytes, dst: bytes) -> Tuple[bytes, bytes]:
return (src, dst) return (src, dst)
def main(*unused_args) -> None: def main() -> None:
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description='Synchronize a directory between an Android device and the ' + description='Synchronize a directory between an Android device and the ' +
'local file system') 'local file system')
@@ -855,4 +839,4 @@ def main(*unused_args) -> None:
if __name__ == '__main__': if __name__ == '__main__':
main(*sys.argv) main()