Refactor script to fix issue #1 and #2

.github/workflows/run_test.sh

@@ -0,0 +1,46 @@
+#!/bin/sh
+set -eu
+
+applicationId=com.viliussutkus89.adb.pull.as
+
+adb shell am start -W $applicationId/$applicationId.MainActivity
+echo
+
+echo "Pulling a single file into current directory"
+./src/adbPullAs/__init__.py $applicationId /data/data/$applicationId/cache/test/activityScreenshot.png
+rm activityScreenshot.png
+echo
+
+echo "Pulling a single file into specified directory"
+mkdir test_tmp
+./src/adbPullAs/__init__.py $applicationId /data/data/$applicationId/cache/test/activityScreenshot.png test_tmp
+rm test_tmp/activityScreenshot.png
+rmdir test_tmp
+echo
+
+echo "Pulling directory into current directory"
+./src/adbPullAs/__init__.py $applicationId /data/data/$applicationId/cache/test
+rm test/activityScreenshot.png
+rm test/subDirectory/file1.txt
+rm test/subDirectory/file2.txt
+rmdir test/subDirectory
+rmdir test
+echo
+
+echo "Pulling directory into specified directory"
+mkdir test_tmp
+./src/adbPullAs/__init__.py $applicationId /data/data/$applicationId/cache/test test_tmp
+rm test_tmp/test/activityScreenshot.png
+rm test_tmp/test/subDirectory/file1.txt
+rm test_tmp/test/subDirectory/file2.txt
+rmdir test_tmp/test/subDirectory
+rmdir test_tmp/test
+rmdir test_tmp
+echo
+
+echo "Asking for the same file twice"
+mkdir test_tmp
+./src/adbPullAs/__init__.py $applicationId /data/data/$applicationId/cache/test/activityScreenshot.png /data/data/$applicationId/cache/test/activityScreenshot.png test_tmp
+rm test_tmp/activityScreenshot.png
+rmdir test_tmp
+echo

.github/workflows/test.yml

@@ -143,10 +143,9 @@ jobs:
             mkdir -p testResults
             adb logcat -c || true
             adb logcat > testResults/logcat.txt &
+
             adb install app-debug.apk
-            adb shell am start -W $applicationId/$applicationId.MainActivity
-            ./src/adbPullAs/__init__.py $applicationId /data/data/$applicationId testResults
-            test -f testResults/cache/activityScreenshot.png
+            .github/workflows/run_test.sh
 
       - uses: actions/upload-artifact@v3
         if: always()

TestApplication/app/src/main/java/com/viliussutkus89/adb/pull/as/MainActivity.kt

@@ -13,15 +13,25 @@ import java.io.FileOutputStream
 class MainActivity : AppCompatActivity(R.layout.activity_main) {
     override fun onResume() {
         super.onResume()
+
+        val testDir = File(cacheDir, "test")
         window.decorView.rootView.doOnLayout { view ->
             val bitmap = Bitmap.createBitmap(view.width, view.height, Bitmap.Config.ARGB_8888)
             view.draw(Canvas(bitmap))
-            val screenshotFile = File(cacheDir, "activityScreenshot.png")
+            val screenshotFile = File(testDir, "activityScreenshot.png")
             lifecycleScope.launch(Dispatchers.IO) {
                 FileOutputStream(screenshotFile).use { out ->
                     bitmap.compress(Bitmap.CompressFormat.PNG, 50, out)
                 }
             }
         }
+        lifecycleScope.launch(Dispatchers.IO) {
+            val subDirectory = File(testDir, "subDirectory").also { it.mkdirs() }
+            for (f in arrayOf("file1.txt", "file2.txt")) {
+                FileOutputStream(File(subDirectory, f)).use { out ->
+                    out.write("TEST".toByteArray())
+                }
+            }
+        }
     }
 }

src/adbPullAs/__init__.py

@@ -22,19 +22,23 @@
 import pathlib
 import subprocess
 import sys
-from pathlib import PurePosixPath, PurePath
+from pathlib import PurePosixPath
 
 
 class FsTest:
-    def __init__(self, package_name):
-        self.package_name = package_name
+    def __init__(self, package_name=None):
+        self.__package_name = package_name
 
     def __implementation(self, test, path):
-        # old Androids exit "adb shell false" with exit code 0... work this around
-        inner_command = f'"{test} {path}" || echo NOT'
-        return not subprocess.run(('adb', 'shell', 'run-as', self.package_name,
-                                   'sh', '-c', inner_command),
-                                  capture_output=True, text=True).stdout
+        args = ('adb', 'shell')
+        if self.__package_name:
+            args += ('run-as', self.__package_name)
+
+        # sh -c || echo NOT needed because
+        # old Androids exit "adb shell false" with exit code 0... workaround this
+        args += ('sh', '-c', f'"{test} {path}" || echo NOT')
+
+        return not subprocess.run(args, capture_output=True, text=True).stdout
 
     def exists(self, path):
         return self.__implementation('test -e', path)
@@ -46,75 +50,69 @@ def is_file(self, path):
         return self.__implementation('test -f', path)
 
 
-class AdbPullAs:
-    def __init__(self, package_name, local):
-        self.fs_test = FsTest(package_name)
-        self.tmp_dir = self.__mkdir_unique('/data/local/tmp/adbPullAs')
-        self.package_name = package_name
-        self.run_as_package = ('adb', 'shell', 'run-as', self.package_name)
-        self.local = local
-
-    def __del__(self):
-        subprocess.run(('adb', 'shell', 'rmdir', self.tmp_dir))
-
+class TmpDir:
     @staticmethod
-    def __mkdir_unique(suggested_path):
+    def mkdir_unique(location, name_prefix):
         i = 0
-        path = suggested_path
-        while len(subprocess.run(('adb', 'shell', 'mkdir', path), capture_output=True).stdout):
+        directory = location / name_prefix
+        while len(subprocess.run(('adb', 'shell', 'mkdir', directory), capture_output=True).stdout):
+            directory = location / f'{name_prefix}-{i}'
             i += 1
-            path = f'{suggested_path}-{i}'
-        return path
+        return directory
 
-    def __pull_file(self, remote, local_dir):
-        filename = remote.name
-        device_tmp_file = PurePosixPath(self.tmp_dir, filename)
+    def __init__(self):
+        self.dir = self.mkdir_unique(PurePosixPath('/data/local/tmp'), 'adbPullAs')
+
+    def __del__(self):
+        if self.dir:
+            subprocess.run(('adb', 'shell', 'rm', '-r', self.dir))
 
-        # cat-pipe from package private to /data/local/tmp
-        subprocess.run(self.run_as_package + ('cat', remote, '>', device_tmp_file), shell=False)
 
-        subprocess.run(('adb', 'pull', device_tmp_file, local_dir))
-        subprocess.run(('adb', 'shell', 'rm', device_tmp_file))
+class AdbPullAs:
+    def __init__(self, package_name):
+        self.__package_fs_test = FsTest(package_name)
+        self.__regular_fs_test = FsTest()
+        self.__run_as_package_args = ('adb', 'shell', 'run-as', package_name)
+
+    def __cache_target(self, remote, cached_remote):
+        ret_val = True
+        if self.__package_fs_test.is_file(remote):
+            subprocess.run(self.__run_as_package_args + ('cat', remote, '>', cached_remote), shell=False)
+        elif self.__package_fs_test.is_directory(remote):
+            if not self.__regular_fs_test.is_directory(cached_remote):
+                subprocess.run(('adb', 'shell', 'mkdir', cached_remote))
+            ls_proc = subprocess.run(self.__run_as_package_args + ('ls', f'{remote}/'), capture_output=True, text=True)
+            for dir_entry in ls_proc.stdout.splitlines():
+                ret_val &= self.__cache_target(
+                    remote / dir_entry,
+                    cached_remote / dir_entry
+                )
+        else:
+            print('Unknown entity type', remote)
+            ret_val = False
+        return ret_val
 
-    def __pull(self, remote_entry_point, sub_items=()):
-        remote_dir = remote_entry_point
-        local_dir = self.local
-        for s in sub_items:
-            remote_dir = PurePosixPath(remote_dir, s)
-            local_dir = PurePath(local_dir, s)
+    def pull(self, remotes, local_destination) -> bool:
+        ret_val = True
+        pull_args = ('adb', 'pull')
 
-        print('Pulling', remote_dir, '->', local_dir)
+        tmp_dir = TmpDir()
+        for remote in remotes:
+            remote = PurePosixPath(remote)
+            cached_remote = TmpDir.mkdir_unique(tmp_dir.dir, 'pull') / remote.name
+            pull_args += (cached_remote, )
+            ret_val &= self.__cache_target(remote, cached_remote)
 
-        if not self.fs_test.exists(remote_dir):
-            print(remote_dir, 'does not exist!', file=sys.stderr)
-            return False
+        if local_destination:
+            pull_args += (local_destination, )
 
-        elif self.fs_test.is_directory(remote_dir):
-            if sub_items:
-                try:
-                    os.mkdir(local_dir)
-                except FileExistsError:
-                    pass
+        ret_val &= subprocess.run(pull_args).returncode == 0
 
-            ls_proc = subprocess.run(self.run_as_package + ('ls', f'{remote_dir}/'), capture_output=True, text=True)
-            for dir_entry in ls_proc.stdout.splitlines():
-                dir_entry_file = PurePosixPath(remote_dir, dir_entry)
-                if self.fs_test.is_file(dir_entry_file):
-                    self.__pull_file(dir_entry_file, local_dir)
-                else:
-                    self.__pull(remote_entry_point, sub_items + (dir_entry, ))
-
-        return True
-
-    def pull(self, remote):
-        remote = PurePosixPath(remote)
-        if self.fs_test.is_file(remote):
-            return self.__pull_file(remote, self.local)
-        else:
-            return self.__pull(remote)
+        return ret_val
 
 
 def main():
+    args = []
     try:
         opts, args = getopt.getopt(sys.argv[1:], 'hv', ['help', 'version'])
         for o, a in opts:
@@ -129,28 +127,26 @@ def main():
         print_usage(sys.stderr)
         exit(os.EX_USAGE)
 
-    args = sys.argv[1:]
-    if 2 > len(args):
+    package_name = None
+    try:
+        package_name = args.pop(0)
+    except IndexError:
         print_usage(sys.stderr)
         exit(os.EX_USAGE)
 
-    package_name = args.pop(0)
-
-    first_remote = args.pop(0)
-    local = args.pop() if len(args) else '.'
-    args.insert(0, first_remote)
-
-    if not os.path.isdir(local):
-        print(f'"{local}" - COMPUTER_DESTINATION_DIR is not a directory', file=sys.stderr)
-        exit(os.EX_NOINPUT)
+    local = args.pop() if len(args) > 1 else None
+    remotes = args
+    if local:
+        if not os.path.exists(local):
+            print(local, '- COMPUTER_DESTINATION_DIR does not exist!', file=sys.stderr)
+            exit(os.EX_NOINPUT)
 
-    apa = AdbPullAs(package_name, local)
-    exit_code = os.EX_OK
-    for remote in args:
-        if not apa.pull(remote):
-            exit_code = os.EX_IOERR
+        if not os.path.isdir(local):
+            print(local, '- COMPUTER_DESTINATION_DIR is not a directory!', file=sys.stderr)
+            exit(os.EX_NOINPUT)
 
-    exit(exit_code)
+    pull_status = AdbPullAs(package_name).pull(remotes, local)
+    exit(os.EX_OK if pull_status else os.EX_IOERR)
 
 
 def print_usage(output_to=sys.stdout):