Repository navigation
fix(AutoCheckinBigGod): 任务退出时清理所属 Frida 进程及子进程 #1951
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,121 @@ | ||
| import subprocess | ||
| import sys | ||
| from types import SimpleNamespace | ||
| from unittest.mock import Mock | ||
|
|
||
| import pytest | ||
|
|
||
| from module.exception import TaskEnd | ||
| from tasks.AutoCheckinBigGod.script_task import ScriptTask | ||
|
|
||
|
|
||
| def task_with_process(): | ||
| task = object.__new__(ScriptTask) | ||
| task._frida_session = subprocess.Popen( | ||
| [sys.executable, '-c', 'import time; time.sleep(60)'], | ||
| stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, | ||
| creationflags=getattr(subprocess, 'CREATE_NO_WINDOW', 0)) | ||
| task._frida_attached_pid = 123 | ||
| task.frida_pid = 123 | ||
| return task | ||
|
|
||
|
|
||
| @pytest.mark.parametrize('exit_kind', ['return', 'task_end', 'error', 'interrupt']) | ||
| def test_run_reaps_owned_process_on_every_exit(exit_kind): | ||
| task = task_with_process() | ||
| proc = task._frida_session | ||
| commands = [] | ||
| task._adb_shell = lambda command: commands.append(command) | ||
| task.device = SimpleNamespace(app_start=Mock()) | ||
| task.session = SimpleNamespace(close=Mock()) | ||
|
|
||
| def checkin(): | ||
| assert proc.poll() is None # The process remains usable during the task. | ||
| task._adb_connected = True | ||
| task._restore_game_after_checkin = True | ||
| if exit_kind == 'task_end': | ||
| raise TaskEnd('AutoCheckinBigGod') | ||
| if exit_kind == 'error': | ||
| raise RuntimeError('failed') | ||
| if exit_kind == 'interrupt': | ||
| raise KeyboardInterrupt() | ||
|
|
||
| task._run_checkin = checkin | ||
| try: | ||
| if exit_kind == 'return': | ||
| task.run() | ||
| else: | ||
| expected = {'task_end': TaskEnd, 'error': RuntimeError, 'interrupt': KeyboardInterrupt}[exit_kind] | ||
| with pytest.raises(expected): | ||
| task.run() | ||
| assert proc.poll() is not None | ||
| assert all(pipe.closed for pipe in (proc.stdin, proc.stdout, proc.stderr)) | ||
| assert task._frida_session is None and task._frida_attached_pid is None | ||
| assert task.frida_pid is None | ||
| assert ['su -c "killall frida-server"'] in commands | ||
| task.device.app_start.assert_called_once() | ||
| task.session.close.assert_called_once() | ||
| finally: | ||
| if proc.poll() is None: | ||
| proc.kill() | ||
| proc.wait(timeout=3) | ||
|
|
||
|
|
||
| def test_failed_adb_check_does_not_touch_unspecified_device(): | ||
| task = object.__new__(ScriptTask) | ||
| task._run_checkin = Mock(side_effect=TaskEnd('AutoCheckinBigGod')) | ||
| task._adb_shell = Mock() | ||
|
Comment on lines
+66
to
+67
There was a problem hiding this comment. 🟡 Medium · 失败的 ADB 检查被绕过 当实际的 请测试真实的连接失败路径，并断言该路径不会发出 ADB 命令。 给 AI 代理的提示在 `tests/test_biggod_frida_cleanup.py` 第 66-67 行：
**失败的 ADB 检查被绕过**
当实际的 `_run_checkin` 在因连接检查失败而抛出异常之前访问 ADB 时，`test_failed_adb_check_does_not_touch_unspecified_device` 将 `_run_checkin` 替换为一个立即抛出异常的 mock。因此，即使实际的 ADB 失败路径在抛出异常前调用了 `_adb_shell` 并访问了未指定的设备，该测试仍会通过。
请测试真实的连接失败路径，并断言该路径不会发出 ADB 命令。Original comment in English🟡 Medium · Failed ADB check is bypassed When the actual Exercise the real failed-connection path and assert that it does not issue ADB commands. Prompt for AI agentsIn `tests/test_biggod_frida_cleanup.py` at lines 66-67:
**Failed ADB check is bypassed**
When the actual `_run_checkin` touches ADB before raising on a failed connection check, `test_failed_adb_check_does_not_touch_unspecified_device` replaces `_run_checkin` with a mock that raises immediately, so the test passes even if the real failed-ADB path calls `_adb_shell` before raising and touches an unspecified device.
Exercise the real failed-connection path and assert that it does not issue ADB commands. |
||
| with pytest.raises(TaskEnd): | ||
| task.run() | ||
| task._adb_shell.assert_not_called() | ||
|
|
||
|
|
||
| def test_stdin_failure_reaps_repl_before_fallback(): | ||
| task = object.__new__(ScriptTask) | ||
| proc = Mock() | ||
| proc.poll.return_value = None | ||
| proc.stdin.write.side_effect = BrokenPipeError() | ||
| task._frida_session = proc | ||
| task._frida_attached_pid = 123 | ||
| def fallback(*args): | ||
| proc.wait.assert_called_once() | ||
| assert task._frida_session is None | ||
| return 'fallback' | ||
| task._run_frida_script_oneshot = fallback | ||
| assert task._run_frida_script(123, 'test') == 'fallback' | ||
| proc.terminate.assert_called_once() | ||
|
|
||
|
|
||
| @pytest.mark.parametrize('error', [OSError('failed'), subprocess.TimeoutExpired('frida', 10)]) | ||
| def test_oneshot_exception_reaps_process(monkeypatch, error): | ||
| task = object.__new__(ScriptTask) | ||
| task._get_frida_exe = lambda: 'frida' | ||
| proc = Mock() | ||
| proc.poll.return_value = None | ||
| proc.communicate.side_effect = error | ||
| monkeypatch.setattr('tasks.AutoCheckinBigGod.script_task.subprocess.Popen', lambda *a, **kw: proc) | ||
| monkeypatch.setattr('tasks.AutoCheckinBigGod.script_task.time.sleep', lambda seconds: None) | ||
| assert task._run_frida_script_oneshot(123, 'test') is None | ||
| proc.terminate.assert_called_once() | ||
| proc.wait.assert_called_once() | ||
|
|
||
|
|
||
| def test_unresponsive_process_is_killed_and_reaped(): | ||
| proc = Mock() | ||
| proc.poll.return_value = None | ||
| proc.wait.side_effect = [subprocess.TimeoutExpired('frida', 3), 0] | ||
| ScriptTask._stop_frida_process(proc) | ||
| proc.terminate.assert_called_once() | ||
| proc.kill.assert_called_once() | ||
| assert proc.wait.call_count == 2 | ||
|
|
||
|
|
||
| @pytest.mark.skipif(sys.platform != 'win32', reason='Windows launcher process tree') | ||
| def test_windows_cleanup_targets_only_owned_process_tree(monkeypatch): | ||
| proc = Mock(pid=12345) | ||
| proc.poll.return_value = None | ||
| taskkill = Mock() | ||
| monkeypatch.setattr('tasks.AutoCheckinBigGod.script_task.subprocess.run', taskkill) | ||
| ScriptTask._stop_frida_process(proc) | ||
| assert taskkill.call_args.args[0] == ['taskkill', '/PID', '12345', '/T', '/F'] | ||
| proc.wait.assert_called_once() | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🟡 Medium · 其他 Frida 客户端会失去服务器
当此任务退出时，如果另一个客户端正在同一 ADB 设备上使用已运行的 Frida 服务器，
_cleanup会执行killall frida-server，却不检查该服务器是否由此任务启动，从而断开另一个客户端并导致其 Frida 操作失败。只有在此任务启动了 Frida 服务器时，才停止该服务器。
给 AI 代理的提示
Original comment in English
🟡 Medium · Other Frida clients lose their server
When another client uses a Frida server already running on the same ADB device when this task exits,
_cleanuprunskillall frida-serverwithout checking whether this task started the server, disconnecting the other client and causing its Frida operations to fail.Only stop the Frida server when this task started it.
Prompt for AI agents