Skip to content

Commit

Permalink
Cleanup tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@JeffLIrion
JeffLIrion committed Jun 30, 2020
1 parent d8bb11f commit d303941
Show file tree
Hide file tree
Showing 2 changed files with 0 additions and 160 deletions.
78 changes: 0 additions & 78 deletions tests/test_adb_device.py
View file
Expand Up @@ -517,35 +517,6 @@ def test_list(self):
self.assertEqual(expected_result, self.device.list('/dir'))
self.assertEqual(expected_bulk_write, self.device._transport._bulk_write)

def _test_push(self, mtime):
self.assertTrue(self.device.connect())
self.device._transport._bulk_write = b''

filedata = b'Ohayou sekai.\nGood morning world!'

# Provide the `bulk_read` return values
self.device._transport._bulk_read = join_messages(AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b''),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(constants.OKAY, data=b''))),
AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))

# Expected `bulk_write` values
expected_bulk_write = join_messages(AdbMessage(command=constants.OPEN, arg0=1, arg1=0, data=b'sync:\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(command=constants.SEND, data=b'/data,33272'),
FileSyncMessage(command=constants.DATA, data=filedata),
FileSyncMessage(command=constants.DONE, arg0=mtime))),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b''),
AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))

with patch('time.time', return_value=mtime):
self.device.push(BytesIO(filedata), '/data', mtime=mtime)
self.assertEqual(expected_bulk_write, self.device._transport._bulk_write)

return True

def _test_push(self):
self.assertTrue(self._test_push(100))

def test_push_fail(self):
self.assertTrue(self.device.connect())
self.device._transport._bulk_write = b''
Expand Down Expand Up @@ -607,7 +578,6 @@ def test_push_file_mtime0(self):
AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b''),
AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))

#with patch('time.time', return_value=mtime):
with patch('adb_shell.adb_device.open', mock_open(read_data=filedata)), patch('time.time', return_value=mtime):
self.device.push('TEST_FILE', '/data', mtime=mtime)
self.assertEqual(expected_bulk_write, self.device._transport._bulk_write)
Expand Down Expand Up @@ -670,28 +640,6 @@ def test_push_dir(self):
with patch('adb_shell.adb_device.open', mock_open(read_data=filedata)), patch('os.path.isdir', lambda x: x == 'TEST_DIR/'), patch('os.listdir', return_value=['TEST_FILE1', 'TEST_FILE2']):
self.device.push('TEST_DIR/', '/data', mtime=mtime)

def _test_pull(self):
self.assertTrue(self.device.connect())
self.device._transport._bulk_write = b''

filedata = b'Ohayou sekai.\nGood morning world!'

# Provide the `bulk_read` return values
self.device._transport._bulk_read = join_messages(AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(command=constants.DATA, data=filedata),
FileSyncMessage(command=constants.DONE))),
AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))

# Expected `bulk_write` values
expected_bulk_write = join_messages(AdbMessage(command=constants.OPEN, arg0=1, arg1=0, data=b'sync:\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(command=constants.RECV, data=b'/data'))),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1),
AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))

self.assertEqual(filedata, self.device.pull('/data'))
self.assertEqual(expected_bulk_write, self.device._transport._bulk_write)

def test_pull_file(self):
self.assertTrue(self.device.connect())
self.device._transport._bulk_write = b''
Expand All @@ -716,38 +664,13 @@ def test_pull_file(self):
self.assertEqual(b''.join([bytes(call.args[0]) for call in m().write.mock_calls]), filedata)
self.assertEqual(expected_bulk_write, self.device._transport._bulk_write)

def _test_pull_file_return_true(self):
self.assertTrue(self.device.connect())
self.device._transport._bulk_write = b''

filedata = b'Ohayou sekai.\nGood morning world!'

# Provide the `bulk_read` return values
self.device._transport._bulk_read = join_messages(AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(command=constants.DATA, data=filedata),
FileSyncMessage(command=constants.DONE))),
AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))

# Expected `bulk_write` values
expected_bulk_write = join_messages(AdbMessage(command=constants.OPEN, arg0=1, arg1=0, data=b'sync:\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(command=constants.RECV, data=b'/data'))),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1),
AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))

with patch('adb_shell.adb_device.open', mock_open()) as m:
self.device.pull('/data', 'TEST_FILE')
self.assertEqual(b''.join([bytes(call.args[0]) for call in m().write.mock_calls]), filedata)
self.assertEqual(expected_bulk_write, self.device._transport._bulk_write)

def test_pull_big_file(self):
self.assertTrue(self.device.connect())
self.device._transport._bulk_write = b''

filedata = b'0' * int(1.5 * constants.MAX_ADB_DATA)

# Provide the `bulk_read` return values

self.device._transport._bulk_read = join_messages(AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(command=constants.DATA, data=filedata),
Expand All @@ -770,7 +693,6 @@ def test_stat(self):
self.device._transport._bulk_write = b''

# Provide the `bulk_read` return values

self.device._transport._bulk_read = join_messages(AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncStatMessage(constants.STAT, 1, 2, 3),
Expand Down
82 changes: 0 additions & 82 deletions tests/test_adb_device_async.py
View file
Expand Up @@ -553,38 +553,6 @@ async def test_list(self):
self.assertEqual(await self.device.list('/dir'), expected_result)
self.assertEqual(self.device._transport._bulk_write, expected_bulk_write)

async def _test_push(self, mtime):
self.assertTrue(await self.device.connect())
self.device._transport._bulk_write = b''

filedata = b'Ohayou sekai.\nGood morning world!'

# Provide the `bulk_read` return values
self.device._transport._bulk_read = join_messages(AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b''),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(constants.OKAY, data=b''))),
AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))

# Expected `bulk_write` values
expected_bulk_write = join_messages(AdbMessage(command=constants.OPEN, arg0=1, arg1=0, data=b'sync:\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(command=constants.SEND, data=b'/data,33272'),
FileSyncMessage(command=constants.DATA, data=filedata),
FileSyncMessage(command=constants.DONE, arg0=mtime))),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b''),
AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))

with patch('time.time', return_value=mtime):
with self.assertRaises(Exception):
await self.device.push(BytesIO(filedata), '/data', mtime=mtime)
# self.assertEqual(self.device._transport._bulk_write, expected_bulk_write)

self.device._transport._bulk_read = []
return True

@awaiter
async def _test_push(self):
self.assertTrue(await self._test_push(100))

@awaiter
async def test_push_fail(self):
self.assertTrue(await self.device.connect())
Expand Down Expand Up @@ -713,29 +681,6 @@ async def test_push_dir(self):
with patch('adb_shell.adb_device_async.open', mock_open(read_data=filedata)), patch('os.path.isdir', lambda x: x == 'TEST_DIR/'), patch('os.listdir', return_value=['TEST_FILE1', 'TEST_FILE2']):
await self.device.push('TEST_DIR/', '/data', mtime=mtime)

@awaiter
async def _test_pull(self):
self.assertTrue(await self.device.connect())
self.device._transport._bulk_write = b''

filedata = b'Ohayou sekai.\nGood morning world!'

# Provide the `bulk_read` return values
self.device._transport._bulk_read = join_messages(AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(command=constants.DATA, data=filedata),
FileSyncMessage(command=constants.DONE))),
AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))

# Expected `bulk_write` values
expected_bulk_write = join_messages(AdbMessage(command=constants.OPEN, arg0=1, arg1=0, data=b'sync:\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(command=constants.RECV, data=b'/data'))),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1),
AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))

self.assertEqual(await self.device.pull('/data'), filedata)
self.assertEqual(self.device._transport._bulk_write, expected_bulk_write)

@awaiter
async def test_pull_file(self):
self.assertTrue(await self.device.connect())
Expand All @@ -761,31 +706,6 @@ async def test_pull_file(self):
# self.assertEqual(b''.join([bytes(call.args[0]) for call in m().write.mock_calls]), filedata)
self.assertEqual(self.device._transport._bulk_write, expected_bulk_write)

@awaiter
async def _test_pull_file_return_true(self):
self.assertTrue(await self.device.connect())
self.device._transport._bulk_write = b''

filedata = b'Ohayou sekai.\nGood morning world!'

# Provide the `bulk_read` return values
self.device._transport._bulk_read = join_messages(AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(command=constants.DATA, data=filedata),
FileSyncMessage(command=constants.DONE))),
AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))

# Expected `bulk_write` values
expected_bulk_write = join_messages(AdbMessage(command=constants.OPEN, arg0=1, arg1=0, data=b'sync:\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(command=constants.RECV, data=b'/data'))),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1),
AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))

with patch('adb_shell.adb_device_async.open', mock_open()) as m:
await self.device.pull('/data', 'TEST_FILE')
# self.assertEqual(b''.join([bytes(call.args[0]) for call in m().write.mock_calls]), filedata)
self.assertEqual(self.device._transport._bulk_write, expected_bulk_write)

@awaiter
async def test_pull_big_file(self):
self.assertTrue(await self.device.connect())
Expand All @@ -794,7 +714,6 @@ async def test_pull_big_file(self):
filedata = b'0' * int(1.5 * constants.MAX_ADB_DATA)

# Provide the `bulk_read` return values

self.device._transport._bulk_read = join_messages(AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(command=constants.DATA, data=filedata),
Expand All @@ -818,7 +737,6 @@ async def test_stat(self):
self.device._transport._bulk_write = b''

# Provide the `bulk_read` return values

self.device._transport._bulk_read = join_messages(AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncStatMessage(constants.STAT, 1, 2, 3),
Expand Down

0 comments on commit d303941

Please sign in to comment.