fix: scope Android cleanup to test executables

Author: TheQmaks

scripts/android_smoke.py

@@ -655,47 +655,83 @@ def _exercise_gadget(
     }
 
 
+def _parse_remote_processes(output: str, remote_executables: Sequence[str]) -> dict[str, list[int]]:
+    matches: dict[str, list[int]] = {path: [] for path in remote_executables}
+    for line in output.splitlines():
+        match = re.search(r"/proc/(\d+)/exe -> (.+)$", line)
+        if match is None:
+            continue
+        target = match.group(2).removesuffix(" (deleted)")
+        if target in matches:
+            matches[target].append(int(match.group(1)))
+    return matches
+
+
+def _remote_signal_command(remote_executable: str, pid: int, signal: str) -> str:
+    body = (
+        f"target=$(readlink /proc/{pid}/exe 2>/dev/null) || exit 0; "
+        f'case "$target" in "{remote_executable}"|"{remote_executable} (deleted)") '
+        f"kill -{signal} {pid} || exit 1 ;; esac; exit 0"
+    )
+    return f"'{body}'"
+
+
 def _cleanup(config: AndroidSmokeConfig, serial: str) -> None:
     processes = (
-        (f"{config.name}-server", f"{config.name}-server"[:15]),
-        ("gadget-loader", "gadget-loader"),
+        (f"{config.name}-server", f"{REMOTE_DIR}/{config.name}-server"),
+        ("gadget-loader", f"{REMOTE_DIR}/gadget-loader"),
     )
     failures: list[str] = []
-    signaled = False
+    remote_executables = tuple(path for _label, path in processes)
 
-    for label, process_comm in processes:
+    def snapshot(stage: str) -> dict[str, list[int]] | None:
         result = root_shell(
             serial,
-            f"pkill -TERM -x {process_comm}",
+            "'ls -l /proc/[0-9]*/exe 2>/dev/null; exit 0'",
             check=False,
         )
-        signaled = signaled or result.returncode == 0
-        if result.returncode not in (0, 1):
-            failures.append(f"stop {label} (exit {result.returncode})")
-
-    if signaled:
-        time.sleep(0.5)
-
-    forced: list[tuple[str, str]] = []
-    for label, process_comm in processes:
-        result = root_shell(serial, f"pgrep -x {process_comm}", check=False)
-        if result.returncode == 0:
-            forced.append((label, process_comm))
-            kill_result = root_shell(serial, f"pkill -KILL -x {process_comm}", check=False)
-            if kill_result.returncode not in (0, 1):
-                failures.append(f"force-stop {label} (exit {kill_result.returncode})")
-        elif result.returncode != 1:
-            failures.append(f"verify {label} (exit {result.returncode})")
-
-    if forced:
-        failures.append("processes did not stop gracefully: " + ", ".join(x[0] for x in forced))
+        if result.returncode != 0:
+            failures.append(f"{stage} (exit {result.returncode})")
+            return None
+        return _parse_remote_processes(result.stdout, remote_executables)
+
+    initial = snapshot("list Android smoke processes")
+    if initial is not None:
+        for label, remote_executable in processes:
+            for pid in initial[remote_executable]:
+                result = root_shell(
+                    serial,
+                    _remote_signal_command(remote_executable, pid, "TERM"),
+                    check=False,
+                )
+                if result.returncode != 0:
+                    failures.append(f"stop {label} (exit {result.returncode})")
+        if any(initial.values()):
+            time.sleep(0.5)
+
+    remaining = snapshot("verify Android smoke processes")
+    forced_labels: list[str] = []
+    if remaining is not None:
+        for label, remote_executable in processes:
+            for pid in remaining[remote_executable]:
+                if label not in forced_labels:
+                    forced_labels.append(label)
+                result = root_shell(
+                    serial,
+                    _remote_signal_command(remote_executable, pid, "KILL"),
+                    check=False,
+                )
+                if result.returncode != 0:
+                    failures.append(f"force-stop {label} (exit {result.returncode})")
+
+    if forced_labels:
+        failures.append("processes did not stop gracefully: " + ", ".join(forced_labels))
         time.sleep(0.2)
-        for label, process_comm in forced:
-            result = root_shell(serial, f"pgrep -x {process_comm}", check=False)
-            if result.returncode == 0:
-                failures.append(f"process remains: {label}")
-            elif result.returncode != 1:
-                failures.append(f"verify {label} after SIGKILL (exit {result.returncode})")
+        final = snapshot("verify Android smoke processes after SIGKILL")
+        if final is not None:
+            for label, remote_executable in processes:
+                if final[remote_executable]:
+                    failures.append(f"process remains: {label}")
 
     remove_result = root_shell(serial, f"rm -rf -- {REMOTE_DIR}", check=False)
     if remove_result.returncode != 0:

tests/test_android_smoke.py

@@ -542,12 +542,24 @@ def fail_run(command: list[str], **_kwargs: object) -> subprocess.CompletedProce
 def test_cleanup_removes_remote_test_directory(monkeypatch: pytest.MonkeyPatch) -> None:
     root_commands: list[str] = []
     adb_commands: list[tuple[str, ...]] = []
+    snapshot_calls = 0
 
     def fake_root_shell(_serial: str, command: str, **_kwargs: object) -> SimpleNamespace:
+        nonlocal snapshot_calls
         root_commands.append(command)
+        stdout = ""
+        if "/proc/[0-9]*/exe" in command:
+            snapshot_calls += 1
+            if snapshot_calls == 1:
+                stdout = (
+                    "lrwxrwxrwx root /proc/101/exe -> "
+                    "/data/local/tmp/phantom-frida-test/oemcodec-server\n"
+                    "lrwxrwxrwx root /proc/102/exe -> "
+                    "/data/local/tmp/phantom-frida-test/gadget-loader\n"
+                )
         return SimpleNamespace(
             returncode=1 if command.startswith(("pkill", "pgrep")) else 0,
-            stdout="",
+            stdout=stdout,
             stderr="",
         )
 
@@ -557,12 +569,21 @@ def fake_adb(_serial: str, *arguments: str, **_kwargs: object) -> SimpleNamespac
 
     monkeypatch.setattr(android_smoke, "root_shell", fake_root_shell)
     monkeypatch.setattr(android_smoke, "adb", fake_adb)
+    monkeypatch.setattr(android_smoke.time, "sleep", lambda _seconds: None)
     config = SimpleNamespace(name="oemcodec", port=27142)
 
     android_smoke._cleanup(config, "SERIAL-1")
 
-    assert "pkill -TERM -x oemcodec-server" in root_commands
-    assert "pkill -TERM -x gadget-loader" in root_commands
+    term_commands = [command for command in root_commands if "kill -TERM" in command]
+    assert len(term_commands) == 2
+    assert all("readlink /proc/" in command for command in term_commands)
+    assert any(
+        "/data/local/tmp/phantom-frida-test/oemcodec-server" in command for command in term_commands
+    )
+    assert any(
+        "/data/local/tmp/phantom-frida-test/gadget-loader" in command for command in term_commands
+    )
+    assert not any(command.startswith("pkill") for command in root_commands)
     assert "rm -rf -- /data/local/tmp/phantom-frida-test" in root_commands
     assert ("forward", "--list") in adb_commands
 
@@ -594,8 +615,8 @@ def fake_adb(_serial: str, *arguments: str, **_kwargs: object) -> SimpleNamespac
     with pytest.raises(android_smoke.SmokeFailure, match="remove remote test directory"):
         android_smoke._cleanup(config, "SERIAL-1")
 
-    assert len([command for command in root_commands if command.startswith("pkill")]) == 2
-    assert len([command for command in root_commands if command.startswith("pgrep")]) == 2
+    assert len([command for command in root_commands if "/proc/[0-9]*/exe" in command]) == 2
+    assert not any(command.startswith(("pkill", "pgrep")) for command in root_commands)
     assert ("forward", "--remove", "tcp:27142") in adb_commands
     assert ("forward", "--remove", "tcp:27143") in adb_commands
     assert ("forward", "--list") in adb_commands
@@ -650,27 +671,25 @@ def fake_root_shell(_serial: str, command: str, **_kwargs: object) -> SimpleName
         android_smoke._cleanup(config, "SERIAL-1")
 
 
-def test_cleanup_uses_kernel_comm_limit_for_long_server_name(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    root_commands: list[str] = []
+def test_remote_signal_is_scoped_to_exact_long_name_executable() -> None:
+    remote_executable = "/data/local/tmp/phantom-frida-test/abcdefghijklmnopqrst-server"
 
-    def fake_root_shell(_serial: str, command: str, **_kwargs: object) -> SimpleNamespace:
-        root_commands.append(command)
-        return SimpleNamespace(
-            returncode=1 if command.startswith(("pkill", "pgrep")) else 0,
-            stdout="",
-            stderr="",
-        )
+    command = android_smoke._remote_signal_command(remote_executable, 123, "TERM")
 
-    monkeypatch.setattr(android_smoke, "root_shell", fake_root_shell)
-    monkeypatch.setattr(
-        android_smoke,
-        "adb",
-        lambda *_args, **_kwargs: SimpleNamespace(returncode=0, stdout="", stderr=""),
-    )
-    config = SimpleNamespace(name="abcdefghijklmnopqrst", port=27142)
+    assert "readlink /proc/123/exe" in command
+    assert remote_executable in command
+    assert "kill -TERM 123" in command
+    assert "pkill" not in command
 
-    android_smoke._cleanup(config, "SERIAL-1")
 
-    assert "pkill -TERM -x abcdefghijklmno" in root_commands
+def test_parse_remote_processes_excludes_same_basename_outside_test_directory() -> None:
+    remote_executable = "/data/local/tmp/phantom-frida-test/oemcodec-server"
+    output = (
+        f"lrwxrwxrwx root /proc/101/exe -> {remote_executable}\n"
+        f"lrwxrwxrwx root /proc/102/exe -> {remote_executable} (deleted)\n"
+        "lrwxrwxrwx root /proc/103/exe -> /other/path/oemcodec-server\n"
+    )
+
+    assert android_smoke._parse_remote_processes(output, (remote_executable,)) == {
+        remote_executable: [101, 102]
+    }