import importlib.util import base64 import json from pathlib import Path import subprocess import sys import tempfile import unittest from unittest import mock RUNNER_PATH = Path(__file__).parents[1] / "runner.py" SPEC = importlib.util.spec_from_file_location("legado_android_runner", RUNNER_PATH) runner = importlib.util.module_from_spec(SPEC) assert SPEC.loader is not None SPEC.loader.exec_module(runner) class ScenarioValidationTest(unittest.TestCase): def test_repository_scenario_is_valid(self): scenario = runner.load_scenario("reader/page-turn") self.assertEqual("long-chapter-v1", scenario["fixture"]) def test_bookshelf_scenario_is_valid(self): scenario = runner.load_scenario("bookshelf/open-reader-back") self.assertEqual("bookshelf", scenario["entry"]["type"]) def test_source_edit_scenario_is_valid(self): scenario = runner.load_scenario("source-edit/predictive-back") self.assertEqual("source_manage", scenario["entry"]["type"]) def test_theme_save_baseline_scenario_is_valid(self): scenario = runner.load_scenario("theme/save-current-baseline") self.assertEqual("theme_config", scenario["entry"]["type"]) def test_theme_export_baseline_scenario_is_valid(self): scenario = runner.load_scenario("theme/export-current-baseline") self.assertEqual("theme_config", scenario["entry"]["type"]) def test_theme_background_scenarios_are_valid(self): for name in ( "theme/select-day-background", "theme/save-background-after-restart", "theme/export-background-after-restart", ): with self.subTest(name=name): scenario = runner.load_scenario(name) self.assertEqual("theme_config", scenario["entry"]["type"]) def test_theme_failure_scenarios_are_valid(self): for name in ( "theme/save-missing-file", "theme/save-missing-content-uri", "theme/export-missing-file", "theme/export-missing-content-uri", ): with self.subTest(name=name): scenario = runner.load_scenario(name) self.assertIn(scenario["themeAsset"], {"missing-file", "missing-content-uri"}) def test_deleted_applied_theme_export_scenario_is_valid(self): scenario = runner.load_scenario("theme/export-current-after-deleting-applied-theme") self.assertNotIn("expectThemeExportFailure", scenario) def test_rss_day_night_scenario_is_valid(self): self.assertEqual( "rss_read", runner.load_scenario("rss/read-day-night")["entry"]["type"], ) def test_rejects_unknown_action(self): scenario = { "schemaVersion": 1, "name": "reader/page-turn", "fixture": "long-chapter-v1", "entry": {"type": "reader"}, "readyMarker": "marker", "actions": [{"type": "tap"}], "assertions": ["reader_ready"], } with self.assertRaises(runner.ScenarioError): runner.validate_scenario(scenario) def test_rejects_missing_fixture(self): with self.assertRaises(runner.ScenarioError): runner.validate_fixture("missing-v1") def test_rejects_invalid_fixture_before_device_work(self): fixture = { "schemaVersion": 1, "id": "broken-v1", "resetPolicy": "replace", "book": { "bookUrl": "debug://book", "name": "broken", "author": "debug", "pageMode": "simulation", "startChapter": 0, "startPosition": 0, }, "chapters": [{"title": "chapter", "readyMarker": "ready", "paragraph": "text", "repeat": 0}], } with self.assertRaises(runner.ScenarioError): runner.validate_fixture_data(fixture, "broken-v1") class CapabilityTest(unittest.TestCase): def test_lists_registered_entries_actions_assertions(self): caps = runner.build_capabilities() self.assertEqual(["turn_page"], caps["entries"]["reader"]["allowedActions"]) self.assertEqual( ["open_fixture_book", "press_back"], caps["entries"]["bookshelf"]["allowedActions"], ) self.assertEqual( ["open_fixture_source", "predictive_back"], caps["entries"]["source_manage"]["allowedActions"], ) self.assertEqual( [ "apply_saved_theme", "delete_saved_theme", "export_current_theme", "open_theme_manage", "save_current_theme", "select_day_background", ], caps["entries"]["theme_config"]["allowedActions"], ) self.assertIn("turn_page", caps["actions"]) self.assertIn("no_fatal_exception", caps["assertions"]) def test_list_command_returns_zero(self): self.assertEqual(0, runner.main(["list"])) def test_parses_skip_build_flag(self): args = runner.parse_args(["run", "reader/page-turn", "--skip-build"]) self.assertTrue(args.skip_build) def test_run_without_skip_build_defaults_false(self): args = runner.parse_args(["run", "reader/page-turn"]) self.assertFalse(args.skip_build) class PreferencesTest(unittest.TestCase): BASE = { "schemaVersion": 1, "name": "bookshelf/miuix-hidden-statusbar", "fixture": "long-chapter-v1", "entry": {"type": "bookshelf"}, "readyMarker": "marker", "record": True, "actions": [{"type": "open_fixture_book"}, {"type": "press_back"}], "assertions": ["bookshelf_ready"], } def test_repository_miuix_scenario_is_valid(self): scenario = runner.load_scenario("bookshelf/miuix-hidden-statusbar") self.assertTrue(scenario["record"]) self.assertEqual("miuix", scenario["preferences"]["composeEngine"]) self.assertTrue(scenario["preferences"]["hideStatusBar"]) def test_accepts_string_and_boolean_preferences(self): scenario = {**self.BASE, "preferences": {"composeEngine": "miuix", "hideStatusBar": True}} runner.validate_scenario(scenario) def test_rejects_non_primitive_preference_value(self): scenario = {**self.BASE, "preferences": {"menuAlpha": 50}} with self.assertRaises(runner.ScenarioError): runner.validate_scenario(scenario) def test_rejects_invalid_preference_key(self): scenario = {**self.BASE, "preferences": {"bad-key": "x"}} with self.assertRaises(runner.ScenarioError): runner.validate_scenario(scenario) def test_rejects_non_boolean_record(self): scenario = {**self.BASE, "record": "yes"} with self.assertRaises(runner.ScenarioError): runner.validate_scenario(scenario) def test_rejects_unknown_theme_asset_fixture(self): scenario = runner.load_scenario("theme/save-background-after-restart") scenario["themeAsset"] = "guess" with self.assertRaises(runner.ScenarioError): runner.validate_scenario(scenario) class FailureParsingTest(unittest.TestCase): def test_ignores_crash_from_another_package(self): logcat = """FATAL EXCEPTION: main Process: com.example.other, PID: 42 java.lang.IllegalStateException: unrelated """ self.assertEqual("", runner.extract_crash(logcat)) def test_parses_app_frame_as_inference(self): crash = """FATAL EXCEPTION: main java.lang.IndexOutOfBoundsException: Index 18 out of bounds at io.legado.app.ui.book.read.ReadBookViewModel.loadChapter(ReadBookViewModel.kt:418) """ failure = runner.parse_failure(crash) self.assertEqual("uncaught_exception", failure["category"]) self.assertEqual("IndexOutOfBoundsException", failure["exceptionType"]) self.assertEqual("medium", failure["suspectedFrame"]["confidence"]) self.assertEqual(418, failure["suspectedFrame"]["line"]) def test_prefers_deepest_caused_by_as_root_cause_candidate(self): crash = """FATAL EXCEPTION: main java.lang.RuntimeException: wrapper Caused by: java.lang.IllegalStateException: middle Caused by: java.lang.IndexOutOfBoundsException: root """ failure = runner.parse_failure(crash) self.assertEqual("IndexOutOfBoundsException", failure["exceptionType"]) self.assertEqual("root", failure["message"]) self.assertEqual(["fatal_exception"], failure["detectedSignals"]) def test_crash_has_priority_over_assertion_failure(self): scenario = runner.load_scenario("reader/page-turn") summary, exit_code = runner.make_summary( session_id="test", scenario=scenario, fixture="long-chapter-v1", build={"status": "passed"}, ready=False, activity_visible=False, content_changed=False, action_count=0, failure={"category": "uncaught_exception", "stacktraceArtifact": "crash.txt"}, ) self.assertEqual(runner.EXIT_CRASH, exit_code) self.assertEqual(runner.EXIT_CRASH, summary["exitCode"]) class CommandDecodingTest(unittest.TestCase): def test_replaces_invalid_utf8_in_command_output(self): result = runner.run_command( [sys.executable, "-c", "import sys; sys.stdout.buffer.write(b'log\\xc0tail')"], ) self.assertEqual("log\ufffdtail", result.stdout) @mock.patch.object(runner, "cleanup_device") @mock.patch.object(runner, "run_command") def test_install_retries_with_streaming_after_timeout(self, run_command, cleanup_device): run_command.side_effect = [ subprocess.TimeoutExpired(["adb", "install"], 120), subprocess.CompletedProcess(["adb", "install"], 0, "Success\n"), ] with tempfile.TemporaryDirectory() as directory: mode = runner.install_apk("device", Path("app.apk"), Path(directory)) install_log = (Path(directory) / "install.log").read_text(encoding="utf-8") self.assertEqual("streaming_retry", mode) self.assertIn("timeout=120s", install_log) cleanup_device.assert_called_once_with("device") class ReaderContentTest(unittest.TestCase): @mock.patch.object(runner, "adb") def test_decodes_debug_state_provider_response(self, adb): payload = base64.b64encode(json.dumps({"themeMode": "2"}).encode()).decode() adb.return_value = subprocess.CompletedProcess([], 0, f"Result: Bundle[{{resultB64={payload}}}]\n") self.assertEqual({"themeMode": "2"}, runner.debug_state("device", "toggleTheme")) def test_extracts_reader_pages_from_logcat(self): logcat = ( "07-21 23:52 26863 26863 I LegadoDebug: [session=x] FIXTURE_READY fixture=y entry=reader\n" "07-21 23:52 26863 26863 I LegadoDebug: READER_PAGE LEGADO_DEBUG_LONG_CHAPTER_V1 page one\n" "07-21 23:52 26863 26863 I LegadoDebug: READER_PAGE page two body\n" ) self.assertEqual( ["LEGADO_DEBUG_LONG_CHAPTER_V1 page one", "page two body"], runner.extract_reader_pages(logcat), ) def test_returns_empty_when_no_reader_page_logged(self): self.assertEqual([], runner.extract_reader_pages("nothing here\n")) @mock.patch.object(runner.time, "sleep") @mock.patch.object(runner, "reader_current_page") def test_waits_until_reader_content_changes(self, reader_current_page, _sleep): reader_current_page.side_effect = ["before", "after"] changed, latest, content = runner.wait_until_reader_content_changed("device", "before", 5) self.assertTrue(changed) self.assertEqual("after", content) self.assertEqual("after", latest) self.assertEqual(2, reader_current_page.call_count) def test_finds_clickable_bookshelf_item_and_center(self): dump = """ """ node = runner.find_node_by_content_prefix(dump, "调试夹具：长章节") self.assertIsNotNone(node) self.assertEqual((200, 400), runner.node_center(node)) def test_finds_node_in_external_picker_package(self): dump = """ """ node = runner.find_node_by_content_prefix( dump, "更多", "com.google.android.photopicker", ) self.assertEqual((996, 825), runner.node_center(node)) def test_finds_edit_button_in_same_source_row(self): dump = """ """ source = runner.find_node_by_text(dump, "Legado Debug Fixture") edit = runner.find_node_in_same_row( dump, source, "io.legato.kazusa.debug:id/iv_edit", ) self.assertEqual((816, 415), runner.node_center(edit)) class EntryExecutionTest(unittest.TestCase): def make_context(self): ctx = mock.Mock() ctx.device = "device" ctx.scenario = {"readyTimeoutSeconds": 30} ctx.records = {} ctx.state = {} ctx.metadata = {} return ctx @mock.patch.object(runner, "find_node_in_same_row") @mock.patch.object(runner, "find_node_by_text") @mock.patch.object(runner, "ui_dump", return_value="source dump") @mock.patch.object(runner, "wait_until_activity_resumed", return_value=(True, "source component")) def test_source_manage_prepare_owns_its_state( self, _wait, _dump, find_text, find_row, ): find_text.return_value = {"bounds": "[1,1][2,2]"} find_row.return_value = {"bounds": "[3,3][4,4]"} ctx = self.make_context() runner.ENTRIES["source_manage"].prepare(ctx) self.assertTrue(ctx.records["source_manage_ready"]) self.assertEqual("Legado Debug Fixture", ctx.state["source_name"]) @mock.patch.object(runner, "wait_until_text_present", return_value=(True, "theme dump", {})) def test_theme_config_prepare_owns_its_state(self, _wait): ctx = self.make_context() runner.ENTRIES["theme_config"].prepare(ctx) self.assertTrue(ctx.records["theme_config_ready"]) self.assertFalse(ctx.records["theme_save_succeeded"]) self.assertEqual("theme dump", ctx.records["ui_before"]) class SessionLogTest(unittest.TestCase): @mock.patch.object(runner, "adb") def test_collects_each_observed_application_pid(self, adb): adb.side_effect = [ subprocess.CompletedProcess([], 0, "full log\n[session=s1] ready\n"), subprocess.CompletedProcess([], 0, "pid 10 log\n"), subprocess.CompletedProcess([], 0, "pid 11 log\n"), ] with tempfile.TemporaryDirectory() as directory: runner.collect_logcat("device", Path(directory), ["10", "11", "10"], "s1") session_log = (Path(directory) / "session-logcat.txt").read_text(encoding="utf-8") self.assertIn("--- pid=10 ---", session_log) self.assertIn("--- pid=11 ---", session_log) self.assertIn("[session=s1] ready", session_log) self.assertEqual(3, adb.call_count) class DeviceStateTest(unittest.TestCase): @mock.patch.object(runner, "adb") def test_detects_awake_device(self, adb): adb.return_value = subprocess.CompletedProcess([], 0, " mWakefulness=Awake\n") self.assertTrue(runner.device_is_awake("device")) @mock.patch.object(runner, "adb") def test_detects_visible_keyguard(self, adb): adb.return_value = subprocess.CompletedProcess([], 0, "KeyguardServiceDelegate\n showing=true\n") self.assertTrue(runner.keyguard_is_showing("device")) if __name__ == "__main__": unittest.main()