Skip to content

Commit

Permalink
adb conn class improved and tests added
Browse files Browse the repository at this point in the history
den4uk committed Nov 14, 2020
1 parent e60b1c4 commit 213adee
Showing 3 changed files with 85 additions and 19 deletions.
28 changes: 18 additions & 10 deletions andriller/adb_conn.py
Original file line number Diff line number Diff line change
@@ -50,26 +50,29 @@ def setup_logging(self, logger=None, log_level=logging.INFO, **kwargs):
def setup(self):
self.logger.debug(f'Platform: {self.platform}')
if self.platform in self.UNIX:
self.adb_bin = self.cmd_shell('which adb') or None
self.adb_bin = self._get_adb_bin()
self.logger.debug(f'Using adb binary: {self.adb_bin}')
else:
self.adb_bin = os.path.join(CODEPATH, 'bin', 'adb.exe')
self._win_startupinfo()
if not self.adb_bin or not os.path.exists(self.adb_bin):
self.logger.warning('ADB binary is not found!')
raise ADBConnError('ADB binary is not found!')
self._is_adb_out_post_v5 = self.cmd_shell(f'{self.adb_bin} exec-out id', code=True) == 0
self._is_adb_out_post_v5 = self._adb_has_exec()

def _win_startupinfo(self):
self.startupinfo = subprocess.STARTUPINFO()
self.startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
self.rmr = b'\r\r\n'

def _opt_use_capture(self) -> bool:
return tuple(sys.version_info) >= (3, 7)

@property
def run_opt(self):
if not self._run_opt:
opt = {'shell': False, 'startupinfo': self.startupinfo}
if tuple(sys.version_info) >= (3, 7):
if self._opt_use_capture():
opt['capture_output'] = True
else:
opt['stdout'] = subprocess.PIPE
@@ -90,7 +93,7 @@ def adb(self, cmd, binary=False, su=False, _for_out=False, **kwargs) -> Union[st
to run `adb pull /path/myfile.txt` do: self.adb('shell id')
"""
cmd = self._get_adb_cmd(cmd, su, _for_out, **kwargs)
run = subprocess.run([self.adb_bin] + cmd, **self.run_opt)
run = subprocess.run([self.adb_bin, *cmd], **self.run_opt)
return self._return_run_output(run, binary)

def adb_out(self, cmd, binary=False, su=False, **kwargs) -> Union[str, bytes]:
@@ -105,17 +108,16 @@ def adb_out(self, cmd, binary=False, su=False, **kwargs) -> Union[str, bytes]:
Example:
to run `adb shell id` do: self.adb_out('id')
"""
return self.adb(cmd, binary=False, su=False, _for_out=True, **kwargs)
return self.adb(cmd, binary=binary, su=su, _for_out=True, **kwargs)

def _get_adb_cmd(self, cmd, su, _for_out, no_log=False, **kwargs) -> List[str]:
def _get_adb_cmd(self, cmd, su, _for_out, **kwargs) -> List[str]:
if isinstance(cmd, str):
cmd = shlex.split(cmd)
if su:
cmd.insert(0, 'su -c')
if _for_out:
cmd.insert(0, 'exec-out' if self._is_adb_out_post_v5 else 'shell')
if no_log is False:
self.logger.debug(f'ADB cmd: {cmd}')
self.logger.debug(f'ADB cmd: {cmd}')
return cmd

def _return_run_output(self, run: subprocess.CompletedProcess, binary: bool):
@@ -160,7 +162,7 @@ def start(self):
self.adb('start-server', timeout=10)

def kill(self):
self.adb('kill-server', timeout=5, no_log=True)
self.adb('kill-server', timeout=5)

@staticmethod
def _file_regex(fp):
@@ -219,14 +221,20 @@ def get_size(self, file_path, **kwargs) -> int:

@timeout(30, use_signals=False)
def cmd_shell(self, cmd, code=False, **kwargs):
self.logger.debug(f'CMD: {cmd}')
self.logger.debug(f'CMD cmd: {cmd}')
run = subprocess.run(shlex.split(cmd), **self.run_opt)
if code:
return run.returncode
else:
if run.stdout:
return run.stdout.decode().strip()

def _get_adb_bin(self):
return self.cmd_shell('which adb') or None

def _adb_has_exec(self) -> bool:
return self.cmd_shell(f'{self.adb_bin} exec-out id', code=True) == 0

def reboot(self, mode=None):
mode = self.MODES.get(mode, '')
self.logger.info(f'Rebooting in {mode}.')
5 changes: 3 additions & 2 deletions andriller/windows.py
Original file line number Diff line number Diff line change
@@ -268,7 +268,7 @@ def __init__(self, **kwargs):
# Text Field + logger
self.TF = tk.Text(
textframe, font=self.FontMono, wrap=tk.WORD, width=65,
bg='white', height=self.conf('window_size'))
height=self.conf('window_size'))
self.TF.bind('<Button-3>', rClicker, add='')
self.TF.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
self.set_logger()
@@ -302,7 +302,8 @@ def __init__(self, **kwargs):
self.conf.check_latest_version(logger=self.logger)

# Setup ADB logging
self.adb.setup_logging(logger=logger, log_level=self.log_level)
# do not pass self.logger as the logger, as the logger's lifetime is shorter, and is not reliable.
self.adb.setup_logging(log_level=self.log_level)

@property
def time_now_local(self):
71 changes: 64 additions & 7 deletions tests/test_adb.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import sys
import pytest
import tempfile
import subprocess
from unittest import mock
from andriller import adb_conn

@@ -10,24 +11,44 @@
@pytest.fixture
def ADB(mocker):
mocker.patch('andriller.adb_conn.ADBConn.kill')
with mock.patch('andriller.adb_conn.ADBConn.cmd_shell', return_value=fake_adb.name):
adb = adb_conn.ADBConn()
mocker.patch('andriller.adb_conn.ADBConn._opt_use_capture', return_value=True)
with mock.patch('andriller.adb_conn.ADBConn._get_adb_bin', return_value=fake_adb.name):
with mock.patch('andriller.adb_conn.ADBConn._adb_has_exec', return_value=True):
adb = adb_conn.ADBConn()
adb_cmd = adb.adb.__func__
setattr(adb, 'adb', lambda *args, **kwargs: adb_cmd(adb, *args, **kwargs))
return adb


@pytest.fixture
def ADB_alt(mocker):
mocker.patch('andriller.adb_conn.ADBConn.kill')
mocker.patch('andriller.adb_conn.ADBConn._opt_use_capture', return_value=False)
with mock.patch('andriller.adb_conn.ADBConn._get_adb_bin', return_value=fake_adb.name):
with mock.patch('andriller.adb_conn.ADBConn._adb_has_exec', return_value=False):
adb = adb_conn.ADBConn()
adb_cmd = adb.adb.__func__
setattr(adb, 'adb', lambda *args, **kwargs: adb_cmd(adb, *args, **kwargs))
return adb


@pytest.fixture
def ADB_win(mocker):
mock_sub = mocker.patch('andriller.adb_conn.subprocess', autospec=True)
mock_sub.STARTUPINFO = mock.MagicMock()
mock_sub.STARTF_USESHOWWINDOW = mock.MagicMock()
mocker.patch('andriller.adb_conn.ADBConn.kill')
mocker.patch('andriller.adb_conn.ADBConn._opt_use_capture', return_value=True)
with mock.patch('sys.platform', return_value='win32'):
adb = adb_conn.ADBConn()
with mock.patch('andriller.adb_conn.ADBConn._get_adb_bin', return_value=fake_adb.name):
with mock.patch('andriller.adb_conn.ADBConn._adb_has_exec', return_value=True):
adb = adb_conn.ADBConn()
return adb


@mock.Mock('subprocess.STARTUPINFO')
def test_init_windows(ADB_win):
assert ADB_win.startupinfo is not None
assert ADB_win.rmr == b'\r\r\n'


@pytest.mark.parametrize('file_path, result', [
@@ -39,7 +60,6 @@ def test_file_regex(file_path, result):
assert adb_conn.ADBConn._file_regex(file_path).match(result)


@pytest.mark.skipif(sys.version_info < (3, 7), reason="requires python3.7 or higher")
def test_adb_simple(ADB, mocker):
output = mock.Mock(stdout=b'lala', returncode=0)
mock_run = mocker.patch('andriller.adb_conn.subprocess.run', return_value=output)
@@ -50,7 +70,6 @@ def test_adb_simple(ADB, mocker):
capture_output=True, shell=False, startupinfo=None)


@pytest.mark.skipif(sys.version_info < (3, 7), reason="requires python3.7 or higher")
def test_adb_simple_su(ADB, mocker):
output = mock.Mock(stdout=b'lala', returncode=0)
mock_run = mocker.patch('andriller.adb_conn.subprocess.run', return_value=output)
@@ -61,7 +80,6 @@ def test_adb_simple_su(ADB, mocker):
capture_output=True, shell=False, startupinfo=None)


@pytest.mark.skipif(sys.version_info < (3, 7), reason="requires python3.7 or higher")
def test_adb_binary(ADB, mocker):
output = mock.Mock(stdout=b'lala', returncode=0)
mock_run = mocker.patch('andriller.adb_conn.subprocess.run', return_value=output)
@@ -70,3 +88,42 @@ def test_adb_binary(ADB, mocker):
assert res == b'lala'
mock_run.assert_called_with([fake_adb.name, 'hello'],
capture_output=True, shell=False, startupinfo=None)


def test_adb_out(ADB, mocker):
output = mock.Mock(stdout=b'uid(1000)', returncode=0)
mock_run = mocker.patch('andriller.adb_conn.subprocess.run', return_value=output)

res = ADB.adb_out('id', binary=False)
assert res == 'uid(1000)'
mock_run.assert_called_with([fake_adb.name, 'shell', 'id'],
capture_output=True, shell=False, startupinfo=None)


def test_adb_out_alt(ADB_alt, mocker):
output = mock.Mock(stdout=b'uid(1000)', returncode=0)
mock_run = mocker.patch('andriller.adb_conn.subprocess.run', return_value=output)

res = ADB_alt.adb_out('id', binary=True)
assert res == b'uid(1000)'
mock_run.assert_called_with([fake_adb.name, 'shell', 'id'],
stdout=subprocess.PIPE, shell=False, startupinfo=None)


def test_adb_out_win(ADB_win, mocker):
output = mock.Mock(stdout=b'uid(1000)\r\r\n', returncode=0)
mock_run = mocker.patch('andriller.adb_conn.subprocess.run', return_value=output)

res = ADB_win.adb_out('id', binary=True)
assert res == b'uid(1000)\n'


def test_adb_out_uses_exec(ADB, mocker):
ADB._is_adb_out_post_v5 = True
output = mock.Mock(stdout=b'uid(1000)', returncode=0)
mock_run = mocker.patch('andriller.adb_conn.subprocess.run', return_value=output)

res = ADB.adb_out('id', binary=False)
assert res == 'uid(1000)'
mock_run.assert_called_with([fake_adb.name, 'exec-out', 'id'],
capture_output=True, shell=False, startupinfo=None)

0 comments on commit 213adee

Please sign in to comment.