#!/usr/bin/env python3 from __future__ import annotations import argparse import json import shutil import subprocess import time from pathlib import Path from typing import Any from urllib.parse import urlencode, urljoin from verify_mobile_browser_controls import ( WebDriverError, execute_js, free_port, request_json, wait_for_driver, wait_for_map_runtime, wait_for_page, ) from verify_candidate_route_progress_browser import ( click_title_route_clear_gate_script, click_title_continue_script, route_completion_notice_script, route_completion_notice_state_script, route_continuation_field_encounter_script, route_continuation_field_encounter_state_script, title_route_clear_gate_state_script, title_state_script, wait_for_route_continuation_field_encounter, ) ROOT = Path(__file__).resolve().parents[1] OUT = ROOT / "out" TARGET = "map5_40f" def load_route_row(target: str = TARGET) -> dict[str, Any]: data = json.loads((OUT / "web_playtest_route.json").read_text(encoding="utf-8")) for row in data.get("routeAssistPathSamples") or []: if row.get("target") == target: return row raise WebDriverError(f"routeAssistPathSamples has no target {target!r}") def load_map(base: str, port: int, session_id: str, params: dict[str, str]) -> str: query = urlencode({**params, "_": str(time.time_ns())}) url = urljoin(base.rstrip("/") + "/", f"/web/game.html?{query}") request_json(port, "POST", f"/session/{session_id}/url", {"url": url}, timeout=30) wait_for_page(port, session_id) wait_for_map_runtime(port, session_id, params.get("map")) return url def deep_route_state_script() -> str: return """ try { if (typeof setupRoutePathSelect === 'function') setupRoutePathSelect(); if (typeof render === 'function') render(); const row = typeof selectedRouteAssistPathRow === 'function' ? selectedRouteAssistPathRow() : null; const nextStep = typeof routeAssistPathNextStep === 'function' ? routeAssistPathNextStep(row) : null; const routeProgress = typeof routeAssistPathProgress === 'function' ? routeAssistPathProgress(row) : null; const transitionProgress = typeof routeAssistPathTransitionProgress === 'function' ? routeAssistPathTransitionProgress(row) : null; const progress = typeof ensurePrototypeProgressState === 'function' ? ensurePrototypeProgressState() : (window.HWANSE_LAST_PROTOTYPE_PROGRESS || null); let payload = null; try { payload = JSON.parse(localStorage.getItem(RUNTIME_SAVE_KEY) || "null"); } catch (error) { payload = null; } let routeControlAction = null; try { routeControlAction = JSON.parse(sessionStorage.getItem('HWANSE_LAST_ROUTE_CONTROL_ACTION') || "null"); } catch (error) { routeControlAction = null; } return { readyState: document.readyState, scene: typeof scene === 'undefined' ? '' : scene, map: typeof map === 'undefined' || !map ? '' : map.name, tile: typeof footTile === 'function' ? footTile() : null, search: location.search, trialTransitions: typeof trialTransitions === 'undefined' ? '' : trialTransitions, selectedRouteGoal: typeof selectedRouteGoal === 'undefined' ? '' : selectedRouteGoal, selectedTransitionTarget: typeof selectedTransitionTarget === 'undefined' ? '' : selectedTransitionTarget, routePathValue: document.getElementById('routePathSelect')?.value || '', routeNextText: document.getElementById('routeNextButton')?.textContent || '', routeNextTitle: document.getElementById('routeNextButton')?.title || '', routeContinuationTarget: document.getElementById('routeNextButton')?.dataset.routeContinuationTarget || '', rowTarget: row?.target || '', rowHopCount: row?.hopCount ?? null, rowPath: row?.path || [], nextStep: nextStep ? { source: nextStep.source || '', target: nextStep.target || '', kind: nextStep.kind || '', status: nextStep.status || '', strictHotspotStatus: nextStep.strictHotspotStatus || '', blockReasons: nextStep.blockReasons || [], } : null, routeProgress, transitionProgress, progressCounts: progress?.counts || {}, progressEvents: progress?.events || [], lastProgressRecord: window.HWANSE_LAST_PROGRESS_RECORD || null, lastRouteCandidateAutoSave: window.HWANSE_LAST_ROUTE_CANDIDATE_AUTO_SAVE || null, lastRouteAssistAutoSave: window.HWANSE_LAST_ROUTE_ASSIST_AUTO_SAVE || null, routeProgressFeedbackLast: window.HWANSE_LAST_ROUTE_PROGRESS_FEEDBACK || null, routeProgressFeedbackRender: window.HWANSE_LAST_ROUTE_PROGRESS_FEEDBACK_RENDER || null, routeControlAction, quickLoadHidden: document.getElementById('quickLoadButton')?.hidden ?? null, quickLoadText: document.getElementById('quickLoadButton')?.textContent || '', routeCompletion: typeof routePrototypeCompletionState === 'function' ? routePrototypeCompletionState() : null, playableGate: typeof publishPrototypeCompletionState === 'function' ? publishPrototypeCompletionState()?.playableGate : null, savedPayloadMap: payload?.map || '', savedPayloadTile: payload?.tile || null, savedRouteState: payload?.routeState || null, savedProgressCounts: payload?.prototypeProgress?.counts || {}, titleContinueLabel: typeof titleContinueLabel === 'function' ? titleContinueLabel() : '', playHudLines: typeof playHudLines === 'function' ? playHudLines() : (window.HWANSE_LAST_PLAY_HUD_LINES || []), originalRoutePromotionImplemented: false, originalStoryFlagRuntimeImplemented: false, }; } catch (error) { return { error: String(error && error.message || error) }; } """ def activate_route_next_script() -> str: return """ try { activeDialogue = null; menuOpen = false; if (typeof setupRoutePathSelect === 'function') setupRoutePathSelect(); if (typeof render === 'function') render(); sessionStorage.removeItem('HWANSE_LAST_ROUTE_CONTROL_ACTION'); sessionStorage.removeItem('HWANSE_LAST_ROUTE_CONTROL_SOUND'); window.HWANSE_LAST_ROUTE_CONTROL_ACTION = null; window.HWANSE_LAST_ROUTE_CONTROL_SOUND = null; const before = { map: map?.name || '', tile: typeof footTile === 'function' ? footTile() : null, selectedRouteGoal, routePathValue: document.getElementById('routePathSelect')?.value || '', routeNextText: document.getElementById('routeNextButton')?.textContent || '', routeNextTitle: document.getElementById('routeNextButton')?.title || '', }; const activated = typeof activateNextRoutePath === 'function' ? activateNextRoutePath() : false; sessionStorage.setItem('__hwanseDeepRouteLastActivation', JSON.stringify({ before, activated, afterSearch: location.search, })); return { before, activated, afterSearch: location.search }; } catch (error) { return { error: String(error && error.message || error) }; } """ def wait_for_route_state(port: int, session_id: str, timeout: float = 10) -> dict[str, Any]: deadline = time.monotonic() + timeout last: dict[str, Any] = {} while time.monotonic() < deadline: state = execute_js(port, session_id, deep_route_state_script(), timeout=4) last = state if ( state.get("readyState") == "complete" and state.get("scene") == "map" and state.get("map") and state.get("trialTransitions") == "routeAssist" and state.get("selectedRouteGoal") == TARGET and state.get("rowTarget") == TARGET and not state.get("error") ): return state time.sleep(0.2) raise WebDriverError(f"deep route state did not stabilize: {last!r}") def route_complete_enough(state: dict[str, Any], expected_candidate_count: int) -> bool: route_progress = state.get("routeProgress") or {} return ( state.get("map") == TARGET and route_progress.get("count") == expected_candidate_count and route_progress.get("completedCount") == expected_candidate_count ) def unique_map_path(states: list[dict[str, Any]]) -> list[str]: path: list[str] = [] for state in states: name = str(state.get("map") or "") if name and (not path or path[-1] != name): path.append(name) return path def tile_text(tile: dict[str, Any] | None) -> str: tile = tile or {} return f"{tile.get('x')},{tile.get('y')}" def path_progress_text(state: dict[str, Any]) -> str: route_progress = state.get("routeProgress") or {} completed = route_progress.get("completedCount") count = route_progress.get("count") if completed == count and count: return f"후보 완료 {completed}/{count}" return f"후보 진행 {completed}/{count}" def route_state_summary(state: dict[str, Any]) -> str: route_progress = state.get("routeProgress") or {} saved_counts = state.get("savedProgressCounts") or {} return ( f"map={state.get('map')}@{tile_text(state.get('tile'))} " f"routeGoal={state.get('selectedRouteGoal')} " f"routePathValue={state.get('routePathValue')} " f"routeNextText={state.get('routeNextText')} " f"routeCandidateCount={route_progress.get('completedCount')} " f"savedRouteCandidateCount={saved_counts.get('route-candidate')} " f"savedPayloadMap={state.get('savedPayloadMap')} " f"pathProgress={path_progress_text(state)} " f"originalRoutePromotionImplemented={state.get('originalRoutePromotionImplemented')}" ) def title_continue_summary(title_before: dict[str, Any], click: dict[str, Any], restored: dict[str, Any]) -> str: label = next( (str(label) for label in title_before.get("titleMenuLabels") or [] if str(label).startswith("이어하기")), "", ) saved_counts = restored.get("savedProgressCounts") or {} return ( f"titleContinue=True label={label} clickOk={click.get('ok')} " f"map={restored.get('map')}@{tile_text(restored.get('tile'))} " f"quickLoadText={restored.get('quickLoadText')} " f"routeCandidateCount={(restored.get('routeProgress') or {}).get('completedCount')} " f"savedRouteCandidateCount={saved_counts.get('route-candidate')} " f"pathProgress={path_progress_text(restored)}" ) def field_encounter_summary(state: dict[str, Any]) -> str: start = state.get("startSnapshot") or {} candidate = start.get("battleCandidate") or {} progress = state.get("progress") or {} counts = progress.get("counts") or {} completion = (state.get("completion") or {}).get("route") or {} auto_save = state.get("victoryAutoSave") or {} return ( f"map={state.get('mapName')} started={state.get('started')} " f"battleId={candidate.get('id')} " f"battleBackground={candidate.get('battleBackground')} " f"fieldEncounterVictoryCount={counts.get('field-encounter-victory')} " f"completionCompleted={completion.get('completed')} " f"autoSource={auto_save.get('source')} " f"originalRoutePromotionImplemented={state.get('originalRoutePromotionImplemented')} " f"originalEncounterRuntimeImplemented={state.get('originalEncounterRuntimeImplemented')}" ) def completion_notice_summary(state: dict[str, Any]) -> str: completion = (state.get("completion") or {}).get("route") or {} playable_gate = (state.get("notice") or {}).get("playableGate") or {} feedback = state.get("routeGateFeedbackLast") or {} return ( f"activeId={state.get('activeId')} " f"completionRouteCompleteCount={completion.get('routeCompleteCount')} " f"playableGate={playable_gate.get('source')} " f"playableGateOpen={playable_gate.get('opened')} " f"playableGateLabel={playable_gate.get('label')} " f"routeGateFeedback={feedback.get('source')}:{feedback.get('text')} " f"routeGateFeedbackRender={bool(state.get('routeGateFeedbackRender') or [])} " f"routeGateSound={feedback.get('routeGateSound')} " f"routeGateSoundSrc={feedback.get('routeGateSoundSrc')} " f"routeGateSoundPlayed={feedback.get('routeGateSoundPlayed')}" ) def completion_title_continue_summary(title_before: dict[str, Any], click: dict[str, Any], restored: dict[str, Any]) -> str: label = next( (str(label) for label in title_before.get("titleMenuLabels") or [] if str(label).startswith("이어하기")), "", ) saved_counts = restored.get("savedProgressCounts") or {} completion = restored.get("routeCompletion") or {} playable_gate = restored.get("playableGate") or {} return ( f"titleContinue=True label={label} clickOk={click.get('ok')} " f"map={restored.get('map')}@{tile_text(restored.get('tile'))} " f"routeCompleteCount={completion.get('routeCompleteCount')} " f"savedRouteCompleteCount={saved_counts.get('route-complete')} " f"playableGateOpen={playable_gate.get('opened')}" ) def route_clear_gate_summary(state: dict[str, Any]) -> str: summary = state.get("summary") or {} completion = (state.get("completion") or {}).get("route") or {} feedback = state.get("routeGateFeedbackLast") or {} auto_save = summary.get("autoSave") or {} payload_counts = state.get("payloadCounts") or {} lines = state.get("activeLines") or [] return ( f"activeId={state.get('activeId')} " f"routeClearCount={summary.get('routeClearCount')} " f"completionRouteClearCount={completion.get('routeClearCount')} " f"payloadRouteClearCount={payload_counts.get('route-clear')} " f"autoSource={auto_save.get('source')} " f"routeGateFeedback={feedback.get('source')}:{feedback.get('text')} " f"routeGateFeedbackRender={bool(state.get('routeGateFeedbackRender') or [])} " f"routeGateSound={feedback.get('routeGateSound')} " f"routeGateSoundSrc={feedback.get('routeGateSoundSrc')} " f"routeGateSoundPlayed={feedback.get('routeGateSoundPlayed')} " f"lines={' / '.join(str(line) for line in lines[:4])}" ) def route_clear_title_continue_summary(title_before: dict[str, Any], click: dict[str, Any], restored: dict[str, Any]) -> str: label = next( (str(label) for label in title_before.get("titleMenuLabels") or [] if str(label).startswith("이어하기")), "", ) saved_counts = restored.get("savedProgressCounts") or {} completion = restored.get("routeCompletion") or {} return ( f"titleContinue=True label={label} clickOk={click.get('ok')} " f"map={restored.get('map')}@{tile_text(restored.get('tile'))} " f"routeCompleteCount={completion.get('routeCompleteCount')} " f"routeClearCount={completion.get('routeClearCount')} " f"savedRouteCompleteCount={saved_counts.get('route-complete')} " f"savedRouteClearCount={saved_counts.get('route-clear')} " f"routeNextText={restored.get('routeNextText')} " f"playableGateOpen={(restored.get('playableGate') or {}).get('opened')}" ) def wait_for_title(port: int, session_id: str, timeout: float = 10) -> dict[str, Any]: deadline = time.monotonic() + timeout last: dict[str, Any] = {} while time.monotonic() < deadline: value = execute_js(port, session_id, title_state_script(), timeout=4) last = value if isinstance(value, dict) else {} if ( last.get("readyState") == "complete" and last.get("scene") == "title" and last.get("hasScreen") is True and last.get("titleLoaded") is True ): return last time.sleep(0.2) raise WebDriverError(f"title did not stabilize: {last!r}") def restore_from_title(base: str, port: int, session_id: str) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: url = urljoin(base.rstrip("/") + "/", "/web/game.html?game=1") request_json(port, "POST", f"/session/{session_id}/url", {"url": url}, timeout=30) wait_for_page(port, session_id) title_before = wait_for_title(port, session_id) click = execute_js(port, session_id, click_title_continue_script(), timeout=4) if not click.get("ok"): raise WebDriverError(f"title continue was not available: title={title_before!r} click={click!r}") wait_for_page(port, session_id) wait_for_map_runtime(port, session_id, TARGET) restored = wait_for_route_state(port, session_id) return title_before, click, restored def wait_for_deep_completion_notice(port: int, session_id: str, expected_candidate_count: int, timeout: float = 10) -> dict[str, Any]: deadline = time.monotonic() + timeout state: dict[str, Any] = {} while time.monotonic() < deadline: value = execute_js(port, session_id, route_completion_notice_state_script(), timeout=4) state = value if isinstance(value, dict) else {} notice = state.get("notice") or {} completion = (state.get("completion") or {}).get("route") or {} feedback = state.get("routeGateFeedbackLast") or {} gate = notice.get("playableGate") or {} if ( state and state.get("error") is None and state.get("activeId") == f"route-complete:{TARGET}" and notice.get("blockId") == f"route-complete:{TARGET}" and completion.get("completed") is True and completion.get("currentMap") == TARGET and completion.get("selectedRouteGoal") == TARGET and completion.get("routeCandidateCount") == expected_candidate_count and completion.get("fieldEncounterVictoryCount") == 1 and completion.get("routeCompleteCount") == 1 and gate.get("source") == "prototype-playable-route-gate" and gate.get("opened") is True and gate.get("label") == f"웹 후보 루트 완료 {TARGET}" and feedback.get("source") == "route-completion-feedback" and feedback.get("text") == f"후보 완료 {TARGET}" and feedback.get("routeGateSound") == "victory" and str(feedback.get("routeGateSoundSrc") or "").endswith("/extract_wlk/12.wav") and feedback.get("routeGateSoundPlayed") is True and state.get("routeGateFeedbackRender") ): return state time.sleep(0.2) raise WebDriverError(f"deep route completion notice did not open: {state!r}") def wait_for_deep_route_clear_gate(port: int, session_id: str, expected_candidate_count: int, timeout: float = 10) -> dict[str, Any]: deadline = time.monotonic() + timeout state: dict[str, Any] = {} while time.monotonic() < deadline: value = execute_js(port, session_id, title_route_clear_gate_state_script(), timeout=4) state = value if isinstance(value, dict) else {} click = state.get("click") or {} title_gate = state.get("titleGate") or {} summary = state.get("summary") or {} title_payload = state.get("titleRouteClearPayload") or {} payload_counts = state.get("payloadCounts") or {} summary_progress = summary.get("progressEvent") or {} summary_auto_save = summary.get("autoSave") or {} saved_progress = (summary_auto_save.get("progress") or {}).get("counts") or {} completion = (state.get("completion") or {}).get("route") or {} playable_gate = (state.get("completion") or {}).get("playableGate") or {} feedback = state.get("routeGateFeedbackLast") or {} menu_commands = state.get("menuCommands") or [] lines = "\n".join(state.get("activeLines") or []) progress_review_lines = "\n".join((state.get("progressReviewBlock") or {}).get("lines") or []) if ( state and click.get("ok") is True and click.get("activated") is True and (click.get("titleItem") or {}).get("key") == "routeClear" and (click.get("titleItem") or {}).get("label") == f"후보 클리어 저장 {TARGET}" and title_gate.get("source") == "title-route-clear-gate" and title_gate.get("loaded") is True and title_gate.get("opened") is True and title_gate.get("map") == TARGET and title_gate.get("activeId") == f"route-clear:{TARGET}" and title_payload.get("goal") == TARGET and title_payload.get("routeCompleteCount") == 1 and title_payload.get("routeClearCount") == 1 and state.get("scene") == "map" and state.get("map") == TARGET and state.get("activeId") == f"route-clear:{TARGET}" and summary.get("blockId") == f"route-clear:{TARGET}" and summary.get("routeClearCount") == 1 and summary_progress.get("kind") == "route-clear" and summary_progress.get("id") == f"route-clear:{TARGET}" and summary_auto_save.get("saved") is True and summary_auto_save.get("source") == "route-clear" and summary_auto_save.get("payloadMap") == TARGET and summary_auto_save.get("duplicateProgress") is False and payload_counts.get("route-complete") == 1 and payload_counts.get("route-clear") == 1 and saved_progress.get("route-complete") == 1 and saved_progress.get("route-clear") == 1 and completion.get("routeCandidateCount") == expected_candidate_count and completion.get("routeCompleteCount") == 1 and completion.get("routeClearCount") == 1 and completion.get("routeClearRecorded") is True and playable_gate.get("source") == "prototype-playable-route-gate" and playable_gate.get("opened") is True and feedback.get("source") == "route-clear-feedback" and feedback.get("text") == f"후보 클리어 {TARGET}" and feedback.get("routeGateSound") == "victory" and str(feedback.get("routeGateSoundSrc") or "").endswith("/extract_wlk/12.wav") and feedback.get("routeGateSoundPlayed") is True and state.get("routeGateFeedbackRender") and any( command.get("command") == "showRouteClearSummary" and command.get("name") == f"후보 클리어 {TARGET}" and command.get("usable") is True for command in menu_commands ) and f"후보 클리어 요약 {TARGET}" in lines and "웹 후보 클리어 저장이 고정되었습니다." in lines and "원본 full-game ending/story flag 완료는 아직 아닙니다." in lines and f"route-clear:{TARGET}" in progress_review_lines and f"클리어 목표 {TARGET}" in progress_review_lines and f"클리어 {TARGET}" in str(state.get("titleContinueLabel") or "") ): return state time.sleep(0.2) raise WebDriverError(f"deep route clear gate did not open: {state!r}") def write_report(report: dict[str, Any]) -> None: OUT.mkdir(parents=True, exist_ok=True) (OUT / "candidate_deep_route_browser_smoke.json").write_text( json.dumps(report, ensure_ascii=False, separators=(",", ":")) + "\n", encoding="utf-8", ) final = report.get("finalState") or {} route_progress = final.get("routeProgress") or {} transition_progress = final.get("transitionProgress") or {} lines = [ "# Candidate Deep Route Browser Smoke", "", f"- status: `{report.get('status')}`", f"- target: `{report.get('target')}`", f"- start url: `{report.get('startUrl')}`", f"- path: `{' -> '.join(report.get('path') or [])}`", f"- activations: `{report.get('activationCount')}`", f"- action count: `{report.get('actionCount')}`", f"- candidate entries: `{report.get('candidateEntryCount')}`", f"- final map: `{final.get('map')}`", f"- candidate progress: `{route_progress.get('completedCount')}/{route_progress.get('count')}`", f"- transition progress: `{transition_progress.get('completedCount')}/{transition_progress.get('count')}`", f"- saved payload: `{final.get('savedPayloadMap')}` / `{(final.get('savedRouteState') or {}).get('selectedRouteGoal')}`", f"- title continue label: `{final.get('titleContinueLabel')}`", f"- original route promotion implemented: `{final.get('originalRoutePromotionImplemented')}`", f"- final: `{report.get('final')}`", f"- title continue: `{report.get('titleContinue')}`", f"- field encounter: `{report.get('fieldEncounter')}`", f"- completion notice: `{report.get('completionNotice')}`", f"- completion title continue: `{report.get('completionTitleContinue')}`", f"- route clear gate: `{report.get('routeClearGate')}`", f"- route clear title continue: `{report.get('routeClearTitleContinue')}`", f"- first step: `{(report.get('stepSummaries') or [''])[0]}`", f"- blocker step: `{(report.get('stepSummaries') or ['', ''])[1]}`", f"- last step: `{(report.get('stepSummaries') or [''])[-1]}`", "", ] (OUT / "candidate_deep_route_browser_smoke.md").write_text("\n".join(lines), encoding="utf-8") def verify_browser(base: str) -> None: route_row = load_route_row(TARGET) steps = route_row.get("steps") or [] expected_candidate_count = sum(1 for step in steps if step.get("kind") != "confirmed") expected_path = route_row.get("path") or [] if expected_candidate_count < 10 or route_row.get("hopCount") != 13: raise WebDriverError(f"unexpected deep route fixture shape: {route_row!r}") driver_path = shutil.which("WebKitWebDriver") if not driver_path: raise WebDriverError("WebKitWebDriver is not installed; install webkit2gtk-driver and run under Xvfb") port = free_port() log_path = OUT / "candidate_deep_route_webkitdriver.log" log_path.parent.mkdir(parents=True, exist_ok=True) with log_path.open("wb") as log: proc = subprocess.Popen( [ driver_path, "--host=127.0.0.1", f"--port={port}", "--replace-on-new-session", ], stdout=log, stderr=subprocess.STDOUT, ) session_id = "" try: wait_for_driver(port, proc) session = request_json( port, "POST", "/session", {"capabilities": {"alwaysMatch": {"browserName": "MiniBrowser"}}}, timeout=30, ) session_id = str((session or {}).get("value", {}).get("sessionId") or "") if not session_id: raise WebDriverError(f"could not create WebKit session: {session!r}") request_json( port, "POST", f"/session/{session_id}/window/rect", {"x": 0, "y": 0, "width": 1180, "height": 820}, timeout=8, ) start_url = load_map( base, port, session_id, { "map": "map1_02b", "startTile": "11,12", "trialTransitions": "routeAssist", "routeGoal": TARGET, }, ) initial = wait_for_route_state(port, session_id) visited = [initial] activations: list[dict[str, Any]] = [] for _ in range(34): current = visited[-1] if route_complete_enough(current, expected_candidate_count): break activation = execute_js(port, session_id, activate_route_next_script(), timeout=4) activations.append(activation) wait_for_page(port, session_id) wait_for_map_runtime(port, session_id) time.sleep(0.25) visited.append(wait_for_route_state(port, session_id)) final = visited[-1] if not route_complete_enough(final, expected_candidate_count): raise WebDriverError(f"deep route did not reach {TARGET}: final={final!r} activations={activations!r}") saved_counts = final.get("savedProgressCounts") or {} if ( final.get("savedPayloadMap") != TARGET or (final.get("savedRouteState") or {}).get("selectedRouteGoal") != TARGET or saved_counts.get("route-candidate") != expected_candidate_count or (final.get("routeProgress") or {}).get("completedIds") != [ f"{step.get('source')}->{step.get('target')}" for step in steps if step.get("kind") != "confirmed" ] or final.get("originalRoutePromotionImplemented") is not False or final.get("originalStoryFlagRuntimeImplemented") is not False ): raise WebDriverError(f"deep route final save/progress state is incomplete: {final!r}") title_before, title_click, title_restore = restore_from_title(base, port, session_id) if not route_complete_enough(title_restore, expected_candidate_count): raise WebDriverError(f"deep route title restore lost route progress: {title_restore!r}") execute_js(port, session_id, route_continuation_field_encounter_script(), timeout=4) field_encounter = wait_for_route_continuation_field_encounter( port, session_id, TARGET, expected_candidate_count, timeout=12, ) execute_js(port, session_id, route_completion_notice_script(), timeout=4) completion_notice = wait_for_deep_completion_notice( port, session_id, expected_candidate_count, timeout=10, ) completion_title_before, completion_title_click, completion_title_restore = restore_from_title( base, port, session_id, ) completion_counts = completion_title_restore.get("savedProgressCounts") or {} completion_route = completion_title_restore.get("routeCompletion") or {} if ( completion_counts.get("route-complete") != 1 or completion_route.get("routeCompleteCount") != 1 or completion_route.get("currentMap") != TARGET ): raise WebDriverError( f"deep route completion title restore lost completion state: {completion_title_restore!r}" ) route_clear_gate_url = urljoin(base.rstrip("/") + "/", "/web/game.html?game=1") request_json( port, "POST", f"/session/{session_id}/url", {"url": route_clear_gate_url}, timeout=30, ) wait_for_page(port, session_id) route_clear_title = wait_for_title(port, session_id) if ( f"후보 클리어 저장 {TARGET}" not in (route_clear_title.get("titleMenuLabels") or []) or "routeClear" not in (route_clear_title.get("titleMenuKeys") or []) ): raise WebDriverError(f"deep route title clear item is missing: {route_clear_title!r}") execute_js(port, session_id, click_title_route_clear_gate_script(), timeout=4) route_clear_gate = wait_for_deep_route_clear_gate( port, session_id, expected_candidate_count, timeout=10, ) route_clear_title_before, route_clear_title_click, route_clear_title_restore = restore_from_title( base, port, session_id, ) route_clear_counts = route_clear_title_restore.get("savedProgressCounts") or {} route_clear_completion = route_clear_title_restore.get("routeCompletion") or {} route_clear_title_label = next( ( str(label) for label in route_clear_title_before.get("titleMenuLabels") or [] if str(label).startswith("이어하기") ), "", ) if ( route_clear_counts.get("route-complete") != 1 or route_clear_counts.get("route-clear") != 1 or route_clear_completion.get("routeClearCount") != 1 or route_clear_completion.get("routeClearRecorded") is not True or route_clear_title_restore.get("routeNextText") != "클리어" or f"클리어 {TARGET}" not in route_clear_title_label or f"완료 {TARGET}" in route_clear_title_label ): raise WebDriverError( "deep route clear title restore lost clear state: " f"title={route_clear_title_before!r} restored={route_clear_title_restore!r}" ) candidate_edges = [ f"{step.get('source')}->{step.get('target')}" for step in steps if step.get("kind") != "confirmed" ] candidate_edge_counts = { edge: index + 1 for index, edge in enumerate(candidate_edges) } step_summaries: list[str] = [] action_index = 1 for step in steps: edge = f"{step.get('source')}->{step.get('target')}" if step.get("kind") == "confirmed": step_summaries.append(f"{action_index}:{edge} kind=confirmed") action_index += 1 continue action_index += 1 step_summaries.append(f"{action_index}:{edge} progressId={edge}") action_index += 1 report = { "status": "passed", "base": base, "target": TARGET, "startUrl": start_url, "path": expected_path, "hopCount": route_row.get("hopCount"), "expectedCandidateCount": expected_candidate_count, "activationCount": len(activations), "actionCount": len(activations), "candidateEntryCount": expected_candidate_count, "visitedMaps": expected_path, "candidateEdges": candidate_edges, "candidateEdgeCounts": candidate_edge_counts, "final": route_state_summary(final), "titleContinue": title_continue_summary(title_before, title_click, title_restore), "fieldEncounter": field_encounter_summary(field_encounter), "completionNotice": completion_notice_summary(completion_notice), "completionTitleContinue": completion_title_continue_summary( completion_title_before, completion_title_click, completion_title_restore, ), "routeClearGate": route_clear_gate_summary(route_clear_gate), "routeClearTitleContinue": route_clear_title_continue_summary( route_clear_title_before, route_clear_title_click, route_clear_title_restore, ), "stepSummaries": step_summaries, "activations": activations, "visited": visited, "finalState": final, "snapshots": { "titleRestore": { **title_restore, "routeCandidateCount": (title_restore.get("routeProgress") or {}).get("completedCount"), }, "fieldEncounter": field_encounter, "completionNotice": completion_notice, "completionTitleRestore": { **completion_title_restore, "routeCandidateCount": (completion_title_restore.get("routeProgress") or {}).get("completedCount"), "routeCompleteCount": (completion_title_restore.get("routeCompletion") or {}).get("routeCompleteCount"), }, "routeClearGateTitle": route_clear_title, "routeClearGate": route_clear_gate, "routeClearTitleRestore": { **route_clear_title_restore, "routeCandidateCount": (route_clear_title_restore.get("routeProgress") or {}).get("completedCount"), "routeCompleteCount": (route_clear_title_restore.get("routeCompletion") or {}).get("routeCompleteCount"), "routeClearCount": (route_clear_title_restore.get("routeCompletion") or {}).get("routeClearCount"), }, }, } write_report(report) print( "ok candidate deep route browser " f"target={TARGET} hops={route_row.get('hopCount')} " f"activations={len(activations)} " f"candidateProgress={(final.get('routeProgress') or {}).get('completedCount')}/" f"{(final.get('routeProgress') or {}).get('count')} " f"saved={final.get('savedPayloadMap')}" ) finally: if session_id: try: request_json(port, "DELETE", f"/session/{session_id}", timeout=5) except Exception: pass proc.terminate() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=5) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--base", default="http://127.0.0.1:8013") args = parser.parse_args() verify_browser(args.base) if __name__ == "__main__": main()