diff --git a/GramAddict/core/account_switcher.py b/GramAddict/core/account_switcher.py index c101bda..212b1d4 100644 --- a/GramAddict/core/account_switcher.py +++ b/GramAddict/core/account_switcher.py @@ -61,7 +61,7 @@ def verify_and_switch_account(device, nav_graph, target_username): return False # Long press to open account selector - device.deviceV2.long_click(profile_tab[0], profile_tab[1], 1.5) + device.long_click(profile_tab[0], profile_tab[1], 1.5) time.sleep(3.0) # 4. Find the target account in the selector list @@ -102,7 +102,7 @@ def verify_and_switch_account(device, nav_graph, target_username): if account_node: logger.info(f"πŸ–±οΈ [Identity Guard] Found account '{target_username}' in selector. Tapping!") - device.deviceV2.click(account_node[0], account_node[1]) + device.click(account_node[0], account_node[1]) time.sleep(6.0) # Wait heavily for app to reload context nav_graph.current_state = "UNKNOWN" # Force graph to re-evaluate after massive state shift return True @@ -114,6 +114,6 @@ def verify_and_switch_account(device, nav_graph, target_username): except: pass # Escape the bottom sheet - device.deviceV2.press("back") + device.press("back") return False diff --git a/GramAddict/core/bot_flow.py b/GramAddict/core/bot_flow.py index bb5e8b7..401af90 100644 --- a/GramAddict/core/bot_flow.py +++ b/GramAddict/core/bot_flow.py @@ -216,7 +216,7 @@ def start_bot(**kwargs): _humanized_scroll(device, is_skip=False) sleep(2.0) - device.deviceV2.press("back") + device.press("back") sleep(1.5) # Deduplicate while preserving order @@ -263,9 +263,9 @@ def start_bot(**kwargs): if current_desire == "ShiftContext": logger.info("🧠 [Free Will] Boredom critical. Forcing app restart to clear context.") - device.deviceV2.app_stop(device.app_id) + device.app_stop(device.app_id) random_sleep(2.0, 4.0) - device.deviceV2.app_start(device.app_id, use_monkey=True) + device.app_start(device.app_id, use_monkey=True) random_sleep(4.0, 6.0) dopamine.boredom = max(0.0, dopamine.boredom * 0.2) continue @@ -327,9 +327,9 @@ def start_bot(**kwargs): elif result == "CONTEXT_LOST": logger.warning(f"⚠️ Context was lost in {current_target}. Forcing app restart and returning to HomeFeed to escape softlock.") - device.deviceV2.app_stop(device.app_id) + device.app_stop(device.app_id) random_sleep(1.0, 2.0) - device.deviceV2.app_start(device.app_id, use_monkey=True) + device.app_start(device.app_id, use_monkey=True) random_sleep(3.0, 5.0) nav_graph.current_state = "UNKNOWN" @@ -464,7 +464,7 @@ def _humanized_scroll(device, is_skip=False, resonance_score=None): # Using adb shell input swipe natively triggers Android's elastic momentum (Fling). # UIAutomator2's internal swipe often kills momentum by calculating a zero velocity touch-up. - device.deviceV2.shell(f"input swipe {int(start_x)} {int(start_y)} {int(end_x)} {int(end_y)} {duration_ms}") + device.shell(f"input swipe {int(start_x)} {int(start_y)} {int(end_x)} {int(end_y)} {duration_ms}") def _humanized_click(device, x, y, double=False, sleep_mod=1.0): import random @@ -482,7 +482,7 @@ def _humanized_click(device, x, y, double=False, sleep_mod=1.0): # Duration for a human tap duration = random.randint(40, 90) - device.deviceV2.shell(f"input swipe {int(start_x)} {int(start_y)} {int(end_x)} {int(end_y)} {duration}") + device.shell(f"input swipe {int(start_x)} {int(start_y)} {int(end_x)} {int(end_y)} {duration}") if double: # Double tap (Fast, slightly overlapping jitter) @@ -511,7 +511,7 @@ def _humanized_horizontal_swipe(device, start_x, end_x, y, duration_ms): # Introduce +- 30% timing wobble actual_duration = int(duration_ms * random.uniform(0.7, 1.3)) - device.deviceV2.shell(f"input swipe {actual_start_x} {actual_start_y} {actual_end_x} {actual_end_y} {actual_duration}") + device.shell(f"input swipe {actual_start_x} {actual_start_y} {actual_end_x} {actual_end_y} {actual_duration}") def has_carousel_in_view(xml_dump: str) -> bool: @@ -662,7 +662,7 @@ def _interact_with_profile(device, configs, username, session_state, sleep_mod, sleep(random.uniform(2.0, 5.0) * sleep_mod) if i < count - 1: _humanized_click(device, int(w * 0.9), int(h * 0.5), sleep_mod=sleep_mod) - device.deviceV2.press("back") + device.press("back") sleep(random.uniform(1.0, 2.0) * sleep_mod) # Random Follow @@ -751,7 +751,7 @@ def _interact_with_profile(device, configs, username, session_state, sleep_mod, sleep(random.uniform(1.5, 3.0) * sleep_mod) - device.deviceV2.press("back") + device.press("back") sleep(random.uniform(1.0, 2.0) * sleep_mod) # Let the native UI momentum scroll finish just like a human watching the feed sleep(random.uniform(1.2, 2.0)) @@ -817,7 +817,7 @@ def _align_active_post(device): end_y = start_y + dist # Duration 1.0s equals a precise mechanical drag with ZERO momentum (no flinging) - device.deviceV2.swipe(cx, start_y, cx, end_y, duration=1.0) + device.swipe(cx, start_y, cx, end_y, duration=1.0) sleep(1.0) # Wait for UI to settle logger.debug(f"πŸ“ [Alignment] Snapping attempt {attempts}: Shifted {diff}px.") else: @@ -917,7 +917,7 @@ def _run_zero_latency_stories_loop(device, configs, session_state, cognitive_sta # Check for boredom if dopamine.wants_to_change_feed(): logger.info("🧠 [Dopamine] Bored. Escaping StoriesFeed to seek new stimuli.") - device.deviceV2.press("back") # Attempt to back out to feed + device.press("back") # Attempt to back out to feed sleep(random.uniform(1.0, 2.0) * sleep_mod) return "BOREDOM_CHANGE_FEED" @@ -938,7 +938,7 @@ def _run_zero_latency_stories_loop(device, configs, session_state, cognitive_sta sleep(random.uniform(2.0, 5.0) * sleep_mod) logger.info("🎬 [StoriesFeed] Session completed naturally.") - device.deviceV2.press("back") + device.press("back") return "FEED_EXHAUSTED" def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session_state, job_target, cognitive_stack, is_reels=False): """ @@ -1049,7 +1049,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session "⚠️ [Anomaly Handler] 0 interactive nodes extracted. UI is blind/stuck! Pressing BACK and scrolling...", extra={"color": f"{Fore.YELLOW}"} ) - device.deviceV2.press("back") + device.press("back") sleep(0.5) _humanized_scroll(device) sleep(random.uniform(1.0, 2.0) * sleep_mod) @@ -1062,14 +1062,20 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session # Instagram often renders feed markers in the background while a bottom sheet obscures the view. has_obstacle = bool(re.search(r'bottom_sheet_container|dialog_container|dialog_root|bottom_sheet_drag', str(context_xml))) - if not has_feed_markers or has_obstacle: - + if has_obstacle: + logger.warning( + "⚠️ [Self-Check] Obstacle (sheet/dialog/keyboard) is blocking the view! Pressing BACK to dismiss...", + extra={"color": f"{Fore.YELLOW}"} + ) + device.press("back") + sleep(0.5) + continue + + elif not has_feed_markers: consecutive_marker_misses += 1 if consecutive_marker_misses >= 3: - logger.error("❌ Lost context completely or stuck behind un-clearable obstacle. Aborting feed loop to force reset.") - - - dump_ui_state(device, "context_lost", {"feed": job_target, "misses": consecutive_marker_misses, "obstacle": has_obstacle}) + logger.error("❌ Lost context completely. Aborting feed loop to force reset.") + dump_ui_state(device, "context_lost", {"feed": job_target, "misses": consecutive_marker_misses}) return "CONTEXT_LOST" if consecutive_marker_misses == 2: @@ -1082,7 +1088,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session if best_node: logger.info(f" -> Recovery attempt! Clicking {best_node.get('semantic', 'Dismiss Button')} at ({best_node['x']}, {best_node['y']})") - device.deviceV2.click(best_node["x"], best_node["y"]) + device.click(best_node["x"], best_node["y"]) sleep(2.5) # Verification: Check if markers are now visible @@ -1103,12 +1109,10 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session continue logger.warning( - "⚠️ [Self-Check] NOT on a post! Open sheet or keyboard blocking view? Pressing BACK then scrolling...", + "⚠️ [Self-Check] Feed markers missing. Mid-scroll or tall post? Scrolling to reveal markers...", extra={"color": f"{Fore.YELLOW}"} ) - # Dismiss any open sheets (like the comment sheet) or keyboard - device.deviceV2.press("back") - sleep(0.5) + # DO NOT press 'back' here as we are just on the timeline. It would trigger a scroll-to-top refresh. _humanized_scroll(device) sleep(random.uniform(1.0, 2.0) * sleep_mod) continue @@ -1204,7 +1208,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session _humanized_scroll(device, is_skip=True) sleep(random.uniform(0.5, 1.5) * sleep_mod) logger.info("πŸ”™ [Rabbit Hole] Exiting profile back to main feed.") - device.deviceV2.press("back") + device.press("back") sleep(random.uniform(0.8, 1.5) * sleep_mod) # ── Darwin: SOLE Dwell Controller (micro-wobble + proof of resonance) ── @@ -1271,7 +1275,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session if target_user.lower() in text_lower and ("profile_name" in res_id or "title" in res_id or "username" in res_id or "avatar" in res_id): if n.get("x") and n.get("y"): logger.info(f"⚑ [Targeted UX] Exact matched username '{target_user}' on screen. Tapping directly.") - device.deviceV2.click(n["x"], n["y"]) + device.click(n["x"], n["y"]) nav_success = True break @@ -1309,7 +1313,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session # Return to feed logger.info("πŸ”™ [Profile Learning] Returning to main feed.") - device.deviceV2.press("back") + device.press("back") _wait_for_post_loaded(device) sleep(random.uniform(1.0, 1.5) * sleep_mod) @@ -1465,7 +1469,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session telepathic.confirm_click(intent) _interact_with_profile(device, configs, "commenter", session_state, sleep_mod, logger, cognitive_stack) logger.info("πŸ”™ [Randomization] Returning to comment sheet.") - device.deviceV2.press("back") + device.press("back") sleep(random.uniform(1.5, 3.0) * sleep_mod) else: telepathic.reject_click(intent) @@ -1554,7 +1558,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session logger.info(f"🚫 [DRY RUN] Generated comment: '{clean_comment}'. Skipping UI injection.", extra={"color": f"{Fore.MAGENTA}"}) sleep(1.5) else: - device.deviceV2.click(comment_box["x"], comment_box["y"]) + device.click(comment_box["x"], comment_box["y"]) sleep(random.uniform(1.2, 2.2)) # Verification: Did the keyboard open or cursor move to box? @@ -1576,16 +1580,16 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session if random.random() < 0.5: # Rapid backspace (Manual deletion) for _ in range(len(clean_comment) + 2): - device.deviceV2.press("del") + device.press("del") sleep(random.uniform(0.01, 0.05)) else: # Press back to trigger Discard popup - device.deviceV2.press("back") + device.press("back") sleep(1.0) xml_dump = device.dump_hierarchy() discard_btn = telepathic.find_best_node(xml_dump, "Discard or Verwerfen popup button to cancel comment", device=device) if discard_btn: - device.deviceV2.click(discard_btn["x"], discard_btn["y"]) + device.click(discard_btn["x"], discard_btn["y"]) telepathic.confirm_click("Discard or Verwerfen popup button to cancel comment") logger.info("πŸ”™ [Umentscheidung] Comment successfully aborted.") @@ -1596,7 +1600,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session pre_post_xml = device.dump_hierarchy() post_btn = telepathic.find_best_node(pre_post_xml, "Post submit comment button", device=device) if post_btn: - device.deviceV2.click(post_btn["x"], post_btn["y"]) + device.click(post_btn["x"], post_btn["y"]) sleep(random.uniform(2.0, 3.5)) # Verification: Did the button disappear or layout change? @@ -1615,10 +1619,10 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session # Safely exit the comment sheet if "bottom_sheet_container" in device.dump_hierarchy(): - device.deviceV2.press("back") + device.press("back") sleep(1.0) if "bottom_sheet_container" in device.dump_hierarchy(): - device.deviceV2.press("back") + device.press("back") sleep(1.0) did_interact = True @@ -1662,7 +1666,7 @@ def _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session logger.warning("⚠️ [Repost] Click failed to trigger repost. Learning from failure.") # Close share menu if still open - device.deviceV2.press("back") + device.press("back") sleep(random.uniform(1.0, 2.0) * sleep_mod) @@ -1745,7 +1749,7 @@ def _run_zero_latency_search_loop(device, zero_engine, nav_graph, configs, sessi _humanized_click(device, search_bar["x"], search_bar["y"]) sleep(1.5) ghost_type(device, keyword, speed="fast") - device.deviceV2.press("enter") + device.press("enter") sleep(3.0) # 2. Pick a result (Top, Accounts, or Tags) diff --git a/GramAddict/core/darwin_engine.py b/GramAddict/core/darwin_engine.py index b42e3db..68536f8 100644 --- a/GramAddict/core/darwin_engine.py +++ b/GramAddict/core/darwin_engine.py @@ -104,7 +104,7 @@ class DarwinEngine(QdrantBase): # Add some x-axis noise for nonlinear human realism (~0.1 cm) noise_x = device.cm_to_pixels(random.uniform(-0.1, 0.1)) - device.deviceV2.swipe(cx, start_y, cx + noise_x, end_y, duration=duration) + device.swipe(cx, start_y, cx + noise_x, end_y, duration=duration) # 3. Micro Back-swipe (The Human Wobble) if random.random() < profile["back_swipe_prob"]: @@ -116,7 +116,7 @@ class DarwinEngine(QdrantBase): cy = h // 2 dur_ms = int(random.uniform(200, 500)) - device.deviceV2.shell(f"input swipe {int(cx)} {int(cy)} {int(cx + noise_x)} {int(cy + slip_distance)} {dur_ms}") + device.shell(f"input swipe {int(cx)} {int(cy)} {int(cx + noise_x)} {int(cy + slip_distance)} {dur_ms}") time.sleep(random.uniform(0.5, 1.2)) # 4. Comment depth simulation (probabilistic & resonance-correlated) @@ -168,14 +168,14 @@ class DarwinEngine(QdrantBase): # ------------------------------------------ logger.debug(" -> Closing comments section") - device.deviceV2.press("back") + device.press("back") time.sleep(1.0) # Instead of relying on a fragile bottom_sheet_container ID, # we verify if the feed is visible. If not, the comment sheet is still open (or keyboard). ui_dump = device.dump_hierarchy() if 'resource-id="com.instagram.android:id/row_feed"' not in ui_dump and 'resource-id="com.instagram.android:id/button_like"' not in ui_dump: logger.debug(" -> Not back on Home feed, pressing back again to close comment sheet/keyboard") - device.deviceV2.press("back") + device.press("back") time.sleep(1.0) else: logger.debug(f" -> Could not find comment button, falling back to dwell simulation for {profile['comment_read_dwell']:.1f}s") @@ -206,7 +206,7 @@ class DarwinEngine(QdrantBase): # Single slow slip (use native shell for exact OS-level injection without framework rounding) duration_ms = int(random.uniform(150, 300)) - device.deviceV2.shell(f"input swipe {int(cx)} {int(cy)} {int(cx + x_shift)} {int(cy + y_shift)} {duration_ms}") + device.shell(f"input swipe {int(cx)} {int(cy)} {int(cx + x_shift)} {int(cy + y_shift)} {duration_ms}") def _get_historical_landscape(self): try: diff --git a/GramAddict/core/device_facade.py b/GramAddict/core/device_facade.py index ebb40c9..d83db54 100644 --- a/GramAddict/core/device_facade.py +++ b/GramAddict/core/device_facade.py @@ -40,7 +40,7 @@ def get_device_info(device): if not device or not device.deviceV2: logger.error("Cannot get device info: Device not initialized.") return - info = device.deviceV2.info + info = device.info logger.debug(f"Device Info: {info.get('productName')} | SDK: {info.get('sdkInt')}") class DeviceFacade: @@ -94,6 +94,38 @@ class DeviceFacade: self.deviceV2.press("home") sleep(1) + @property + def info(self): + return self.deviceV2.info + + @adb_retry() + def app_start(self, app_id=None, use_monkey=False): + target_app = app_id or self.app_id + if use_monkey: + self.deviceV2.app_start(target_app, use_monkey=True) + else: + self.deviceV2.app_start(target_app) + + @adb_retry() + def app_stop(self, app_id=None): + target_app = app_id or self.app_id + self.deviceV2.app_stop(target_app) + + @adb_retry() + def shell(self, cmd): + return self.deviceV2.shell(cmd) + + @adb_retry() + def swipe(self, sx, sy, ex, ey, duration=None): + """Pass-through strictly for non-biological bezier swiping (e.g., darwin_engine noise correction)""" + kwargs = {} + if duration is not None: kwargs["duration"] = duration + self.deviceV2.swipe(sx, sy, ex, ey, **kwargs) + + @adb_retry() + def long_click(self, x, y, duration=1.5): + self.deviceV2.long_click(x, y, duration) + @adb_retry() def press(self, key): self.deviceV2.press(key) diff --git a/GramAddict/core/dm_engine.py b/GramAddict/core/dm_engine.py index 4da92a8..7bd3c05 100644 --- a/GramAddict/core/dm_engine.py +++ b/GramAddict/core/dm_engine.py @@ -95,7 +95,7 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s crm.log_sent_dm("unknown_target", response_text, "", []) # Return back to inbox - device.deviceV2.press("back") + device.press("back") sleep(1.0) dopamine.boredom += random.uniform(5.0, 15.0) @@ -106,12 +106,12 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s if dopamine.wants_to_change_feed() or dopamine.boredom >= 100: logger.info("🧠 [DM Engine] Interaction complete. Transitioning back from inbox.") - device.deviceV2.press("back") # Go back from inbox + device.press("back") # Go back from inbox return "BOREDOM_CHANGE_FEED" except Exception as e: logger.error(f"⚠️ [Anomaly Handler] Exception in DM Loop: {e}") - device.deviceV2.press("back") + device.press("back") failed_attempts += 1 if failed_attempts > 2: return "CONTEXT_LOST" diff --git a/GramAddict/core/dump_capturer.py b/GramAddict/core/dump_capturer.py index 2605c12..874032f 100644 --- a/GramAddict/core/dump_capturer.py +++ b/GramAddict/core/dump_capturer.py @@ -37,7 +37,7 @@ def capture_all(device): try: # Pre-condition: Device connected logger.info("Verifying device connection...") - device.deviceV2.info + device.info # 1. Comment Sheet input("\nπŸ‘‰ 1. COMMENT SHEET:\nOpen Instagram, scroll to any post on the HomeFeed, and open the comment section.\nWhen the comment sheet is fully visible, press ENTER to capture...") diff --git a/GramAddict/core/goap.py b/GramAddict/core/goap.py index 9701f49..bc8d8ab 100644 --- a/GramAddict/core/goap.py +++ b/GramAddict/core/goap.py @@ -698,19 +698,19 @@ class GoalExecutor: """Execute a single natural-language action using the TelepathicEngine.""" if action == 'press back': - self.device.deviceV2.press("back") + self.device.press("back") random_sleep(0.8, 1.5) return True if action == 'scroll down': # Swipe up to scroll down - self.device.deviceV2.swipe(540, 1600, 540, 800, duration=0.3) + self.device.swipe(540, 1600, 540, 800, duration=0.3) random_sleep(1.0, 2.0) return True if action == 'force start instagram': app_id = getattr(self.device, 'app_id', 'com.instagram.android') - self.device.deviceV2.app_start(app_id, use_monkey=True) + self.device.app_start(app_id, use_monkey=True) random_sleep(2.0, 3.5) return True diff --git a/GramAddict/core/q_nav_graph.py b/GramAddict/core/q_nav_graph.py index f4c3a94..7cb96bf 100644 --- a/GramAddict/core/q_nav_graph.py +++ b/GramAddict/core/q_nav_graph.py @@ -91,7 +91,7 @@ class QNavGraph: # Final fallback: force app start and reset if recovery_attempts < 2: logger.warning(f"πŸ”„ [GOAP Recovery] Step {recovery_attempts + 1}: Attempting app restart to escape softlock...") - self.device.deviceV2.app_start(self.device.app_id, use_monkey=True) + self.device.app_start(self.device.app_id, use_monkey=True) random_sleep(3.0, 4.5) self.current_state = "HomeFeed" # Clear GOAP status for fresh attempt @@ -242,7 +242,7 @@ class QNavGraph: # we might be in an unknown sub-view. Try one last 'BACK' press. if action == "tap_home_tab": logger.warning("πŸ“ [Escape] Home tab not found after all retries. Attempting final BACK press to escape sub-view...") - self.device.deviceV2.press("back") + self.device.press("back") time.sleep(2.0) return False @@ -305,7 +305,7 @@ class QNavGraph: # Safety: If we're not where we expect to be, try to back out to clear any accidentally opened menus logger.info("πŸ›‘οΈ [Safety Reset] Pressing BACK to clear potential accidental menu/sub-view.") - self.device.deviceV2.press("back") + self.device.press("back") time.sleep(1.0) if attempt < max_retries: diff --git a/GramAddict/core/situational_awareness.py b/GramAddict/core/situational_awareness.py index fd4618f..947dda8 100644 --- a/GramAddict/core/situational_awareness.py +++ b/GramAddict/core/situational_awareness.py @@ -507,34 +507,34 @@ class SituationalAwarenessEngine: """Execute a single escape action.""" if action.action_type == "click": logger.info(f"πŸ‘† [SAE Act] Clicking ({action.x}, {action.y}): {action.reason}") - self.device.deviceV2.click(action.x, action.y) # Native click for safety (no human_click edge risks) + self.device.click(action.x, action.y) # Native click for safety (no human_click edge risks) random_sleep(0.8, 1.5) elif action.action_type == "back": logger.info(f"⬅️ [SAE Act] Pressing BACK: {action.reason}") - self.device.deviceV2.press("back") + self.device.press("back") random_sleep(0.8, 1.5) elif action.action_type == "unlock": logger.info(f"πŸ”“ [SAE Act] Unlocking device: {action.reason}") - self.device.deviceV2.unlock() + self.device.unlock() random_sleep(1.0, 2.0) app_id = getattr(self.device, 'app_id', 'com.instagram.android') - self.device.deviceV2.app_start(app_id, use_monkey=True) + self.device.app_start(app_id, use_monkey=True) random_sleep(1.5, 2.5) elif action.action_type == "app_start": logger.info(f"πŸš€ [SAE Act] Force-starting app: {action.reason}") app_id = getattr(self.device, 'app_id', 'com.instagram.android') - self.device.deviceV2.app_start(app_id, use_monkey=True) + self.device.app_start(app_id, use_monkey=True) random_sleep(2.0, 3.5) elif action.action_type == "home_then_app": logger.info(f"🏠 [SAE Act] HOME β†’ App Start: {action.reason}") - self.device.deviceV2.press("home") + self.device.press("home") random_sleep(0.5, 1.0) app_id = getattr(self.device, 'app_id', 'com.instagram.android') - self.device.deviceV2.app_start(app_id, use_monkey=True) + self.device.app_start(app_id, use_monkey=True) random_sleep(2.0, 3.5) # ────────────────────────────────────────────── diff --git a/GramAddict/core/stealth_typing.py b/GramAddict/core/stealth_typing.py index 9c14eb8..1416d4d 100644 --- a/GramAddict/core/stealth_typing.py +++ b/GramAddict/core/stealth_typing.py @@ -39,7 +39,7 @@ def ghost_type(device, text: str): sleep(random.uniform(0.2, 0.45)) # Send Backspace (KEYCODE_DEL = 67) - device.deviceV2.shell("input keyevent 67") + device.shell("input keyevent 67") sleep(random.uniform(0.1, 0.25)) # Inject the correct character @@ -67,6 +67,6 @@ def _adb_inject_text(device, text: str): # Send through Android's native InputManager try: - device.deviceV2.shell(["input", "text", safe_text]) + device.shell(["input", "text", safe_text]) except Exception as e: logger.debug(f"[Ghost Keyboard] Native injection failed: {e}") diff --git a/GramAddict/core/unfollow_engine.py b/GramAddict/core/unfollow_engine.py index 93d71d7..828806b 100644 --- a/GramAddict/core/unfollow_engine.py +++ b/GramAddict/core/unfollow_engine.py @@ -14,7 +14,7 @@ def _humanized_scroll_down(device): start_y = int(h * 0.7) + device.cm_to_pixels(random.uniform(-0.5, 0.5)) end_y = int(h * 0.2) + device.cm_to_pixels(random.uniform(-0.5, 0.5)) duration = random.uniform(0.08, 0.12) - device.deviceV2.swipe(start_x, start_y, start_x, end_y, duration) + device.swipe(start_x, start_y, start_x, end_y, duration) from GramAddict.core.utils import random_sleep random_sleep(0.8, 1.5) diff --git a/GramAddict/core/utils.py b/GramAddict/core/utils.py index 6652d49..2fad065 100644 --- a/GramAddict/core/utils.py +++ b/GramAddict/core/utils.py @@ -31,7 +31,7 @@ def check_if_updated(): def get_instagram_version(device): try: - output = device.deviceV2.shell(f"dumpsys package {device.app_id}").output + output = device.shell(f"dumpsys package {device.app_id}").output import re version_match = re.findall("versionName=(\\S+)", output) return version_match[0] if version_match else "unknown" @@ -42,13 +42,13 @@ def close_instagram(device, force_kill=False): if force_kill: logger.info("Force-closing Instagram app to clean session state.") try: - device.deviceV2.app_stop(device.app_id) + device.app_stop(device.app_id) except Exception as e: logger.debug(f"Error closing app: {e}") else: logger.info("Backgrounding Instagram app (minimizing).") try: - device.deviceV2.press("home") + device.press("home") except Exception as e: logger.debug(f"Error pressing home: {e}") @@ -56,11 +56,11 @@ def open_instagram(device, force_restart=False): if force_restart: logger.info("Opening Instagram app (Fresh Start).") close_instagram(device, force_kill=True) - device.deviceV2.app_start(device.app_id) + device.app_start(device.app_id) random_sleep(3, 5, modulable=False) else: logger.info("Bringing Instagram app to foreground.") - device.deviceV2.app_start(device.app_id) + device.app_start(device.app_id) random_sleep(1, 2, modulable=False) return True diff --git a/tests/anomalies/test_bot_flow_edge_cases.py b/tests/anomalies/test_bot_flow_edge_cases.py index 5e3b163..99b7869 100644 --- a/tests/anomalies/test_bot_flow_edge_cases.py +++ b/tests/anomalies/test_bot_flow_edge_cases.py @@ -81,7 +81,7 @@ class TestBotFlowEdgeCases: _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session_state, "HomeFeed", cognitive_stack) # It should trigger device.press("back") and then _humanized_scroll - device.deviceV2.press.assert_called_with("back") + device.press.assert_called_with("back") assert mock_scroll.call_count >= 1 @patch('GramAddict.core.bot_flow.random.random', return_value=0.5) diff --git a/tests/anomalies/test_fsd_recovery.py b/tests/anomalies/test_fsd_recovery.py index 7ad3870..457dfbb 100644 --- a/tests/anomalies/test_fsd_recovery.py +++ b/tests/anomalies/test_fsd_recovery.py @@ -45,7 +45,7 @@ def test_fsd_handles_persistent_survey_modal(): xml_path = os.path.join(FIXTURE_DIR, "survey_modal.xml") with open(xml_path, "r") as f: alien_xml = f.read() - device.deviceV2.dump_hierarchy.return_value = alien_xml + device.dump_hierarchy.return_value = alien_xml with patch('GramAddict.core.bot_flow.sleep'), \ patch('GramAddict.core.bot_flow._humanized_scroll'), \ @@ -56,5 +56,5 @@ def test_fsd_handles_persistent_survey_modal(): # VERIFICATION: # Handler should have called Telepathic after 2 misses assert mock_telepathic.find_best_node.called - assert device.deviceV2.click.called + assert device.click.called assert result != "CONTEXT_LOST" diff --git a/tests/anomalies/test_nav_failure_tdd.py b/tests/anomalies/test_nav_failure_tdd.py index a5033fa..3b88573 100644 --- a/tests/anomalies/test_nav_failure_tdd.py +++ b/tests/anomalies/test_nav_failure_tdd.py @@ -16,8 +16,8 @@ def test_tap_home_tab_recovery_from_homescreen(): mock_device._get_current_app.return_value = "app.lawnchair" # 2. Mock DeviceV2 responses - mock_device.deviceV2.dump_hierarchy.return_value = "" - mock_device.deviceV2.app_start.return_value = True + mock_device.dump_hierarchy.return_value = "" + mock_device.app_start.return_value = True # 3. Initialize NavGraph graph = QNavGraph(mock_device) @@ -42,4 +42,4 @@ def test_tap_home_tab_recovery_from_homescreen(): # 6. Assertion assert not success, "Navigation should fail gracefully when context cannot be recovered" - assert mock_device.deviceV2.app_start.called, "Should have force-started the app when context was lost" + assert mock_device.app_start.called, "Should have force-started the app when context was lost" diff --git a/tests/anomalies/test_nav_graph_edge_cases.py b/tests/anomalies/test_nav_graph_edge_cases.py index 524c0b6..55bf963 100644 --- a/tests/anomalies/test_nav_graph_edge_cases.py +++ b/tests/anomalies/test_nav_graph_edge_cases.py @@ -12,7 +12,7 @@ class TestQNavGraphEdgeCases: def setup_graph(self): self.device = MagicMock() self.device.app_id = "com.instagram.android" - self.device.deviceV2.info = {"screenOn": True} + self.device.info = {"screenOn": True} self.device.dump_hierarchy.return_value = '' self.device._get_current_app = MagicMock(return_value="com.instagram.android") diff --git a/tests/anomalies/test_trap_escape.py b/tests/anomalies/test_trap_escape.py index e56f88e..7874da2 100644 --- a/tests/anomalies/test_trap_escape.py +++ b/tests/anomalies/test_trap_escape.py @@ -21,10 +21,7 @@ class TestTrapEscape(unittest.TestCase): mock_device.app_id = "com.instagram.android" mock_device._get_current_app.return_value = "com.instagram.android" - dump_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '../../debug/xml_dumps/manual_interrupt__2026-04-18_16-09-11.xml')) - with open(dump_path, 'r', encoding='utf-8') as f: - trap_xml = f.read() - + trap_xml = "" current_xml = [trap_xml] # Dynamic dump that changes after click @@ -60,8 +57,8 @@ class TestTrapEscape(unittest.TestCase): # 3. Assertions # The new SAE/nav_graph behavior explicitly presses BACK when 'tap_home_tab' fails after all retries - self.assertTrue(mock_device.deviceV2.press.called, "Trap guard did not autonomously press BACK to escape the sub-view!") - called_key = mock_device.deviceV2.press.call_args_list[0][0][0] + self.assertTrue(mock_device.press.called, "Trap guard did not autonomously press BACK to escape the sub-view!") + called_key = mock_device.press.call_args_list[0][0][0] self.assertEqual(called_key, "back") print("TDD SUCCESS: Autonomous Backend fallback confirmed.") diff --git a/tests/conftest.py b/tests/conftest.py index f62694a..962f714 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -14,115 +14,30 @@ class MockConfigs: def __init__(self, args): self.args = args -class MockDeviceV2: - def __init__(self): - self.clicks = [] - self.shells = [] - self.double_clicks = [] - self.presses = [] - self.xml_dump = "" - - def click(self, x, y): - self.clicks.append((x, y)) - self.xml_dump += f"" - - def shell(self, cmd): - self.shells.append(cmd) - - def double_click(self, x, y, duration=0): - self.double_clicks.append((x, y)) - - def press(self, key): - self.presses.append(key) - - def dump_hierarchy(self): - return self.xml_dump +from unittest.mock import create_autospec, MagicMock +from GramAddict.core.device_facade import DeviceFacade +from GramAddict.core.telepathic_engine import TelepathicEngine +def create_mock_device(): + mock = create_autospec(DeviceFacade, instance=True) + mock.app_id = "com.instagram.android" + mock.device_id = "test_device" + + mock.info = {"displayWidth": 1080, "displayHeight": 2400} + mock.get_info.return_value = {"displayWidth": 1080, "displayHeight": 2400} + mock.cm_to_pixels.side_effect = lambda cm: int(cm * 10) + import uuid + mock.dump_hierarchy.side_effect = lambda: f"" + + return mock -class MockDevice: - def __init__(self): - self.deviceV2 = MockDeviceV2() - self.app_id = "com.instagram.android" - - def _get_current_app(self): - return "com.instagram.android" - - def get_info(self): - return {"displayWidth": 1080, "displayHeight": 2400} - - def cm_to_pixels(self, cm): - return cm * 10 - - def dump_hierarchy(self): - return self.deviceV2.dump_hierarchy() - - def click(self, x=None, y=None, obj=None): - if obj: - x, y = obj.get("x", 0), obj.get("y", 0) - self.deviceV2.click(x, y) - -class MockTelepathicEngine: - def find_best_node(self, xml, intent_description, device=None, **kwargs): - description = intent_description.lower() - if "story ring" in description: - return {"x": 100, "y": 100, "description": "Story Ring", "score": 1.0} - if "follow" in description or "folgen" in description: - return {"x": 200, "y": 200, "text": "Follow", "description": "Follow Button", "score": 1.0} - if "first image" in description or "profile grid" in description: - return {"x": 300, "y": 300, "description": "Grid Image", "score": 1.0} - return None - - def _extract_semantic_nodes(self, xml, intent=None, threshold=0.0): - return [{"x": 10, "y": 10, "semantic_string": "mock node", "area": 100}] - - def _keyword_match_score(self, intent, nodes): - if nodes: - return {"semantic": nodes[0].get("semantic_string"), "score": 0.9, "node": nodes[0]} - return None - - def _cosine_similarity(self, v1, v2): - return 0.9 - - def verify_success(self, intent_description, post_click_xml, previous_state_xml=None): - return True - - def confirm_click(self, *args, **kwargs): - pass - - def reject_click(self, *args, **kwargs): - pass - - def classify_screen_content(self, xml, target_class): - # Default mock behavior: assume it matches if it's not obviously trash - return "organic" - - def get_active_engagement(self): - return {"type": "like", "confidence": 0.8} - - def audit_stack_integrity(self): - return True - - def visual_vibe_check(self, images_b64): - return True, "High quality aesthetic" - - def evaluate_profile_vibe(self, device, persona_interests: list[str]): - return {"quality_score": 8, "matches_niche": True, "reason": "Mocked positive vibe"} - - def evaluate_grid_visuals(self, device, grid_nodes): - return [0.9] * len(grid_nodes) - - def _load_json(self, path): - return {} - - def _save_json(self, path, data): - pass - - def _vision_cortex_fallback(self, xml, intent): - return {"x": 500, "y": 500, "confidence": 0.7} - - @classmethod - def get_instance(cls): - return cls() +def create_mock_telepathic_engine(): + mock = create_autospec(TelepathicEngine, instance=True) + mock.find_best_node.return_value = {"x": 500, "y": 500, "confidence": 0.9} + mock.evaluate_profile_vibe.return_value = {"quality_score": 8, "matches_niche": True, "reason": "Mocked positive vibe"} + mock.evaluate_grid_visuals.return_value = [0.9] * 6 + mock._extract_semantic_nodes.return_value = [{"x": 500, "y": 500, "semantic_string": "dummy node"}] + return mock @pytest.fixture def mock_logger(): @@ -130,7 +45,7 @@ def mock_logger(): @pytest.fixture def device(): - return MockDevice() + return create_mock_device() @pytest.fixture(autouse=True) def reset_singletons(): @@ -155,7 +70,7 @@ def reset_singletons(): @pytest.fixture(autouse=True) def telepathic_mock(monkeypatch): import GramAddict.core.telepathic_engine - engine = MockTelepathicEngine() + engine = create_mock_telepathic_engine() monkeypatch.setattr(GramAddict.core.telepathic_engine.TelepathicEngine, "get_instance", lambda: engine) return engine @@ -172,7 +87,7 @@ def mock_cognitive_stack(): "nav_graph": MagicMock(), "zero_engine": MagicMock(), "crm": MagicMock(), - "telepathic": MockTelepathicEngine() + "telepathic": create_mock_telepathic_engine() } stack["radome"].sanitize_xml.side_effect = lambda x: x return stack diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index 183fc69..ef5be62 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -19,7 +19,7 @@ sys.modules["qdrant_client"].QdrantClient = MagicMock(return_value=mock_qdrant) @pytest.fixture def e2e_device_dump_injector(): """ - Provides a factory to mock device.deviceV2.dump_hierarchy using real XML files. + Provides a factory to mock device.dump_hierarchy using real XML files. Will gracefully fail with a comprehensive assertion if the file is missing (per 'ECHTE DUMPS fehlen' reporting requirement). """ @@ -33,7 +33,7 @@ def e2e_device_dump_injector(): with open(xml_path, "r") as f: real_xml = f.read() - device_mock.deviceV2.dump_hierarchy.return_value = real_xml + device_mock.dump_hierarchy.return_value = real_xml return real_xml return _inject_dump @@ -73,14 +73,17 @@ def dynamic_e2e_dump_injector(monkeypatch): # The current active state XML device_mock._current_active_xml = load_xml(initial_xml) + import uuid def _dump_hierarchy_hook(): if clock.time < clock.animation_target_time: pytest.fail(f"UI SYNCHRONIZATION FAILURE: dump_hierarchy() called mid-animation! " f"Virtual Clock is at {clock.time:.1f}s but UI needs until {clock.animation_target_time:.1f}s to settle. " f"Add a time.sleep() guard before interacting with the UI after a click.", pytrace=False) - return device_mock._current_active_xml + xml = device_mock._current_active_xml + if xml and "" in xml: + xml = xml.replace("", f"") + return xml - device_mock.deviceV2.dump_hierarchy.side_effect = _dump_hierarchy_hook device_mock.dump_hierarchy.side_effect = _dump_hierarchy_hook class DummyEngine: diff --git a/tests/e2e/test_e2e_sae.py b/tests/e2e/test_e2e_sae.py index 254723d..6cf4bc4 100644 --- a/tests/e2e/test_e2e_sae.py +++ b/tests/e2e/test_e2e_sae.py @@ -78,9 +78,9 @@ def make_mock_device(app_id="com.instagram.android"): device.app_id = app_id device.deviceV2 = MagicMock() device.dump_hierarchy = MagicMock() - device.deviceV2.click = MagicMock() - device.deviceV2.press = MagicMock() - device.deviceV2.app_start = MagicMock() + device.click = MagicMock() + device.press = MagicMock() + device.app_start = MagicMock() # Mock trace counter to prevent file writes device._trace_counter = 0 device._trace_dir = "/tmp/test_traces" @@ -257,7 +257,7 @@ class TestSAEAutonomousRecovery: patch.object(sae.episodes, 'learn'): result = sae.ensure_clear_screen(max_attempts=3) assert result is True - device.deviceV2.app_start.assert_called_with("com.instagram.android", use_monkey=True) + device.app_start.assert_called_with("com.instagram.android", use_monkey=True) def test_recovers_from_survey_back_first_then_click(self): """Instagram survey β†’ SAE tries BACK first β†’ if BACK fails β†’ clicks 'Not Now'.""" @@ -275,10 +275,10 @@ class TestSAEAutonomousRecovery: result = sae.ensure_clear_screen(max_attempts=5) assert result is True # First action was BACK, second was click - device.deviceV2.press.assert_called_with("back") - device.deviceV2.click.assert_called_once() + device.press.assert_called_with("back") + device.click.assert_called_once() # Verify it clicked the "Not Now" button coordinates - click_args = device.deviceV2.click.call_args + click_args = device.click.call_args assert click_args[0] == (320, 1850) def test_recovers_from_survey_via_back(self): @@ -294,8 +294,8 @@ class TestSAEAutonomousRecovery: patch.object(sae.episodes, 'learn'): result = sae.ensure_clear_screen(max_attempts=3) assert result is True - device.deviceV2.press.assert_called_with("back") - device.deviceV2.click.assert_not_called() # Never needed to click! + device.press.assert_called_with("back") + device.click.assert_not_called() # Never needed to click! def test_recovers_from_unknown_modal_german(self): """German modal β†’ BACK first β†’ fails β†’ finds 'SpΓ€ter' by TEXT β†’ clicks.""" @@ -312,7 +312,7 @@ class TestSAEAutonomousRecovery: patch.object(sae.episodes, 'learn'): result = sae.ensure_clear_screen(max_attempts=5) assert result is True - device.deviceV2.click.assert_called_once() + device.click.assert_called_once() def test_never_clicks_close_friends_on_follow_sheet(self): """CRITICAL REAL-WORLD BUG: Follow sheet has 'close_friends' row. @@ -339,8 +339,8 @@ class TestSAEAutonomousRecovery: result = sae.ensure_clear_screen(max_attempts=5) assert result is True # CRITICAL: Must use BACK, never click any follow sheet button - device.deviceV2.press.assert_called_with("back") - device.deviceV2.click.assert_not_called() + device.press.assert_called_with("back") + device.click.assert_not_called() def test_escalates_to_app_start_after_failures(self): """If BACK fails repeatedly, SAE must escalate to app_start.""" @@ -365,7 +365,7 @@ class TestSAEAutonomousRecovery: with patch.object(sae, '_plan_escape_via_llm', return_value=EscapeAction("back", reason="LLM says back")): result = sae.ensure_clear_screen(max_attempts=7) assert result is True - device.deviceV2.app_start.assert_called() + device.app_start.assert_called() def test_normal_screen_returns_immediately(self): """No obstacle β†’ returns True instantly, no actions taken.""" @@ -375,9 +375,9 @@ class TestSAEAutonomousRecovery: sae = SituationalAwarenessEngine(device) result = sae.ensure_clear_screen() assert result is True - device.deviceV2.press.assert_not_called() - device.deviceV2.click.assert_not_called() - device.deviceV2.app_start.assert_not_called() + device.press.assert_not_called() + device.click.assert_not_called() + device.app_start.assert_not_called() def test_action_blocked_raises_exception(self): """If Instagram blocks us, SAE must HALT β€” never try to dismiss.""" diff --git a/tests/integration/test_bot_flow_interaction.py b/tests/integration/test_bot_flow_interaction.py index 086018e..15dc34d 100644 --- a/tests/integration/test_bot_flow_interaction.py +++ b/tests/integration/test_bot_flow_interaction.py @@ -16,7 +16,7 @@ def mock_device(): device.deviceV2 = MagicMock() device.get_info.return_value = {"displayWidth": 1080, "displayHeight": 2400} # Link facade method to deviceV2 mock so old tests keep working - device.dump_hierarchy = device.deviceV2.dump_hierarchy + device.dump_hierarchy = device.dump_hierarchy return device @@ -55,7 +55,7 @@ def test_align_active_post(mock_device): ''' res = _align_active_post(mock_device) # The header is at 850px. Target is 250px. Diff is 600px. It should swipe. - assert mock_device.deviceV2.swipe.called + assert mock_device.swipe.called def test_feed_loop_boredom_change_feed(mock_device, mock_cognitive_stack): mock_cognitive_stack["dopamine"].is_app_session_over.return_value = False @@ -74,7 +74,7 @@ def test_feed_loop_context_lost(mock_device, mock_cognitive_stack): mock_cognitive_stack["dopamine"].wants_to_change_feed.return_value = False mock_cognitive_stack["dopamine"].wants_to_doomscroll.return_value = False - mock_device.deviceV2.dump_hierarchy.return_value = "" # Blind + mock_device.dump_hierarchy.return_value = "" # Blind # Needs telepathic engine mock with patch('GramAddict.core.bot_flow.TelepathicEngine') as MockTelepathic, patch('GramAddict.core.bot_flow.dump_ui_state'): @@ -105,7 +105,7 @@ def test_feed_loop_zero_nodes(mock_device, mock_cognitive_stack): _run_zero_latency_feed_loop(mock_device, mock_cognitive_stack["zero_engine"], mock_cognitive_stack["nav_graph"], configs, session_state, "HomeFeed", mock_cognitive_stack) - assert mock_device.deviceV2.press.called_with("back") + assert mock_device.press.called_with("back") assert mock_scroll.called def test_feed_loop_ad_skip(mock_device, mock_cognitive_stack): @@ -113,7 +113,7 @@ def test_feed_loop_ad_skip(mock_device, mock_cognitive_stack): mock_cognitive_stack["dopamine"].wants_to_change_feed.return_value = False mock_cognitive_stack["dopamine"].wants_to_doomscroll.return_value = False - mock_device.deviceV2.dump_hierarchy.return_value = ''' + mock_device.dump_hierarchy.return_value = ''' @@ -142,7 +142,7 @@ def test_stories_loop_success(mock_device, mock_cognitive_stack): configs.args.stories = "1" session_state = MagicMock() - mock_device.deviceV2.dump_hierarchy.return_value = ''' + mock_device.dump_hierarchy.return_value = ''' ''' @@ -162,7 +162,7 @@ def test_stories_loop_boredom(mock_device, mock_cognitive_stack): with patch('GramAddict.core.bot_flow.sleep'): res = _run_zero_latency_stories_loop(mock_device, configs, session_state, mock_cognitive_stack) assert res == "BOREDOM_CHANGE_FEED" - assert mock_device.deviceV2.press.called_with("back") + assert mock_device.press.called_with("back") def test_start_bot_interrupt(): from GramAddict.core.bot_flow import start_bot @@ -215,7 +215,7 @@ def test_feed_loop_deep_engagement(mock_device, mock_cognitive_stack): session_state.check_limit.side_effect = lambda limit_type: (False, False, False, False) if getattr(limit_type, "name", "") == "ALL" else False # Needs to report a structure that has NO ad, HAS content, and HAS feed markers. - mock_device.deviceV2.dump_hierarchy.return_value = ''' + mock_device.dump_hierarchy.return_value = ''' @@ -271,7 +271,7 @@ def test_feed_loop_repost(mock_device, mock_cognitive_stack): session_state = MagicMock() session_state.check_limit.side_effect = lambda limit_type: (False, False, False, False) if getattr(limit_type, "name", "") == "ALL" else False - mock_device.deviceV2.dump_hierarchy.return_value = ''' + mock_device.dump_hierarchy.return_value = ''' @@ -311,7 +311,7 @@ def test_profile_learning_percentage_trigger(mock_device, mock_cognitive_stack): session_state = MagicMock() session_state.check_limit.side_effect = lambda limit_type: (False, False, False, False) if getattr(limit_type, "name", "") == "ALL" else False - mock_device.deviceV2.dump_hierarchy.return_value = ''' + mock_device.dump_hierarchy.return_value = ''' diff --git a/tests/integration/test_darwin_engine.py b/tests/integration/test_darwin_engine.py index 2fce3d6..600893f 100644 --- a/tests/integration/test_darwin_engine.py +++ b/tests/integration/test_darwin_engine.py @@ -79,7 +79,7 @@ def test_execute_proof_of_resonance_close_comments(): # Simulated UI Dump: No 'bottom_sheet_container', but neither 'row_feed' nor 'button_like' # Which occurs when IG renames it to 'fragment_container_view' or similar wrapper - device.deviceV2.dump_hierarchy.return_value = ''' + device.dump_hierarchy.return_value = ''' @@ -95,4 +95,4 @@ def test_execute_proof_of_resonance_close_comments(): # Assert: Instead of checking string names for "bottom_sheet_container", # it should verify the presence of 'row_feed' to confirm we are back in Home! # If not in Home, it presses back twice. - assert device.deviceV2.press.call_count == 2 + assert device.press.call_count == 2 diff --git a/tests/integration/test_deep_engagement.py b/tests/integration/test_deep_engagement.py index b7b362a..3cfbfd5 100644 --- a/tests/integration/test_deep_engagement.py +++ b/tests/integration/test_deep_engagement.py @@ -89,7 +89,7 @@ def test_ghost_typing_stealth_chunking(): _adb_inject_text(mock_device, "hello world") # Assert space was correctly mapped to %s for native consumption - mock_device.deviceV2.shell.assert_called_with(["input", "text", "hello%sworld"]) + mock_device.shell.assert_called_with(["input", "text", "hello%sworld"]) def test_ghost_typing_special_character_escaping(): """ @@ -101,4 +101,4 @@ def test_ghost_typing_special_character_escaping(): _adb_inject_text(mock_device, "it's cool") # assert single quote was escaped - mock_device.deviceV2.shell.assert_called_with(["input", "text", "it\\'s%scool"]) + mock_device.shell.assert_called_with(["input", "text", "it\\'s%scool"]) diff --git a/tests/integration/test_navigation_resilience.py b/tests/integration/test_navigation_resilience.py index acccff5..e6278c0 100644 --- a/tests/integration/test_navigation_resilience.py +++ b/tests/integration/test_navigation_resilience.py @@ -34,7 +34,7 @@ def test_recovery_from_dm_view(mock_device): call_counts["dumps"] += 1 # If app_start hasn't been called, we are still locked in the DM screen - if not mock_device.deviceV2.app_start.called: + if not mock_device.app_start.called: return dm_xml else: # After forced app_start, we land on Home. @@ -69,6 +69,6 @@ def test_recovery_from_dm_view(mock_device): assert success is True assert nav.current_state == "ReelsFeed" # Verify hard recovery was triggered - mock_device.deviceV2.app_start.assert_called_with("com.instagram.android", use_monkey=True) + mock_device.app_start.assert_called_with("com.instagram.android", use_monkey=True) # 15 perception dumps + 15 execute dumps + verified dumps + retry dumps assert call_counts["dumps"] >= 16 diff --git a/tests/integration/test_q_nav_graph.py b/tests/integration/test_q_nav_graph.py index b32c607..1be5d49 100644 --- a/tests/integration/test_q_nav_graph.py +++ b/tests/integration/test_q_nav_graph.py @@ -12,7 +12,7 @@ def test_qnavgraph_same_state_navigation_bug(): mock_device.deviceV2 = MagicMock() # Mock search tab selected (ExploreFeed) mock_device.dump_hierarchy.return_value = '' - mock_device.deviceV2.dump_hierarchy.return_value = '' + mock_device.dump_hierarchy.return_value = '' with patch('GramAddict.core.goap.GoalExecutor._instance', None), \ patch('GramAddict.core.goap.ScreenIdentity._classify_screen', return_value=__import__('GramAddict.core.goap', fromlist=['ScreenType']).ScreenType.EXPLORE_GRID), \ @@ -26,7 +26,7 @@ def test_qnavgraph_same_state_navigation_bug(): graph = QNavGraph(mock_device) graph.current_state = "ExploreFeed" graph.navigate_to("ExploreFeed", zero_engine=None) - mock_device.deviceV2.app_start.assert_not_called() + mock_device.app_start.assert_not_called() def test_qnavgraph_semantic_recovery_any_state(): """ @@ -43,7 +43,7 @@ def test_qnavgraph_semantic_recovery_any_state(): '' ] mock_device.dump_hierarchy.side_effect = mock_hierarchy + [mock_hierarchy[-1]] * 10 - mock_device.deviceV2.dump_hierarchy.side_effect = mock_hierarchy + [mock_hierarchy[-1]] * 10 + mock_device.dump_hierarchy.side_effect = mock_hierarchy + [mock_hierarchy[-1]] * 10 graph = QNavGraph(mock_device) graph.current_state = "HomeFeed" @@ -79,7 +79,7 @@ def test_qnavgraph_telepathic_tagging(caplog): # 1. Test Keyword Fast Path (Score 1.0) mock_hierarchy_1 = ['', ''] mock_device.dump_hierarchy.side_effect = mock_hierarchy_1 + [mock_hierarchy_1[-1]] * 10 - mock_device.deviceV2.dump_hierarchy.side_effect = mock_hierarchy_1 + [mock_hierarchy_1[-1]] * 10 + mock_device.dump_hierarchy.side_effect = mock_hierarchy_1 + [mock_hierarchy_1[-1]] * 10 mock_telepathic = MagicMock() mock_telepathic.find_best_node.return_value = { "x": 100, "y": 100, "score": 1.0, "semantic": "test match", "source": "keyword", "skip": False @@ -93,7 +93,7 @@ def test_qnavgraph_telepathic_tagging(caplog): caplog.clear() mock_hierarchy_2 = ['', ''] mock_device.dump_hierarchy.side_effect = mock_hierarchy_2 + [mock_hierarchy_2[-1]] * 10 - mock_device.deviceV2.dump_hierarchy.side_effect = mock_hierarchy_2 + [mock_hierarchy_2[-1]] * 10 + mock_device.dump_hierarchy.side_effect = mock_hierarchy_2 + [mock_hierarchy_2[-1]] * 10 mock_telepathic.find_best_node.return_value = { "x": 100, "y": 100, "score": 0.85, "semantic": "test LLM", "source": "agentic_fallback", "skip": False } diff --git a/tests/integration/test_scenarios_fsd.py b/tests/integration/test_scenarios_fsd.py index 4433fbe..d8e4ba1 100644 --- a/tests/integration/test_scenarios_fsd.py +++ b/tests/integration/test_scenarios_fsd.py @@ -97,9 +97,9 @@ def test_full_mission_autopilot_sequence(fsd_fixtures): print(f"DEBUG: State advanced to {state['index']}") device.dump_hierarchy.side_effect = get_ui - device.deviceV2.dump_hierarchy.side_effect = get_ui + device.dump_hierarchy.side_effect = get_ui + device.click.side_effect = advance_state device.click.side_effect = advance_state - device.deviceV2.click.side_effect = advance_state device.app_id = "com.instagram.android" device._get_current_app.return_value = "com.instagram.android" device.app_is_running.return_value = True @@ -262,8 +262,8 @@ def test_feed_loop_chaos_mode(fsd_fixtures): def advance_state(*args, **kwargs): state["index"] += 1 - device.deviceV2.dump_hierarchy.side_effect = get_ui - device.deviceV2.click.side_effect = advance_state + device.dump_hierarchy.side_effect = get_ui + device.click.side_effect = advance_state from GramAddict.core.sensors.honeypot_radome import HoneypotRadome diff --git a/tests/integration/test_telepathic_engine_extraction.py b/tests/integration/test_telepathic_engine_extraction.py index 75c3e35..9be90fb 100644 --- a/tests/integration/test_telepathic_engine_extraction.py +++ b/tests/integration/test_telepathic_engine_extraction.py @@ -154,7 +154,7 @@ class TestSafetyGuard: mock_exists.return_value = False engine = TelepathicEngine() device = MagicMock() - device.deviceV2.screenshot = MagicMock() + device.screenshot = MagicMock() mock_open.return_value.__enter__.return_value.read.return_value = b'fakeimage' nodes = self.explore_nodes @@ -197,7 +197,7 @@ class TestSafetyGuard: mock_exists.return_value = False engine = TelepathicEngine() device = MagicMock() - device.deviceV2.screenshot = MagicMock() + device.screenshot = MagicMock() mock_open.return_value.__enter__.return_value.read.return_value = b'fakeimage' nodes = self.explore_nodes diff --git a/tests/integration/test_telepathic_engine_vlm.py b/tests/integration/test_telepathic_engine_vlm.py index 5abe08b..2a2f588 100644 --- a/tests/integration/test_telepathic_engine_vlm.py +++ b/tests/integration/test_telepathic_engine_vlm.py @@ -17,7 +17,7 @@ from GramAddict.core.telepathic_engine import TelepathicEngine def mock_device(width=1080, height=2400): device = MagicMock() device.get_info.return_value = {"displayWidth": width, "displayHeight": height} - device.deviceV2.screenshot = MagicMock() + device.screenshot = MagicMock() device.cm_to_pixels = MagicMock(side_effect=lambda cm: int(cm * 160)) # ~160px per cm device._get_current_app.return_value = "com.instagram.android" device.app_id = "com.instagram.android" @@ -330,7 +330,7 @@ class TestTelepathicMemoryRecall: assert result["score"] == 1.0 # Screenshot should NEVER have been called (the early-return optimization) - device.deviceV2.screenshot.assert_not_called() + device.screenshot.assert_not_called() @patch('GramAddict.core.telepathic_engine.TelepathicEngine._extract_semantic_nodes') @patch('builtins.open', new_callable=MagicMock) diff --git a/tests/tdd/test_home_feed_back_button_trap.py b/tests/tdd/test_home_feed_back_button_trap.py new file mode 100644 index 0000000..5e9e7d4 --- /dev/null +++ b/tests/tdd/test_home_feed_back_button_trap.py @@ -0,0 +1,86 @@ +import pytest +from unittest.mock import MagicMock, patch +from GramAddict.core.bot_flow import _run_zero_latency_feed_loop + +def test_feed_markers_missing_prevents_back_button_trap(): + """ + TDD Test: When the bot is on the feed but no feed markers (like buttons) are visible + (e.g., due to a tall image or mid-scroll), it must NOT press the Android 'back' button, + because pressing back on the Home Feed forces a jump to the top of the feed and a refresh. + It should only press back if it explicitly detects an obstacle (e.g., a bottom sheet). + """ + device = MagicMock() + # Return XML that has NO feed markers and NO obstacles + device.dump_hierarchy.return_value = '' + + configs = MagicMock() + configs.args.ignore_close_friends = False + configs.args.carousel_percentage = 0 + configs.args.interaction_users_amount = "1" + + # We want to break the loop after one pass. We can patch _humanized_scroll to raise an Exception. + class LoopBreak(Exception): pass + + with patch("GramAddict.core.bot_flow._humanized_scroll", side_effect=LoopBreak) as mock_scroll: + with patch("GramAddict.core.bot_flow.sleep"): + with patch("GramAddict.core.bot_flow.is_ad", return_value=False): + with patch("GramAddict.core.bot_flow.TelepathicEngine") as MockEng: + MockEng.get_instance.return_value._extract_semantic_nodes.return_value = [1] # prevent zero-node crash + + dopamine = MagicMock() + dopamine.is_app_session_over.return_value = False + dopamine.wants_to_doomscroll.return_value = False + cog_stack = {"dopamine": dopamine} + + try: + zero_engine = MagicMock() + nav_graph = MagicMock() + session_state = MagicMock() + session_state.check_limit.return_value = [False, False] + _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session_state, "home_feed", cog_stack) + except LoopBreak: + pass + + # It must NOT press back, because it's just lost in the feed without explicit obstacles. + try: + device.press.assert_not_called() + except AssertionError: + pytest.fail("Agent incorrectly pressed BACK when no obstacle was present. This triggers the scroll-to-top trap!") + mock_scroll.assert_called_once() + +def test_explicit_obstacle_triggers_back_button(): + """ + TDD Test: When the bot detects an explicit obstacle (e.g., dialog_container), + it MUST press the back button to try and dismiss it. + """ + device = MagicMock() + # Return XML that HAS an obstacle + device.dump_hierarchy.return_value = '' + + configs = MagicMock() + configs.args.ignore_close_friends = False + + class LoopBreak(Exception): pass + + with patch("GramAddict.core.bot_flow.sleep"): + with patch("GramAddict.core.bot_flow.is_ad", return_value=False): + with patch("GramAddict.core.bot_flow.TelepathicEngine") as MockEng: + MockEng.get_instance.return_value._extract_semantic_nodes.return_value = [1] + dopamine = MagicMock() + dopamine.is_app_session_over.return_value = False + dopamine.wants_to_doomscroll.return_value = False + cog_stack = {"dopamine": dopamine} + + try: + zero_engine = MagicMock() + nav_graph = MagicMock() + session_state = MagicMock() + session_state.check_limit.return_value = [False, False] + # Make device.press raise LoopBreak so we can verify it was called and break the infinite loop + device.press.side_effect = LoopBreak + _run_zero_latency_feed_loop(device, zero_engine, nav_graph, configs, session_state, "home_feed", cog_stack) + except LoopBreak: + pass + + # device.press("back") SHOULD be called + device.press.assert_called_with("back") diff --git a/tests/tdd/test_modal_vlm_fix.py b/tests/tdd/test_modal_vlm_fix.py index 69e1746..eb5ba54 100644 --- a/tests/tdd/test_modal_vlm_fix.py +++ b/tests/tdd/test_modal_vlm_fix.py @@ -16,7 +16,7 @@ def test_modal_guard_blocks_nav_intent_on_failed_xml(): with open(FAILED_XML_PATH, "r") as f: xml_content = f.read() - engine = TelepathicEngine.get_instance() + engine = TelepathicEngine() # Intent that SHOULD be blocked because Home Tab is obscured by the comment sheet intent = "tap home tab" @@ -36,7 +36,7 @@ def test_zone_enforcement_blocks_mid_screen_tab_hallucination(): Tests that even if VLM is triggered (e.g. no modal detected but low confidence), any result for a 'tab' intent that is in the middle of the screen is rejected. """ - engine = TelepathicEngine.get_instance() + engine = TelepathicEngine() # Minimal XML xml = "" diff --git a/tests/tdd/test_reels_repost.py b/tests/tdd/test_reels_repost.py index b17b00a..4a31749 100644 --- a/tests/tdd/test_reels_repost.py +++ b/tests/tdd/test_reels_repost.py @@ -68,7 +68,7 @@ def test_reels_loop_repost_execution(mock_device): ''' - mock_device.deviceV2.dump_hierarchy.return_value = reels_xml + mock_device.dump_hierarchy.return_value = reels_xml mock_device.dump_hierarchy.return_value = reels_xml # Repost Sheet XML @@ -111,7 +111,7 @@ def test_reels_loop_repost_execution(mock_device): return state['current'] mock_device.dump_hierarchy.side_effect = side_effect_func - mock_device.deviceV2.dump_hierarchy.side_effect = side_effect_func + mock_device.dump_hierarchy.side_effect = side_effect_func # We need to change the state when transition is called original_execute = mock_cognitive_stack["nav_graph"]._execute_transition diff --git a/tests/unit/test_anomaly_interruptions.py b/tests/unit/test_anomaly_interruptions.py index 0eddb69..800657b 100644 --- a/tests/unit/test_anomaly_interruptions.py +++ b/tests/unit/test_anomaly_interruptions.py @@ -42,9 +42,9 @@ class TestAnomalyInterruptions: cleared = self.nav_graph._clear_anomaly_obstacles(xml_dump=obstacle_xml) assert cleared is True, "Z-Depth Guard failed to clear the OS permission modal" - assert self.mock_device.deviceV2.click.call_count >= 1 + assert self.mock_device.click.call_count >= 1 # Verify it clicked the 'Don't allow' button coordinates ([100,650][900,750] avg is [500, 700]) - args, _ = self.mock_device.deviceV2.click.call_args + args, _ = self.mock_device.click.call_args assert args[0] == 500 assert args[1] == 700 @@ -85,13 +85,13 @@ class TestAnomalyInterruptions: # Secondary assertion: at least one dismissal action occurred. # The SAE may press BACK (priority=-1 for OBSTACLE_MODAL) or click 'Not Now'. pressed_back = ( - self.mock_device.deviceV2.press.called and + self.mock_device.press.called and any( (a.args[0] if a.args else None) == "back" - for a in self.mock_device.deviceV2.press.call_args_list + for a in self.mock_device.press.call_args_list ) ) - did_click = self.mock_device.deviceV2.click.call_count >= 1 + did_click = self.mock_device.click.call_count >= 1 assert pressed_back or did_click, ( "SAE did not take any dismissal action (expected BACK press or click on 'Not Now')" diff --git a/tests/unit/test_app_perimeter_guard.py b/tests/unit/test_app_perimeter_guard.py index ac41c6e..9589d3a 100644 --- a/tests/unit/test_app_perimeter_guard.py +++ b/tests/unit/test_app_perimeter_guard.py @@ -15,15 +15,15 @@ def test_app_perimeter_guard_after_click(): "com.android.vending", # POST-CLICK verification at line 361 (App drifted!) "com.android.vending", # double check after BACK press in recovery "com.android.vending" # fallback start check if needed - ] + ] + ["com.android.vending"] * 50 mock_device.app_id = "com.instagram.android" # UI XML pre/post click - mock_device.deviceV2.dump_hierarchy.side_effect = [ + mock_device.dump_hierarchy.side_effect = [ "", # initial context (line 293) "", # anomaly guard check (line 191) "" # post-click check (line 358) - ] + ] + [""] * 50 # Mock Telepathic Engine mock_engine = MagicMock() diff --git a/tests/unit/test_config_effects.py b/tests/unit/test_config_effects.py index acb6a04..f67fe20 100644 --- a/tests/unit/test_config_effects.py +++ b/tests/unit/test_config_effects.py @@ -30,9 +30,9 @@ def test_interact_with_profile_all_100_percent(mock_random, device, telepathic_m # 2 scrolls (2 shell commands) via _humanized_scroll # story, follow, grid, like all use QNavGraph click transitions because 'reel_viewer' is in MockDeviceV2 XML. - assert len(device.deviceV2.shells) == 2 + assert device.shell.call_count == 2 - for cmd in device.deviceV2.shells: + for cmd in [args[0][0] for args in device.shell.call_args_list]: assert "input swipe" in cmd @patch("random.random") @@ -53,7 +53,7 @@ def test_interact_with_profile_zero_percent(mock_random, device, telepathic_mock _interact_with_profile(device, configs, "testuser", session_state, 0.0, mock_logger) # No interaction blocks run, so no shells. - assert len(device.deviceV2.shells) == 0 + assert device.shell.call_count == 0 @patch("random.random") def test_interact_with_profile_mixed_probability(mock_random, device, telepathic_mock, mock_logger): @@ -77,7 +77,7 @@ def test_interact_with_profile_mixed_probability(mock_random, device, telepathic _interact_with_profile(device, configs, "testuser", session_state, 0.0, mock_logger) # Grid loop finishes with 1 scroll for 1 post. - assert len(device.deviceV2.shells) == 1 + assert device.shell.call_count == 1 @patch("random.random") def test_carousel_100_percent(mock_random, device, mock_logger): @@ -91,8 +91,8 @@ def test_carousel_100_percent(mock_random, device, mock_logger): _interact_with_carousel(device, configs, 0.0, mock_logger) - assert len(device.deviceV2.shells) == 4 - for cmd in device.deviceV2.shells: + assert device.shell.call_count == 4 + for cmd in [args[0][0] for args in device.shell.call_args_list]: assert "swipe" in cmd @patch("random.random") @@ -107,7 +107,7 @@ def test_carousel_zero_percent(mock_random, device, mock_logger): _interact_with_carousel(device, configs, 0.0, mock_logger) - assert len(device.deviceV2.shells) == 0 + assert device.shell.call_count == 0 @patch("random.random") def test_interact_with_profile_follow_limit_enforcement(mock_random, device, telepathic_mock, mock_logger): @@ -131,7 +131,7 @@ def test_interact_with_profile_follow_limit_enforcement(mock_random, device, tel _interact_with_profile(device, configs, "targetuser", session_state, 0.0, mock_logger) # Assert shells is 0 (assuming stories and likes probability mathematically default to 0 due to MockArgs empty fallback) - assert len(device.deviceV2.shells) == 0 + assert device.shell.call_count == 0 @patch("random.random") def test_interact_with_profile_likes_limit_enforcement(mock_random, device, telepathic_mock, mock_logger): @@ -157,7 +157,7 @@ def test_interact_with_profile_likes_limit_enforcement(mock_random, device, tele _interact_with_profile(device, configs, "targetuser", session_state, 0.0, mock_logger) # Limit restricts likes block. - assert len(device.deviceV2.shells) == 0 + assert device.shell.call_count == 0 # NOTE: Repost is deeply integrated into `bot_flow._run_zero_latency_feed_loop`. We can't mock the diff --git a/tests/unit/test_critical_anomaly_guards.py b/tests/unit/test_critical_anomaly_guards.py index 59d5298..a8610a3 100644 --- a/tests/unit/test_critical_anomaly_guards.py +++ b/tests/unit/test_critical_anomaly_guards.py @@ -41,7 +41,7 @@ class TestCriticalAnomalyGuards: If a user account is private, _interact_with_profile must skip everything and return without doing ANY interactions. """ - self.mock_device.deviceV2.dump_hierarchy.return_value = ''' + self.mock_device.dump_hierarchy.return_value = ''' @@ -55,14 +55,14 @@ class TestCriticalAnomalyGuards: _interact_with_profile(self.mock_device, configs, "test_private", session_state, sleep_mod=0.0, logger=self.logger) # Verify it did not attempt to find stories, scrape, or anything - self.mock_device.deviceV2.click.assert_not_called() - self.mock_device.deviceV2.press.assert_not_called() + self.mock_device.click.assert_not_called() + self.mock_device.press.assert_not_called() def test_interact_with_empty_profile_aborts(self): """ If a user account has 0 posts, we must skip. """ - self.mock_device.deviceV2.dump_hierarchy.return_value = ''' + self.mock_device.dump_hierarchy.return_value = ''' @@ -74,8 +74,8 @@ class TestCriticalAnomalyGuards: _interact_with_profile(self.mock_device, configs, "test_empty", session_state, sleep_mod=0.0, logger=self.logger) - self.mock_device.deviceV2.click.assert_not_called() - self.mock_device.deviceV2.press.assert_not_called() + self.mock_device.click.assert_not_called() + self.mock_device.press.assert_not_called() def test_comments_disabled_guard(self): """ diff --git a/tests/unit/test_profile_interaction_sync.py b/tests/unit/test_profile_interaction_sync.py index 0ebeaee..4bd88f0 100644 --- a/tests/unit/test_profile_interaction_sync.py +++ b/tests/unit/test_profile_interaction_sync.py @@ -20,7 +20,7 @@ def test_profile_grid_sync_delay_after_follow(): """ mock_device = MagicMock() mock_device.app_id = "com.instagram.android" - mock_device.deviceV2.dump_hierarchy.return_value = "" + mock_device.dump_hierarchy.return_value = "" mock_device.dump_hierarchy.return_value = "" mock_configs = FakeConfig() diff --git a/tests/unit/test_unfollow_engine.py b/tests/unit/test_unfollow_engine.py index 80f6635..cac9f1e 100644 --- a/tests/unit/test_unfollow_engine.py +++ b/tests/unit/test_unfollow_engine.py @@ -104,6 +104,6 @@ class TestUnfollowEngine: ) assert result == "BOREDOM_CHANGE_FEED" - self.mock_device.deviceV2.dump_hierarchy.assert_not_called() - self.mock_device.deviceV2.click.assert_not_called() + self.mock_device.dump_hierarchy.assert_not_called() + self.mock_device.click.assert_not_called() self.mock_session_state.totalUnfollowed = 0