From d303941a41bc31f6f849ddc132317c1c98e11bed Mon Sep 17 00:00:00 2001 From: Jeff Irion Date: Mon, 29 Jun 2020 20:01:32 -0700 Subject: [PATCH] Cleanup tests --- tests/test_adb_device.py | 78 -------------------------------- tests/test_adb_device_async.py | 82 ---------------------------------- 2 files changed, 160 deletions(-) diff --git a/tests/test_adb_device.py b/tests/test_adb_device.py index 2eef96c..978c037 100644 --- a/tests/test_adb_device.py +++ b/tests/test_adb_device.py @@ -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'' @@ -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) @@ -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'' @@ -716,30 +664,6 @@ 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'' @@ -747,7 +671,6 @@ 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), @@ -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), diff --git a/tests/test_adb_device_async.py b/tests/test_adb_device_async.py index 9a689e5..5c177fe 100644 --- a/tests/test_adb_device_async.py +++ b/tests/test_adb_device_async.py @@ -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()) @@ -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()) @@ -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()) @@ -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), @@ -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),